From 397150bcd34e6d14edf8722b07e88374ed1d0a89 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 14:05:18 +0300
Subject: [PATCH 01/13] p2p: include peer name in send errors

---
 p2p/sender.go      | 16 ++++++-------
 p2p/sender_test.go | 58 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 66 insertions(+), 8 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 3006c03fb..9b8fd5e0d 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -315,14 +315,14 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 	// Circuit relay connections are transient
 	s, err := p2pNode.NewStream(network.WithAllowLimitedConn(ctx, ""), peerID, o.protocols...)
 	if err != nil {
-		return errors.Wrap(err, "new stream", z.Any("protocols", o.protocols))
+		return errors.Wrap(err, "new stream", z.Any("protocols", o.protocols), z.Str("peer", PeerName(peerID)))
 	}
 	defer s.Close()
 
 	protoLabel = string(s.Protocol())
 
 	if err := s.SetDeadline(time.Now().Add(o.sendTimeout)); err != nil {
-		return errors.Wrap(err, "set deadline")
+		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
 	writeFunc, ok := o.writersByProtocol[s.Protocol()]
@@ -341,7 +341,7 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 	t0 := time.Now()
 
 	if err = writer.WriteMsg(req); err != nil {
-		return errors.Wrap(err, "write request", z.Any("protocol", s.Protocol()))
+		return errors.Wrap(err, "write request", z.Any("protocol", s.Protocol()), z.Str("peer", PeerName(peerID)))
 	}
 
 	if err := s.CloseWrite(); err != nil {
@@ -354,12 +354,12 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		if isCanceledStreamErr(err) {
 			log.Debug(ctx, "Closing write of canceled stream", z.Err(err), z.Any("protocol", s.Protocol()))
 		} else {
-			return errors.Wrap(err, "close write", z.Any("protocol", s.Protocol()))
+			return errors.Wrap(err, "close write", z.Any("protocol", s.Protocol()), z.Str("peer", PeerName(peerID)))
 		}
 	}
 
 	if err = reader.ReadMsg(resp); err != nil {
-		return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol()))
+		return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol()), z.Str("peer", PeerName(peerID)))
 	}
 
 	o.rttCallback(time.Since(t0))
@@ -389,14 +389,14 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 	// Circuit relay connections are transient
 	s, err := p2pNode.NewStream(network.WithAllowLimitedConn(ctx, ""), peerID, o.protocols...)
 	if err != nil {
-		return errors.Wrap(err, "p2pNode stream")
+		return errors.Wrap(err, "p2pNode stream", z.Str("peer", PeerName(peerID)))
 	}
 	defer s.Close()
 
 	protoLabel = string(s.Protocol())
 
 	if err := s.SetDeadline(time.Now().Add(o.sendTimeout)); err != nil {
-		return errors.Wrap(err, "set deadline")
+		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
 	writeFunc, ok := o.writersByProtocol[s.Protocol()]
@@ -405,7 +405,7 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 	}
 
 	if err = writeFunc(s).WriteMsg(msg); err != nil {
-		return errors.Wrap(err, "write message", z.Any("protocol", s.Protocol()))
+		return errors.Wrap(err, "write message", z.Any("protocol", s.Protocol()), z.Str("peer", PeerName(peerID)))
 	}
 
 	return nil
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 21a646357..48a64e908 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -16,9 +16,12 @@ import (
 	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/obolnetwork/charon/app/log"
+	"github.com/obolnetwork/charon/app/z"
 	pbv1 "github.com/obolnetwork/charon/core/corepb/v1"
 	"github.com/obolnetwork/charon/p2p"
 	"github.com/obolnetwork/charon/testutil"
@@ -145,6 +148,61 @@ func TestWithSendTimeout(t *testing.T) {
 	}
 }
 
+// errFields extracts the structured fields of an error into a map.
+func errFields(t *testing.T, err error) map[string]any {
+	t.Helper()
+
+	fielder, ok := err.(interface{ Fields() []z.Field })
+	require.True(t, ok, "error does not have structured fields")
+
+	enc := zapcore.NewMapObjectEncoder()
+
+	for _, field := range fielder.Fields() {
+		field(func(zf zap.Field) {
+			zf.AddTo(enc)
+		})
+	}
+
+	return enc.Fields
+}
+
+func TestSendErrorContainsPeerField(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	peerName := p2p.PeerName(server.ID())
+
+	t.Run("send new stream error", func(t *testing.T) {
+		// No handler registered on the server, so protocol negotiation fails.
+		err := p2p.Send(ctx, client, "unknown", server.ID(), &pbv1.Duty{Slot: 1})
+		require.Error(t, err)
+		require.Equal(t, peerName, errFields(t, err)["peer"])
+	})
+
+	t.Run("send receive new stream error", func(t *testing.T) {
+		err := p2p.SendReceive(ctx, client, server.ID(), new(pbv1.Duty), new(pbv1.Duty), "unknown")
+		require.Error(t, err)
+		require.Equal(t, peerName, errFields(t, err)["peer"])
+	})
+
+	t.Run("send write error", func(t *testing.T) {
+		protocolID := protocol.ID("testprotocol-peerfield")
+		p2p.RegisterHandler("test", server, protocolID,
+			func() proto.Message { return new(pbv1.Duty) },
+			func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) {
+				return nil, false, nil
+			})
+
+		// The negative send timeout sets an already-expired stream deadline, failing the write.
+		err := p2p.Send(ctx, client, protocolID, server.ID(), &pbv1.Duty{Slot: 1},
+			p2p.WithSendTimeout(-time.Second))
+		require.Error(t, err)
+		require.Equal(t, peerName, errFields(t, err)["peer"])
+	})
+}
+
 func TestSend(t *testing.T) {
 	var (
 		undelimID = protocol.ID("undelimited")

From 8b691610c0420475300d2b89d13eacb84bd2952a Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 14:21:27 +0300
Subject: [PATCH 02/13] dkg: retry transient p2p send failures during
 ceremonies

---
 dkg/bcast/client.go            |  4 +-
 dkg/bcast/helpers.go           |  5 +++
 dkg/bcast/impl_test.go         | 67 +++++++++++++++++++++++++++++++
 dkg/exchanger.go               | 18 ++++++++-
 dkg/exchanger_internal_test.go | 62 +++++++++++++++++++++++++++++
 dkg/frostp2p.go                |  2 +-
 dkg/pedersen/board.go          |  7 +++-
 dkg/pedersen/board_test.go     | 58 +++++++++++++++++++++++++++
 p2p/sender.go                  | 63 +++++++++++++++++++++++++++--
 p2p/sender_test.go             | 72 ++++++++++++++++++++++++++++++++++
 testutil/random.go             | 46 ++++++++++++++++++++++
 11 files changed, 395 insertions(+), 9 deletions(-)

diff --git a/dkg/bcast/client.go b/dkg/bcast/client.go
index 25c410faf..1c53d7bde 100644
--- a/dkg/bcast/client.go
+++ b/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.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")
 		}
diff --git a/dkg/bcast/helpers.go b/dkg/bcast/helpers.go
index ac6f4a122..d7dfb951e 100644
--- a/dkg/bcast/helpers.go
+++ b/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 = 2
 )
 
 // hashFunc is a function that hashes a message ID and a any-wrapped protobuf message.
diff --git a/dkg/bcast/impl_test.go b/dkg/bcast/impl_test.go
index f50b9d353..e57d12659 100644
--- a/dkg/bcast/impl_test.go
+++ b/dkg/bcast/impl_test.go
@@ -150,6 +150,73 @@ func TestBCast(t *testing.T) {
 	assertResults(t, p0Result, peers[0])
 }
 
+// TestBCastRetriesTransientSendFailures ensures a reliable-broadcast survives transient
+// p2p send failures instead of aborting the whole ceremony on the first one.
+func TestBCastRetriesTransientSendFailures(t *testing.T) {
+	const (
+		n     = 3
+		msgID = "msgID"
+	)
+
+	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 sends fail transiently: one signature request per peer plus one message send.
+	tcpNodes[0] = testutil.NewFlakyHost(tcpNodes[0], n)
+
+	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))
+	}
+}
+
 // TestBCastSessionHashMismatch ensures that messages signed in one session
 // cannot be verified in another, binding broadcasts to the cluster session.
 func TestBCastSessionHashMismatch(t *testing.T) {
diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index 6d96e6ad8..027977e5a 100644
--- a/dkg/exchanger.go
+++ b/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,20 @@ const (
 	// Do not add new values greater than sigDepositData.
 )
 
+// 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.
+const sendRetries = 2
+
+// sendRetrying wraps p2p.Send with sendRetries retries. It implements p2p.SendFunc.
+func sendRetrying(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID,
+	msg proto.Message, opts ...p2p.SendRecvOption,
+) error {
+	return p2p.Send(ctx, p2pNode, protoID, peerID, msg, append(opts, p2p.WithRetries(sendRetries))...)
+}
+
 // 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 +127,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, sendRetrying, peerIdx, peers, verifyShareIdx, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
 		sigTypes: st,
 		sigData: dataByPubkey{
 			store: sigTypeStore{},
diff --git a/dkg/exchanger_internal_test.go b/dkg/exchanger_internal_test.go
index 31f4c4f83..6a653267b 100644
--- a/dkg/exchanger_internal_test.go
+++ b/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.
diff --git a/dkg/frostp2p.go b/dkg/frostp2p.go
index 896332ced..4e5a897d8 100644
--- a/dkg/frostp2p.go
+++ b/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.WithRetries(sendRetries))
 		if err != nil {
 			return nil, nil, err
 		}
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index 0e4ead7e0..7b8b2b1bc 100644
--- a/dkg/pedersen/board.go
+++ b/dkg/pedersen/board.go
@@ -56,6 +56,11 @@ type ValidatorPubKeyShare struct {
 
 const (
 	protocolID = "/charon/dkg/pedersen/1.0.0"
+
+	// sendRetries is the number of additional p2p send attempts. A DKG ceremony cannot
+	// complete if a bundle is dropped, so transient failures (e.g. a stalled relay
+	// connection) are retried instead. Receivers deduplicate re-delivered messages.
+	sendRetries = 2
 )
 
 var (
@@ -254,7 +259,7 @@ func (b *Board) broadcastP2P(ctx context.Context, msgID string, msg proto.Messag
 			continue
 		}
 
-		if err := b.sender.SendAsync(ctx, b.host, protocol.ID(msgID), peerID, msg); err != nil {
+		if err := b.sender.SendAsync(ctx, b.host, protocol.ID(msgID), peerID, msg, p2p.WithRetries(sendRetries)); err != nil {
 			return errors.Wrap(err, "p2p send", z.Str("msg", msgID), z.Str("to", peerID.String()))
 		}
 	}
diff --git a/dkg/pedersen/board_test.go b/dkg/pedersen/board_test.go
index 97a8c07fb..f012d4da5 100644
--- a/dkg/pedersen/board_test.go
+++ b/dkg/pedersen/board_test.go
@@ -4,6 +4,7 @@ package pedersen_test
 
 import (
 	"testing"
+	"time"
 
 	"github.com/drand/kyber"
 	kdkg "github.com/drand/kyber/share/dkg"
@@ -15,6 +16,63 @@ import (
 	"github.com/obolnetwork/charon/testutil"
 )
 
+// TestBoardRetriesTransientSendFailures ensures board bundle broadcasts survive transient
+// p2p send failures instead of silently dropping the bundle for a peer.
+func TestBoardRetriesTransientSendFailures(t *testing.T) {
+	const (
+		numNodes  = 3
+		threshold = 2
+	)
+
+	var (
+		peers   []peer.ID
+		peerMap = make(map[peer.ID]cluster.NodeIdx)
+		session = testutil.RandomArray32()
+	)
+
+	nodes := make([]*pedersen.TestNode, numNodes)
+	for i := range numNodes {
+		nodes[i] = pedersen.NewTestNode(t, i)
+		peerMap[nodes[i].NodeHost.ID()] = nodes[i].NodeIdx
+		peers = append(peers, nodes[i].NodeHost.ID())
+	}
+
+	pedersen.ConnectTestNodes(t, nodes)
+
+	// Node 0's first send to each peer fails transiently.
+	nodes[0].NodeHost = testutil.NewFlakyHost(nodes[0].NodeHost, numNodes-1)
+
+	for i := range nodes {
+		nodes[i].InitBoard(t, threshold, peers, peerMap, session[:])
+	}
+
+	dealBundle := kdkg.DealBundle{
+		DealerIndex: 0,
+		Deals: []kdkg.Deal{
+			{
+				ShareIndex:     1,
+				EncryptedShare: []byte{1, 2, 3},
+			},
+		},
+		Public: []kyber.Point{
+			pedersen.RandomPoint(t),
+		},
+		SessionID: []byte("sessionID"),
+		Signature: []byte{13, 14, 15},
+	}
+
+	nodes[0].Board.PushDeals(&dealBundle)
+
+	for i := range nodes {
+		select {
+		case received := <-nodes[i].Board.IncomingDeal():
+			require.Equal(t, []byte{13, 14, 15}, received.Signature)
+		case <-time.After(10 * time.Second):
+			require.Fail(t, "timed out waiting for deal bundle", "node %d", i)
+		}
+	}
+}
+
 func TestBoard(t *testing.T) {
 	const (
 		numNodes  = 4
diff --git a/p2p/sender.go b/p2p/sender.go
index 9b8fd5e0d..1ac50a7ea 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -17,6 +17,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/obolnetwork/charon/app/errors"
+	"github.com/obolnetwork/charon/app/expbackoff"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
 )
@@ -202,6 +203,7 @@ type sendRecvOpts struct {
 	rttCallback       func(time.Duration)
 	receiveTimeout    time.Duration
 	sendTimeout       time.Duration
+	retries           int    // Number of additional send attempts after a failure.
 	metricTopic       string // Optional sub-protocol label for the send_duration metric.
 }
 
@@ -219,6 +221,17 @@ func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 	}
 }
 
+// WithRetries returns an option that retries a failed send up to the given number of
+// additional attempts, each on a fresh stream, backing off briefly in between.
+// Only use it for protocols whose handlers tolerate duplicate delivery, since a send
+// that failed on the sender side may still have been delivered (e.g. DKG ceremony
+// messages, which are deduplicated by all receivers).
+func WithRetries(retries int) func(*sendRecvOpts) {
+	return func(opts *sendRecvOpts) {
+		opts.retries = retries
+	}
+}
+
 // WithSendMetricTopic returns an option that adds a topic label to the send_duration metric.
 // Use a small bounded set of values (e.g. a message-type name) to keep metric cardinality low.
 func WithSendMetricTopic(topic string) func(*sendRecvOpts) {
@@ -288,6 +301,27 @@ func defaultSendRecvOpts(pID protocol.ID) sendRecvOpts {
 	}
 }
 
+// withRetries calls fn and retries it up to the given number of additional times,
+// backing off briefly between attempts. It stops early once the context is done.
+func withRetries(ctx context.Context, retries int, fn func() error) error {
+	var err error
+
+	for attempt := 0; ; attempt++ {
+		err = fn()
+		if err == nil || attempt >= retries || ctx.Err() != nil {
+			return err
+		}
+
+		timer := time.NewTimer(expbackoff.Backoff(expbackoff.FastConfig, attempt))
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return err
+		case <-timer.C:
+		}
+	}
+}
+
 // 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.
@@ -295,8 +329,6 @@ func defaultSendRecvOpts(pID protocol.ID) sendRecvOpts {
 func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 	req, resp proto.Message, pID protocol.ID, opts ...SendRecvOption,
 ) error {
-	tStart := time.Now()
-
 	if !isZeroProto(resp) {
 		return errors.New("bug: response proto must be zero value")
 	}
@@ -306,6 +338,20 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
+	return withRetries(ctx, o.retries, func() error {
+		// A failed attempt may have partially populated the response.
+		proto.Reset(resp)
+
+		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o)
+	})
+}
+
+// sendReceive is a single SendReceive attempt.
+func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
+	req, resp proto.Message, pID protocol.ID, o sendRecvOpts,
+) error {
+	tStart := time.Now()
+
 	protoLabel := string(pID) // Updated to the negotiated protocol once NewStream succeeds.
 
 	defer func() {
@@ -371,13 +417,22 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID, msg proto.Message,
 	opts ...SendRecvOption,
 ) error {
-	t0 := time.Now()
-
 	o := defaultSendRecvOpts(protoID)
 	for _, opt := range opts {
 		opt(&o)
 	}
 
+	return withRetries(ctx, o.retries, func() error {
+		return send(ctx, p2pNode, protoID, peerID, msg, o)
+	})
+}
+
+// send is a single Send attempt.
+func send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID, msg proto.Message,
+	o sendRecvOpts,
+) error {
+	t0 := time.Now()
+
 	protoLabel := string(protoID)
 	if len(o.protocols) > 0 {
 		protoLabel = string(o.protocols[0])
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 48a64e908..28bb0808c 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -203,6 +203,78 @@ func TestSendErrorContainsPeerField(t *testing.T) {
 	})
 }
 
+func TestSendRetries(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	received := make(chan *pbv1.Duty, 1)
+	protocolID := protocol.ID("testprotocol-retries")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			duty, ok := req.(*pbv1.Duty)
+			require.True(t, ok)
+
+			received <- duty
+
+			return nil, false, nil
+		})
+
+	t.Run("no retries by default", func(t *testing.T) {
+		flaky := testutil.NewFlakyHost(client, 1)
+		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 1})
+		require.Error(t, err)
+		require.Equal(t, 1, flaky.Calls())
+	})
+
+	t.Run("retries transient failures", func(t *testing.T) {
+		flaky := testutil.NewFlakyHost(client, 2)
+		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 2}, p2p.WithRetries(2))
+		require.NoError(t, err)
+		require.Equal(t, 3, flaky.Calls())
+
+		select {
+		case duty := <-received:
+			require.EqualValues(t, 2, duty.GetSlot())
+		case <-time.After(5 * time.Second):
+			require.Fail(t, "timed out waiting for message")
+		}
+	})
+
+	t.Run("fails when retries exhausted", func(t *testing.T) {
+		flaky := testutil.NewFlakyHost(client, 3)
+		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 3}, p2p.WithRetries(2))
+		require.ErrorContains(t, err, "transient stream failure")
+		require.Equal(t, 3, flaky.Calls())
+	})
+}
+
+func TestSendReceiveRetries(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	protocolID := protocol.ID("testprotocol-sendrecv-retries")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			duty, ok := req.(*pbv1.Duty)
+			require.True(t, ok)
+
+			return &pbv1.Duty{Slot: duty.GetSlot() + 1}, true, nil
+		})
+
+	flaky := testutil.NewFlakyHost(client, 2)
+	resp := new(pbv1.Duty)
+	err := p2p.SendReceive(ctx, flaky, server.ID(), &pbv1.Duty{Slot: 41}, resp, protocolID, p2p.WithRetries(2))
+	require.NoError(t, err)
+	require.Equal(t, 3, flaky.Calls())
+	require.EqualValues(t, 42, resp.GetSlot())
+}
+
 func TestSend(t *testing.T) {
 	var (
 		undelimID = protocol.ID("undelimited")
diff --git a/testutil/random.go b/testutil/random.go
index b79ad4cec..e5f3a5819 100644
--- a/testutil/random.go
+++ b/testutil/random.go
@@ -4,6 +4,7 @@
 package testutil
 
 import (
+	"context"
 	"crypto/ecdsa"
 	crand "crypto/rand"
 	"fmt"
@@ -11,6 +12,7 @@ import (
 	"math/rand"
 	"net"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -34,11 +36,15 @@ import (
 	"github.com/libp2p/go-libp2p"
 	p2pcrypto "github.com/libp2p/go-libp2p/core/crypto"
 	"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"
 	quic "github.com/libp2p/go-libp2p/p2p/transport/quic" //nolint:revive // Must be imported with alias
 	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
 	"github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
 
+	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/core"
 	"github.com/obolnetwork/charon/eth2util"
 	"github.com/obolnetwork/charon/eth2util/enr"
@@ -1527,6 +1533,46 @@ func CreateHost(t *testing.T, addr *net.TCPAddr, opts ...libp2p.Option) host.Hos
 	return CreateHostWithIdentity(t, addr, pkey, opts...)
 }
 
+// NewFlakyHost wraps a host, failing the first `failures` outbound NewStream calls
+// with a transient error. It is used to test p2p send retry behavior.
+func NewFlakyHost(h host.Host, failures int) *FlakyHost {
+	return &FlakyHost{Host: h, failures: failures}
+}
+
+// FlakyHost is a host whose first outbound NewStream calls fail with a transient error.
+type FlakyHost struct {
+	host.Host
+
+	mu       sync.Mutex
+	failures int
+	calls    int
+}
+
+func (h *FlakyHost) NewStream(ctx context.Context, peerID peer.ID, pIDs ...protocol.ID) (network.Stream, error) {
+	h.mu.Lock()
+	h.calls++
+	fail := h.failures > 0
+
+	if fail {
+		h.failures--
+	}
+	h.mu.Unlock()
+
+	if fail {
+		return nil, errors.New("transient stream failure")
+	}
+
+	return h.Host.NewStream(ctx, peerID, pIDs...)
+}
+
+// Calls returns the number of NewStream calls made on the host.
+func (h *FlakyHost) Calls() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	return h.calls
+}
+
 func CreateHostWithIdentity(t *testing.T, addr *net.TCPAddr, secret *k1.PrivateKey, opts ...libp2p.Option) host.Host {
 	t.Helper()
 

From ecd8fb2e460962319129386e0d0a65746c0cbf65 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 17:33:27 +0300
Subject: [PATCH 03/13] p2p: address copilot review on send retries

---
 dkg/bcast/impl_test.go | 104 +++++++++++++++++++++++------------------
 docs/metrics.md        |   2 +-
 p2p/metrics.go         |   2 +-
 p2p/sender.go          |   8 ++--
 p2p/sender_test.go     |  10 ++++
 testutil/random.go     |  19 ++++++--
 6 files changed, 91 insertions(+), 54 deletions(-)

diff --git a/dkg/bcast/impl_test.go b/dkg/bcast/impl_test.go
index e57d12659..7dbeb0b2a 100644
--- a/dkg/bcast/impl_test.go
+++ b/dkg/bcast/impl_test.go
@@ -10,6 +10,7 @@ import (
 	"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"
@@ -151,69 +152,82 @@ func TestBCast(t *testing.T) {
 }
 
 // TestBCastRetriesTransientSendFailures ensures a reliable-broadcast survives transient
-// p2p send failures instead of aborting the whole ceremony on the first one.
+// 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"
 	)
 
-	var (
-		ctx      = context.Background()
-		secrets  []*k1.PrivateKey
-		tcpNodes []host.Host
-		peers    []peer.ID
-	)
+	// 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 range n {
-		secret, err := k1.GeneratePrivateKey()
-		require.NoError(t, err)
+	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
+			)
 
-		secrets = append(secrets, secret)
+			for range n {
+				secret, err := k1.GeneratePrivateKey()
+				require.NoError(t, err)
 
-		tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
-		tcpNodes = append(tcpNodes, tcpNode)
+				secrets = append(secrets, secret)
 
-		peers = append(peers, tcpNode.ID())
-	}
+				tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+				tcpNodes = append(tcpNodes, tcpNode)
 
-	for i := range n {
-		for j := range n {
-			tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
-		}
-	}
+				peers = append(peers, tcpNode.ID())
+			}
 
-	// Node 0's first sends fail transiently: one signature request per peer plus one message send.
-	tcpNodes[0] = testutil.NewFlakyHost(tcpNodes[0], n)
+			for i := range n {
+				for j := range n {
+					tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
+				}
+			}
 
-	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")
-		}
+			// Node 0's first send to each peer in this phase fails transiently.
+			tcpNodes[0] = testutil.NewFlakyHost(tcpNodes[0], n-1, test.flakyProtocol)
 
-		return nil
-	}
+			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")
+				}
 
-	var bcasts []bcast.BroadcastFunc
+				return nil
+			}
 
-	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)
-	}
+			var bcasts []bcast.BroadcastFunc
 
-	msg := timestamppb.Now()
-	err := bcasts[0](ctx, msgID, msg)
-	require.NoError(t, err)
+			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)
+			}
 
-	for range n - 1 {
-		require.True(t, proto.Equal(msg, <-received))
+			msg := timestamppb.Now()
+			err := bcasts[0](ctx, msgID, msg)
+			require.NoError(t, err)
+
+			for range n - 1 {
+				require.True(t, proto.Equal(msg, <-received))
+			}
+		})
 	}
 }
 
diff --git a/docs/metrics.md b/docs/metrics.md
index 97d28d849..391656867 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -110,7 +110,7 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `p2p_reachability_status` | Gauge | Current libp2p reachability status of this node as detected by autonat: unknown(0), public(1) or private(2). |  |
 | `p2p_relay_connection_types` | Gauge | Current number of libp2p connections by relay, type (`direct` or `relay`), and protocol (`tcp`, `quic`). Note that peers may have multiple connections. | `peer, type, protocol` |
 | `p2p_relay_connections` | Gauge | Connected relays by name | `peer` |
-| `p2p_send_duration_seconds` | Histogram | Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) calls, by peer, protocol, and topic. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller. | `peer, protocol, topic` |
+| `p2p_send_duration_seconds` | Histogram | Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) attempts, by peer, protocol, and topic. Sends with retries enabled observe each attempt separately, excluding backoff. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller. | `peer, protocol, topic` |
 | `relay_p2p_active_connections` | Gauge | Current number of active connections by peer and cluster | `peer, peer_cluster` |
 | `relay_p2p_connection_total` | Counter | Total number of new connections by peer and cluster | `peer, peer_cluster` |
 | `relay_p2p_network_receive_bytes_total` | Counter | Total number of network bytes received from the peer and cluster | `peer, peer_cluster` |
diff --git a/p2p/metrics.go b/p2p/metrics.go
index 6f4a04398..b7049c36c 100644
--- a/p2p/metrics.go
+++ b/p2p/metrics.go
@@ -36,7 +36,7 @@ var (
 	sendDurations = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: "p2p",
 		Name:      "send_duration_seconds",
-		Help:      "Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) calls, by peer, protocol, and topic. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller.",
+		Help:      "Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) attempts, by peer, protocol, and topic. Sends with retries enabled observe each attempt separately, excluding backoff. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller.",
 		Buckets:   prometheus.ExponentialBuckets(0.0001, 2, 18), // 0.1ms .. ~13.1s, covers the ~7s default SendReceive timeout
 	}, []string{"peer", "protocol", "topic"})
 
diff --git a/p2p/sender.go b/p2p/sender.go
index 1ac50a7ea..f56f386c6 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -302,13 +302,15 @@ func defaultSendRecvOpts(pID protocol.ID) sendRecvOpts {
 }
 
 // withRetries calls fn and retries it up to the given number of additional times,
-// backing off briefly between attempts. It stops early once the context is done.
+// backing off briefly between attempts. Cancellation stops retrying and surfaces
+// as the context error (with the last attempt error attached as a field), so
+// callers can detect it with errors.Is.
 func withRetries(ctx context.Context, retries int, fn func() error) error {
 	var err error
 
 	for attempt := 0; ; attempt++ {
 		err = fn()
-		if err == nil || attempt >= retries || ctx.Err() != nil {
+		if err == nil || attempt >= retries {
 			return err
 		}
 
@@ -316,7 +318,7 @@ func withRetries(ctx context.Context, retries int, fn func() error) error {
 		select {
 		case <-ctx.Done():
 			timer.Stop()
-			return err
+			return errors.Wrap(ctx.Err(), "aborting send retries", z.Err(err))
 		case <-timer.C:
 		}
 	}
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 28bb0808c..3074918d6 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -249,6 +249,16 @@ func TestSendRetries(t *testing.T) {
 		require.ErrorContains(t, err, "transient stream failure")
 		require.Equal(t, 3, flaky.Calls())
 	})
+
+	t.Run("cancellation surfaces as context error", func(t *testing.T) {
+		cancelledCtx, cancel := context.WithCancel(ctx)
+		cancel()
+
+		flaky := testutil.NewFlakyHost(client, 10)
+		err := p2p.Send(cancelledCtx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 4}, p2p.WithRetries(5))
+		require.ErrorIs(t, err, context.Canceled)
+		require.Equal(t, 1, flaky.Calls())
+	})
 }
 
 func TestSendReceiveRetries(t *testing.T) {
diff --git a/testutil/random.go b/testutil/random.go
index e5f3a5819..a1a911f2c 100644
--- a/testutil/random.go
+++ b/testutil/random.go
@@ -11,6 +11,7 @@ import (
 	"math"
 	"math/rand"
 	"net"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -1534,24 +1535,34 @@ func CreateHost(t *testing.T, addr *net.TCPAddr, opts ...libp2p.Option) host.Hos
 }
 
 // NewFlakyHost wraps a host, failing the first `failures` outbound NewStream calls
-// with a transient error. It is used to test p2p send retry behavior.
-func NewFlakyHost(h host.Host, failures int) *FlakyHost {
-	return &FlakyHost{Host: h, failures: failures}
+// with a transient error. If protocols are provided, only NewStream calls for those
+// protocols fail; other calls pass through. It is used to test p2p send retry behavior.
+func NewFlakyHost(h host.Host, failures int, protocols ...protocol.ID) *FlakyHost {
+	return &FlakyHost{Host: h, failures: failures, protocols: protocols}
 }
 
 // FlakyHost is a host whose first outbound NewStream calls fail with a transient error.
 type FlakyHost struct {
 	host.Host
 
+	protocols []protocol.ID
+
 	mu       sync.Mutex
 	failures int
 	calls    int
 }
 
 func (h *FlakyHost) NewStream(ctx context.Context, peerID peer.ID, pIDs ...protocol.ID) (network.Stream, error) {
+	match := len(h.protocols) == 0
+	for _, pID := range pIDs {
+		if slices.Contains(h.protocols, pID) {
+			match = true
+		}
+	}
+
 	h.mu.Lock()
 	h.calls++
-	fail := h.failures > 0
+	fail := match && h.failures > 0
 
 	if fail {
 		h.failures--

From a9ee2d32662f10434c5fed2d968bfa77dfb17c80 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 17:59:14 +0300
Subject: [PATCH 04/13] dkg: add frost round 1 send retry test

---
 dkg/frostp2p_internal_test.go | 86 +++++++++++++++++++++++++++++++++++
 1 file changed, 86 insertions(+)

diff --git a/dkg/frostp2p_internal_test.go b/dkg/frostp2p_internal_test.go
index 2e72dae8d..21389f752 100644
--- a/dkg/frostp2p_internal_test.go
+++ b/dkg/frostp2p_internal_test.go
@@ -5,17 +5,103 @@ package dkg
 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/stretchr/testify/require"
 
 	"github.com/obolnetwork/charon/cluster"
+	"github.com/obolnetwork/charon/dkg/bcast"
 	pb "github.com/obolnetwork/charon/dkg/dkgpb/v1"
 	"github.com/obolnetwork/charon/testutil"
 )
 
+// TestFrostRound1RetriesTransientSendFailures ensures the frost round-1 p2p transport
+// survives transient send failures instead of aborting the whole ceremony on the first one.
+func TestFrostRound1RetriesTransientSendFailures(t *testing.T) {
+	const (
+		nodes     = 3
+		threshold = 2
+		numVals   = 2
+		dkgCtx    = "test round 1 retries"
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	var (
+		secrets []*k1.PrivateKey
+		hosts   []host.Host
+		peers   []peer.ID
+		peerMap = make(map[peer.ID]cluster.NodeIdx)
+	)
+
+	for i := range nodes {
+		secret, err := k1.GeneratePrivateKey()
+		require.NoError(t, err)
+
+		secrets = append(secrets, secret)
+
+		h := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+		hosts = append(hosts, h)
+		peers = append(peers, h.ID())
+		peerMap[h.ID()] = cluster.NodeIdx{PeerIdx: i, ShareIdx: i + 1}
+	}
+
+	for i := range nodes {
+		for j := range nodes {
+			if i == j {
+				continue
+			}
+
+			hosts[i].Peerstore().AddAddrs(peers[j], hosts[j].Addrs(), peerstore.PermanentAddrTTL)
+		}
+	}
+
+	// Node 0's first round-1 p2p send to each peer fails transiently,
+	// leaving the reliable-broadcast phase untouched.
+	hosts[0] = testutil.NewFlakyHost(hosts[0], nodes-1, round1P2PID)
+
+	var transports []*frostP2P
+
+	for i := range nodes {
+		bcastComp := bcast.New(hosts[i], peers, secrets[i], []byte("session hash"))
+
+		tp, err := newFrostP2P(hosts[i], peerMap, bcastComp, threshold, numVals)
+		require.NoError(t, err)
+
+		transports = append(transports, tp)
+	}
+
+	errChan := make(chan error, nodes)
+
+	for i := range nodes {
+		go func(node int) {
+			validators, err := newFrostParticipants(numVals, nodes, threshold, uint32(node+1), dkgCtx)
+			if err != nil {
+				errChan <- err
+				return
+			}
+
+			castR1, p2pR1, err := round1(validators)
+			if err != nil {
+				errChan <- err
+				return
+			}
+
+			_, _, err = transports[node].Round1(ctx, castR1, p2pR1)
+			errChan <- err
+		}(i)
+	}
+
+	for range nodes {
+		require.NoError(t, <-errChan)
+	}
+}
+
 func TestBcastCallback(t *testing.T) {
 	const (
 		n         = 4

From 1ba73c4a72f12c00917c0fdcd81b7fab63cbf184 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 18:42:25 +0300
Subject: [PATCH 05/13] dkg: apply 60s p2p send timeout to ceremony sends

---
 dkg/bcast/client.go    |  2 +-
 dkg/bcast/impl_test.go | 72 ++++++++++++++++++++++++++++++++++++++++++
 dkg/exchanger.go       | 38 ++++++++++++++--------
 dkg/frostp2p.go        |  2 +-
 dkg/pedersen/board.go  |  8 ++++-
 5 files changed, 106 insertions(+), 16 deletions(-)

diff --git a/dkg/bcast/client.go b/dkg/bcast/client.go
index 1c53d7bde..5f6c9d504 100644
--- a/dkg/bcast/client.go
+++ b/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, p2p.WithRetries(sendRetries))
+		err := c.sendRecvFunc(ctx, c.p2pNode, pID, sigReq, sigResp, protocolIDSig, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 
 		return sigResp, err
 	})
diff --git a/dkg/bcast/impl_test.go b/dkg/bcast/impl_test.go
index 7dbeb0b2a..b39016453 100644
--- a/dkg/bcast/impl_test.go
+++ b/dkg/bcast/impl_test.go
@@ -5,6 +5,7 @@ package bcast_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	k1 "github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -231,6 +232,77 @@ func TestBCastRetriesTransientSendFailures(t *testing.T) {
 	}
 }
 
+// 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) {
diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index 027977e5a..c0f785164 100644
--- a/dkg/exchanger.go
+++ b/dkg/exchanger.go
@@ -40,18 +40,30 @@ const (
 	// Do not add new values greater than sigDepositData.
 )
 
-// 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.
-const sendRetries = 2
-
-// sendRetrying wraps p2p.Send with sendRetries retries. It implements p2p.SendFunc.
-func sendRetrying(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID,
-	msg proto.Message, opts ...p2p.SendRecvOption,
-) error {
-	return p2p.Send(ctx, p2pNode, protoID, peerID, msg, append(opts, p2p.WithRetries(sendRetries))...)
+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 = 2
+
+	// sendTimeout is the p2p send deadline per attempt during DKG ceremonies. 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.
@@ -127,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, sendRetrying, 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{},
diff --git a/dkg/frostp2p.go b/dkg/frostp2p.go
index 4e5a897d8..24bca7de6 100644
--- a/dkg/frostp2p.go
+++ b/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, p2p.WithRetries(sendRetries))
+		err := p2p.Send(ctx, f.p2pNode, round1P2PID, pID, p2pMsg, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 		if err != nil {
 			return nil, nil, err
 		}
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index 7b8b2b1bc..d727f7408 100644
--- a/dkg/pedersen/board.go
+++ b/dkg/pedersen/board.go
@@ -8,6 +8,7 @@ import (
 	"crypto/sha256"
 	"path"
 	"sync"
+	"time"
 
 	kdkg "github.com/drand/kyber/share/dkg"
 	"github.com/libp2p/go-libp2p/core/host"
@@ -61,6 +62,11 @@ const (
 	// complete if a bundle is dropped, so transient failures (e.g. a stalled relay
 	// connection) are retried instead. Receivers deduplicate re-delivered messages.
 	sendRetries = 2
+
+	// sendTimeout is the p2p send deadline per attempt. 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
 )
 
 var (
@@ -259,7 +265,7 @@ func (b *Board) broadcastP2P(ctx context.Context, msgID string, msg proto.Messag
 			continue
 		}
 
-		if err := b.sender.SendAsync(ctx, b.host, protocol.ID(msgID), peerID, msg, p2p.WithRetries(sendRetries)); err != nil {
+		if err := b.sender.SendAsync(ctx, b.host, protocol.ID(msgID), peerID, msg, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries)); err != nil {
 			return errors.Wrap(err, "p2p send", z.Str("msg", msgID), z.Str("to", peerID.String()))
 		}
 	}

From 2dbd757e13c4ca5933f527e1ec4b92801d7f0bb4 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:00:23 +0300
Subject: [PATCH 06/13] dkg: increase ceremony send retries to 5

---
 dkg/bcast/helpers.go  | 2 +-
 dkg/exchanger.go      | 2 +-
 dkg/pedersen/board.go | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/dkg/bcast/helpers.go b/dkg/bcast/helpers.go
index d7dfb951e..cdb5a4af8 100644
--- a/dkg/bcast/helpers.go
+++ b/dkg/bcast/helpers.go
@@ -24,7 +24,7 @@ const (
 	// 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 = 2
+	sendRetries = 5
 )
 
 // hashFunc is a function that hashes a message ID and a any-wrapped protobuf message.
diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index c0f785164..ecf9e919f 100644
--- a/dkg/exchanger.go
+++ b/dkg/exchanger.go
@@ -46,7 +46,7 @@ const (
 	// 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 = 2
+	sendRetries = 5
 
 	// sendTimeout is the p2p send deadline per attempt during DKG ceremonies. Ceremony
 	// peers may lag behind (e.g. slower operators or relay-routed connections), so sends
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index d727f7408..e719598e3 100644
--- a/dkg/pedersen/board.go
+++ b/dkg/pedersen/board.go
@@ -61,7 +61,7 @@ const (
 	// sendRetries is the number of additional p2p send attempts. A DKG ceremony cannot
 	// complete if a bundle is dropped, so transient failures (e.g. a stalled relay
 	// connection) are retried instead. Receivers deduplicate re-delivered messages.
-	sendRetries = 2
+	sendRetries = 5
 
 	// sendTimeout is the p2p send deadline per attempt. Ceremony peers may lag behind
 	// (e.g. slower operators or relay-routed connections), so sends get much more

From 2cd1e5c7265a2d5f589c89d75569cc62fadbebcc Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:23:10 +0300
Subject: [PATCH 07/13] p2p: bound send retries by the send timeout budget

---
 dkg/exchanger.go      |  6 +++---
 dkg/pedersen/board.go |  6 +++---
 p2p/sender.go         | 47 ++++++++++++++++++++++++++++---------------
 p2p/sender_test.go    | 13 ++++++++++++
 4 files changed, 50 insertions(+), 22 deletions(-)

diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index ecf9e919f..4cb194047 100644
--- a/dkg/exchanger.go
+++ b/dkg/exchanger.go
@@ -48,9 +48,9 @@ const (
 	// is harmless.
 	sendRetries = 5
 
-	// sendTimeout is the p2p send deadline per attempt during DKG ceremonies. 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 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
 )
 
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index e719598e3..c3b725f9f 100644
--- a/dkg/pedersen/board.go
+++ b/dkg/pedersen/board.go
@@ -63,9 +63,9 @@ const (
 	// connection) are retried instead. Receivers deduplicate re-delivered messages.
 	sendRetries = 5
 
-	// sendTimeout is the p2p send deadline per attempt. 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 is the total p2p send budget, 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
 )
 
diff --git a/p2p/sender.go b/p2p/sender.go
index f56f386c6..1f9c1ecd4 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -215,6 +215,8 @@ func WithReceiveTimeout(timeout time.Duration) func(*sendRecvOpts) {
 }
 
 // WithSendTimeout returns an option for SendReceive that sets a timeout for sending messages.
+// The timeout is the total wall-clock budget for the call, including all retry attempts
+// and backoff when combined with WithRetries.
 func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
 		opts.sendTimeout = timeout
@@ -222,9 +224,10 @@ func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 }
 
 // WithRetries returns an option that retries a failed send up to the given number of
-// additional attempts, each on a fresh stream, backing off briefly in between.
-// Only use it for protocols whose handlers tolerate duplicate delivery, since a send
-// that failed on the sender side may still have been delivered (e.g. DKG ceremony
+// additional attempts, each on a fresh stream, backing off briefly in between. The whole
+// sequence shares the send timeout as its total budget, so retries never extend a send
+// beyond it. Only use it for protocols whose handlers tolerate duplicate delivery, since
+// a send that failed on the sender side may still have been delivered (e.g. DKG ceremony
 // messages, which are deduplicated by all receivers).
 func WithRetries(retries int) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
@@ -302,10 +305,11 @@ func defaultSendRecvOpts(pID protocol.ID) sendRecvOpts {
 }
 
 // withRetries calls fn and retries it up to the given number of additional times,
-// backing off briefly between attempts. Cancellation stops retrying and surfaces
-// as the context error (with the last attempt error attached as a field), so
-// callers can detect it with errors.Is.
-func withRetries(ctx context.Context, retries int, fn func() error) error {
+// backing off briefly between attempts. The whole sequence (attempts and backoff)
+// is bounded by the deadline, so retries never extend a send beyond its configured
+// send timeout. Cancellation stops retrying and surfaces as the context error (with
+// the last attempt error attached as a field), so callers can detect it with errors.Is.
+func withRetries(ctx context.Context, retries int, deadline time.Time, fn func() error) error {
 	var err error
 
 	for attempt := 0; ; attempt++ {
@@ -314,7 +318,12 @@ func withRetries(ctx context.Context, retries int, fn func() error) error {
 			return err
 		}
 
-		timer := time.NewTimer(expbackoff.Backoff(expbackoff.FastConfig, attempt))
+		backoff := expbackoff.Backoff(expbackoff.FastConfig, attempt)
+		if !time.Now().Add(backoff).Before(deadline) {
+			return err // Budget exhausted, another attempt would fail its deadline immediately.
+		}
+
+		timer := time.NewTimer(backoff)
 		select {
 		case <-ctx.Done():
 			timer.Stop()
@@ -340,17 +349,20 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
-	return withRetries(ctx, o.retries, func() error {
+	// The send timeout is the total budget for the call: all attempts share one deadline.
+	deadline := time.Now().Add(o.sendTimeout)
+
+	return withRetries(ctx, o.retries, deadline, func() error {
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o)
+		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, deadline)
 	})
 }
 
 // sendReceive is a single SendReceive attempt.
 func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
-	req, resp proto.Message, pID protocol.ID, o sendRecvOpts,
+	req, resp proto.Message, pID protocol.ID, o sendRecvOpts, deadline time.Time,
 ) error {
 	tStart := time.Now()
 
@@ -369,7 +381,7 @@ func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 
 	protoLabel = string(s.Protocol())
 
-	if err := s.SetDeadline(time.Now().Add(o.sendTimeout)); err != nil {
+	if err := s.SetDeadline(deadline); err != nil {
 		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
@@ -424,14 +436,17 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 		opt(&o)
 	}
 
-	return withRetries(ctx, o.retries, func() error {
-		return send(ctx, p2pNode, protoID, peerID, msg, o)
+	// The send timeout is the total budget for the call: all attempts share one deadline.
+	deadline := time.Now().Add(o.sendTimeout)
+
+	return withRetries(ctx, o.retries, deadline, func() error {
+		return send(ctx, p2pNode, protoID, peerID, msg, o, deadline)
 	})
 }
 
 // send is a single Send attempt.
 func send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID, msg proto.Message,
-	o sendRecvOpts,
+	o sendRecvOpts, deadline time.Time,
 ) error {
 	t0 := time.Now()
 
@@ -452,7 +467,7 @@ func send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 
 	protoLabel = string(s.Protocol())
 
-	if err := s.SetDeadline(time.Now().Add(o.sendTimeout)); err != nil {
+	if err := s.SetDeadline(deadline); err != nil {
 		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 3074918d6..f5329d757 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -250,6 +250,19 @@ func TestSendRetries(t *testing.T) {
 		require.Equal(t, 3, flaky.Calls())
 	})
 
+	t.Run("retries bounded by send timeout", func(t *testing.T) {
+		flaky := testutil.NewFlakyHost(client, 100)
+		t0 := time.Now()
+
+		// The send timeout is the total budget for all attempts, so the retry
+		// sequence must stop long before all 8 retries (~7s of backoff) elapse.
+		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 5},
+			p2p.WithRetries(8), p2p.WithSendTimeout(200*time.Millisecond))
+		require.ErrorContains(t, err, "transient stream failure")
+		require.Less(t, time.Since(t0), 3*time.Second)
+		require.Less(t, flaky.Calls(), 9)
+	})
+
 	t.Run("cancellation surfaces as context error", func(t *testing.T) {
 		cancelledCtx, cancel := context.WithCancel(ctx)
 		cancel()

From 1c34f0efc0dd7d4ae124924821166b60dd901fe0 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:32:48 +0300
Subject: [PATCH 08/13] p2p: bound stream creation by the send timeout budget

---
 p2p/sender.go      | 14 ++++++++--
 p2p/sender_test.go | 65 +++++++++++++++++++++++++++++++++++++++++++---
 2 files changed, 74 insertions(+), 5 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 1f9c1ecd4..9e55fd508 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -349,9 +349,14 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
-	// The send timeout is the total budget for the call: all attempts share one deadline.
+	// The send timeout is the total budget for the call: stream creation and all
+	// attempts share one deadline, also enforced via the context so dialing and
+	// protocol negotiation cannot block past it.
 	deadline := time.Now().Add(o.sendTimeout)
 
+	ctx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+
 	return withRetries(ctx, o.retries, deadline, func() error {
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
@@ -436,9 +441,14 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 		opt(&o)
 	}
 
-	// The send timeout is the total budget for the call: all attempts share one deadline.
+	// The send timeout is the total budget for the call: stream creation and all
+	// attempts share one deadline, also enforced via the context so dialing and
+	// protocol negotiation cannot block past it.
 	deadline := time.Now().Add(o.sendTimeout)
 
+	ctx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+
 	return withRetries(ctx, o.retries, deadline, func() error {
 		return send(ctx, p2pNode, protoID, peerID, msg, o, deadline)
 	})
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index f5329d757..ab4078746 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -12,6 +12,7 @@ import (
 	"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/peerstore"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -124,10 +125,9 @@ func TestSendReceiveClosesStreamOnError(t *testing.T) {
 func TestWithSendTimeout(t *testing.T) {
 	servers := []host.Host{testutil.CreateHost(t, testutil.AvailableAddr(t)), testutil.CreateQUICHost(t, testutil.AvailableUDPAddr(t))}
 	clients := []host.Host{testutil.CreateHost(t, testutil.AvailableAddr(t)), testutil.CreateQUICHost(t, testutil.AvailableUDPAddr(t))}
-	errors := []string{"deadline reached", "deadline exceeded"}
 
 	for i := range len(servers) {
-		client, server, errorStr := clients[i], servers[i], errors[i]
+		client, server := clients[i], servers[i]
 
 		client.Peerstore().AddAddrs(server.ID(), server.Addrs(), time.Hour)
 
@@ -141,10 +141,13 @@ func TestWithSendTimeout(t *testing.T) {
 				return nil, false, nil
 			})
 
+		// Depending on how far the tiny budget lets the call get, the deadline error
+		// surfaces at dialing ("context deadline exceeded"), or on the stream
+		// ("deadline reached" on TCP, "deadline exceeded" on QUIC).
 		err := p2p.SendReceive(context.Background(), client, server.ID(),
 			new(pbv1.Duty), new(pbv1.Duty), protocolID, p2p.WithSendTimeout(sendTimeout))
 		require.Error(t, err)
-		require.ErrorContains(t, err, errorStr)
+		require.ErrorContains(t, err, "deadline")
 	}
 }
 
@@ -274,6 +277,62 @@ func TestSendRetries(t *testing.T) {
 	})
 }
 
+// blockingHost wraps a host whose NewStream blocks until the context is done,
+// simulating a hung dial or protocol negotiation.
+type blockingHost struct {
+	host.Host
+}
+
+func (h blockingHost) NewStream(ctx context.Context, _ peer.ID, _ ...protocol.ID) (network.Stream, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func TestSendTimeoutBoundsStreamCreation(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	blocking := blockingHost{Host: client}
+
+	tests := []struct {
+		name string
+		send func() error
+	}{
+		{
+			name: "send",
+			send: func() error {
+				return p2p.Send(ctx, blocking, "proto", server.ID(), &pbv1.Duty{Slot: 1},
+					p2p.WithSendTimeout(100*time.Millisecond), p2p.WithRetries(2))
+			},
+		},
+		{
+			name: "send receive",
+			send: func() error {
+				return p2p.SendReceive(ctx, blocking, server.ID(), new(pbv1.Duty), new(pbv1.Duty), "proto",
+					p2p.WithSendTimeout(100*time.Millisecond), p2p.WithRetries(2))
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			errChan := make(chan error, 1)
+			go func() {
+				errChan <- test.send()
+			}()
+
+			select {
+			case err := <-errChan:
+				require.ErrorIs(t, err, context.DeadlineExceeded)
+			case <-time.After(2 * time.Second):
+				require.Fail(t, "send blocked past its timeout budget on stream creation")
+			}
+		})
+	}
+}
+
 func TestSendReceiveRetries(t *testing.T) {
 	ctx := context.Background()
 	server := testutil.CreateHost(t, testutil.AvailableAddr(t))

From 96bd9d5c92b3db43dfff29e3ed933b492e971c69 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 21:51:33 +0300
Subject: [PATCH 09/13] p2p: cap each send attempt at its slice of the timeout
 budget

---
 p2p/sender.go      | 32 +++++++++++++++++++++++++-------
 p2p/sender_test.go | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 58 insertions(+), 7 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 9e55fd508..30d09a2d7 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -226,9 +226,10 @@ func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 // WithRetries returns an option that retries a failed send up to the given number of
 // additional attempts, each on a fresh stream, backing off briefly in between. The whole
 // sequence shares the send timeout as its total budget, so retries never extend a send
-// beyond it. Only use it for protocols whose handlers tolerate duplicate delivery, since
-// a send that failed on the sender side may still have been delivered (e.g. DKG ceremony
-// messages, which are deduplicated by all receivers).
+// beyond it; each attempt is capped at its slice of the budget, so a stalled stream
+// cannot starve the remaining attempts. Only use it for protocols whose handlers tolerate
+// duplicate delivery, since a send that failed on the sender side may still have been
+// delivered (e.g. DKG ceremony messages, which are deduplicated by all receivers).
 func WithRetries(retries int) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
 		opts.retries = retries
@@ -351,8 +352,11 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 
 	// The send timeout is the total budget for the call: stream creation and all
 	// attempts share one deadline, also enforced via the context so dialing and
-	// protocol negotiation cannot block past it.
+	// protocol negotiation cannot block past it. Each attempt gets a slice of the
+	// budget, so a stalled stream cannot consume it all and a retry on a fresh
+	// stream can still occur.
 	deadline := time.Now().Add(o.sendTimeout)
+	attemptTimeout := o.sendTimeout / time.Duration(o.retries+1)
 
 	ctx, cancel := context.WithDeadline(ctx, deadline)
 	defer cancel()
@@ -361,10 +365,21 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, deadline)
+		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, attemptDeadline(attemptTimeout, deadline))
 	})
 }
 
+// attemptDeadline returns the deadline for a single send attempt: its slice of the
+// total budget, capped by the overall deadline.
+func attemptDeadline(attemptTimeout time.Duration, overall time.Time) time.Time {
+	deadline := time.Now().Add(attemptTimeout)
+	if deadline.After(overall) {
+		return overall
+	}
+
+	return deadline
+}
+
 // sendReceive is a single SendReceive attempt.
 func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 	req, resp proto.Message, pID protocol.ID, o sendRecvOpts, deadline time.Time,
@@ -443,14 +458,17 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 
 	// The send timeout is the total budget for the call: stream creation and all
 	// attempts share one deadline, also enforced via the context so dialing and
-	// protocol negotiation cannot block past it.
+	// protocol negotiation cannot block past it. Each attempt gets a slice of the
+	// budget, so a stalled stream cannot consume it all and a retry on a fresh
+	// stream can still occur.
 	deadline := time.Now().Add(o.sendTimeout)
+	attemptTimeout := o.sendTimeout / time.Duration(o.retries+1)
 
 	ctx, cancel := context.WithDeadline(ctx, deadline)
 	defer cancel()
 
 	return withRetries(ctx, o.retries, deadline, func() error {
-		return send(ctx, p2pNode, protoID, peerID, msg, o, deadline)
+		return send(ctx, p2pNode, protoID, peerID, msg, o, attemptDeadline(attemptTimeout, deadline))
 	})
 }
 
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index ab4078746..904df4edd 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -333,6 +333,39 @@ func TestSendTimeoutBoundsStreamCreation(t *testing.T) {
 	}
 }
 
+// TestSendReceiveRetriesAfterStall ensures a stalled attempt only consumes its slice of
+// the total send budget, leaving room for a retry on a fresh stream — the incident mode
+// where a stream stalls until its I/O deadline.
+func TestSendReceiveRetriesAfterStall(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	var calls atomic.Int32
+
+	protocolID := protocol.ID("testprotocol-stall-retries")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			if calls.Add(1) == 1 {
+				time.Sleep(3 * time.Second) // Stall the first attempt past its deadline.
+			}
+
+			duty, ok := req.(*pbv1.Duty)
+			require.True(t, ok)
+
+			return &pbv1.Duty{Slot: duty.GetSlot() + 1}, true, nil
+		})
+
+	resp := new(pbv1.Duty)
+	err := p2p.SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 41}, resp, protocolID,
+		p2p.WithSendTimeout(2*time.Second), p2p.WithRetries(2))
+	require.NoError(t, err)
+	require.EqualValues(t, 42, resp.GetSlot())
+	require.GreaterOrEqual(t, calls.Load(), int32(2))
+}
+
 func TestSendReceiveRetries(t *testing.T) {
 	ctx := context.Background()
 	server := testutil.CreateHost(t, testutil.AvailableAddr(t))

From 089aa18a302cf4588e6e02a1592c3b40457205b5 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Fri, 4 Sep 2026 22:01:18 +0300
Subject: [PATCH 10/13] p2p: bound each attempt's dial by its deadline slice

---
 p2p/sender.go      | 18 ++++++++++++++--
 p2p/sender_test.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+), 2 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 30d09a2d7..24acae304 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -365,7 +365,14 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, attemptDeadline(attemptTimeout, deadline))
+		// Bound the attempt's dialing and negotiation by its deadline as well,
+		// so a hung dial cannot consume the remaining attempts' budget.
+		attemptDL := attemptDeadline(attemptTimeout, deadline)
+
+		attemptCtx, cancel := context.WithDeadline(ctx, attemptDL)
+		defer cancel()
+
+		return sendReceive(attemptCtx, p2pNode, peerID, req, resp, pID, o, attemptDL)
 	})
 }
 
@@ -468,7 +475,14 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 	defer cancel()
 
 	return withRetries(ctx, o.retries, deadline, func() error {
-		return send(ctx, p2pNode, protoID, peerID, msg, o, attemptDeadline(attemptTimeout, deadline))
+		// Bound the attempt's dialing and negotiation by its deadline as well,
+		// so a hung dial cannot consume the remaining attempts' budget.
+		attemptDL := attemptDeadline(attemptTimeout, deadline)
+
+		attemptCtx, cancel := context.WithDeadline(ctx, attemptDL)
+		defer cancel()
+
+		return send(attemptCtx, p2pNode, protoID, peerID, msg, o, attemptDL)
 	})
 }
 
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 904df4edd..82e8cbec8 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -333,6 +333,58 @@ func TestSendTimeoutBoundsStreamCreation(t *testing.T) {
 	}
 }
 
+// blockOnceHost wraps a host whose first NewStream blocks until the context is done,
+// simulating a hung dial on the first attempt only.
+type blockOnceHost struct {
+	host.Host
+
+	calls atomic.Int32
+}
+
+func (h *blockOnceHost) NewStream(ctx context.Context, peerID peer.ID, pIDs ...protocol.ID) (network.Stream, error) {
+	if h.calls.Add(1) == 1 {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+
+	return h.Host.NewStream(ctx, peerID, pIDs...)
+}
+
+// TestSendRetriesAfterStalledDial ensures a hung dial only consumes its slice of the
+// total send budget, leaving room for a retry.
+func TestSendRetriesAfterStalledDial(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	received := make(chan *pbv1.Duty, 1)
+	protocolID := protocol.ID("testprotocol-stalled-dial")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			duty, ok := req.(*pbv1.Duty)
+			require.True(t, ok)
+
+			received <- duty
+
+			return nil, false, nil
+		})
+
+	blocking := &blockOnceHost{Host: client}
+	err := p2p.Send(ctx, blocking, protocolID, server.ID(), &pbv1.Duty{Slot: 7},
+		p2p.WithSendTimeout(2*time.Second), p2p.WithRetries(2))
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, blocking.calls.Load(), int32(2))
+
+	select {
+	case duty := <-received:
+		require.EqualValues(t, 7, duty.GetSlot())
+	case <-time.After(5 * time.Second):
+		require.Fail(t, "timed out waiting for message")
+	}
+}
+
 // TestSendReceiveRetriesAfterStall ensures a stalled attempt only consumes its slice of
 // the total send budget, leaving room for a retry on a fresh stream — the incident mode
 // where a stream stalls until its I/O deadline.

From 7b1738d5727abe68c7fd3d86d63c2309149c2505 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Sat, 5 Sep 2026 00:05:54 +0300
Subject: [PATCH 11/13] p2p: give each SendReceive attempt the full timeout,
 clamp negative retries

---
 p2p/sender.go      | 19 ++++++++--------
 p2p/sender_test.go | 57 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 66 insertions(+), 10 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 24acae304..5b10c67d2 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -232,7 +232,7 @@ func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 // delivered (e.g. DKG ceremony messages, which are deduplicated by all receivers).
 func WithRetries(retries int) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
-		opts.retries = retries
+		opts.retries = max(retries, 0)
 	}
 }
 
@@ -350,13 +350,12 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
-	// The send timeout is the total budget for the call: stream creation and all
-	// attempts share one deadline, also enforced via the context so dialing and
-	// protocol negotiation cannot block past it. Each attempt gets a slice of the
-	// budget, so a stalled stream cannot consume it all and a retry on a fresh
-	// stream can still occur.
-	deadline := time.Now().Add(o.sendTimeout)
-	attemptTimeout := o.sendTimeout / time.Duration(o.retries+1)
+	// Unlike a one-way Send, each SendReceive attempt gets the full send timeout, not a
+	// slice of it: the response wait is a legitimate long operation (a peer may take up to
+	// its receive timeout to process and reply), so slicing it across retries would abort
+	// valid slow responses. The overall deadline therefore spans all attempts.
+	attemptTimeout := o.sendTimeout
+	deadline := time.Now().Add(attemptTimeout * time.Duration(o.retries+1))
 
 	ctx, cancel := context.WithDeadline(ctx, deadline)
 	defer cancel()
@@ -365,8 +364,8 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		// Bound the attempt's dialing and negotiation by its deadline as well,
-		// so a hung dial cannot consume the remaining attempts' budget.
+		// Bound each attempt (dialing, negotiation and the response wait) by its own
+		// deadline so a hung attempt cannot consume the remaining attempts' budget.
 		attemptDL := attemptDeadline(attemptTimeout, deadline)
 
 		attemptCtx, cancel := context.WithDeadline(ctx, attemptDL)
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 82e8cbec8..420349818 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -385,6 +385,63 @@ func TestSendRetriesAfterStalledDial(t *testing.T) {
 	}
 }
 
+// TestSendReceiveAllowsSlowResponse ensures each SendReceive attempt gets the full send
+// timeout for the response wait, not a slice of it, so a peer that legitimately responds
+// slower than sendTimeout/(retries+1) is not aborted.
+func TestSendReceiveAllowsSlowResponse(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	var calls atomic.Int32
+
+	protocolID := protocol.ID("testprotocol-slow-response")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			calls.Add(1)
+			// Respond slower than a sliced budget (1s/6 ≈ 166ms) would allow, but
+			// well within the full 1s send timeout.
+			time.Sleep(400 * time.Millisecond)
+
+			duty, ok := req.(*pbv1.Duty)
+			require.True(t, ok)
+
+			return &pbv1.Duty{Slot: duty.GetSlot() + 1}, true, nil
+		})
+
+	resp := new(pbv1.Duty)
+	err := p2p.SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 41}, resp, protocolID,
+		p2p.WithSendTimeout(time.Second), p2p.WithRetries(5))
+	require.NoError(t, err)
+	require.EqualValues(t, 42, resp.GetSlot())
+	require.EqualValues(t, 1, calls.Load(), "slow but valid response must succeed on the first attempt")
+}
+
+func TestWithRetriesClampsNegative(t *testing.T) {
+	ctx := context.Background()
+	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	protocolID := protocol.ID("testprotocol-negative-retries")
+	p2p.RegisterHandler("test", server, protocolID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) {
+			return nil, false, nil
+		})
+
+	// A negative retry count must not panic (it would divide by zero when computing the
+	// per-attempt slice) and behaves as zero retries.
+	require.NotPanics(t, func() {
+		flaky := testutil.NewFlakyHost(client, 1)
+		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 1}, p2p.WithRetries(-1))
+		require.Error(t, err)
+		require.Equal(t, 1, flaky.Calls())
+	})
+}
+
 // TestSendReceiveRetriesAfterStall ensures a stalled attempt only consumes its slice of
 // the total send budget, leaving room for a retry on a fresh stream — the incident mode
 // where a stream stalls until its I/O deadline.

From 969e27a7c5c075a0984adf9c990d17f0e33ab0a5 Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Sat, 5 Sep 2026 00:23:14 +0300
Subject: [PATCH 12/13] p2p: bound SendReceive by its send timeout without
 slicing the response wait

---
 p2p/sender.go      | 23 +++++++++--------------
 p2p/sender_test.go | 36 ++++++++++++++----------------------
 2 files changed, 23 insertions(+), 36 deletions(-)

diff --git a/p2p/sender.go b/p2p/sender.go
index 5b10c67d2..8fca52d9b 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -350,12 +350,14 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
-	// Unlike a one-way Send, each SendReceive attempt gets the full send timeout, not a
-	// slice of it: the response wait is a legitimate long operation (a peer may take up to
-	// its receive timeout to process and reply), so slicing it across retries would abort
-	// valid slow responses. The overall deadline therefore spans all attempts.
-	attemptTimeout := o.sendTimeout
-	deadline := time.Now().Add(attemptTimeout * time.Duration(o.retries+1))
+	// The send timeout is the total budget for the whole call. Unlike a one-way Send, each
+	// SendReceive attempt may use the full remaining budget rather than a fixed slice: the
+	// response wait is a legitimate long operation (a peer may take up to its receive
+	// timeout to reply), so slicing it would abort valid slow responses. Retries therefore
+	// only fire on attempts that fail fast enough to leave budget (e.g. dial errors); a
+	// stalled attempt consumes the budget and is not retried, since a delivered request to
+	// a slow peer must be waited out, not re-sent.
+	deadline := time.Now().Add(o.sendTimeout)
 
 	ctx, cancel := context.WithDeadline(ctx, deadline)
 	defer cancel()
@@ -364,14 +366,7 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		// Bound each attempt (dialing, negotiation and the response wait) by its own
-		// deadline so a hung attempt cannot consume the remaining attempts' budget.
-		attemptDL := attemptDeadline(attemptTimeout, deadline)
-
-		attemptCtx, cancel := context.WithDeadline(ctx, attemptDL)
-		defer cancel()
-
-		return sendReceive(attemptCtx, p2pNode, peerID, req, resp, pID, o, attemptDL)
+		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, deadline)
 	})
 }
 
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 420349818..6df244796 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -442,37 +442,29 @@ func TestWithRetriesClampsNegative(t *testing.T) {
 	})
 }
 
-// TestSendReceiveRetriesAfterStall ensures a stalled attempt only consumes its slice of
-// the total send budget, leaving room for a retry on a fresh stream — the incident mode
-// where a stream stalls until its I/O deadline.
-func TestSendReceiveRetriesAfterStall(t *testing.T) {
+// TestSendReceiveBoundedBySendTimeout ensures the whole SendReceive call, retries included,
+// is bounded by the send timeout: a stalled attempt consumes the budget and is not retried
+// past it (a delivered request to a slow peer is waited out, not re-sent).
+func TestSendReceiveBoundedBySendTimeout(t *testing.T) {
 	ctx := context.Background()
 	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
 	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
 	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
 
-	var calls atomic.Int32
-
-	protocolID := protocol.ID("testprotocol-stall-retries")
+	protocolID := protocol.ID("testprotocol-sendrecv-bounded")
 	p2p.RegisterHandler("test", server, protocolID,
 		func() proto.Message { return new(pbv1.Duty) },
-		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
-			if calls.Add(1) == 1 {
-				time.Sleep(3 * time.Second) // Stall the first attempt past its deadline.
-			}
-
-			duty, ok := req.(*pbv1.Duty)
-			require.True(t, ok)
-
-			return &pbv1.Duty{Slot: duty.GetSlot() + 1}, true, nil
+		func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) {
+			time.Sleep(3 * time.Second) // Always stall past the budget.
+			return new(pbv1.Duty), true, nil
 		})
 
-	resp := new(pbv1.Duty)
-	err := p2p.SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 41}, resp, protocolID,
-		p2p.WithSendTimeout(2*time.Second), p2p.WithRetries(2))
-	require.NoError(t, err)
-	require.EqualValues(t, 42, resp.GetSlot())
-	require.GreaterOrEqual(t, calls.Load(), int32(2))
+	t0 := time.Now()
+	err := p2p.SendReceive(ctx, client, server.ID(), new(pbv1.Duty), new(pbv1.Duty), protocolID,
+		p2p.WithSendTimeout(500*time.Millisecond), p2p.WithRetries(5))
+	require.Error(t, err)
+	// Bounded by the 500ms budget, not 6 × 500ms, despite 5 retries.
+	require.Less(t, time.Since(t0), 2*time.Second)
 }
 
 func TestSendReceiveRetries(t *testing.T) {

From 892a5b4a98031532753a25bada9493a63e9fbe1f Mon Sep 17 00:00:00 2001
From: kalo <24719519+KaloyanTanev@users.noreply.github.com>
Date: Tue, 8 Sep 2026 13:04:33 +0300
Subject: [PATCH 13/13] p2p: make send timeout per-attempt instead of a shared
 budget

---
 dkg/exchanger.go      |   6 +--
 dkg/pedersen/board.go |   6 +--
 p2p/sender.go         | 110 +++++++++++++++---------------------------
 p2p/sender_test.go    |  45 +++++------------
 4 files changed, 57 insertions(+), 110 deletions(-)

diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index 4cb194047..3a3b9b765 100644
--- a/dkg/exchanger.go
+++ b/dkg/exchanger.go
@@ -48,9 +48,9 @@ const (
 	// 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 is the per-attempt p2p send timeout during DKG ceremonies. 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
 )
 
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index c3b725f9f..641ac732d 100644
--- a/dkg/pedersen/board.go
+++ b/dkg/pedersen/board.go
@@ -63,9 +63,9 @@ const (
 	// connection) are retried instead. Receivers deduplicate re-delivered messages.
 	sendRetries = 5
 
-	// sendTimeout is the total p2p send budget, 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 is the per-attempt p2p send timeout. 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
 )
 
diff --git a/p2p/sender.go b/p2p/sender.go
index 8fca52d9b..0280ca2e9 100644
--- a/p2p/sender.go
+++ b/p2p/sender.go
@@ -215,8 +215,8 @@ func WithReceiveTimeout(timeout time.Duration) func(*sendRecvOpts) {
 }
 
 // WithSendTimeout returns an option for SendReceive that sets a timeout for sending messages.
-// The timeout is the total wall-clock budget for the call, including all retry attempts
-// and backoff when combined with WithRetries.
+// The timeout applies per attempt, so with WithRetries the whole call can take up to
+// (retries+1) times this timeout plus the inter-attempt backoff.
 func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
 		opts.sendTimeout = timeout
@@ -224,12 +224,11 @@ func WithSendTimeout(timeout time.Duration) func(*sendRecvOpts) {
 }
 
 // WithRetries returns an option that retries a failed send up to the given number of
-// additional attempts, each on a fresh stream, backing off briefly in between. The whole
-// sequence shares the send timeout as its total budget, so retries never extend a send
-// beyond it; each attempt is capped at its slice of the budget, so a stalled stream
-// cannot starve the remaining attempts. Only use it for protocols whose handlers tolerate
-// duplicate delivery, since a send that failed on the sender side may still have been
-// delivered (e.g. DKG ceremony messages, which are deduplicated by all receivers).
+// additional attempts, each on a fresh stream with the full send timeout, backing off
+// briefly in between. Only use it for protocols whose handlers tolerate duplicate delivery,
+// since a send that failed on the sender side may still have been delivered (e.g. DKG
+// ceremony messages, which are deduplicated by all receivers). Negative values are clamped
+// to zero (no retries); left unclamped they would make the retry loop skip the send entirely.
 func WithRetries(retries int) func(*sendRecvOpts) {
 	return func(opts *sendRecvOpts) {
 		opts.retries = max(retries, 0)
@@ -306,25 +305,20 @@ func defaultSendRecvOpts(pID protocol.ID) sendRecvOpts {
 }
 
 // withRetries calls fn and retries it up to the given number of additional times,
-// backing off briefly between attempts. The whole sequence (attempts and backoff)
-// is bounded by the deadline, so retries never extend a send beyond its configured
-// send timeout. Cancellation stops retrying and surfaces as the context error (with
-// the last attempt error attached as a field), so callers can detect it with errors.Is.
-func withRetries(ctx context.Context, retries int, deadline time.Time, fn func() error) error {
+// backing off briefly between attempts. Each attempt is independently timed by fn (via
+// the send timeout), so the whole call takes up to (retries+1) attempts plus the backoffs.
+// Cancellation of ctx stops retrying and surfaces as the context error (with the last
+// attempt error attached as a field), so callers can detect it with errors.Is.
+func withRetries(ctx context.Context, retries int, fn func() error) error {
 	var err error
 
-	for attempt := 0; ; attempt++ {
+	for attempt := range retries + 1 {
 		err = fn()
-		if err == nil || attempt >= retries {
+		if err == nil || attempt == retries {
 			return err
 		}
 
-		backoff := expbackoff.Backoff(expbackoff.FastConfig, attempt)
-		if !time.Now().Add(backoff).Before(deadline) {
-			return err // Budget exhausted, another attempt would fail its deadline immediately.
-		}
-
-		timer := time.NewTimer(backoff)
+		timer := time.NewTimer(expbackoff.Backoff(expbackoff.FastConfig, attempt))
 		select {
 		case <-ctx.Done():
 			timer.Stop()
@@ -332,6 +326,8 @@ func withRetries(ctx context.Context, retries int, deadline time.Time, fn func()
 		case <-timer.C:
 		}
 	}
+
+	return err // Unreachable: the final iteration (attempt == retries) always returns.
 }
 
 // SendReceive sends and receives a libp2p request and response message
@@ -350,40 +346,17 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		opt(&o)
 	}
 
-	// The send timeout is the total budget for the whole call. Unlike a one-way Send, each
-	// SendReceive attempt may use the full remaining budget rather than a fixed slice: the
-	// response wait is a legitimate long operation (a peer may take up to its receive
-	// timeout to reply), so slicing it would abort valid slow responses. Retries therefore
-	// only fire on attempts that fail fast enough to leave budget (e.g. dial errors); a
-	// stalled attempt consumes the budget and is not retried, since a delivered request to
-	// a slow peer must be waited out, not re-sent.
-	deadline := time.Now().Add(o.sendTimeout)
-
-	ctx, cancel := context.WithDeadline(ctx, deadline)
-	defer cancel()
-
-	return withRetries(ctx, o.retries, deadline, func() error {
+	return withRetries(ctx, o.retries, func() error {
 		// A failed attempt may have partially populated the response.
 		proto.Reset(resp)
 
-		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o, deadline)
+		return sendReceive(ctx, p2pNode, peerID, req, resp, pID, o)
 	})
 }
 
-// attemptDeadline returns the deadline for a single send attempt: its slice of the
-// total budget, capped by the overall deadline.
-func attemptDeadline(attemptTimeout time.Duration, overall time.Time) time.Time {
-	deadline := time.Now().Add(attemptTimeout)
-	if deadline.After(overall) {
-		return overall
-	}
-
-	return deadline
-}
-
-// sendReceive is a single SendReceive attempt.
+// sendReceive is a single SendReceive attempt, timed out after the send timeout.
 func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
-	req, resp proto.Message, pID protocol.ID, o sendRecvOpts, deadline time.Time,
+	req, resp proto.Message, pID protocol.ID, o sendRecvOpts,
 ) error {
 	tStart := time.Now()
 
@@ -393,6 +366,13 @@ func sendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID,
 		sendDurations.WithLabelValues(PeerName(peerID), protoLabel, o.metricTopic).Observe(time.Since(tStart).Seconds())
 	}()
 
+	// The send timeout bounds this attempt: the context covers dialing and protocol
+	// negotiation, the stream deadline covers the write and response read.
+	deadline := time.Now().Add(o.sendTimeout)
+
+	ctx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+
 	// Circuit relay connections are transient
 	s, err := p2pNode.NewStream(network.WithAllowLimitedConn(ctx, ""), peerID, o.protocols...)
 	if err != nil {
@@ -457,32 +437,14 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 		opt(&o)
 	}
 
-	// The send timeout is the total budget for the call: stream creation and all
-	// attempts share one deadline, also enforced via the context so dialing and
-	// protocol negotiation cannot block past it. Each attempt gets a slice of the
-	// budget, so a stalled stream cannot consume it all and a retry on a fresh
-	// stream can still occur.
-	deadline := time.Now().Add(o.sendTimeout)
-	attemptTimeout := o.sendTimeout / time.Duration(o.retries+1)
-
-	ctx, cancel := context.WithDeadline(ctx, deadline)
-	defer cancel()
-
-	return withRetries(ctx, o.retries, deadline, func() error {
-		// Bound the attempt's dialing and negotiation by its deadline as well,
-		// so a hung dial cannot consume the remaining attempts' budget.
-		attemptDL := attemptDeadline(attemptTimeout, deadline)
-
-		attemptCtx, cancel := context.WithDeadline(ctx, attemptDL)
-		defer cancel()
-
-		return send(attemptCtx, p2pNode, protoID, peerID, msg, o, attemptDL)
+	return withRetries(ctx, o.retries, func() error {
+		return send(ctx, p2pNode, protoID, peerID, msg, o)
 	})
 }
 
-// send is a single Send attempt.
+// send is a single Send attempt, timed out after the send timeout.
 func send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID, msg proto.Message,
-	o sendRecvOpts, deadline time.Time,
+	o sendRecvOpts,
 ) error {
 	t0 := time.Now()
 
@@ -494,6 +456,14 @@ func send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe
 	defer func() {
 		sendDurations.WithLabelValues(PeerName(peerID), protoLabel, o.metricTopic).Observe(time.Since(t0).Seconds())
 	}()
+
+	// The send timeout bounds this attempt: the context covers dialing and protocol
+	// negotiation, the stream deadline covers the message write.
+	deadline := time.Now().Add(o.sendTimeout)
+
+	ctx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
+
 	// Circuit relay connections are transient
 	s, err := p2pNode.NewStream(network.WithAllowLimitedConn(ctx, ""), peerID, o.protocols...)
 	if err != nil {
diff --git a/p2p/sender_test.go b/p2p/sender_test.go
index 6df244796..dd998729b 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -253,17 +253,19 @@ func TestSendRetries(t *testing.T) {
 		require.Equal(t, 3, flaky.Calls())
 	})
 
-	t.Run("retries bounded by send timeout", func(t *testing.T) {
-		flaky := testutil.NewFlakyHost(client, 100)
+	t.Run("each attempt gets the full send timeout", func(t *testing.T) {
+		// blocking dials never succeed, so every attempt runs until its own send
+		// timeout. The send timeout is per attempt, so the total spans all attempts.
+		const perAttempt = 300 * time.Millisecond
+
+		blocking := blockingHost{Host: client}
 		t0 := time.Now()
 
-		// The send timeout is the total budget for all attempts, so the retry
-		// sequence must stop long before all 8 retries (~7s of backoff) elapse.
-		err := p2p.Send(ctx, flaky, protocolID, server.ID(), &pbv1.Duty{Slot: 5},
-			p2p.WithRetries(8), p2p.WithSendTimeout(200*time.Millisecond))
-		require.ErrorContains(t, err, "transient stream failure")
-		require.Less(t, time.Since(t0), 3*time.Second)
-		require.Less(t, flaky.Calls(), 9)
+		err := p2p.Send(ctx, blocking, protocolID, server.ID(), &pbv1.Duty{Slot: 5},
+			p2p.WithRetries(2), p2p.WithSendTimeout(perAttempt))
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+		// 3 attempts each get the full timeout (not a sliced share of it).
+		require.GreaterOrEqual(t, time.Since(t0), 3*perAttempt)
 	})
 
 	t.Run("cancellation surfaces as context error", func(t *testing.T) {
@@ -442,31 +444,6 @@ func TestWithRetriesClampsNegative(t *testing.T) {
 	})
 }
 
-// TestSendReceiveBoundedBySendTimeout ensures the whole SendReceive call, retries included,
-// is bounded by the send timeout: a stalled attempt consumes the budget and is not retried
-// past it (a delivered request to a slow peer is waited out, not re-sent).
-func TestSendReceiveBoundedBySendTimeout(t *testing.T) {
-	ctx := context.Background()
-	server := testutil.CreateHost(t, testutil.AvailableAddr(t))
-	client := testutil.CreateHost(t, testutil.AvailableAddr(t))
-	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
-
-	protocolID := protocol.ID("testprotocol-sendrecv-bounded")
-	p2p.RegisterHandler("test", server, protocolID,
-		func() proto.Message { return new(pbv1.Duty) },
-		func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) {
-			time.Sleep(3 * time.Second) // Always stall past the budget.
-			return new(pbv1.Duty), true, nil
-		})
-
-	t0 := time.Now()
-	err := p2p.SendReceive(ctx, client, server.ID(), new(pbv1.Duty), new(pbv1.Duty), protocolID,
-		p2p.WithSendTimeout(500*time.Millisecond), p2p.WithRetries(5))
-	require.Error(t, err)
-	// Bounded by the 500ms budget, not 6 × 500ms, despite 5 retries.
-	require.Less(t, time.Since(t0), 2*time.Second)
-}
-
 func TestSendReceiveRetries(t *testing.T) {
 	ctx := context.Background()
 	server := testutil.CreateHost(t, testutil.AvailableAddr(t))