dkg: retry transient p2p send failures during ceremonies

dkg/bcast/client.go

@@ -65,7 +65,7 @@ func (c *client) Broadcast(ctx context.Context, msgID string, msg proto.Message)
 
 	fork, join, cancel := forkjoin.New(ctx, func(ctx context.Context, pID peer.ID) (*pb.BCastSigResponse, error) {
 		sigResp := new(pb.BCastSigResponse)
-		err := c.sendRecvFunc(ctx, c.p2pNode, pID, sigReq, sigResp, protocolIDSig)
+		err := c.sendRecvFunc(ctx, c.p2pNode, pID, sigReq, sigResp, protocolIDSig, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 
 		return sigResp, err
 	})
@@ -133,7 +133,7 @@ func (c *client) Broadcast(ctx context.Context, msgID string, msg proto.Message)
 			continue // Skip self.
 		}
 
-		err := c.sendFunc(ctx, c.p2pNode, protocolIDMsg, pID, bcastMsg, p2p.WithSendTimeout(sendTimeout))
+		err := c.sendFunc(ctx, c.p2pNode, protocolIDMsg, pID, bcastMsg, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 		if err != nil {
 			return errors.Wrap(err, "send message")
 		}

dkg/bcast/helpers.go

@@ -20,6 +20,11 @@ const (
 	protocolIDMsg    = protocolIDPrefix + "/msg"
 	receiveTimeout   = time.Minute                    // Allow for peers to be out of sync, with some sending messages much earlier and having to wait.
 	sendTimeout      = receiveTimeout + 2*time.Second // Allow for server to timeout first.
+
+	// sendRetries is the number of additional p2p send attempts. A DKG ceremony aborts on the
+	// first failed exchange, so transient failures (e.g. a stalled relay connection) are retried
+	// instead of failing the whole ceremony. Receivers deduplicate re-delivered messages.
+	sendRetries = 5
 )
 
 // hashFunc is a function that hashes a message ID and a any-wrapped protobuf message.

dkg/bcast/impl_test.go

@@ -5,11 +5,13 @@ package bcast_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	k1 "github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -150,6 +152,157 @@ func TestBCast(t *testing.T) {
 	assertResults(t, p0Result, peers[0])
 }
 
+// TestBCastRetriesTransientSendFailures ensures a reliable-broadcast survives transient
+// p2p send failures in each phase instead of aborting the whole ceremony on the first one.
+func TestBCastRetriesTransientSendFailures(t *testing.T) {
+	const (
+		n     = 3
+		msgID = "msgID"
+	)
+
+	// The bcast protocol IDs are unexported; keep in sync with helpers.go.
+	tests := []struct {
+		name          string
+		flakyProtocol protocol.ID
+	}{
+		{name: "signature request phase", flakyProtocol: "/charon/dkg/bcast/2.0.0/sig"},
+		{name: "message send phase", flakyProtocol: "/charon/dkg/bcast/2.0.0/msg"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				ctx      = context.Background()
+				secrets  []*k1.PrivateKey
+				tcpNodes []host.Host
+				peers    []peer.ID
+			)
+
+			for range n {
+				secret, err := k1.GeneratePrivateKey()
+				require.NoError(t, err)
+
+				secrets = append(secrets, secret)
+
+				tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+				tcpNodes = append(tcpNodes, tcpNode)
+
+				peers = append(peers, tcpNode.ID())
+			}
+
+			for i := range n {
+				for j := range n {
+					tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
+				}
+			}
+
+			// Node 0's first send to each peer in this phase fails transiently.
+			tcpNodes[0] = testutil.NewFlakyHost(tcpNodes[0], n-1, test.flakyProtocol)
+
+			received := make(chan proto.Message, n)
+			callback := func(_ context.Context, _ peer.ID, _ string, msg proto.Message) error {
+				received <- msg
+				return nil
+			}
+			checkMessage := func(_ context.Context, _ peer.ID, msgAny *anypb.Any) error {
+				var ts timestamppb.Timestamp
+				if err := msgAny.UnmarshalTo(&ts); err != nil {
+					return errors.Wrap(err, "anypb error")
+				}
+
+				return nil
+			}
+
+			var bcasts []bcast.BroadcastFunc
+
+			for i := range n {
+				bcastFunc := bcast.New(tcpNodes[i], peers, secrets[i], []byte("session hash"))
+				bcastFunc.RegisterMessageIDFuncs(msgID, callback, checkMessage)
+				bcasts = append(bcasts, bcastFunc.Broadcast)
+			}
+
+			msg := timestamppb.Now()
+			err := bcasts[0](ctx, msgID, msg)
+			require.NoError(t, err)
+
+			for range n - 1 {
+				require.True(t, proto.Equal(msg, <-received))
+			}
+		})
+	}
+}
+
+// TestBCastSigRequestAllowsSlowPeer ensures a signature request tolerates a peer that
+// responds slower than the default p2p send timeout (7s), since ceremony peers may
+// lawfully lag by up to the bcast receive timeout (1min) while catching up.
+func TestBCastSigRequestAllowsSlowPeer(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping slow-peer timeout test in short mode")
+	}
+
+	const (
+		n     = 2
+		msgID = "msgID"
+		// slowDelay must exceed the 7s default p2p send timeout to prove the
+		// bcast send timeout is applied to signature requests.
+		slowDelay = 8 * time.Second
+	)
+
+	var (
+		ctx      = context.Background()
+		secrets  []*k1.PrivateKey
+		tcpNodes []host.Host
+		peers    []peer.ID
+		bcasts   []bcast.BroadcastFunc
+	)
+
+	for range n {
+		secret, err := k1.GeneratePrivateKey()
+		require.NoError(t, err)
+
+		secrets = append(secrets, secret)
+
+		tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+		tcpNodes = append(tcpNodes, tcpNode)
+
+		peers = append(peers, tcpNode.ID())
+	}
+
+	for i := range n {
+		for j := range n {
+			tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
+		}
+	}
+
+	callback := func(context.Context, peer.ID, string, proto.Message) error {
+		return nil
+	}
+
+	for i := range n {
+		delay := time.Duration(0)
+		if i == 1 {
+			delay = slowDelay // Node 1 is slow to check (and so sign) node 0's message.
+		}
+
+		checkMessage := func(_ context.Context, _ peer.ID, msgAny *anypb.Any) error {
+			time.Sleep(delay)
+
+			var ts timestamppb.Timestamp
+			if err := msgAny.UnmarshalTo(&ts); err != nil {
+				return errors.Wrap(err, "anypb error")
+			}
+
+			return nil
+		}
+
+		bcastFunc := bcast.New(tcpNodes[i], peers, secrets[i], []byte("session hash"))
+		bcastFunc.RegisterMessageIDFuncs(msgID, callback, checkMessage)
+		bcasts = append(bcasts, bcastFunc.Broadcast)
+	}
+
+	require.NoError(t, bcasts[0](ctx, msgID, timestamppb.Now()))
+}
+
 // TestBCastSessionHashMismatch ensures that messages signed in one session
 // cannot be verified in another, binding broadcasts to the cluster session.
 func TestBCastSessionHashMismatch(t *testing.T) {

dkg/exchanger.go

@@ -11,6 +11,8 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/z"
@@ -38,6 +40,32 @@ const (
 	// Do not add new values greater than sigDepositData.
 )
 
+const (
+	// sendRetries is the number of additional p2p send attempts during DKG ceremonies.
+	// A DKG aborts on the first failed exchange, so transient failures (e.g. a stalled
+	// relay connection) are retried instead of failing the whole ceremony. All DKG
+	// receive paths deduplicate messages, so a retry of an already-delivered message
+	// is harmless.
+	sendRetries = 5
+
+	// sendTimeout is the total p2p send budget during DKG ceremonies, shared by all
+	// retry attempts. Ceremony peers may lag behind (e.g. slower operators or
+	// relay-routed connections), so sends get much more headroom than the 7s p2p default.
+	sendTimeout = time.Minute
+)
+
+// newRetryingSendFunc returns a p2p.SendFunc wrapping p2p.Send with the given send
+// timeout and sendRetries retries.
+func newRetryingSendFunc(timeout time.Duration) p2p.SendFunc {
+	return func(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID,
+		msg proto.Message, opts ...p2p.SendRecvOption,
+	) error {
+		opts = append(opts, p2p.WithSendTimeout(timeout), p2p.WithRetries(sendRetries))
+
+		return p2p.Send(ctx, p2pNode, protoID, peerID, msg, opts...)
+	}
+}
+
 // sigTypeStore is a shorthand for a map of sigType to map of core.PubKey to slice of core.ParSignedData.
 type sigTypeStore map[sigType]map[core.PubKey][]core.ParSignedData
 
@@ -111,7 +139,7 @@ func newExchanger(p2pNode host.Host, peerIdx int, peers []peer.ID, peerMap map[p
 	ex := &exchanger{
 		// threshold is len(peers) to wait until we get all the partial sigs from all the peers per DV
 		sigdb:    parsigdb.NewMemDB(len(peers), noopDeadliner{}, parsigdb.NewMemDBMetadata(0, time.Now())), // metadata timestamps are used for metrics, irrelevant for DKG
-		sigex:    parsigex.NewParSigEx(p2pNode, p2p.Send, peerIdx, peers, verifyShareIdx, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
+		sigex:    parsigex.NewParSigEx(p2pNode, newRetryingSendFunc(timeout), peerIdx, peers, verifyShareIdx, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
 		sigTypes: st,
 		sigData: dataByPubkey{
 			store: sigTypeStore{},

dkg/exchanger_internal_test.go

@@ -194,6 +194,68 @@ func TestExchanger(t *testing.T) {
 	require.Len(t, actual, len(expectedSigTypes))
 }
 
+// TestExchangerRetriesTransientSendFailures ensures the DKG signature exchange survives
+// transient p2p send failures instead of aborting the whole ceremony on the first one.
+func TestExchangerRetriesTransientSendFailures(t *testing.T) {
+	ctx := context.Background()
+
+	const nodes = 3
+
+	pubkey := testutil.RandomCorePubKey(t)
+
+	var (
+		peers     []peer.ID
+		hosts     []host.Host
+		hostsInfo []peer.AddrInfo
+	)
+
+	for range nodes {
+		h := testutil.CreateHost(t, testutil.AvailableAddr(t))
+		hostsInfo = append(hostsInfo, peer.AddrInfo{ID: h.ID(), Addrs: h.Addrs()})
+		peers = append(peers, h.ID())
+		hosts = append(hosts, h)
+	}
+
+	for i := range nodes {
+		for j := range nodes {
+			if i == j {
+				continue
+			}
+
+			hosts[i].Peerstore().AddAddrs(hostsInfo[j].ID, hostsInfo[j].Addrs, peerstore.PermanentAddrTTL)
+		}
+	}
+
+	// Node 0's first send to each peer fails transiently.
+	hosts[0] = testutil.NewFlakyHost(hosts[0], nodes-1)
+
+	var exchangers []*exchanger
+
+	for i := range nodes {
+		ex, err := newExchanger(hosts[i], i, peers, positionalPeerMap(peers), []sigType{sigLock}, 8*time.Second)
+		require.NoError(t, err)
+
+		exchangers = append(exchangers, ex)
+	}
+
+	errChan := make(chan error, nodes)
+
+	for i := range nodes {
+		go func(node int) {
+			set := core.ParSignedDataSet{
+				pubkey: core.NewPartialSignature(testutil.RandomCoreSignature(), node+1),
+			}
+
+			_, err := exchangers[node].exchange(ctx, sigLock, set)
+			errChan <- err
+		}(i)
+	}
+
+	for range nodes {
+		require.NoError(t, <-errChan)
+	}
+}
+
 // TestExchangerPushPsigsNeverBlocks fires pushPsigs repeatedly with no exchange call draining
 // results, which must not block. A previous implementation used a shared size-1 channel that
 // deadlocked once full, especially when pushPsigs ran synchronously on the exchange goroutine.

dkg/frostp2p.go

@@ -305,7 +305,7 @@ func (f *frostP2P) Round1(ctx context.Context, castR1 map[msgKey]frost.Round1Bca
 			return nil, nil, errors.New("bug: unexpected p2p message to self")
 		}
 
-		err := p2p.Send(ctx, f.p2pNode, round1P2PID, pID, p2pMsg)
+		err := p2p.Send(ctx, f.p2pNode, round1P2PID, pID, p2pMsg, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 		if err != nil {
 			return nil, nil, err
 		}