p2p: retry send once on relay errors (#1465)

Retry sends once on relay errors. This should solve 99% of the `stream reset` warnings of infosync and peerinfo protocols when using relay connections. This only now became an issue since these protocols do send-receive (other protocols did send and receive separately). 

category: bug 
ticket: none

p2p/sender.go

@@ -113,7 +113,10 @@ func (s *Sender) SendAsync(parent context.Context, tcpNode host.Host, protoID pr
 		// Clone the context since parent context may be closed soon.
 		ctx := log.CopyFields(context.Background(), parent)
 		ctx = log.WithCtx(ctx, z.Str("protocol", string(protoID)))
-		err := Send(ctx, tcpNode, protoID, peerID, msg) // TODO(corver): Retry once on relay errors.
+
+		err := withRelayRetry(func() error {
+			return Send(ctx, tcpNode, protoID, peerID, msg)
+		})
 		s.addResult(ctx, peerID, err)
 	}()
 
@@ -126,12 +129,24 @@ func (s *Sender) SendAsync(parent context.Context, tcpNode host.Host, protoID pr
 // It implements SendReceiveFunc.
 func (s *Sender) SendReceive(ctx context.Context, tcpNode host.Host, peerID peer.ID, req, resp proto.Message, protocols ...protocol.ID) error {
 	ctx = log.WithCtx(ctx, z.Any("protocol", protocols))
-	err := SendReceive(ctx, tcpNode, peerID, req, resp, protocols...)
+	err := withRelayRetry(func() error {
+		return SendReceive(ctx, tcpNode, peerID, req, resp, protocols...)
+	})
 	s.addResult(ctx, peerID, err)
 
 	return err
 }
 
+// withRelayRetry wraps a function and retries it once if the error is a relay error.
+func withRelayRetry(fn func() error) error {
+	err := fn()
+	if isRelayError(err) { // Retry once if relay error
+		err = fn()
+	}
+
+	return err
+}
+
 // SendReceive sends and receives a libp2p request and response message
 // pair synchronously and then closes the stream.
 // The provided response proto will be populated if err is nil.

p2p/sender_internal_test.go

@@ -17,9 +17,14 @@ package p2p
 
 import (
 	"context"
+	"sync"
 	"testing"
+	"time"
 
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/stretchr/testify/require"
 
 	"github.com/obolnetwork/charon/app/errors"
@@ -67,3 +72,41 @@ func TestSenderAddResult(t *testing.T) {
 	// INFO P2P sending recovered {"peer": "better-week"}
 	// INFO P2P sending failing {"peer": "better-week"}
 }
+
+func TestSenderRetry(t *testing.T) {
+	sender := new(Sender)
+	ctx := context.Background()
+
+	h := new(testHost)
+	err := sender.SendReceive(ctx, h, "", nil, nil, "")
+	require.ErrorIs(t, err, network.ErrReset)
+	require.Equal(t, 2, h.Count())
+
+	h = new(testHost)
+	err = sender.SendAsync(ctx, h, "", "", nil)
+	require.Nil(t, err)
+	require.Eventually(t, func() bool {
+		return h.Count() == 2
+	}, time.Second, time.Millisecond)
+}
+
+type testHost struct {
+	host.Host
+	mu    sync.Mutex
+	count int
+}
+
+func (h *testHost) Count() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	return h.count
+}
+
+func (h *testHost) NewStream(context.Context, peer.ID, ...protocol.ID) (network.Stream, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.count++
+
+	return nil, network.ErrReset
+}