fix(puller): non-blocking peer disconnect

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 x 1s, stalling all queued topology-change notifications for the
same duration.

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.

Covered by TestRadiusDecreaseLiveness (synctest).

Author: misaakidis

pkg/puller/puller.go

@@ -222,12 +222,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())
@@ -582,12 +583,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

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/log"
@@ -507,6 +508,73 @@ func TestContinueSyncing(t *testing.T) {
 	}
 }
 
+// TestRadiusDecreaseLiveness 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 up the
+// restart by cancelDelay; a non-blocking one restarts immediately.
+func TestRadiusDecreaseLiveness(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		const (
+			bins          = 4
+			initialRadius = 2 // peer within depth: only bins >= 2 sync initially
+			newRadius     = 0 // radius decrease: all bins must be resynced
+			// cancelDelay models a slow drain of the in-flight Sync after its
+			// context is cancelled; > 0 and < Close's 10s drain timeout.
+			cancelDelay = time.Second
+		)
+
+		base := swarm.RandAddress(t)
+		// peer PO matches initialRadius so it participates both before and after.
+		peerAddr := swarm.RandAddressAt(t, base, initialRadius)
+
+		rs := resMock.NewReserve(resMock.WithRadius(initialRadius))
+		ps := mockps.NewPullSync(
+			mockps.WithCursors(make([]uint64, bins), 0),
+			mockps.WithSyncCancelDelay(cancelDelay),
+		)
+		kad := kadMock.NewMockKademlia(
+			kadMock.WithEachPeerRevCalls(kadMock.AddrTuple{Addr: peerAddr, PO: initialRadius}),
+		)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		p := puller.New(base, mock.NewStateStore(), kad, rs, ps, nil, log.Noop, puller.Options{Bins: bins})
+		p.Start(ctx)
+
+		// Wait for syncing to start (peer is within depth at radius=2).
+		kad.Trigger()
+		synctest.Wait()
+		if got := ps.TotalSyncCalls(); got < 1 {
+			t.Fatalf("expected syncing to start, got %d sync calls", got)
+		}
+		snapshot := ps.TotalSyncCalls()
+
+		// Radius decrease tears down the peer's workers and restarts them.
+		rs.SetStorageRadius(newRadius)
+		start := time.Now()
+		kad.Trigger()
+		synctest.Wait()
+
+		// A new worker must have started without the manager blocking on the
+		// in-flight drain. Use Errorf not Fatalf so we still reach Close: on the
+		// bug the manager is blocked waiting for the drain and Close's cancel
+		// is what lets the bubble exit.
+		if ps.TotalSyncCalls() <= snapshot {
+			t.Errorf("no new sync worker started after radius decrease (still %d); "+
+				"manager blocked draining the in-flight goroutine during disconnect", ps.TotalSyncCalls())
+		}
+		if elapsed := time.Since(start); elapsed >= cancelDelay {
+			t.Errorf("manager blocked on in-flight drain: elapsed %v, want < %v", elapsed, cancelDelay)
+		}
+
+		// Close inside the bubble so the sleeping goroutines drain.
+		if err := p.Close(); err != nil {
+			t.Errorf("close puller: %v", err)
+		}
+	})
+}
+
 func TestPeerGone(t *testing.T) {
 	t.Parallel()
 

pkg/pullsync/mock/pullsync.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/pullsync"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
@@ -36,6 +37,15 @@ func WithReplies(replies ...SyncReply) Option {
 	})
 }
 
+// WithSyncCancelDelay adds an artificial delay after context cancellation in
+// the blocking Sync path (when no reply is configured), simulating slow TCP
+// teardown latency.
+func WithSyncCancelDelay(d time.Duration) Option {
+	return optionFunc(func(p *PullSyncMock) {
+		p.cancelDelay = d
+	})
+}
+
 func toID(a swarm.Address, bin uint8, start uint64) string {
 	return fmt.Sprintf("%s-%d-%d", a, bin, start)
 }
@@ -51,11 +61,13 @@ type SyncReply struct {
 type PullSyncMock struct {
 	mtx             sync.Mutex
 	syncCalls       []SyncReply
+	totalSyncCalls  int // every Sync() entry, including blocking ones
 	syncErr         error
 	cursors         []uint64
 	epoch           uint64
 	getCursorsPeers []swarm.Address
 	replies         map[string][]SyncReply
+	cancelDelay     time.Duration // extra delay after ctx cancellation in blocking path
 
 	quit chan struct{}
 }
@@ -74,6 +86,8 @@ func NewPullSync(opts ...Option) *PullSyncMock {
 func (p *PullSyncMock) Sync(ctx context.Context, peer swarm.Address, bin uint8, start uint64) (topmost uint64, count int, err error) {
 	p.mtx.Lock()
 
+	p.totalSyncCalls++
+
 	id := toID(peer, bin, start)
 	replies := p.replies[id]
 
@@ -84,11 +98,24 @@ func (p *PullSyncMock) Sync(ctx context.Context, peer swarm.Address, bin uint8,
 		p.mtx.Unlock()
 		return reply.Topmost, reply.Count, p.syncErr
 	}
+
+	cancelDelay := p.cancelDelay
 	p.mtx.Unlock()
+
 	<-ctx.Done()
+	if cancelDelay > 0 {
+		time.Sleep(cancelDelay)
+	}
 	return 0, 0, ctx.Err()
 }
 
+// TotalSyncCalls returns the total number of Sync() invocations, including blocking ones.
+func (p *PullSyncMock) TotalSyncCalls() int {
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	return p.totalSyncCalls
+}
+
 func (p *PullSyncMock) GetCursors(_ context.Context, peer swarm.Address) ([]uint64, uint64, error) {
 	p.mtx.Lock()
 	defer p.mtx.Unlock()