feat: reserve-expanding sync eligibility for redistribution game (SWIP-44)

pkg/api/api_test.go

@@ -712,7 +712,7 @@ func createRedistributionAgentService(
 		postageContract,
 		stakingContract,
 		mockstorer.NewReserve(),
-		func() bool { return true },
+		func(uint8) bool { return true },
 		time.Millisecond*10,
 		blocksPerRound,
 		blocksPerPhase,

pkg/api/status_test.go

@@ -74,7 +74,6 @@ func TestGetStatus(t *testing.T) {
 			jsonhttptest.WithExpectedJSONResponse(ssr),
 		)
 	})
-
 }
 
 // TestGetStatusPeersIncludesBootnodes is a regression test for
@@ -136,7 +135,7 @@ type statusSnapshotMock struct {
 	committedDepth          uint8
 }
 
-func (m *statusSnapshotMock) SyncRate() float64                  { return m.syncRate }
+func (m *statusSnapshotMock) SyncRateWithinRadius() float64      { return m.syncRate }
 func (m *statusSnapshotMock) ReserveSize() int                   { return m.reserveSize }
 func (m *statusSnapshotMock) StorageRadius() uint8               { return m.storageRadius }
 func (m *statusSnapshotMock) Commitment() (uint64, error)        { return m.commitment, nil }

pkg/node/node.go

@@ -221,7 +221,7 @@ const (
 	minPaymentThreshold           = 2 * refreshRate           // minimal accepted payment threshold of full nodes
 	maxPaymentThreshold           = 24 * refreshRate          // maximal accepted payment threshold of full nodes
 	mainnetNetworkID              = uint64(1)                 //
-	reserveWakeUpDuration         = 15 * time.Minute          // time to wait before waking up reserveWorker
+	reserveWakeUpDuration         = 15 * time.Minute          // time to wait before waking up reserveWorker; must exceed puller.DefaultHistRateWindow
 	reserveMinEvictCount          = 1_000
 	cacheMinEvictCount            = 10_000
 	maxAllowedDoubling            = 1
@@ -1289,7 +1289,7 @@ func NewBee(
 
 			reserveThreshold := reserveCapacity >> 1
 			isFullySynced := func() bool {
-				return pullerService.SyncRate() == 0 && saludService.IsHealthy() && localStore.ReserveSize() >= reserveThreshold
+				return pullerService.IsReserveSynced(localStore.StorageRadius()) && saludService.IsHealthy() && localStore.ReserveSize() >= reserveThreshold
 			}
 
 			syncCheckTicker := time.NewTicker(2 * time.Second)
@@ -1301,7 +1301,7 @@ func NewBee(
 						return
 					case <-syncCheckTicker.C:
 						synced := isFullySynced()
-						logger.Debug("sync status check", "synced", synced, "reserveSize", localStore.ReserveSize(), "syncRate", pullerService.SyncRate())
+						logger.Debug("sync status check", "synced", synced, "reserveSize", localStore.ReserveSize(), "syncRate", pullerService.SyncRateWithinRadius())
 						if synced {
 							fullSyncTime := pullSyncStartTime.Sub(t)
 							logger.Info("full sync done", "duration", fullSyncTime)
@@ -1326,10 +1326,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(
@@ -1340,7 +1340,7 @@ func NewBee(
 				postageStampContractService,
 				stakingContract,
 				localStore,
-				isFullySynced,
+				isReserveSynced,
 				o.BlockTime,
 				storageincentives.DefaultBlocksPerRound,
 				storageincentives.DefaultBlocksPerPhase,

pkg/puller/metrics.go

@@ -15,7 +15,7 @@ type metrics struct {
 	SyncedCounter         *prometheus.CounterVec // number of synced chunks
 	SyncWorkerErrCounter  prometheus.Counter     // count number of errors
 	MaxUintErrCounter     prometheus.Counter     // how many times we got maxuint as topmost
-	PullsyncRate          prometheus.GaugeFunc   // rate of historical syncing
+	PullsyncRate          prometheus.GaugeFunc   // rate of historical syncing within the storage radius
 }
 
 func newMetrics(pullsyncRate func() float64) metrics {
@@ -56,7 +56,7 @@ func newMetrics(pullsyncRate func() float64) metrics {
 			Namespace: m.Namespace,
 			Subsystem: subsystem,
 			Name:      "pullsync_rate",
-			Help:      "Rate of historical syncing in chunks.",
+			Help:      "Rate of historical syncing in chunks within the storage radius.",
 		}, pullsyncRate),
 	}
 }

pkg/puller/mock/puller.go

@@ -6,8 +6,8 @@ package mock
 
 import "context"
 
-type mockSyncer struct{ rate float64 }
+type mockSyncer struct{ synced bool }
 
-func NewMockRateReporter(r float64) *mockSyncer { return &mockSyncer{r} }
-func (m *mockSyncer) SyncRate() float64         { return m.rate }
-func (m *mockSyncer) Start(context.Context)     {}
+func NewMockSyncer(synced bool) *mockSyncer      { return &mockSyncer{synced} }
+func (m *mockSyncer) IsReserveSynced(uint8) bool { return m.synced }
+func (m *mockSyncer) Start(context.Context)      {}

pkg/puller/puller.go

@@ -68,6 +68,9 @@ func countErrors(err error) int {
 }
 
 const (
+	// DefaultHistRateWindow must stay shorter than the reserve worker wakeup
+	// (reserveWakeUpDuration in pkg/node), so a full window of syncing has
+	// passed before the radius is decreased again.
 	DefaultHistRateWindow = time.Minute * 10
 
 	IntervalPrefix = "sync_interval"
@@ -104,7 +107,7 @@ type Puller struct {
 
 	bins uint8 // how many bins do we support
 
-	rate *rate.Rate // rate of historical syncing
+	binRate [swarm.MaxBins]*rate.Rate // rate of historical syncing per bin
 
 	start sync.Once
 
@@ -123,25 +126,27 @@ func New(
 ) *Puller {
 	bins := swarm.MaxBins
 	if o.Bins != 0 {
-		bins = o.Bins
+		bins = min(o.Bins, swarm.MaxBins)
 	}
-	histRate := rate.New(DefaultHistRateWindow)
 	p := &Puller{
 		base:        addr,
 		statestore:  stateStore,
 		topology:    topology,
 		radius:      reserveState,
 		syncer:      pullSync,
-		metrics:     newMetrics(histRate.Rate),
 		logger:      logger.WithName(loggerName).Register(),
 		syncPeers:   make(map[string]*syncPeer),
 		bins:        bins,
 		blockLister: blockLister,
-		rate:        histRate,
 		cancel:      func() { /* Noop, since the context is initialized in the Start(). */ },
 		limiter:     ratelimit.NewLimiter(ratelimit.Every(time.Second/maxChunksPerSecond), maxChunksPerSecond),
 	}
 
+	for i := range p.binRate {
+		p.binRate[i] = rate.New(DefaultHistRateWindow)
+	}
+	p.metrics = newMetrics(p.SyncRateWithinRadius)
+
 	return p
 }
 
@@ -155,8 +160,41 @@ func (p *Puller) Start(ctx context.Context) {
 	})
 }
 
-func (p *Puller) SyncRate() float64 {
-	return p.rate.Rate()
+// SyncRateWithinRadius returns the historical sync rate of the bins within the storage radius.
+func (p *Puller) SyncRateWithinRadius() float64 {
+	var r float64
+	for bin := p.radius.StorageRadius(); bin < p.bins; bin++ {
+		r += p.binRate[bin].Rate()
+	}
+	return r
+}
+
+// IsReserveSynced returns true if no bin at or above depth is historically syncing.
+// A depth beyond the tracked bins is never reported as synced.
+func (p *Puller) IsReserveSynced(depth uint8) bool {
+	if depth >= p.bins {
+		return false
+	}
+	for bin := depth; bin < p.bins; bin++ {
+		if p.binRate[bin].Rate() > 0 {
+			return false
+		}
+	}
+	return true
+}
+
+// localBin returns our bin for chunks in bin of a peer at proximity peerPO.
+// For bin == peerPO our bin is only known to be deeper, so the top bin is used
+// and every depth check counts the chunks.
+func (p *Puller) localBin(peerPO, bin uint8) uint8 {
+	switch {
+	case bin < peerPO:
+		return bin
+	case bin > peerPO:
+		return peerPO
+	default:
+		return p.bins - 1
+	}
 }
 
 func (p *Puller) manage(ctx context.Context) {
@@ -391,7 +429,7 @@ func (p *Puller) syncPeerBin(parentCtx context.Context, peer *syncPeer, bin uint
 
 			if isHistorical {
 				p.metrics.SyncedCounter.WithLabelValues("historical").Add(float64(count))
-				p.rate.Add(count)
+				p.binRate[p.localBin(peer.po, bin)].Add(count)
 			} else {
 				p.metrics.SyncedCounter.WithLabelValues("live").Add(float64(count))
 			}

pkg/puller/puller_test.go

@@ -93,6 +93,70 @@ func TestSyncOutsideDepth(t *testing.T) {
 	waitSyncCalledBins(t, pullsync, addr2, 0)
 }
 
+func TestSyncRates(t *testing.T) {
+	t.Parallel()
+
+	const radius = 2
+
+	// the peer bin is PO(peer, chunk); the rate must be recorded in our bin PO(base, chunk)
+	for _, tc := range []struct {
+		name     string
+		peerPO   uint8
+		bin      uint8
+		syncedAt uint8 // lowest depth reported as synced, 5 = none
+	}{
+		{name: "outer peer", peerPO: 1, bin: 1, syncedAt: 5},             // our bin unknown, >= radius
+		{name: "sister peer same bin", peerPO: 2, bin: 2, syncedAt: 5},   // our bin unknown, > 2
+		{name: "sister peer deeper bin", peerPO: 2, bin: 3, syncedAt: 3}, // our bin 2
+		{name: "neighbor shallower bin", peerPO: 4, bin: 2, syncedAt: 3}, // our bin 2
+		{name: "neighbor deeper bin", peerPO: 3, bin: 4, syncedAt: 4},    // our bin 3
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			addr := swarm.RandAddress(t)
+			p, _, kad, _ := newPuller(t, opts{
+				kad:      []kadMock.Option{kadMock.WithEachPeerRevCalls(kadMock.AddrTuple{Addr: addr, PO: tc.peerPO})},
+				pullSync: []mockps.Option{mockps.WithCursors([]uint64{1000, 1000, 1000, 1000, 1000}, 0), mockps.WithReplies(mockps.SyncReply{Bin: tc.bin, Start: 1, Topmost: 1000, Count: 100, Peer: addr})},
+				bins:     5,
+				rs:       resMock.NewReserve(resMock.WithRadius(radius)),
+			})
+			if !p.IsReserveSynced(radius) {
+				t.Fatal("expected reserve to be synced before any syncing")
+			}
+			kad.Trigger()
+
+			err := spinlock.Wait(time.Second, func() bool {
+				return p.SyncRateWithinRadius() > 0
+			})
+			if err != nil {
+				t.Fatal("expected a non-zero sync rate within radius")
+			}
+			for depth := uint8(radius); depth < 5; depth++ {
+				if got, want := p.IsReserveSynced(depth), depth >= tc.syncedAt; got != want {
+					t.Fatalf("depth %d: got synced %v, want %v", depth, got, want)
+				}
+			}
+			if p.IsReserveSynced(5) {
+				t.Fatal("expected a depth beyond the tracked bins not to be synced")
+			}
+		})
+	}
+
+	t.Run("bins above max", func(t *testing.T) {
+		t.Parallel()
+
+		p, _, _, _ := newPuller(t, opts{bins: swarm.MaxBins + 8, rs: resMock.NewReserve(resMock.WithRadius(2))})
+		if !p.IsReserveSynced(swarm.MaxBins - 1) {
+			t.Fatal("expected the last bin to be tracked")
+		}
+		if p.IsReserveSynced(swarm.MaxBins) {
+			t.Fatal("expected bins to be capped at swarm.MaxBins")
+		}
+		_ = p.SyncRateWithinRadius()
+	})
+}
+
 func TestSyncIntervals(t *testing.T) {
 	t.Parallel()