fix(pullsync): cap Topmost at contiguous frontier to prevent mid-offer gap advancement

collectAddrs tracks the highest BinID reachable from start without a gap
(contiguousEnd). When the server's store has internal gaps (e.g. BinIDs
{3,7,11} with start=3), all matching chunks are still included in the
offer for eager client-side storage, but Topmost is capped at contiguousEnd
(3) so the client's sync interval does not advance past missing BinIDs.

The cap only fires when contiguousEnd >= start (at least one historical
chunk extended the frontier). This avoids capping when the offer contains
only live chunks (BinID > historicalCursor), where topmost is already
bounded by the existing historicalCursor cap.

Adds TestSync_MidOfferGapCapsTopmostAtContiguousFrontier to cover this case.

Author: misaakidis

pkg/pullsync/pullsync.go

@@ -461,11 +461,15 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
 		}
 
 		var (
-			chs     []*storer.BinC
-			topmost uint64
-			timer   *time.Timer
-			timerC  <-chan time.Time
+			chs          []*storer.BinC
+			topmost      uint64
+			timer        *time.Timer
+			timerC       <-chan time.Time
+			contiguousEnd uint64 // highest BinID contiguous from start; only meaningful when start > 0
 		)
+		if start > 0 {
+			contiguousEnd = start - 1
+		}
 		chC, unsub, errC := s.store.SubscribeBin(ctx, bin, start)
 		defer func() {
 			unsub()
@@ -495,6 +499,13 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
 					break LOOP
 				}
 
+				// Track the contiguous frontier: the highest BinID reachable
+				// from start without a gap. Used after the loop to cap Topmost
+				// so the client's interval does not advance past missing BinIDs.
+				if c.BinID == contiguousEnd+1 {
+					contiguousEnd = c.BinID
+				}
+
 				chs = append(chs, &storer.BinC{Address: c.Address, BatchID: c.BatchID, StampHash: c.StampHash})
 				if c.BinID > topmost {
 					topmost = c.BinID
@@ -527,6 +538,16 @@ func (s *Syncer) collectAddrs(ctx context.Context, bin uint8, start uint64) ([]*
 			topmost = historicalCursor
 		}
 
+		// Cap topmost at the contiguous frontier. All collected chunks are
+		// included in the offer for eager client-side storage, but Topmost is
+		// bounded to the highest BinID reachable from start without a gap.
+		// The client stores every chunk in one round trip; subsequent round
+		// trips advance the interval bookkeeping via leading-gap empty offers
+		// and empty Want bitvectors — no chunk data is retransmitted.
+		if start > 0 && len(chs) > 0 && contiguousEnd >= start && contiguousEnd < topmost {
+			topmost = contiguousEnd
+		}
+
 		return &collectAddrsResult{chs: chs, topmost: topmost}, nil
 	})
 	if err != nil {

pkg/pullsync/pullsync_test.go

@@ -440,6 +440,51 @@ func TestSync_HistoricalGapReturnsEmptyOfferAtBoundary(t *testing.T) {
 	})
 }
 
+// TestSync_MidOfferGapCapsTopmostAtContiguousFrontier verifies that when the
+// server's offer contains an internal gap (chunks at BinIDs {3,7,11} with
+// start=3), Topmost is capped at the contiguous frontier (3) so the client's
+// interval does not advance past the gap. All chunks are still included in
+// the offer for eager storage — no chunk data is retransmitted on later rounds.
+func TestSync_MidOfferGapCapsTopmostAtContiguousFrontier(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ch1 := testingc.GenerateTestRandomChunk()
+		ch2 := testingc.GenerateTestRandomChunk()
+		ch3 := testingc.GenerateTestRandomChunk()
+
+		makeResult := func(c swarm.Chunk, binID uint64) *storer.BinC {
+			h, err := c.Stamp().Hash()
+			if err != nil {
+				t.Fatal(err)
+			}
+			return &storer.BinC{Address: c.Address(), BatchID: c.Stamp().BatchID(), BinID: binID, StampHash: h}
+		}
+
+		// BinIDs: 3, 7, 11 — gaps at [4,6] and [8,10].
+		results := []*storer.BinC{makeResult(ch1, 3), makeResult(ch2, 7), makeResult(ch3, 11)}
+		const cursor = uint64(11)
+
+		var (
+			ps, _         = newPullSync(t, nil, 10, mock.WithSubscribeResp(results, nil), mock.WithChunks(ch1, ch2, ch3), mock.WithCursors([]uint64{cursor}, 0))
+			recorder      = streamtest.New(streamtest.WithProtocols(ps.Protocol()))
+			psClient, db  = newPullSync(t, recorder, 0)
+		)
+
+		topmost, count, err := psClient.Sync(context.Background(), swarm.ZeroAddress, 0, 3)
+		if err != nil {
+			t.Fatal(err)
+		}
+		// Topmost must be the contiguous frontier (3), not the max BinID (11).
+		if topmost != 3 {
+			t.Fatalf("topmost: got %d, want 3 (contiguous frontier)", topmost)
+		}
+		// All three chunks must be delivered and stored in this single round trip.
+		if count != 3 {
+			t.Fatalf("count: got %d, want 3 (all chunks delivered eagerly)", count)
+		}
+		haveChunks(t, db, ch1, ch2, ch3)
+	})
+}
+
 // TestSync_OverwriteNewerChunkDoesNotBlockInterval verifies that
 // ErrOverwriteNewerChunk does not prevent interval advancement.
 // The chunk is already present in the reserve with a newer stamp, so it is