diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index 02de8ce6d68..9d24c5688a9 100644 --- a/pkg/api/api_test.go +++ b/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, diff --git a/pkg/api/status_test.go b/pkg/api/status_test.go index 5844e86547e..7590b32a653 100644 --- a/pkg/api/status_test.go +++ b/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 } diff --git a/pkg/node/node.go b/pkg/node/node.go index b9a93612e3e..b44edeebab9 100644 --- a/pkg/node/node.go +++ b/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, diff --git a/pkg/puller/metrics.go b/pkg/puller/metrics.go index 952f8f6967e..768face8623 100644 --- a/pkg/puller/metrics.go +++ b/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), } } diff --git a/pkg/puller/mock/puller.go b/pkg/puller/mock/puller.go index ff8ef10a9ce..dfeee18a29c 100644 --- a/pkg/puller/mock/puller.go +++ b/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) {} diff --git a/pkg/puller/puller.go b/pkg/puller/puller.go index d437a131e3f..41db01d9a8b 100644 --- a/pkg/puller/puller.go +++ b/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)) } diff --git a/pkg/puller/puller_test.go b/pkg/puller/puller_test.go index 29687138b9c..2e4d01c8b57 100644 --- a/pkg/puller/puller_test.go +++ b/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() diff --git a/pkg/rate/rate.go b/pkg/rate/rate.go index 0af41fb031f..d4329513e5e 100644 --- a/pkg/rate/rate.go +++ b/pkg/rate/rate.go @@ -2,69 +2,99 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -package rate - // Package rate is a thread-safe rate tracker with per second resolution. -// Under the hood, it uses a moving window to interpolate a rate. +// Under the hood, it sums counts over a sliding window split into buckets. +// A count stays in the rate for at least one window and at most one window +// (rounded up to whole buckets) plus one bucket, so the rate reaches zero no +// earlier than one window after the last Add. +package rate import ( "sync" "time" ) -const milliInSeconds = 1000 +const ( + milliInSeconds = 1000 + bucketsPerWindow = 30 +) + +// bucket holds the count added during one bucketSize slice of time. id is the +// time in milliseconds divided by bucketSize, so it identifies which slice the +// count belongs to. +type bucket struct { + id int64 + count int +} type Rate struct { mtx sync.Mutex - windows map[int64]int - windowSize int64 // window size in milliseconds + buckets []bucket // ring buffer indexed by id modulo its length + windowSize int64 // window size in milliseconds + bucketSize int64 // bucket size in milliseconds now func() time.Time } // New returns a new rate tracker with a defined window size that must be greater than one millisecond. -func New(windowsSize time.Duration) *Rate { +func New(windowSize time.Duration) *Rate { + // a window shorter than 1ms would make the rate divide by zero + ws := max(windowSize.Milliseconds(), 1) + // split the window into bucketsPerWindow buckets; windows shorter than + // bucketsPerWindow milliseconds get 1ms buckets + bs := max(ws/bucketsPerWindow, 1) return &Rate{ - windows: make(map[int64]int), - windowSize: windowsSize.Milliseconds(), + // one slot per bucket in the window, rounded up so the ring covers the + // whole window, plus the current, partly elapsed one + // e.g. 1s window: 33ms buckets, ceil(1000/33)=31 slots + 1 = 32 + buckets: make([]bucket, (ws+bs-1)/bs+1), + windowSize: ws, + bucketSize: bs, now: func() time.Time { return time.Now() }, } } -// add uses the current time and rounds it down to a window -// and increments the window's value. +// Add increments the bucket for the current time by count. +// Counts that are not positive are ignored. func (r *Rate) Add(count int) { + if count <= 0 { + return + } + r.mtx.Lock() defer r.mtx.Unlock() - defer r.cleanup() - window := r.now().UnixMilli() / r.windowSize - r.windows[window] += count + id := r.now().UnixMilli() / r.bucketSize + b := &r.buckets[r.index(id)] + // the slot still holds a bucket that has left the window: start over + if b.id != id { + *b = bucket{id: id} + } + b.count += count } -// rate uses the current window and previous windows' counter to compute a moving-window rate in seconds. -// the rate is computed by first calculating how far along the current time is in the current window -// as a ratio between 0 and 1.0. Then, the sum of currentWindowCount + (1 - ratio) * previousWindowCounter -// is returned as the interpolated counter between the two windows. +// Rate returns the per second rate over the most recent window. func (r *Rate) Rate() float64 { r.mtx.Lock() defer r.mtx.Unlock() - defer r.cleanup() - now := r.now().UnixMilli() - window := now / r.windowSize + current := r.now().UnixMilli() / r.bucketSize + // the ring holds exactly the buckets from oldest to current + oldest := current - int64(len(r.buckets)) + 1 - interpolate := 1 - float64(now-(window*r.windowSize))/float64(r.windowSize) + var sum int + for _, b := range r.buckets { + // stale slots are skipped and overwritten by the next Add that lands + // on them; ids ahead of current are left over from a backward clock jump + if b.id >= oldest && b.id <= current { + sum += b.count + } + } - return milliInSeconds * ((float64(r.windows[window])) + ((interpolate) * float64(r.windows[window-1]))) / float64(r.windowSize) + return milliInSeconds * float64(sum) / float64(r.windowSize) } -// cleanup removes windows older than the most recent two windows. -// Must be called under lock. -func (r *Rate) cleanup() { - window := r.now().UnixMilli() / r.windowSize - for k := range r.windows { - if k <= window-2 { - delete(r.windows, k) - } - } +// index returns the ring slot for a bucket id. +func (r *Rate) index(id int64) int { + n := int64(len(r.buckets)) + return int(((id % n) + n) % n) } diff --git a/pkg/rate/rate_test.go b/pkg/rate/rate_test.go index c9cbbf91782..45eac0421eb 100644 --- a/pkg/rate/rate_test.go +++ b/pkg/rate/rate_test.go @@ -27,11 +27,14 @@ func TestRateFirstBucket(t *testing.T) { } } -// TestIgnoreOldBuckets tests that the buckets older than the most recent two buckets are ignored in rate calculation. +// TestIgnoreOldBuckets tests that counts older than one window (rounded up to +// whole buckets) plus one bucket are ignored in rate calculation. func TestIgnoreOldBuckets(t *testing.T) { t.Parallel() windowSize := 1000 * time.Millisecond + bucket := (windowSize / 30).Truncate(time.Millisecond) + covered := coveredWindow(windowSize, bucket) r := 100 rate := rate.New(windowSize) @@ -39,8 +42,7 @@ func TestIgnoreOldBuckets(t *testing.T) { rate.SetTimeFunc(func() time.Time { return setTime(windowSize) }) rate.Add(10) - // windowSize * 3 ensures that the previous bucket is ignored - rate.SetTimeFunc(func() time.Time { return setTime(windowSize * 3) }) + rate.SetTimeFunc(func() time.Time { return setTime(windowSize + covered + bucket) }) rate.Add(r) got := rate.Rate() @@ -49,37 +51,104 @@ func TestIgnoreOldBuckets(t *testing.T) { } } +// TestRate tests that all counts within the window contribute at full weight, +// non-positive counts are ignored, and buckets left ahead of a backward clock +// jump do not count. func TestRate(t *testing.T) { t.Parallel() - // windowSizeMs := 1000 - windowSize := 1000 * time.Millisecond - const r = 100 + windowSize := 3000 * time.Millisecond + rate := rate.New(windowSize) + + for _, at := range []time.Duration{0, 1000 * time.Millisecond, 2500 * time.Millisecond} { + rate.SetTimeFunc(func() time.Time { return setTime(windowSize + at) }) + rate.Add(100) + rate.Add(0) + rate.Add(-50) + } + + if got, want := rate.Rate(), 100.0; got != want { + t.Fatalf("got %v, want %v", got, want) + } + + rate.SetTimeFunc(func() time.Time { return setTime(windowSize - time.Second) }) + if got, want := rate.Rate(), 0.0; got != want { + t.Fatalf("after a backward clock jump: got %v, want %v", got, want) + } +} + +// TestRateDecaysLinearly tests that the rate drops by an equal step as each +// bucket leaves the window, starting one bucket after the window has passed. +func TestRateDecaysLinearly(t *testing.T) { + t.Parallel() - // tc represents the different ratios that will applied to the previous bucket in the Rate calculation - // eg: the ratio is x, so (1 - 1/x) will be applied to the previous bucket - for _, tc := range []struct { - ratio int - rate float64 - }{ - {ratio: 2, rate: 150}, - {ratio: 4, rate: 175}, - {ratio: 5, rate: 180}, - {ratio: 10, rate: 190}, - {ratio: 100, rate: 199}, - } { + const buckets = 30 + windowSize := 30 * time.Second + bucket := windowSize / buckets - rate := rate.New(windowSize) + rate := rate.New(windowSize) - rate.SetTimeFunc(func() time.Time { return setTime(windowSize) }) - rate.Add(r) + start := windowSize + for i := range buckets { + rate.SetTimeFunc(func() time.Time { return setTime(start + time.Duration(i)*bucket) }) + rate.Add(10) + } + last := start + (buckets-1)*bucket - rate.SetTimeFunc(func() time.Time { return setTime(windowSize*2 + windowSize/time.Duration(tc.ratio)) }) - rate.Add(r) + full := rate.Rate() + if want := 10.0; full != want { + t.Fatalf("full window: got %v, want %v", full, want) + } + step := full / buckets + for i := 1; i <= buckets; i++ { + rate.SetTimeFunc(func() time.Time { return setTime(last + time.Duration(i+1)*bucket) }) got := rate.Rate() - if got != tc.rate { - t.Fatalf("ratio %v, got %v, want %v", tc.ratio, got, tc.rate) + want := full - float64(i)*step + if diff := got - want; diff > 1e-9 || diff < -1e-9 { + t.Fatalf("%d buckets after the last add: got %v, want %v", i, got, want) + } + } +} + +// TestRateReachesZeroWithinWindow tests that a count stays in the rate for a +// full window and is gone one window (rounded up to whole buckets) plus one +// bucket after the last Add, wherever in a bucket that Add landed. The 1s +// window does not divide into whole buckets. +func TestRateReachesZeroWithinWindow(t *testing.T) { + t.Parallel() + + for _, windowSize := range []time.Duration{5 * time.Minute, time.Second} { + bucket := max(windowSize/30, time.Millisecond).Truncate(time.Millisecond) + covered := coveredWindow(windowSize, bucket) + + for _, offset := range []time.Duration{ + 0, + time.Millisecond, + windowSize / 10, + windowSize / 2, + windowSize - time.Millisecond, + // the last millisecond of a bucket, where a count lives the shortest + bucket - windowSize%bucket - time.Millisecond, + } { + rate := rate.New(windowSize) + addAt := windowSize + offset + + rate.SetTimeFunc(func() time.Time { return setTime(addAt) }) + rate.Add(1000) + if rate.Rate() == 0 { + t.Fatalf("window %v offset %v: rate is zero straight after an add", windowSize, offset) + } + + rate.SetTimeFunc(func() time.Time { return setTime(addAt + windowSize - time.Millisecond) }) + if rate.Rate() == 0 { + t.Fatalf("window %v offset %v: count expired before a full window passed", windowSize, offset) + } + + rate.SetTimeFunc(func() time.Time { return setTime(addAt + covered + bucket) }) + if got := rate.Rate(); got != 0 { + t.Fatalf("window %v offset %v: one covered window plus one bucket after the last add, got %v, want 0", windowSize, offset, got) + } } } } @@ -87,3 +156,9 @@ func TestRate(t *testing.T) { func setTime(ms time.Duration) time.Time { return time.UnixMilli(ms.Milliseconds()) } + +// coveredWindow returns the window rounded up to whole buckets. +func coveredWindow(window, bucket time.Duration) time.Duration { + ws, bs := window.Milliseconds(), bucket.Milliseconds() + return time.Duration((ws+bs-1)/bs*bs) * time.Millisecond +} diff --git a/pkg/status/status.go b/pkg/status/status.go index c37d7f34856..cd5c8f79214 100644 --- a/pkg/status/status.go +++ b/pkg/status/status.go @@ -34,7 +34,7 @@ type Snapshot pb.Snapshot // SyncReporter defines the interface to report syncing rate. type SyncReporter interface { - SyncRate() float64 + SyncRateWithinRadius() float64 } // Reserve defines the reserve storage related information required. @@ -109,7 +109,7 @@ func (s *Service) LocalSnapshot() (*Snapshot, error) { } if s.sync != nil { - syncRate = s.sync.SyncRate() + syncRate = s.sync.SyncRateWithinRadius() } commitment, err := s.chainState.Commitment() diff --git a/pkg/status/status_test.go b/pkg/status/status_test.go index ae4b336e9f8..c6c7b9afc84 100644 --- a/pkg/status/status_test.go +++ b/pkg/status/status_test.go @@ -314,10 +314,10 @@ type statusSnapshotMock struct { *pb.Snapshot } -func (m *statusSnapshotMock) SyncRate() float64 { return m.PullsyncRate } -func (m *statusSnapshotMock) ReserveSize() int { return int(m.Snapshot.ReserveSize) } -func (m *statusSnapshotMock) StorageRadius() uint8 { return uint8(m.Snapshot.StorageRadius) } -func (m *statusSnapshotMock) Commitment() (uint64, error) { return m.BatchCommitment, nil } +func (m *statusSnapshotMock) SyncRateWithinRadius() float64 { return m.PullsyncRate } +func (m *statusSnapshotMock) ReserveSize() int { return int(m.Snapshot.ReserveSize) } +func (m *statusSnapshotMock) StorageRadius() uint8 { return uint8(m.Snapshot.StorageRadius) } +func (m *statusSnapshotMock) Commitment() (uint64, error) { return m.BatchCommitment, nil } func (m *statusSnapshotMock) GetChainState() *postage.ChainState { return &postage.ChainState{Block: m.LastSyncedBlock} } diff --git a/pkg/storageincentives/agent.go b/pkg/storageincentives/agent.go index 1473de8698e..dbe1010c9f6 100644 --- a/pkg/storageincentives/agent.go +++ b/pkg/storageincentives/agent.go @@ -66,7 +66,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 @@ -85,7 +85,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, @@ -104,7 +104,7 @@ func New(overlay swarm.Address, contract: contract, batchExpirer: batchExpirer, store: store, - fullSyncedFunc: fullSyncedFunc, + reserveSyncedFunc: reserveSyncedFunc, blocksPerRound: blocksPerRound, quit: make(chan struct{}), redistributionStatuser: redistributionStatuser, @@ -225,7 +225,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() @@ -425,8 +425,15 @@ 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") + salt, err := a.contract.ReserveSalt(ctx) + if err != nil { + return false, fmt.Errorf("reserve salt: %w", err) + } + + // in a sister neighbourhood round the sample comes from our bin PO(overlay, anchor) + depth := min(committedDepth, swarm.Proximity(a.overlay.Bytes(), salt)) + if !a.reserveSyncedFunc(depth) { + a.logger.Info("skipping round because reserve is not synced", "depth", depth, "round", round) return false, nil } @@ -445,7 +452,7 @@ func (a *Agent) handleSample(ctx context.Context, round uint64) (bool, error) { } now := time.Now() - sample, err := a.makeSample(ctx, committedDepth) + sample, err := a.makeSample(ctx, salt, committedDepth) if err != nil { return false, err } @@ -491,12 +498,7 @@ func (a *Agent) reserveSampleAndHash(ctx context.Context, anchor []byte, depth u return res, err } -func (a *Agent) makeSample(ctx context.Context, committedDepth uint8) (SampleData, error) { - salt, err := a.contract.ReserveSalt(ctx) - if err != nil { - return SampleData{}, err - } - +func (a *Agent) makeSample(ctx context.Context, salt []byte, committedDepth uint8) (SampleData, error) { res, err := a.reserveSampleAndHash(ctx, salt, committedDepth) if err != nil { return SampleData{}, err diff --git a/pkg/storageincentives/agent_test.go b/pkg/storageincentives/agent_test.go index 8da336e992e..8cd783dad3b 100644 --- a/pkg/storageincentives/agent_test.go +++ b/pkg/storageincentives/agent_test.go @@ -5,6 +5,7 @@ package storageincentives_test import ( + "bytes" "context" "errors" "math/big" @@ -35,14 +36,16 @@ 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 + sisterAnchor bool // round anchor at PO radius from the overlay instead of within committed depth }{ { name: "3 blocks per phase, same block number returns twice", @@ -91,6 +94,49 @@ 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, + // radius 8 + doubling 1 = committed depth 9 + isReserveSynced: func(depth uint8) bool { 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 { return depth > 9 }, + }, { + name: "no expected calls - sister anchor with sub-depth syncing", + blocksPerRound: 12, + blocksPerPhase: 4, + incrementBy: 2, + expectedCalls: false, + limit: 144, + balance: bigBalance, + doubling: 1, + isReserveSynced: func(depth uint8) bool { return depth >= 9 }, + sisterAnchor: true, + }, { + name: "expected calls - sister anchor with reserve synced at radius", + blocksPerRound: 12, + blocksPerPhase: 4, + incrementBy: 2, + expectedCalls: true, + limit: 144, + balance: bigBalance, + doubling: 1, + isReserveSynced: func(depth uint8) bool { return depth >= 8 }, + sisterAnchor: true, }, } @@ -113,8 +159,12 @@ func TestAgent(t *testing.T) { var radius uint8 = 8 contract := &mockContract{t: t, expectedRadius: radius + tc.doubling} + if tc.sisterAnchor { + contract.anchor = bytes.Clone(addr.Bytes()) + contract.anchor[radius/8] ^= 0x80 >> (radius % 8) + } - 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 @@ -459,6 +509,7 @@ func createService( blocksPerPhase uint64, radius uint8, doubling uint8, + isReserveSynced ...func(uint8) bool, ) (*storageincentives.Agent, error) { t.Helper() @@ -476,6 +527,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, @@ -483,7 +539,7 @@ func createService( postageContract, stakingContract, reserve, - func() bool { return true }, + syncedFunc, time.Millisecond*100, blocksPerRound, blocksPerPhase, @@ -568,6 +624,7 @@ type mockContract struct { isPlayingCount int beforeIsPlaying func() beforeCommit func() + anchor []byte } // getCalls returns a snapshot of the calls list @@ -599,7 +656,7 @@ func (m *mockContract) playingCount() int { } func (m *mockContract) ReserveSalt(context.Context) ([]byte, error) { - return nil, nil + return m.anchor, nil } func (m *mockContract) IsPlaying(_ context.Context, r uint8) (bool, error) { diff --git a/pkg/storer/compact_test.go b/pkg/storer/compact_test.go index 4d9a691008e..ccef4301e24 100644 --- a/pkg/storer/compact_test.go +++ b/pkg/storer/compact_test.go @@ -36,7 +36,7 @@ func TestCompact(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - st.StartReserveWorker(ctx, pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + st.StartReserveWorker(ctx, pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) <-readyC batches := []*postage.Batch{postagetesting.MustNewBatch(), postagetesting.MustNewBatch(), postagetesting.MustNewBatch()} @@ -137,7 +137,7 @@ func TestCompactNoEvictions(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - st.StartReserveWorker(ctx, pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + st.StartReserveWorker(ctx, pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) <-readyC batches := []*postage.Batch{postagetesting.MustNewBatch(), postagetesting.MustNewBatch(), postagetesting.MustNewBatch()} diff --git a/pkg/storer/recover_test.go b/pkg/storer/recover_test.go index 2942c49f73f..53f11627ae2 100644 --- a/pkg/storer/recover_test.go +++ b/pkg/storer/recover_test.go @@ -42,7 +42,7 @@ func TestRecoveryPrunesCorruptedChunks(t *testing.T) { t.Fatalf("New: %v", err) } readyC := make(chan struct{}) - st.StartReserveWorker(ctx, pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + st.StartReserveWorker(ctx, pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) <-readyC goodChunk := chunk.GenerateTestRandomChunk().WithStamp(postagetesting.MustNewBatchStamp(batch.ID)) @@ -140,7 +140,7 @@ func TestRecoveryPrunesUnreadableChunks(t *testing.T) { t.Fatalf("New: %v", err) } readyC := make(chan struct{}) - st.StartReserveWorker(ctx, pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + st.StartReserveWorker(ctx, pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) <-readyC goodChunk := chunk.GenerateTestRandomChunk().WithStamp(postagetesting.MustNewBatchStamp(batch.ID)) diff --git a/pkg/storer/reserve.go b/pkg/storer/reserve.go index b6799ea1cee..7add3f594da 100644 --- a/pkg/storer/reserve.go +++ b/pkg/storer/reserve.go @@ -37,8 +37,7 @@ var ( ) type Syncer interface { - // Number of active historical syncing jobs. - SyncRate() float64 + IsReserveSynced(depth uint8) bool Start(context.Context) } @@ -173,7 +172,9 @@ func (db *DB) reserveWorker(ctx context.Context, ready chan<- struct{}) { continue } - if count < threshold(db.reserve.Capacity()) && db.syncer.SyncRate() == 0 && radius > db.reserveOptions.minimumRadius { + // Check from bin 0: bins below the radius only hold counts from just before + // unreserve raised it, and those must still block an immediate decrease. + if count < threshold(db.reserve.Capacity()) && db.syncer.IsReserveSynced(0) && radius > db.reserveOptions.minimumRadius { radius-- if err := db.reserve.SetRadius(radius); err != nil { db.logger.Error(err, "reserve set radius") diff --git a/pkg/storer/reserve_test.go b/pkg/storer/reserve_test.go index 4c83df534dc..88ee27fd6da 100644 --- a/pkg/storer/reserve_test.go +++ b/pkg/storer/reserve_test.go @@ -76,7 +76,7 @@ func TestIndexCollision(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -92,7 +92,7 @@ func TestIndexCollision(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -176,7 +176,7 @@ func TestReplaceOldIndex(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -192,7 +192,7 @@ func TestReplaceOldIndex(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -212,7 +212,7 @@ func TestEvictBatch(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - st.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + st.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -378,7 +378,7 @@ func TestUnreserveCap(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -395,7 +395,7 @@ func TestUnreserveCap(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(0), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(0), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -416,7 +416,7 @@ func TestNetworkRadius(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(1), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(1), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -435,7 +435,7 @@ func TestNetworkRadius(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(1), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(1), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -482,7 +482,7 @@ func TestRadiusManager(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(3), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(3), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -516,14 +516,14 @@ func TestRadiusManager(t *testing.T) { waitForRadius(t, storer.Reserve(), 0) }) - t.Run("radius doesn't change due to non-zero pull rate", func(t *testing.T) { + t.Run("radius doesn't change while reserve is syncing", func(t *testing.T) { t.Parallel() storer, err := diskStorer(t, dbTestOps(baseAddr, 10, nil, nil, time.Millisecond*500))() if err != nil { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(1), networkRadiusFunc(3), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(false), networkRadiusFunc(3), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -531,8 +531,35 @@ func TestRadiusManager(t *testing.T) { } waitForRadius(t, storer.Reserve(), 3) }) + + t.Run("radius doesn't decrease while bins below it are syncing", func(t *testing.T) { + t.Parallel() + storer, err := diskStorer(t, dbTestOps(baseAddr, 10, nil, nil, time.Millisecond*100))() + if err != nil { + t.Fatal(err) + } + readyC := make(chan struct{}) + // bin 2 synced just before the radius was raised to 3 + storer.StartReserveWorker(context.Background(), depthSyncer{syncedFrom: 3}, networkRadiusFunc(3), readyC) + select { + case <-readyC: + case <-t.Context().Done(): + t.Fatal("start reserve worker timeout") + } + // the empty reserve is under the threshold on every tick + time.Sleep(time.Second) + if got := storer.Reserve().Radius(); got != 3 { + t.Fatalf("got radius %d, want 3", got) + } + }) } +// depthSyncer reports the reserve synced only at or above syncedFrom. +type depthSyncer struct{ syncedFrom uint8 } + +func (s depthSyncer) IsReserveSynced(depth uint8) bool { return depth >= s.syncedFrom } +func (depthSyncer) Start(context.Context) {} + func TestSubscribeBin(t *testing.T) { t.Parallel() @@ -814,7 +841,7 @@ func TestNeighborhoodStats(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(responsibiliyDepth), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(responsibiliyDepth), readyC) select { case <-readyC: case <-t.Context().Done(): @@ -835,7 +862,7 @@ func TestNeighborhoodStats(t *testing.T) { t.Fatal(err) } readyC := make(chan struct{}) - storer.StartReserveWorker(context.Background(), pullerMock.NewMockRateReporter(0), networkRadiusFunc(responsibiliyDepth), readyC) + storer.StartReserveWorker(context.Background(), pullerMock.NewMockSyncer(true), networkRadiusFunc(responsibiliyDepth), readyC) select { case <-readyC: case <-t.Context().Done():