fix(p2p): resolve map race, nil iterator panic, and retry drops in replicator

internal/db/p2p/replicator.go

@@ -260,8 +260,7 @@ func (p *P2P) pushHeadsForDoc(
 			return NewErrMarshalBlock(err, docID, head.cid.String())
 		}
 
-		ctx, cancel := context.WithTimeout(ctx, networkRequestTimeout)
-		defer cancel()
+		reqCtx, reqCancel := context.WithTimeout(ctx, networkRequestTimeout)
 		pushLogReq := protocol.PushLogRequest{
 			DocID:        docID,
 			CID:          head.cid.Bytes(),
@@ -270,10 +269,12 @@ func (p *P2P) pushHeadsForDoc(
 			Block:        rawblock,
 		}
 
-		if _, err := p.replicatorProtocol.SendRequest(ctx, pushLogReq, peerID); err != nil {
+		_, sendErr := p.replicatorProtocol.SendRequest(reqCtx, pushLogReq, peerID)
+		reqCancel()
+		if sendErr != nil {
 			log.ErrorE(
 				"Failed to push doc heads. Handling replicator failure",
-				err,
+				sendErr,
 				corelog.Any("DocID", docID),
 			)
 			err := p.handleReplicatorFailure(ctx, peerID, docID)
@@ -384,6 +385,13 @@ func (p *P2P) ListReplicators(ctx context.Context) ([]client.Replicator, error)
 func (p *P2P) pushLogToReplicators(lg event.Update) {
 	p.repMu.Lock()
 	reps, exists := p.replicators[lg.CollectionID]
+	var peerIDs []string
+	if exists {
+		peerIDs = make([]string, 0, len(reps))
+		for peerID := range reps {
+			peerIDs = append(peerIDs, peerID)
+		}
+	}
 	p.repMu.Unlock()
 
 	for _, handler := range p.pushHandlers {
@@ -394,34 +402,32 @@ func (p *P2P) pushLogToReplicators(lg event.Update) {
 		}
 	}
 
-	if exists {
-		for peerID := range reps {
-			go func() {
-				ctx, cancel := context.WithTimeout(p.ctx, networkRequestTimeout)
-				defer cancel()
-				pushLogReq := protocol.PushLogRequest{
-					DocID:        lg.DocID,
-					CID:          lg.Cid.Bytes(),
-					CollectionID: lg.CollectionID,
-					Creator:      p.host.ID(),
-					Block:        lg.Block,
-				}
-				if _, err := p.replicatorProtocol.SendRequest(ctx, pushLogReq, peerID); err != nil {
-					log.ErrorE(
-						"Failed pushing log",
-						err,
-						corelog.String("DocID", lg.DocID),
-						corelog.Any("CID", lg.Cid),
-						corelog.Any("PeerID", peerID))
-					if !lg.IsRetry {
-						err = p.handleReplicatorFailure(ctx, peerID, lg.DocID)
-						if err != nil {
-							log.ErrorE("Failed to handle replicator failure.", err)
-						}
+	for _, peerID := range peerIDs {
+		go func() {
+			ctx, cancel := context.WithTimeout(p.ctx, networkRequestTimeout)
+			defer cancel()
+			pushLogReq := protocol.PushLogRequest{
+				DocID:        lg.DocID,
+				CID:          lg.Cid.Bytes(),
+				CollectionID: lg.CollectionID,
+				Creator:      p.host.ID(),
+				Block:        lg.Block,
+			}
+			if _, err := p.replicatorProtocol.SendRequest(ctx, pushLogReq, peerID); err != nil {
+				log.ErrorE(
+					"Failed pushing log",
+					err,
+					corelog.String("DocID", lg.DocID),
+					corelog.Any("CID", lg.Cid),
+					corelog.Any("PeerID", peerID))
+				if !lg.IsRetry {
+					err = p.handleReplicatorFailure(p.ctx, peerID, lg.DocID)
+					if err != nil {
+						log.ErrorE("Failed to handle replicator failure.", err)
 					}
 				}
-			}()
-		}
+			}
+		}()
 	}
 }
 
@@ -483,6 +489,9 @@ func (p *P2P) handleReplicatorFailure(ctx context.Context, peerID, docID string)
 }
 
 func (p *P2P) handleCompletedReplicatorRetry(ctx context.Context, peerID string, success bool) error {
+	p.handleRetryMutex.Lock()
+	defer p.handleRetryMutex.Unlock()
+
 	// Check if context is cancelled before attempting database operations.
 	// This prevents attempts to write to a closed database during shutdown.
 	if ctx.Err() != nil {
@@ -607,6 +616,7 @@ func (p *P2P) retryReplicators(ctx context.Context) {
 			return
 		}
 		log.ErrorContextE(ctx, "Failed iterate replicator retry ID keys", err)
+		return
 	}
 	defer closeQueryResults(iter)
 	now := time.Now()
@@ -878,16 +888,17 @@ func (p *P2P) retryDoc(ctx context.Context, peerID string, docID string) error {
 			return NewErrMarshalBlock(err, docID, head.cid.String())
 		}
 
-		ctx, cancel := context.WithTimeout(ctx, networkRequestTimeout)
-		defer cancel()
+		reqCtx, reqCancel := context.WithTimeout(ctx, networkRequestTimeout)
 		pushLogReq := protocol.PushLogRequest{
 			DocID:        docID,
 			CID:          head.cid.Bytes(),
 			CollectionID: head.block.Delta.GetCollectionVersionID(),
 			Creator:      p.host.ID(),
 			Block:        rawblock,
 		}
-		if _, err := p.replicatorProtocol.SendRequest(ctx, pushLogReq, peerID); err != nil {
+		_, err = p.replicatorProtocol.SendRequest(reqCtx, pushLogReq, peerID)
+		reqCancel()
+		if err != nil {
 			return NewErrSendReplicatorRequest(err, peerID, docID)
 		}
 	}

internal/db/p2p/replicator_test.go

@@ -0,0 +1,97 @@
+// Copyright 2026 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package p2p
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/sourcenetwork/defradb/event"
+	"github.com/sourcenetwork/defradb/internal/db/p2p/protocol"
+)
+
+type dummyHost struct {
+	SimpleMockHost
+}
+
+func (d *dummyHost) Connect(ctx context.Context, addresses []string) error {
+	return nil
+}
+
+type dummyPushProtocol struct{}
+
+func (d *dummyPushProtocol) SendRequest(
+	ctx context.Context,
+	req protocol.PushLogRequest,
+	peer string,
+) (protocol.PushLogReply, error) {
+	return protocol.PushLogReply{}, nil
+}
+
+func TestPushLogToReplicators_ConcurrentMapAccess(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	p := &P2P{
+		ctx:                ctx,
+		host:               &dummyHost{},
+		replicators:        make(map[string]map[string][]string),
+		replicatorProtocol: &dummyPushProtocol{},
+	}
+
+	colID := "col1"
+	p.replicators[colID] = map[string][]string{
+		"peer1": {"addr1"},
+		"peer2": {"addr2"},
+	}
+
+	var wg sync.WaitGroup
+	// Concurrently read and push logs
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			p.pushLogToReplicators(event.Update{
+				CollectionID: colID,
+				DocID:        fmt.Sprintf("doc-%d", idx),
+			})
+		}(i)
+	}
+
+	// Concurrently mutate replicators
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			p.updateReplicators(
+				ctx,
+				fmt.Sprintf("peer-%d", idx),
+				[]string{fmt.Sprintf("addr-%d", idx)},
+				map[string]struct{}{colID: {}},
+			)
+		}(i)
+	}
+
+	wg.Wait()
+}
+
+func TestHandleCompletedReplicatorRetry_ContextCanceled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	p := &P2P{}
+	err := p.handleCompletedReplicatorRetry(ctx, "peer1", true)
+	assert.ErrorIs(t, err, context.Canceled)
+}