feat: eligibility for incentives during reserve-expanding pullsync

Author: Alok Nerurkar

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,

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,

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 {

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) {}

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)
 		})
 	}

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()
 

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
 	}