fix(puller): back off after non-fatal sync errors

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.

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.

Covered by TestSyncErrorBackoff (synctest).

Author: misaakidis

pkg/puller/export_test.go

@@ -6,7 +6,10 @@ package puller
 
 import "github.com/ethersphere/bee/v2/pkg/swarm"
 
-var PeerIntervalKey = peerIntervalKey
+var (
+	PeerIntervalKey  = peerIntervalKey
+	SyncRetryBackoff = syncRetryBackoff
+)
 
 func (p *Puller) IsSyncing(addr swarm.Address) bool {
 	p.syncPeersMtx.Lock()

pkg/puller/puller.go

@@ -75,6 +75,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 {
@@ -384,6 +386,11 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint
 				}
 				errCount := countErrors(err)
 				p.logger.Debug("syncWorker interval failed", "error_count", errCount, "example_error", errors.Unwrap(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)

pkg/puller/puller_test.go

@@ -508,6 +508,57 @@ func TestContinueSyncing(t *testing.T) {
 	}
 }
 
+// TestSyncErrorBackoff verifies that a non-fatal sync error is followed by a
+// backoff before the next retry. Under synctest the worker blocks after one
+// failed call; without the backoff it would spin and consume both replies, so
+// the "exactly one call" check fails without the fix.
+func TestSyncErrorBackoff(t *testing.T) {
+	t.Parallel()
+
+	addr := swarm.RandAddress(t)
+
+	synctest.Test(t, func(t *testing.T) {
+		// Topmost=0 keeps top < start so the interval never advances and the loop
+		// retries with the same start value. Two replies cover the failed retry.
+		ps := mockps.NewPullSync(
+			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},
+			),
+		)
+		kad := kadMock.NewMockKademlia(
+			kadMock.WithEachPeerRevCalls(kadMock.AddrTuple{Addr: addr, PO: 0}),
+		)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		p := puller.New(swarm.RandAddress(t), mock.NewStateStore(), kad, resMock.NewReserve(resMock.WithRadius(0)), ps, nil, log.Noop, puller.Options{Bins: 1})
+		p.Start(ctx)
+
+		kad.Trigger()
+
+		// One failed call, then the worker blocks in the backoff.
+		synctest.Wait()
+		if got := len(ps.SyncCalls(addr)); got != 1 {
+			t.Fatalf("expected worker to back off after 1 failed call, got %d calls", got)
+		}
+
+		// Sleeping out the backoff releases exactly one retry.
+		time.Sleep(puller.SyncRetryBackoff)
+		synctest.Wait()
+		if got := len(ps.SyncCalls(addr)); got != 2 {
+			t.Fatalf("expected exactly one retry after the backoff, got %d calls", got)
+		}
+
+		if err := p.Close(); err != nil {
+			t.Errorf("close puller: %v", err)
+		}
+	})
+}
+
 // TestRadiusDecreaseRecalcPeersLiveness verifies that after a radius decrease a
 // new sync worker starts without the manager blocking while the disconnected
 // peer's in-flight worker drains. Under synctest, a blocking disconnect holds