fix(puller): non-blocking peer disconnect and sync error backoff

When onChange held syncPeersMtx and called disconnectPeer, the inner
peer.stop() cancelled per-bin goroutine contexts and then called
peer.wg.Wait(). Live goroutines blocked in Sync() → ReadMsgWithContext
and only unblocked after pageTimeout (1s) per stream. For N peers
disconnecting during a radius decrease, the outer lock was held for up
to N×1s, stalling all queued topology-change notifications for the same
duration.

When syncer.Sync returned a non-fatal error (connection reset, protocol
error, stream timeout), the goroutine fell through to limiter.WaitN with
count=0 and looped immediately. Any persistent non-fatal error caused a
tight CPU spin until the peer disconnected or context was cancelled.

Split syncPeer.stop() into cancelBins() (cancel all per-bin contexts,
clear the map, no wait) and stop() (cancel + wait, used only in
Close()). disconnectPeer now calls cancelBins(): the peer is removed
from the sync map immediately and its goroutines drain in the
background. Close() already calls p.wg.Wait(), so shutdown correctness
is unchanged.

Add syncRetryBackoff (1s) with a ctx.Done()-escape after any non-fatal
sync error before the next retry. This bounds the retry rate to ≤1/s
per goroutine under persistent errors.

Author: misaakidis

pkg/puller/puller.go

@@ -42,6 +42,8 @@ const (
 	maxChunksPerSecond = 1000 // roughly 4 MB/s
 
 	maxPODelta = 2 // the lowest level of proximity order (of peers) subtracted from the storage radius allowed for chunk syncing.
+
+	syncRetryBackoff = time.Second // minimum wait between retries after a non-fatal sync error
 )
 
 type Options struct {
@@ -189,12 +191,13 @@ func (p *Puller) manage(ctx context.Context) {
 }
 
 // disconnectPeer cancels all existing syncing and removes the peer entry from the syncing map.
+// Goroutines drain in the background; the caller is not blocked waiting for them.
 // Must be called under lock.
 func (p *Puller) disconnectPeer(addr swarm.Address) {
 	p.logger.Debug("disconnecting peer", "peer_address", addr)
 	if peer, ok := p.syncPeers[addr.ByteString()]; ok {
 		peer.mtx.Lock()
-		peer.stop()
+		peer.cancelBins()
 		peer.mtx.Unlock()
 	}
 	delete(p.syncPeers, addr.ByteString())
@@ -349,6 +352,11 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint
 					return
 				}
 				p.logger.Debug("syncWorker interval failed", "error", err, "peer_address", address, "bin", bin, "cursor", cursor, "start", start, "topmost", top)
+				select {
+				case <-time.After(syncRetryBackoff):
+				case <-ctx.Done():
+					return
+				}
 			}
 
 			_ = p.limiter.WaitN(ctx, count)
@@ -548,12 +556,19 @@ func newSyncPeer(addr swarm.Address, bins, po uint8) *syncPeer {
 	}
 }
 
-// called when peer disconnects or on shutdown, cleans up ongoing sync operations
-func (p *syncPeer) stop() {
+// cancelBins cancels all per-bin sync contexts and clears the cancel-func map.
+// Does not wait for goroutines to exit. Must be called under peer.mtx.
+func (p *syncPeer) cancelBins() {
 	for bin, c := range p.binCancelFuncs {
 		c()
 		delete(p.binCancelFuncs, bin)
 	}
+}
+
+// stop cancels all per-bin sync contexts and waits for goroutines to exit.
+// Used on shutdown. Must be called under peer.mtx.
+func (p *syncPeer) stop() {
+	p.cancelBins()
 	p.wg.Wait()
 }
 

pkg/puller/puller_test.go

@@ -507,6 +507,59 @@ func TestContinueSyncing(t *testing.T) {
 	}
 }
 
+// TestSyncErrorBackoff verifies that a non-fatal sync error is followed by a
+// backoff before the next retry, bounding the retry rate to roughly 1/s.
+func TestSyncErrorBackoff(t *testing.T) {
+	t.Parallel()
+
+	addr := swarm.RandAddress(t)
+
+	// Use Topmost=0 so that top < start and the interval is never advanced,
+	// causing the loop to retry with the same start value each time.
+	// Provide two replies so we can observe two successive sync calls.
+	_, _, kad, ps := newPuller(t, opts{
+		kad: []kadMock.Option{
+			kadMock.WithEachPeerRevCalls(kadMock.AddrTuple{Addr: addr, PO: 0}),
+		},
+		pullSync: []mockps.Option{
+			mockps.WithCursors([]uint64{100}, 0),
+			mockps.WithSyncError(errors.New("stream error")),
+			mockps.WithReplies(
+				mockps.SyncReply{Bin: 0, Start: 1, Topmost: 0, Peer: addr},
+				mockps.SyncReply{Bin: 0, Start: 1, Topmost: 0, Peer: addr},
+			),
+		},
+		bins: 1,
+		rs:   resMock.NewReserve(resMock.WithRadius(0)),
+	})
+
+	time.Sleep(100 * time.Millisecond)
+	kad.Trigger()
+
+	// wait for the first call
+	err := spinlock.Wait(2*time.Second, func() bool {
+		return len(ps.SyncCalls(addr)) >= 1
+	})
+	if err != nil {
+		t.Fatal("timed out waiting for first sync call")
+	}
+	t1 := time.Now()
+
+	// wait for the second call — must be separated by at least syncRetryBackoff
+	err = spinlock.Wait(3*time.Second, func() bool {
+		return len(ps.SyncCalls(addr)) >= 2
+	})
+	if err != nil {
+		t.Fatal("timed out waiting for second sync call")
+	}
+	elapsed := time.Since(t1)
+
+	const minBackoff = 800 * time.Millisecond // allow 200ms tolerance below syncRetryBackoff
+	if elapsed < minBackoff {
+		t.Fatalf("retry happened too fast: elapsed %v, want >= %v", elapsed, minBackoff)
+	}
+}
+
 func TestPeerGone(t *testing.T) {
 	t.Parallel()