From c1d94c976876a3d6e96f4c245086aa447ef9b1f5 Mon Sep 17 00:00:00 2001 From: kewei Date: Wed, 12 Aug 2026 02:39:52 +0800 Subject: [PATCH 1/2] cl/network: fix blob history backfill boundary --- cl/phase1/network/blob_downloader.go | 27 +++-- .../network/blob_downloader_boundary_test.go | 110 ++++++++++++++++++ 2 files changed, 127 insertions(+), 10 deletions(-) create mode 100644 cl/phase1/network/blob_downloader_boundary_test.go diff --git a/cl/phase1/network/blob_downloader.go b/cl/phase1/network/blob_downloader.go index a50c35d391a..d87dd583a59 100644 --- a/cl/phase1/network/blob_downloader.go +++ b/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 } diff --git a/cl/phase1/network/blob_downloader_boundary_test.go b/cl/phase1/network/blob_downloader_boundary_test.go new file mode 100644 index 00000000000..1ab81ec07c5 --- /dev/null +++ b/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 . + +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) } From f3d0e04792aceb709de0d1df1adfa4c9cac570e5 Mon Sep 17 00:00:00 2001 From: kewei Date: Wed, 12 Aug 2026 16:42:20 +0800 Subject: [PATCH 2/2] cl/network, cl/rpc: bound blob sidecar retries --- cl/phase1/network/blobs.go | 105 +++++++++++++++++++------------- cl/phase1/network/blobs_test.go | 78 ++++++++++++++++++++++++ cl/rpc/rpc.go | 13 ++++ cl/rpc/rpc_test.go | 64 +++++++++++++++++++ 4 files changed, 219 insertions(+), 41 deletions(-) create mode 100644 cl/phase1/network/blobs_test.go diff --git a/cl/phase1/network/blobs.go b/cl/phase1/network/blobs.go index d9b45c70712..d66ddaff317 100644 --- a/cl/phase1/network/blobs.go +++ b/cl/phase1/network/blobs.go @@ -19,13 +19,11 @@ 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" ) @@ -33,6 +31,12 @@ 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 } diff --git a/cl/phase1/network/blobs_test.go b/cl/phase1/network/blobs_test.go new file mode 100644 index 00000000000..b6fdc94d0ca --- /dev/null +++ b/cl/phase1/network/blobs_test.go @@ -0,0 +1,78 @@ +// 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. + +package network + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/cl/cltypes" + "github.com/erigontech/erigon/cl/cltypes/solid" +) + +type blockingBlobPeerClient struct { + calls atomic.Int64 + inFlight atomic.Int64 + maxFlight atomic.Int64 +} + +type failingBlobPeerClient struct{ calls atomic.Int64 } + +func (*failingBlobPeerClient) Peers() (uint64, error) { return 1, nil } + +func (c *failingBlobPeerClient) SendBlobsSidecarByIdentifierReq(context.Context, *solid.ListSSZ[*cltypes.BlobIdentifier]) ([]*cltypes.BlobSidecar, string, error) { + c.calls.Add(1) + return nil, "peer", errors.New("resource unavailable") +} + +func (*blockingBlobPeerClient) Peers() (uint64, error) { return 1, nil } + +func (c *blockingBlobPeerClient) SendBlobsSidecarByIdentifierReq(ctx context.Context, _ *solid.ListSSZ[*cltypes.BlobIdentifier]) ([]*cltypes.BlobSidecar, string, error) { + c.calls.Add(1) + inFlight := c.inFlight.Add(1) + defer c.inFlight.Add(-1) + for { + maxFlight := c.maxFlight.Load() + if inFlight <= maxFlight || c.maxFlight.CompareAndSwap(maxFlight, inFlight) { + break + } + } + <-ctx.Done() + return nil, "peer", ctx.Err() +} + +func TestRequestBlobsFranticallyBoundsConcurrentRequests(t *testing.T) { + client := &blockingBlobPeerClient{} + req := solid.NewStaticListSSZ[*cltypes.BlobIdentifier](0, 40) + req.Append(&cltypes.BlobIdentifier{}) + ctx, cancel := context.WithTimeout(t.Context(), 350*time.Millisecond) + defer cancel() + + _, err := RequestBlobsFrantically(ctx, client, req) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.LessOrEqual(t, client.maxFlight.Load(), int64(2)) + require.LessOrEqual(t, client.calls.Load(), int64(2)) +} + +func TestRequestBlobsFranticallyBacksOffAfterFailures(t *testing.T) { + client := &failingBlobPeerClient{} + req := solid.NewStaticListSSZ[*cltypes.BlobIdentifier](0, 40) + req.Append(&cltypes.BlobIdentifier{}) + ctx, cancel := context.WithTimeout(t.Context(), 750*time.Millisecond) + defer cancel() + + _, err := RequestBlobsFrantically(ctx, client, req) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.LessOrEqual(t, client.calls.Load(), int64(4)) +} diff --git a/cl/rpc/rpc.go b/cl/rpc/rpc.go index 9dac6a11221..74e25f85bb6 100644 --- a/cl/rpc/rpc.go +++ b/cl/rpc/rpc.go @@ -24,6 +24,7 @@ import ( "fmt" "io" "strings" + "sync" "time" "github.com/c2h5oh/datasize" @@ -70,6 +71,9 @@ type BeaconRpcP2P struct { ethClock eth_clock.EthereumClock columnDataPeers *columnDataPeers + + blobSidecarRequestsOnce sync.Once + blobSidecarRequests chan struct{} } // NewBeaconRpcP2P creates a new BeaconRpcP2P struct and returns a pointer to it. @@ -242,6 +246,15 @@ func (b *BeaconRpcP2P) SendExecutionPayloadEnvelopesByRootReq(ctx context.Contex // SendBeaconBlocksByRangeReq retrieves blocks range from beacon chain. func (b *BeaconRpcP2P) SendBlobsSidecarByIdentifierReq(ctx context.Context, req *solid.ListSSZ[*cltypes.BlobIdentifier]) ([]*cltypes.BlobSidecar, string, error) { + b.blobSidecarRequestsOnce.Do(func() { + b.blobSidecarRequests = make(chan struct{}, 2) + }) + select { + case b.blobSidecarRequests <- struct{}{}: + defer func() { <-b.blobSidecarRequests }() + case <-ctx.Done(): + return nil, "", ctx.Err() + } var buffer buffer.Buffer if err := ssz_snappy.EncodeAndWrite(&buffer, req); err != nil { return nil, "", err diff --git a/cl/rpc/rpc_test.go b/cl/rpc/rpc_test.go index 46ab29211ce..149d83dc9ad 100644 --- a/cl/rpc/rpc_test.go +++ b/cl/rpc/rpc_test.go @@ -3,13 +3,16 @@ package rpc import ( "bytes" "context" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/require" "google.golang.org/grpc" "github.com/erigontech/erigon/cl/clparams" "github.com/erigontech/erigon/cl/cltypes" + "github.com/erigontech/erigon/cl/cltypes/solid" "github.com/erigontech/erigon/cl/sentinel/communication/ssz_snappy" "github.com/erigontech/erigon/cl/utils/eth_clock" "github.com/erigontech/erigon/common" @@ -22,6 +25,23 @@ type blockResponseSentinel struct { bannedPeer string } +type blockingBlobSentinel struct { + sentinelproto.SentinelClient + active atomic.Int64 + max atomic.Int64 + enter chan struct{} +} + +func (s *blockingBlobSentinel) SendRequest(ctx context.Context, _ *sentinelproto.RequestData, _ ...grpc.CallOption) (*sentinelproto.ResponseData, error) { + active := s.active.Add(1) + for current := s.max.Load(); active > current && !s.max.CompareAndSwap(current, active); current = s.max.Load() { + } + s.enter <- struct{}{} + <-ctx.Done() + s.active.Add(-1) + return nil, ctx.Err() +} + func (s *blockResponseSentinel) SendRequest(context.Context, *sentinelproto.RequestData, ...grpc.CallOption) (*sentinelproto.ResponseData, error) { return &sentinelproto.ResponseData{ Data: s.response, @@ -61,6 +81,50 @@ func TestMaxRequestPayloadsFallback(t *testing.T) { require.ErrorContains(t, err, "17") } +func TestBlobSidecarByRootRequestsShareConcurrencyLimit(t *testing.T) { + sentinel := &blockingBlobSentinel{enter: make(chan struct{}, 3)} + client := &BeaconRpcP2P{ctx: t.Context(), sentinel: sentinel, beaconConfig: &clparams.MainnetBeaconConfig} + req := solid.NewStaticListSSZ[*cltypes.BlobIdentifier](1, 40) + req.Append(&cltypes.BlobIdentifier{}) + contexts := make([]context.Context, 3) + cancels := make([]context.CancelFunc, 3) + contexts[0], cancels[0] = context.WithCancel(t.Context()) + contexts[1], cancels[1] = context.WithCancel(t.Context()) + contexts[2], cancels[2] = context.WithCancel(t.Context()) + defer cancels[0]() + defer cancels[1]() + defer cancels[2]() + done := make(chan struct{}, 3) + launch := func(index int) { + go func() { + _, _, _ = client.SendBlobsSidecarByIdentifierReq(contexts[index], req) + done <- struct{}{} + }() + } + launch(0) + <-sentinel.enter + launch(1) + <-sentinel.enter + launch(2) + select { + case <-sentinel.enter: + t.Fatal("third blob request crossed the shared concurrency boundary") + case <-time.After(100 * time.Millisecond): + } + cancels[0]() + select { + case <-sentinel.enter: + case <-time.After(time.Second): + t.Fatal("waiting blob request did not acquire a released permit") + } + require.Equal(t, int64(2), sentinel.max.Load()) + cancels[1]() + cancels[2]() + for range 3 { + <-done + } +} + func TestSendBeaconBlocksByRangeReqRejectsForkSchemaSlotMismatch(t *testing.T) { cfg := clparams.MainnetBeaconConfig cfg.InitializeForkSchedule()