From e86755773f12df10fe5f0b0267c08c0a53a016d4 Mon Sep 17 00:00:00 2001 From: Alok Nerurkar Date: Wed, 26 Aug 2026 14:42:33 +0530 Subject: [PATCH 1/2] feat: eligibility for incentives during reserve-expanding pullsync --- pkg/api/api_test.go | 7 ++-- pkg/node/node.go | 19 +++++----- pkg/puller/export_test.go | 2 +- pkg/puller/mock/puller.go | 48 ++++++++++++++++++++++--- pkg/puller/puller.go | 37 ++++++++++++++++++- pkg/puller/puller_test.go | 55 +++++++++++++++++++++++++++-- pkg/storageincentives/agent.go | 15 ++++---- pkg/storageincentives/agent_test.go | 54 ++++++++++++++++++++++------ 8 files changed, 199 insertions(+), 38 deletions(-) diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index 23782bcbc6b..9ac68bdc4f2 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -22,6 +22,9 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/gorilla/websocket" + "resenje.org/web" + "github.com/ethersphere/bee/v2/pkg/accesscontrol" mockac "github.com/ethersphere/bee/v2/pkg/accesscontrol/mock" accountingmock "github.com/ethersphere/bee/v2/pkg/accounting/mock" @@ -70,8 +73,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/transaction/backendmock" transactionmock "github.com/ethersphere/bee/v2/pkg/transaction/mock" "github.com/ethersphere/bee/v2/pkg/util/testutil" - "github.com/gorilla/websocket" - "resenje.org/web" ) var ( @@ -702,7 +703,7 @@ func createRedistributionAgentService( postageContract, stakingContract, mockstorer.NewReserve(), - func() bool { return true }, + func(uint8) bool { return true }, time.Millisecond*10, blocksPerRound, blocksPerPhase, diff --git a/pkg/node/node.go b/pkg/node/node.go index 9ef7704b9dd..e4ecb990fe2 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -26,6 +26,13 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/hashicorp/go-multierror" + ma "github.com/multiformats/go-multiaddr" + "github.com/prometheus/client_golang/prometheus" + "golang.org/x/crypto/sha3" + "golang.org/x/net/idna" + "golang.org/x/sync/errgroup" + bee "github.com/ethersphere/bee/v2" "github.com/ethersphere/bee/v2/pkg/accesscontrol" "github.com/ethersphere/bee/v2/pkg/accounting" @@ -81,12 +88,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/util/ioutil" "github.com/ethersphere/bee/v2/pkg/util/nbhdutil" "github.com/ethersphere/bee/v2/pkg/util/syncutil" - "github.com/hashicorp/go-multierror" - ma "github.com/multiformats/go-multiaddr" - "github.com/prometheus/client_golang/prometheus" - "golang.org/x/crypto/sha3" - "golang.org/x/net/idna" - "golang.org/x/sync/errgroup" ) // LoggerName is the tree path name of the logger for this package. @@ -1323,10 +1324,10 @@ func NewBee( redistributionContract := redistribution.New(swarmAddress, overlayEthAddress, logger, transactionService, redistributionContractAddress, abiutil.MustParseABI(chainCfg.RedistributionABI), contractGasLimit) - isFullySynced := func() bool { + isReserveSynced := func(depth uint8) bool { reserveThreshold := reserveCapacity * 5 / 10 logger.Debug("Sync status check evaluated", "stabilized", detector.IsStabilized()) - return localStore.ReserveSize() >= reserveThreshold && pullerService.SyncRate() == 0 && detector.IsStabilized() + return localStore.ReserveSize() >= reserveThreshold && pullerService.IsReserveSynced(depth) && detector.IsStabilized() } agent, err = storageincentives.New( @@ -1337,7 +1338,7 @@ func NewBee( postageStampContractService, stakingContract, localStore, - isFullySynced, + isReserveSynced, o.BlockTime, storageincentives.DefaultBlocksPerRound, storageincentives.DefaultBlocksPerPhase, diff --git a/pkg/puller/export_test.go b/pkg/puller/export_test.go index b57ae396b3f..e115862b378 100644 --- a/pkg/puller/export_test.go +++ b/pkg/puller/export_test.go @@ -15,7 +15,7 @@ func (p *Puller) IsSyncing(addr swarm.Address) bool { return ok } -func (p *Puller) IsBinSyncing(addr swarm.Address, bin uint8) bool { +func (p *Puller) IsPeerBinSyncing(addr swarm.Address, bin uint8) bool { p.syncPeersMtx.Lock() defer p.syncPeersMtx.Unlock() if peer, ok := p.syncPeers[addr.ByteString()]; ok { diff --git a/pkg/puller/mock/puller.go b/pkg/puller/mock/puller.go index ff8ef10a9ce..29005a76f76 100644 --- a/pkg/puller/mock/puller.go +++ b/pkg/puller/mock/puller.go @@ -6,8 +6,48 @@ package mock import "context" -type mockSyncer struct{ rate float64 } +type Option func(*mockSyncer) -func NewMockRateReporter(r float64) *mockSyncer { return &mockSyncer{r} } -func (m *mockSyncer) SyncRate() float64 { return m.rate } -func (m *mockSyncer) Start(context.Context) {} +type mockSyncer struct { + rate float64 + isReserveSyncedFunc func(depth uint8) bool + isBinSyncingFunc func(bin uint8) bool +} + +func WithReserveSynced(f func(depth uint8) bool) Option { + return func(m *mockSyncer) { + m.isReserveSyncedFunc = f + } +} + +func WithBinSyncing(f func(bin uint8) bool) Option { + return func(m *mockSyncer) { + m.isBinSyncingFunc = f + } +} + +func NewMockRateReporter(r float64, opts ...Option) *mockSyncer { + m := &mockSyncer{rate: r} + for _, opt := range opts { + opt(m) + } + return m +} + +func (m *mockSyncer) SyncRate() float64 { return m.rate } + +func (m *mockSyncer) IsReserveSynced(depth uint8) bool { + if m.isReserveSyncedFunc != nil { + return m.isReserveSyncedFunc(depth) + } + return true +} + +func (m *mockSyncer) IsBinSyncing(bin uint8) bool { + if m.isBinSyncingFunc != nil { + return m.isBinSyncingFunc(bin) + } + return false +} + +func (m *mockSyncer) Start(context.Context) {} diff --git a/pkg/puller/puller.go b/pkg/puller/puller.go index 7f416e640f2..9da7a3b62a5 100644 --- a/pkg/puller/puller.go +++ b/pkg/puller/puller.go @@ -16,6 +16,8 @@ import ( "sync" "time" + ratelimit "golang.org/x/time/rate" + "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/p2p" "github.com/ethersphere/bee/v2/pkg/puller/intervalstore" @@ -26,7 +28,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer" "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/ethersphere/bee/v2/pkg/topology" - ratelimit "golang.org/x/time/rate" ) // loggerName is the tree path name of the logger for this package. @@ -106,6 +107,9 @@ type Puller struct { rate *rate.Rate // rate of historical syncing + activeHistSyncs [swarm.MaxBins]int + activeHistSyncsMu sync.RWMutex + start sync.Once limiter *ratelimit.Limiter @@ -159,6 +163,28 @@ func (p *Puller) SyncRate() float64 { return p.rate.Rate() } +// IsBinSyncing returns true if any peer is actively running historical sync for the given bin. +func (p *Puller) IsBinSyncing(bin uint8) bool { + if bin >= p.bins { + return false + } + p.activeHistSyncsMu.RLock() + defer p.activeHistSyncsMu.RUnlock() + return p.activeHistSyncs[bin] > 0 +} + +// IsReserveSynced returns true if no bins at or above the given depth are actively syncing. +func (p *Puller) IsReserveSynced(depth uint8) bool { + p.activeHistSyncsMu.RLock() + defer p.activeHistSyncsMu.RUnlock() + for bin := depth; bin < p.bins; bin++ { + if p.activeHistSyncs[bin] > 0 { + return false + } + } + return true +} + func (p *Puller) manage(ctx context.Context) { defer p.wg.Done() @@ -409,9 +435,18 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint } if cursor > 0 { + p.activeHistSyncsMu.Lock() + p.activeHistSyncs[bin]++ + p.activeHistSyncsMu.Unlock() + peer.wg.Add(1) p.wg.Add(1) safe.Go(p.logger, "puller-sync-historical", func() { + defer func() { + p.activeHistSyncsMu.Lock() + p.activeHistSyncs[bin]-- + p.activeHistSyncsMu.Unlock() + }() sync(true, peer.address, cursor) }) } diff --git a/pkg/puller/puller_test.go b/pkg/puller/puller_test.go index 29687138b9c..1bc8582df6a 100644 --- a/pkg/puller/puller_test.go +++ b/pkg/puller/puller_test.go @@ -11,6 +11,8 @@ import ( "testing" "time" + "github.com/google/go-cmp/cmp" + "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/puller" "github.com/ethersphere/bee/v2/pkg/puller/intervalstore" @@ -23,7 +25,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/swarm" kadMock "github.com/ethersphere/bee/v2/pkg/topology/kademlia/mock" "github.com/ethersphere/bee/v2/pkg/util/testutil" - "github.com/google/go-cmp/cmp" ) // test that adding one peer starts syncing @@ -462,10 +463,10 @@ func TestRadiusIncrease(t *testing.T) { rs.SetStorageRadius(2) kad.Trigger() time.Sleep(100 * time.Millisecond) - if !p.IsBinSyncing(addr, 1) { + if !p.IsPeerBinSyncing(addr, 1) { t.Fatalf("peer is not syncing but should") } - if p.IsBinSyncing(addr, 2) { + if p.IsPeerBinSyncing(addr, 2) { t.Fatalf("peer is syncing but shouldn't") } } @@ -642,6 +643,54 @@ type opts struct { syncSleepDur time.Duration } +func TestIsReserveSynced(t *testing.T) { + t.Parallel() + + var ( + addr = swarm.RandAddress(t) + cursors = []uint64{1000, 1000, 1000, 1000} + replies = []mockps.SyncReply{ + {Bin: 1, Start: 1, Topmost: 500, Peer: addr}, // partial sync, historical stays active + } + ) + + p, _, kad, pullsync := newPuller(t, opts{ + kad: []kadMock.Option{ + kadMock.WithEachPeerRevCalls( + kadMock.AddrTuple{Addr: addr, PO: 1}, + ), + }, + pullSync: []mockps.Option{ + mockps.WithCursors(cursors, 0), + mockps.WithReplies(replies...), + }, + bins: 4, + rs: resMock.NewReserve(resMock.WithRadius(2)), + }) + + kad.Trigger() + waitCursorsCalled(t, pullsync, addr) + waitSyncCalledBins(t, pullsync, addr, 1) + + // While bin 1 has an active historical sync and radius is 2: + err := spinlock.Wait(time.Second, func() bool { + return p.IsBinSyncing(1) + }) + if err != nil { + t.Fatal("expected bin 1 to be syncing") + } + + if p.IsBinSyncing(2) { + t.Fatal("expected bin 2 not to be syncing") + } + if !p.IsReserveSynced(2) { + t.Fatal("expected reserve to be synced for depth 2 when only bin 1 is syncing") + } + if p.IsReserveSynced(1) { + t.Fatal("expected reserve NOT to be synced for depth 1 when bin 1 is syncing") + } +} + func newPuller(t *testing.T, ops opts) (*puller.Puller, storage.StateStorer, *kadMock.Mock, *mockps.PullSyncMock) { t.Helper() diff --git a/pkg/storageincentives/agent.go b/pkg/storageincentives/agent.go index 4a1c0a6e994..29b0d2aab9e 100644 --- a/pkg/storageincentives/agent.go +++ b/pkg/storageincentives/agent.go @@ -16,6 +16,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "resenje.org/singleflight" + "github.com/ethersphere/bee/v2/pkg/crypto" "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/postage" @@ -28,7 +30,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer" "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/ethersphere/bee/v2/pkg/transaction" - "resenje.org/singleflight" ) const loggerName = "storageincentives" @@ -64,7 +65,7 @@ type Agent struct { batchExpirer postagecontract.PostageBatchExpirer redistributionStatuser staking.RedistributionStatuser store storer.Reserve - fullSyncedFunc func() bool + reserveSyncedFunc func(depth uint8) bool overlay swarm.Address quit chan struct{} wg sync.WaitGroup @@ -82,7 +83,7 @@ func New(overlay swarm.Address, batchExpirer postagecontract.PostageBatchExpirer, redistributionStatuser staking.RedistributionStatuser, store storer.Reserve, - fullSyncedFunc func() bool, + reserveSyncedFunc func(depth uint8) bool, blockTime time.Duration, blocksPerRound, blocksPerPhase uint64, @@ -101,7 +102,7 @@ func New(overlay swarm.Address, contract: contract, batchExpirer: batchExpirer, store: store, - fullSyncedFunc: fullSyncedFunc, + reserveSyncedFunc: reserveSyncedFunc, blocksPerRound: blocksPerRound, quit: make(chan struct{}), redistributionStatuser: redistributionStatuser, @@ -221,7 +222,7 @@ func (a *Agent) start(blockTime time.Duration, blocksPerRound, blocksPerPhase ui a.logger.Info("entered new phase", "phase", currentPhase.String(), "round", round, "block", block) a.state.SetCurrentEvent(currentPhase, round) - a.state.SetFullySynced(a.fullSyncedFunc()) + a.state.SetFullySynced(a.reserveSyncedFunc(a.store.StorageRadius())) a.state.SetHealthy(a.health.IsHealthy()) safe.Go(a.logger, "storageincentives-purge-stale-round-data", func() { a.state.purgeStaleRoundData() @@ -416,8 +417,8 @@ func (a *Agent) handleSample(ctx context.Context, round uint64) (bool, error) { a.metrics.NeighborhoodSelected.Inc() a.logger.Info("neighbourhood chosen", "round", round) - if !a.state.IsFullySynced() { - a.logger.Info("skipping round because node is not fully synced") + if !a.reserveSyncedFunc(committedDepth) { + a.logger.Info("skipping round because reserve is not synced", "depth", committedDepth, "round", round) return false, nil } diff --git a/pkg/storageincentives/agent_test.go b/pkg/storageincentives/agent_test.go index 6449ede9059..99035063b52 100644 --- a/pkg/storageincentives/agent_test.go +++ b/pkg/storageincentives/agent_test.go @@ -15,6 +15,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/postage" contractMock "github.com/ethersphere/bee/v2/pkg/postage/postagecontract/mock" @@ -35,14 +36,15 @@ func TestAgent(t *testing.T) { bigBalance := big.NewInt(4_000_000_000) tests := []struct { - name string - blocksPerRound uint64 - blocksPerPhase uint64 - incrementBy uint64 - limit uint64 - expectedCalls bool - balance *big.Int - doubling uint8 + name string + blocksPerRound uint64 + blocksPerPhase uint64 + incrementBy uint64 + limit uint64 + expectedCalls bool + balance *big.Int + doubling uint8 + isReserveSynced func(depth uint8) bool }{ { name: "3 blocks per phase, same block number returns twice", @@ -91,6 +93,32 @@ func TestAgent(t *testing.T) { limit: 144, balance: big.NewInt(0), doubling: 1, + }, { + name: "expected calls - sub-depth syncing but reserve synced for committed depth", + blocksPerRound: 12, + blocksPerPhase: 4, + incrementBy: 2, + expectedCalls: true, + limit: 144, + balance: bigBalance, + doubling: 1, + isReserveSynced: func(depth uint8) bool { + // Synced at or above committed depth (8 + 1 = 9) + return depth >= 9 + }, + }, { + name: "no expected calls - reserve not synced at committed depth", + blocksPerRound: 12, + blocksPerPhase: 4, + incrementBy: 2, + expectedCalls: false, + limit: 144, + balance: bigBalance, + doubling: 1, + isReserveSynced: func(depth uint8) bool { + // Not synced at committed depth (9) + return depth > 9 + }, }, } @@ -114,7 +142,7 @@ func TestAgent(t *testing.T) { contract := &mockContract{t: t, expectedRadius: radius + tc.doubling} - service, _ := createService(t, addr, backend, contract, tc.blocksPerRound, tc.blocksPerPhase, radius, tc.doubling) + service, _ := createService(t, addr, backend, contract, tc.blocksPerRound, tc.blocksPerPhase, radius, tc.doubling, tc.isReserveSynced) testutil.CleanupCloser(t, service) <-wait @@ -169,6 +197,7 @@ func createService( blocksPerPhase uint64, radius uint8, doubling uint8, + isReserveSynced ...func(uint8) bool, ) (*storageincentives.Agent, error) { t.Helper() @@ -186,6 +215,11 @@ func createService( resMock.WithCapacityDoubling(int(doubling)), ) + syncedFunc := func(uint8) bool { return true } + if len(isReserveSynced) > 0 && isReserveSynced[0] != nil { + syncedFunc = isReserveSynced[0] + } + return storageincentives.New( addr, common.Address{}, backend, @@ -193,7 +227,7 @@ func createService( postageContract, stakingContract, reserve, - func() bool { return true }, + syncedFunc, time.Millisecond*100, blocksPerRound, blocksPerPhase, From 92c8a7f0ac9573da7e965962084fa09d2eade51f Mon Sep 17 00:00:00 2001 From: Alok Nerurkar Date: Tue, 1 Sep 2026 18:25:04 +0530 Subject: [PATCH 2/2] fix: comments --- pkg/node/node.go | 2 +- pkg/puller/puller.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/node/node.go b/pkg/node/node.go index e4ecb990fe2..a2fbabc0aa9 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -1327,7 +1327,7 @@ func NewBee( isReserveSynced := func(depth uint8) bool { reserveThreshold := reserveCapacity * 5 / 10 logger.Debug("Sync status check evaluated", "stabilized", detector.IsStabilized()) - return localStore.ReserveSize() >= reserveThreshold && pullerService.IsReserveSynced(depth) && detector.IsStabilized() + return localStore.ReserveSizeWithinRadius() >= uint64(reserveThreshold) && pullerService.IsReserveSynced(depth) && detector.IsStabilized() } agent, err = storageincentives.New( diff --git a/pkg/puller/puller.go b/pkg/puller/puller.go index 9da7a3b62a5..4ccf3d4ee7c 100644 --- a/pkg/puller/puller.go +++ b/pkg/puller/puller.go @@ -434,7 +434,7 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint } } - if cursor > 0 { + if cursor > 0 && bin < swarm.MaxBins { p.activeHistSyncsMu.Lock() p.activeHistSyncs[bin]++ p.activeHistSyncsMu.Unlock()