diff --git a/dkg/bcast/client.go b/dkg/bcast/client.go
index 25c410faf..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)
+		err := c.sendRecvFunc(ctx, c.p2pNode, pID, sigReq, sigResp, protocolIDSig, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 
 		return sigResp, err
 	})
@@ -133,7 +133,7 @@ func (c *client) Broadcast(ctx context.Context, msgID string, msg proto.Message)
 			continue // Skip self.
 		}
 
-		err := c.sendFunc(ctx, c.p2pNode, protocolIDMsg, pID, bcastMsg, p2p.WithSendTimeout(sendTimeout))
+		err := c.sendFunc(ctx, c.p2pNode, protocolIDMsg, pID, bcastMsg, p2p.WithSendTimeout(sendTimeout), p2p.WithRetries(sendRetries))
 		if err != nil {
 			return errors.Wrap(err, "send message")
 		}
diff --git a/dkg/bcast/helpers.go b/dkg/bcast/helpers.go
index ac6f4a122..cdb5a4af8 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 = 5
 )
 
 // 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..b39016453 100644
--- a/dkg/bcast/impl_test.go
+++ b/dkg/bcast/impl_test.go
@@ -5,11 +5,13 @@ package bcast_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	k1 "github.com/decred/dcrd/dcrec/secp256k1/v4"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -150,6 +152,157 @@ func TestBCast(t *testing.T) {
 	assertResults(t, p0Result, peers[0])
 }
 
+// TestBCastRetriesTransientSendFailures ensures a reliable-broadcast survives transient
+// p2p send failures in each phase instead of aborting the whole ceremony on the first one.
+func TestBCastRetriesTransientSendFailures(t *testing.T) {
+	const (
+		n     = 3
+		msgID = "msgID"
+	)
+
+	// The bcast protocol IDs are unexported; keep in sync with helpers.go.
+	tests := []struct {
+		name          string
+		flakyProtocol protocol.ID
+	}{
+		{name: "signature request phase", flakyProtocol: "/charon/dkg/bcast/2.0.0/sig"},
+		{name: "message send phase", flakyProtocol: "/charon/dkg/bcast/2.0.0/msg"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				ctx      = context.Background()
+				secrets  []*k1.PrivateKey
+				tcpNodes []host.Host
+				peers    []peer.ID
+			)
+
+			for range n {
+				secret, err := k1.GeneratePrivateKey()
+				require.NoError(t, err)
+
+				secrets = append(secrets, secret)
+
+				tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+				tcpNodes = append(tcpNodes, tcpNode)
+
+				peers = append(peers, tcpNode.ID())
+			}
+
+			for i := range n {
+				for j := range n {
+					tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
+				}
+			}
+
+			// Node 0's first send to each peer in this phase fails transiently.
+			tcpNodes[0] = testutil.NewFlakyHost(tcpNodes[0], n-1, test.flakyProtocol)
+
+			received := make(chan proto.Message, n)
+			callback := func(_ context.Context, _ peer.ID, _ string, msg proto.Message) error {
+				received <- msg
+				return nil
+			}
+			checkMessage := func(_ context.Context, _ peer.ID, msgAny *anypb.Any) error {
+				var ts timestamppb.Timestamp
+				if err := msgAny.UnmarshalTo(&ts); err != nil {
+					return errors.Wrap(err, "anypb error")
+				}
+
+				return nil
+			}
+
+			var bcasts []bcast.BroadcastFunc
+
+			for i := range n {
+				bcastFunc := bcast.New(tcpNodes[i], peers, secrets[i], []byte("session hash"))
+				bcastFunc.RegisterMessageIDFuncs(msgID, callback, checkMessage)
+				bcasts = append(bcasts, bcastFunc.Broadcast)
+			}
+
+			msg := timestamppb.Now()
+			err := bcasts[0](ctx, msgID, msg)
+			require.NoError(t, err)
+
+			for range n - 1 {
+				require.True(t, proto.Equal(msg, <-received))
+			}
+		})
+	}
+}
+
+// TestBCastSigRequestAllowsSlowPeer ensures a signature request tolerates a peer that
+// responds slower than the default p2p send timeout (7s), since ceremony peers may
+// lawfully lag by up to the bcast receive timeout (1min) while catching up.
+func TestBCastSigRequestAllowsSlowPeer(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping slow-peer timeout test in short mode")
+	}
+
+	const (
+		n     = 2
+		msgID = "msgID"
+		// slowDelay must exceed the 7s default p2p send timeout to prove the
+		// bcast send timeout is applied to signature requests.
+		slowDelay = 8 * time.Second
+	)
+
+	var (
+		ctx      = context.Background()
+		secrets  []*k1.PrivateKey
+		tcpNodes []host.Host
+		peers    []peer.ID
+		bcasts   []bcast.BroadcastFunc
+	)
+
+	for range n {
+		secret, err := k1.GeneratePrivateKey()
+		require.NoError(t, err)
+
+		secrets = append(secrets, secret)
+
+		tcpNode := testutil.CreateHostWithIdentity(t, testutil.AvailableAddr(t), secret)
+		tcpNodes = append(tcpNodes, tcpNode)
+
+		peers = append(peers, tcpNode.ID())
+	}
+
+	for i := range n {
+		for j := range n {
+			tcpNodes[i].Peerstore().AddAddrs(tcpNodes[j].ID(), tcpNodes[j].Addrs(), peerstore.PermanentAddrTTL)
+		}
+	}
+
+	callback := func(context.Context, peer.ID, string, proto.Message) error {
+		return nil
+	}
+
+	for i := range n {
+		delay := time.Duration(0)
+		if i == 1 {
+			delay = slowDelay // Node 1 is slow to check (and so sign) node 0's message.
+		}
+
+		checkMessage := func(_ context.Context, _ peer.ID, msgAny *anypb.Any) error {
+			time.Sleep(delay)
+
+			var ts timestamppb.Timestamp
+			if err := msgAny.UnmarshalTo(&ts); err != nil {
+				return errors.Wrap(err, "anypb error")
+			}
+
+			return nil
+		}
+
+		bcastFunc := bcast.New(tcpNodes[i], peers, secrets[i], []byte("session hash"))
+		bcastFunc.RegisterMessageIDFuncs(msgID, callback, checkMessage)
+		bcasts = append(bcasts, bcastFunc.Broadcast)
+	}
+
+	require.NoError(t, bcasts[0](ctx, msgID, timestamppb.Now()))
+}
+
 // TestBCastSessionHashMismatch ensures that messages signed in one session
 // cannot be verified in another, binding broadcasts to the cluster session.
 func TestBCastSessionHashMismatch(t *testing.T) {
diff --git a/dkg/exchanger.go b/dkg/exchanger.go
index 6d96e6ad8..3a3b9b765 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,32 @@ const (
 	// Do not add new values greater than sigDepositData.
 )
 
+const (
+	// sendRetries is the number of additional p2p send attempts during DKG ceremonies.
+	// A DKG aborts on the first failed exchange, so transient failures (e.g. a stalled
+	// relay connection) are retried instead of failing the whole ceremony. All DKG
+	// receive paths deduplicate messages, so a retry of an already-delivered message
+	// is harmless.
+	sendRetries = 5
+
+	// sendTimeout is the 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
+)
+
+// newRetryingSendFunc returns a p2p.SendFunc wrapping p2p.Send with the given send
+// timeout and sendRetries retries.
+func newRetryingSendFunc(timeout time.Duration) p2p.SendFunc {
+	return func(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID peer.ID,
+		msg proto.Message, opts ...p2p.SendRecvOption,
+	) error {
+		opts = append(opts, p2p.WithSendTimeout(timeout), p2p.WithRetries(sendRetries))
+
+		return p2p.Send(ctx, p2pNode, protoID, peerID, msg, opts...)
+	}
+}
+
 // sigTypeStore is a shorthand for a map of sigType to map of core.PubKey to slice of core.ParSignedData.
 type sigTypeStore map[sigType]map[core.PubKey][]core.ParSignedData
 
@@ -111,7 +139,7 @@ func newExchanger(p2pNode host.Host, peerIdx int, peers []peer.ID, peerMap map[p
 	ex := &exchanger{
 		// threshold is len(peers) to wait until we get all the partial sigs from all the peers per DV
 		sigdb:    parsigdb.NewMemDB(len(peers), noopDeadliner{}, parsigdb.NewMemDBMetadata(0, time.Now())), // metadata timestamps are used for metrics, irrelevant for DKG
-		sigex:    parsigex.NewParSigEx(p2pNode, p2p.Send, peerIdx, peers, verifyShareIdx, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
+		sigex:    parsigex.NewParSigEx(p2pNode, newRetryingSendFunc(timeout), peerIdx, peers, verifyShareIdx, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
 		sigTypes: st,
 		sigData: dataByPubkey{
 			store: sigTypeStore{},
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..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)
+		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/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
diff --git a/dkg/pedersen/board.go b/dkg/pedersen/board.go
index 0e4ead7e0..641ac732d 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"
@@ -56,6 +57,16 @@ 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 = 5
+
+	// 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
 )
 
 var (
@@ -254,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); 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()))
 		}
 	}
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/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 3006c03fb..0280ca2e9 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.
 }
 
@@ -213,12 +215,26 @@ func WithReceiveTimeout(timeout time.Duration) func(*sendRecvOpts) {
 }
 
 // WithSendTimeout returns an option for SendReceive that sets a timeout for sending messages.
+// 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
 	}
 }
 
+// WithRetries returns an option that retries a failed send up to the given number of
+// 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)
+	}
+}
+
 // 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 +304,32 @@ 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. 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 := range retries + 1 {
+		err = fn()
+		if err == nil || attempt == retries {
+			return err
+		}
+
+		timer := time.NewTimer(expbackoff.Backoff(expbackoff.FastConfig, attempt))
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return errors.Wrap(ctx.Err(), "aborting send retries", z.Err(err))
+		case <-timer.C:
+		}
+	}
+
+	return err // Unreachable: the final iteration (attempt == retries) always returns.
+}
+
 // 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 +337,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,23 +346,44 @@ 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, 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,
+) error {
+	tStart := time.Now()
+
 	protoLabel := string(pID) // Updated to the negotiated protocol once NewStream succeeds.
 
 	defer func() {
 		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 {
-		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")
+	if err := s.SetDeadline(deadline); err != nil {
+		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
 	writeFunc, ok := o.writersByProtocol[s.Protocol()]
@@ -341,7 +402,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 +415,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))
@@ -371,13 +432,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, 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,
+) error {
+	t0 := time.Now()
+
 	protoLabel := string(protoID)
 	if len(o.protocols) > 0 {
 		protoLabel = string(o.protocols[0])
@@ -386,17 +456,25 @@ 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 {
-		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")
+	if err := s.SetDeadline(deadline); err != nil {
+		return errors.Wrap(err, "set deadline", z.Str("peer", PeerName(peerID)))
 	}
 
 	writeFunc, ok := o.writersByProtocol[s.Protocol()]
@@ -405,7 +483,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..dd998729b 100644
--- a/p2p/sender_test.go
+++ b/p2p/sender_test.go
@@ -12,13 +12,17 @@ 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"
 	"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"
@@ -121,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)
 
@@ -138,13 +141,333 @@ 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")
+	}
+}
+
+// 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 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())
+	})
+
+	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()
+
+		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) {
+		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())
+	})
+}
+
+// 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")
+			}
+		})
+	}
+}
+
+// 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")
 	}
 }
 
+// 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())
+	})
+}
+
+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..a1a911f2c 100644
--- a/testutil/random.go
+++ b/testutil/random.go
@@ -4,13 +4,16 @@
 package testutil
 
 import (
+	"context"
 	"crypto/ecdsa"
 	crand "crypto/rand"
 	"fmt"
 	"math"
 	"math/rand"
 	"net"
+	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -34,11 +37,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 +1534,56 @@ 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. 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 := match && 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()