cl/network, cl/rpc: fix blob history backfill boundary

cl/phase1/network/blob_downloader.go

@@ -40,7 +40,6 @@ const (
 	blobLogInterval             = 30 * time.Second
 	blobBackfillWarningInterval = 4 * time.Minute
 	blocksBatchSize             = uint64(8)
-	minPeersForBlobDownload     = 16
 	// bounds a fulu block's column recovery; columns past the custody window are
 	// unfetchable and would otherwise block forever.
 	blobColumnBackfillTimeout = 30 * time.Second
@@ -56,16 +55,25 @@ type PeerDasGetter interface {
 	GetPeerDas() das.PeerDas
 }
 
+type blobPeerCounter interface {
+	Peers() (uint64, error)
+}
+
+type blobSnapshotReader interface {
+	FrozenBlobs() uint64
+}
+
 // BlobHistoryDownloader downloads blob history backwards from a head slot
 type BlobHistoryDownloader struct {
 	ctx context.Context
 
 	beaconCfg   *clparams.BeaconChainConfig
 	rpc         *rpc.BeaconRpcP2P
+	peerCounter blobPeerCounter
 	indiciesDB  kv.RoDB
 	blobStorage blob_storage.BlobStorage
 	blockReader freezeblocks.BeaconSnapshotReader
-	sn          *freezeblocks.CaplinSnapshots
+	sn          blobSnapshotReader
 
 	syncedChecker SyncedChecker
 	peerDasGetter PeerDasGetter
@@ -113,6 +121,7 @@ func NewBlobHistoryDownloader(
 		ctx:                       ctx,
 		beaconCfg:                 beaconCfg,
 		rpc:                       rpc,
+		peerCounter:               rpc,
 		indiciesDB:                indiciesDB,
 		blobStorage:               blobStorage,
 		blockReader:               blockReader,
@@ -207,13 +216,13 @@ func (b *BlobHistoryDownloader) downloadOnce(shouldLog bool) error {
 	startSlot := currentSlot
 
 	// Check peer count before proceeding
-	peers, err := b.rpc.Peers()
+	peers, err := b.peerCounter.Peers()
 	if err != nil {
 		b.logger.Warn("[BlobHistoryDownloader] Failed to get peer count", "err", err)
 		return nil
 	}
-	if peers < minPeersForBlobDownload {
-		b.logger.Warn("[BlobHistoryDownloader] Skipping iteration due to low peer count", "peers", peers, "required", minPeersForBlobDownload)
+	if peers == 0 {
+		b.logger.Warn("[BlobHistoryDownloader] Skipping iteration because no peers are available")
 		return nil
 	}
 
@@ -238,16 +247,14 @@ func (b *BlobHistoryDownloader) downloadOnce(shouldLog bool) error {
 		b.logger.Info("[BlobHistoryDownloader] Downloading blobs backwards", "slot", currentSlot)
 	}
 
-	for currentSlot >= targetSlot {
-		if currentSlot <= b.sn.FrozenBlobs() {
-			break
-		}
+	firstUnfrozenSlot := max(targetSlot, b.sn.FrozenBlobs())
+	for currentSlot >= firstUnfrozenSlot {
 		if !b.syncedChecker.Synced() {
 			time.Sleep(5 * time.Second)
 			continue
 		}
 
-		batch, visited, err := b.collectIncompleteBlocks(currentSlot, targetSlot)
+		batch, visited, err := b.collectIncompleteBlocks(currentSlot, firstUnfrozenSlot)
 		if err != nil {
 			return err
 		}

cl/phase1/network/blob_downloader_boundary_test.go

@@ -0,0 +1,110 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package network
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/cl/clparams"
+	"github.com/erigontech/erigon/cl/cltypes"
+	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/db/kv/dbcfg"
+	"github.com/erigontech/erigon/db/kv/memdb"
+	"github.com/erigontech/erigon/db/snapshotsync/freezeblocks"
+)
+
+func TestBlobHistoryDownloaderProcessesFirstUnfrozenSlot(t *testing.T) {
+	const firstUnfrozenSlot = uint64(100)
+	wantErr := errors.New("first unfrozen slot visited")
+	reader := &boundaryBlockReader{err: wantErr}
+	downloader := newBoundaryDownloader(t, firstUnfrozenSlot, firstUnfrozenSlot, firstUnfrozenSlot, 1, reader)
+
+	require.ErrorIs(t, downloader.downloadOnce(false), wantErr)
+	require.Equal(t, []uint64{firstUnfrozenSlot}, reader.slots)
+}
+
+func TestBlobHistoryDownloaderBatchStopsAtFrozenBoundary(t *testing.T) {
+	const firstUnfrozenSlot = uint64(100)
+	reader := &boundaryBlockReader{}
+	downloader := newBoundaryDownloader(t, firstUnfrozenSlot+1, firstUnfrozenSlot, 0, 1, reader)
+
+	require.NoError(t, downloader.downloadOnce(false))
+	require.Equal(t, []uint64{firstUnfrozenSlot + 1, firstUnfrozenSlot}, reader.slots)
+}
+
+func TestBlobHistoryDownloaderRunsWithAvailablePeer(t *testing.T) {
+	const slot = uint64(100)
+	wantErr := errors.New("available peer used")
+	reader := &boundaryBlockReader{err: wantErr}
+	downloader := newBoundaryDownloader(t, slot, 0, slot, 1, reader)
+
+	require.ErrorIs(t, downloader.downloadOnce(false), wantErr)
+}
+
+func TestBlobHistoryDownloaderWaitsWithoutPeers(t *testing.T) {
+	reader := &boundaryBlockReader{}
+	downloader := newBoundaryDownloader(t, 100, 0, 100, 0, reader)
+
+	require.NoError(t, downloader.downloadOnce(false))
+	require.Empty(t, reader.slots)
+}
+
+func newBoundaryDownloader(t *testing.T, headSlot, frozenBlobs, targetSlot, peers uint64, reader freezeblocks.BeaconSnapshotReader) *BlobHistoryDownloader {
+	t.Helper()
+	downloader := &BlobHistoryDownloader{
+		ctx:           t.Context(),
+		beaconCfg:     &clparams.MainnetBeaconConfig,
+		peerCounter:   boundaryPeerCounter(peers),
+		indiciesDB:    memdb.NewTestDB(t, dbcfg.ChainDB),
+		blockReader:   reader,
+		sn:            boundarySnapshot(frozenBlobs),
+		syncedChecker: boundarySyncedChecker(true),
+		targetSlot:    targetSlot,
+		archiveBlobs:  true,
+		logger:        log.New(),
+	}
+	downloader.headSlot.Store(headSlot)
+	return downloader
+}
+
+type boundaryBlockReader struct {
+	freezeblocks.BeaconSnapshotReader
+	slots []uint64
+	err   error
+}
+
+func (r *boundaryBlockReader) ReadBeaconBlockBodyBySlot(_ context.Context, _ kv.Tx, slot uint64) (*cltypes.SignedBeaconBlock, error) {
+	r.slots = append(r.slots, slot)
+	return nil, r.err
+}
+
+type boundaryPeerCounter uint64
+
+func (p boundaryPeerCounter) Peers() (uint64, error) { return uint64(p), nil }
+
+type boundarySnapshot uint64
+
+func (s boundarySnapshot) FrozenBlobs() uint64 { return uint64(s) }
+
+type boundarySyncedChecker bool
+
+func (s boundarySyncedChecker) Synced() bool { return bool(s) }

cl/phase1/network/blobs.go

@@ -19,20 +19,24 @@ package network
 import (
 	"context"
 	"errors"
-	"sync/atomic"
 	"time"
 
 	"github.com/erigontech/erigon/cl/clparams"
 	"github.com/erigontech/erigon/cl/cltypes"
 	"github.com/erigontech/erigon/cl/cltypes/solid"
-	"github.com/erigontech/erigon/cl/rpc"
 	"github.com/erigontech/erigon/common/log/v3"
 )
 
 var ErrTimeout = errors.New("timeout")
 
 var requestBlobBatchExpiration = 15 * time.Second
 
+const (
+	initialBlobRequestBackoff = 100 * time.Millisecond
+	maxBlobRequestBackoff     = 2 * time.Second
+	maxConcurrentBlobRequests = 2
+)
+
 // This is just a bunch of functions to handle blobs
 
 // BlobsIdentifiersFromBlocks returns a list of blob identifiers from a list of blocks, which should then be forwarded to the network.
@@ -71,52 +75,71 @@ type PeerAndSidecars struct {
 	Responses []*cltypes.BlobSidecar
 }
 
-// RequestBlobsFrantically requests blobs from the network frantically.
-func RequestBlobsFrantically(ctx context.Context, r *rpc.BeaconRpcP2P, req *solid.ListSSZ[*cltypes.BlobIdentifier]) (*PeerAndSidecars, error) {
-	var atomicResp atomic.Value
+type BlobPeerClient interface {
+	Peers() (uint64, error)
+	SendBlobsSidecarByIdentifierReq(context.Context, *solid.ListSSZ[*cltypes.BlobIdentifier]) ([]*cltypes.BlobSidecar, string, error)
+}
 
-	atomicResp.Store(&PeerAndSidecars{})
-	timer := time.NewTimer(requestBlobBatchExpiration)
-	defer timer.Stop()
-	reqInterval := time.NewTicker(100 * time.Millisecond)
-	defer reqInterval.Stop()
-Loop:
+// RequestBlobsFrantically requests blobs from the network frantically.
+func RequestBlobsFrantically(ctx context.Context, r BlobPeerClient, req *solid.ListSSZ[*cltypes.BlobIdentifier]) (*PeerAndSidecars, error) {
+	type requestResult struct {
+		responses []*cltypes.BlobSidecar
+		peer      string
+		err       error
+	}
+	attemptCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	expiration := time.NewTimer(requestBlobBatchExpiration)
+	defer expiration.Stop()
+	retry := time.NewTimer(0)
+	defer retry.Stop()
+	retryC := retry.C
+	results := make(chan requestResult, maxConcurrentBlobRequests)
+	inFlight := 0
+	backoff := initialBlobRequestBackoff
+	resetRetry := func(delay time.Duration) {
+		if !retry.Stop() {
+			select {
+			case <-retry.C:
+			default:
+			}
+		}
+		retry.Reset(delay)
+		retryC = retry.C
+	}
+	launch := func() {
+		inFlight++
+		go func() {
+			responses, peer, err := r.SendBlobsSidecarByIdentifierReq(attemptCtx, req)
+			results <- requestResult{responses: responses, peer: peer, err: err}
+		}()
+	}
 	for {
 		select {
-		case <-reqInterval.C:
-			go func() {
-				if len(atomicResp.Load().(*PeerAndSidecars).Responses) > 0 {
-					return
-				}
-				// this is so we do not get stuck on a side-fork
-				responses, pid, err := r.SendBlobsSidecarByIdentifierReq(ctx, req)
-				if err != nil {
-					log.Trace("RequestBlobsFrantically: error", "err", err, "peer", pid)
-					return
-				}
-				if responses == nil {
-					log.Trace("RequestBlobsFrantically: response is nil", "peer", pid)
-					return
-				}
-				if len(atomicResp.Load().(*PeerAndSidecars).Responses) > 0 {
-					return
-				}
-				atomicResp.Store(&PeerAndSidecars{
-					Peer:      pid,
-					Responses: responses,
-				})
-			}()
+		case <-retryC:
+			launch()
+			if inFlight < maxConcurrentBlobRequests {
+				resetRetry(initialBlobRequestBackoff)
+			} else {
+				retryC = nil
+			}
+		case result := <-results:
+			inFlight--
+			if result.err == nil && len(result.responses) > 0 {
+				return &PeerAndSidecars{Peer: result.peer, Responses: result.responses}, nil
+			}
+			if result.err != nil {
+				log.Trace("RequestBlobsFrantically: error", "err", result.err, "peer", result.peer)
+			} else {
+				log.Trace("RequestBlobsFrantically: response is empty", "peer", result.peer)
+			}
+			backoff = min(backoff*2, maxBlobRequestBackoff)
+			resetRetry(backoff)
 		case <-ctx.Done():
 			return nil, ctx.Err()
-		case <-timer.C:
+		case <-expiration.C:
 			log.Trace("RequestBlobsFrantically: timeout")
 			return nil, ErrTimeout
-		default:
-			if len(atomicResp.Load().(*PeerAndSidecars).Responses) > 0 {
-				break Loop
-			}
-			time.Sleep(10 * time.Millisecond)
 		}
 	}
-	return atomicResp.Load().(*PeerAndSidecars), nil
 }