diff --git a/cli/operator/eventsync.go b/cli/operator/eventsync.go index 616887e121..0c95d3af50 100644 --- a/cli/operator/eventsync.go +++ b/cli/operator/eventsync.go @@ -140,11 +140,12 @@ func syncContractEvents( // Sync ongoing registry events in the background. Crash if it stops: the node can't operate // without staying current with Ethereum events, and until reorg handling exists, restarting // from persisted state is safer than continuing on possibly-incorrect state. + ongoingFromBlock := fromBlock.Uint64() go func() { - err := eventSyncer.SyncOngoing(ctx, fromBlock.Uint64()) + err := eventSyncer.SyncOngoing(ctx, ongoingFromBlock) if err != nil && !errors.Is(err, context.Canceled) { logger.Fatal("failed syncing ongoing registry events", - zap.Uint64("last_processed_block", lastProcessedBlock), + zap.Uint64("from_block", ongoingFromBlock), zap.Error(err), ) } diff --git a/doppelganger/doppelganger.go b/doppelganger/doppelganger.go index ab3378d4d1..903ef3a8c9 100644 --- a/doppelganger/doppelganger.go +++ b/doppelganger/doppelganger.go @@ -182,7 +182,7 @@ func (h *handler) Start(ctx context.Context) error { select { case <-ctx.Done(): return ctx.Err() - case <-ticker.Next(): + case <-ticker.Advance(): currentSlot := ticker.Slot() currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) diff --git a/hprobe/prober.go b/hprobe/prober.go index cfdc35c0a2..d9cda8ab10 100644 --- a/hprobe/prober.go +++ b/hprobe/prober.go @@ -58,8 +58,9 @@ func (p *HealthProber) ProbeAll(ctx context.Context) error { defer wg.Done() err := p.probeComponent(ctx, n) - if err != nil { - // Relay the error and quit early. + if err != nil && !errors.Is(err, context.Canceled) { + // Relay the error and quit early. Skip context.Canceled: it means + // a sibling failure triggered our cancel(), not a real probe fault. errsCh <- fmt.Errorf("probe component %s: %w", name, err) cancel() } diff --git a/ibft/storage/store.go b/ibft/storage/store.go index 311283a688..38aff4fb4c 100644 --- a/ibft/storage/store.go +++ b/ibft/storage/store.go @@ -61,7 +61,7 @@ func (i *ParticipantStorage) PruneContinuously(ctx context.Context, slotTickerPr select { case <-ctx.Done(): return - case <-ticker.Next(): + case <-ticker.Advance(): threshold := ticker.Slot() - retain - 1 count, err := i.removeSlotAt(threshold) if err != nil { diff --git a/ibft/storage/store_test.go b/ibft/storage/store_test.go index 0abeda512a..b728aff049 100644 --- a/ibft/storage/store_test.go +++ b/ibft/storage/store_test.go @@ -186,7 +186,7 @@ func TestSlotCleanupJob(t *testing.T) { close(mockTimeChan) }) - ticker.EXPECT().Next().Return(mockTimeChan).AnyTimes() + ticker.EXPECT().Advance().Return(mockTimeChan).AnyTimes() ticker.EXPECT().Slot().DoAndReturn(func() phase0.Slot { return <-mockSlotChan }).AnyTimes() diff --git a/operator/duties/attester.go b/operator/duties/attester.go index b66975db61..d7464be3ac 100644 --- a/operator/duties/attester.go +++ b/operator/duties/attester.go @@ -76,7 +76,7 @@ func (h *AttesterHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -84,7 +84,7 @@ func (h *AttesterHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() // advances h.ticker + next = h.ticker.Advance() // advances h.ticker currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) nextEpoch := currentEpoch + 1 diff --git a/operator/duties/committee.go b/operator/duties/committee.go index 1ddda27ba4..69e3c5a642 100644 --- a/operator/duties/committee.go +++ b/operator/duties/committee.go @@ -51,7 +51,7 @@ func (h *CommitteeHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -59,7 +59,7 @@ func (h *CommitteeHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() + next = h.ticker.Advance() currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) currentPeriod := h.beaconConfig.EstimatedSyncCommitteePeriodAtEpoch(currentEpoch) diff --git a/operator/duties/proposer.go b/operator/duties/proposer.go index 5bc399f365..5d2513a6ee 100644 --- a/operator/duties/proposer.go +++ b/operator/duties/proposer.go @@ -67,7 +67,7 @@ func (h *ProposerHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -75,7 +75,7 @@ func (h *ProposerHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() // advances h.ticker + next = h.ticker.Advance() // advances h.ticker currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) nextEpoch := currentEpoch + 1 diff --git a/operator/duties/scheduler.go b/operator/duties/scheduler.go index 2bf09c53a3..404360b892 100644 --- a/operator/duties/scheduler.go +++ b/operator/duties/scheduler.go @@ -351,7 +351,7 @@ func (s *Scheduler) SlotTicker(ctx context.Context) { select { case <-ctx.Done(): return - case <-s.ticker.Next(): + case <-s.ticker.Advance(): slot := s.ticker.Slot() delay := s.beaconConfig.IntervalDuration() diff --git a/operator/duties/scheduler_test.go b/operator/duties/scheduler_test.go index e0b064f837..0bba0d5b93 100644 --- a/operator/duties/scheduler_test.go +++ b/operator/duties/scheduler_test.go @@ -84,7 +84,7 @@ func (m *MockSlotTicker) start(ctx context.Context) { }() } -func (m *MockSlotTicker) Next() <-chan time.Time { +func (m *MockSlotTicker) Advance() <-chan time.Time { return m.timeChan } @@ -482,7 +482,7 @@ func TestScheduler_Run(t *testing.T) { s.dutyHandlers = []dutyHandler{mockDutyHandler1, mockDutyHandler2} mockBeaconNode.EXPECT().SubscribeToHeadEvents(ctx, "duty_scheduler", gomock.Any()).Return(nil) - mockTicker.EXPECT().Next().Return(nil).AnyTimes() + mockTicker.EXPECT().Advance().Return(nil).AnyTimes() // setup mock duty handler expectations for _, mockDutyHandler := range s.dutyHandlers { @@ -535,7 +535,7 @@ func TestScheduler_Regression_IndicesChangeStuck(t *testing.T) { // add multiple mock duty handlers s.dutyHandlers = []dutyHandler{NewValidatorRegistrationHandler(nil)} mockBeaconNode.EXPECT().SubscribeToHeadEvents(ctx, "duty_scheduler", gomock.Any()).Return(nil) - mockTicker.EXPECT().Next().Return(nil).AnyTimes() + mockTicker.EXPECT().Advance().Return(nil).AnyTimes() err := s.Start(ctx) require.NoError(t, err) t.Cleanup(func() { diff --git a/operator/duties/sync_committee.go b/operator/duties/sync_committee.go index cfaca99838..304c678f2c 100644 --- a/operator/duties/sync_committee.go +++ b/operator/duties/sync_committee.go @@ -84,7 +84,7 @@ func (h *SyncCommitteeHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -92,7 +92,7 @@ func (h *SyncCommitteeHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() + next = h.ticker.Advance() currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) currentPeriod := h.beaconConfig.EstimatedSyncCommitteePeriodAtEpoch(currentEpoch) diff --git a/operator/duties/validator_registration.go b/operator/duties/validator_registration.go index 5fc7304907..e6da7e9795 100644 --- a/operator/duties/validator_registration.go +++ b/operator/duties/validator_registration.go @@ -117,7 +117,7 @@ func (h *ValidatorRegistrationHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -125,7 +125,7 @@ func (h *ValidatorRegistrationHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() + next = h.ticker.Advance() currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) slotNumber := uint64(currentSlot)%h.beaconConfig.SlotsPerEpoch + 1 diff --git a/operator/duties/voluntary_exit.go b/operator/duties/voluntary_exit.go index 6d296dc3c0..8538973b4e 100644 --- a/operator/duties/voluntary_exit.go +++ b/operator/duties/voluntary_exit.go @@ -111,7 +111,7 @@ func (h *VoluntaryExitHandler) HandleDuties(ctx context.Context) { h.logger.Info("starting duty handler") defer h.logger.Info("duty handler exited") - next := h.ticker.Next() + next := h.ticker.Advance() for { select { case <-ctx.Done(): @@ -119,7 +119,7 @@ func (h *VoluntaryExitHandler) HandleDuties(ctx context.Context) { case <-next: currentSlot := h.ticker.Slot() - next = h.ticker.Next() + next = h.ticker.Advance() currentEpoch := h.beaconConfig.EstimatedEpochAtSlot(currentSlot) slotNumber := uint64(currentSlot)%h.beaconConfig.SlotsPerEpoch + 1 diff --git a/operator/dutytracer/collector.go b/operator/dutytracer/collector.go index 73d9a48a8e..ef13663e3d 100644 --- a/operator/dutytracer/collector.go +++ b/operator/dutytracer/collector.go @@ -127,7 +127,7 @@ func (c *Collector) Start(ctx context.Context, tickerProvider slotticker.Provide select { case <-ctx.Done(): return - case <-ticker.Next(): + case <-ticker.Advance(): currentSlot := ticker.Slot() c.evict(currentSlot) } @@ -1304,7 +1304,7 @@ func (c *Collector) startScheduleFiller(ctx context.Context, tickerProvider slot select { case <-ctx.Done(): return - case <-t.Next(): + case <-t.Advance(): slot := t.Slot() // Enqueue current slot quickly; if queue is full, drop to avoid backpressure. select { diff --git a/operator/fee_recipient/controller.go b/operator/fee_recipient/controller.go index a6b3722575..e130e82e25 100644 --- a/operator/fee_recipient/controller.go +++ b/operator/fee_recipient/controller.go @@ -104,7 +104,7 @@ func (rc *recipientController) submitPreparationsOnSchedule(ctx context.Context) select { case <-ctx.Done(): return - case <-ticker.Next(): + case <-ticker.Advance(): slot := ticker.Slot() // Check if this is the middle slot of the epoch if firstTimeSubmitted && uint64(slot)%rc.beaconConfig.SlotsPerEpoch != (rc.beaconConfig.SlotsPerEpoch/2) { diff --git a/operator/fee_recipient/controller_test.go b/operator/fee_recipient/controller_test.go index d77688a884..67ce8d123d 100644 --- a/operator/fee_recipient/controller_test.go +++ b/operator/fee_recipient/controller_test.go @@ -126,7 +126,7 @@ func TestSubmitProposal(t *testing.T) { ticker := mocks.NewMockSlotTicker(ctrl) mockTimeChan := make(chan time.Time) mockSlotChan := make(chan phase0.Slot) - ticker.EXPECT().Next().Return(mockTimeChan).AnyTimes() + ticker.EXPECT().Advance().Return(mockTimeChan).AnyTimes() ticker.EXPECT().Slot().DoAndReturn(func() phase0.Slot { return <-mockSlotChan }).AnyTimes() frCtrl.beaconClient = client @@ -174,7 +174,7 @@ func TestSubmitProposal(t *testing.T) { ticker := mocks.NewMockSlotTicker(ctrl) mockTimeChan := make(chan time.Time, 1) - ticker.EXPECT().Next().Return(mockTimeChan).AnyTimes() + ticker.EXPECT().Advance().Return(mockTimeChan).AnyTimes() ticker.EXPECT().Slot().Return(phase0.Slot(100)).AnyTimes() frCtrl.beaconClient = client diff --git a/operator/slotticker/mocks/slotticker.go b/operator/slotticker/mocks/slotticker.go index b79e9bf4ce..41ac8a6c00 100644 --- a/operator/slotticker/mocks/slotticker.go +++ b/operator/slotticker/mocks/slotticker.go @@ -41,18 +41,18 @@ func (m *MockSlotTicker) EXPECT() *MockSlotTickerMockRecorder { return m.recorder } -// Next mocks base method. -func (m *MockSlotTicker) Next() <-chan time.Time { +// Advance mocks base method. +func (m *MockSlotTicker) Advance() <-chan time.Time { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Next") + ret := m.ctrl.Call(m, "Advance") ret0, _ := ret[0].(<-chan time.Time) return ret0 } -// Next indicates an expected call of Next. -func (mr *MockSlotTickerMockRecorder) Next() *gomock.Call { +// Advance indicates an expected call of Advance. +func (mr *MockSlotTickerMockRecorder) Advance() *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Next", reflect.TypeOf((*MockSlotTicker)(nil).Next)) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Advance", reflect.TypeOf((*MockSlotTicker)(nil).Advance)) } // Slot mocks base method. diff --git a/operator/slotticker/slotticker.go b/operator/slotticker/slotticker.go index 6255bfd0d5..cec5de7c13 100644 --- a/operator/slotticker/slotticker.go +++ b/operator/slotticker/slotticker.go @@ -18,11 +18,11 @@ type Provider func() SlotTicker // Note, the caller is RESPONSIBLE for calling Next method periodically in order for // SlotTicker to advance forward (to keep ticking) to newer slots. type SlotTicker interface { - // Next returns a channel that will relay 1 tick signaling that "freshest" slot has started. + // Advance returns a channel that will relay 1 tick signaling that "freshest" slot has started. // It advances slot number SlotTicker keeps track of (potentially jumping several slots ahead) // and returns a channel that will signal once the time corresponding to that "freshest" slot // comes. - Next() <-chan time.Time + Advance() <-chan time.Time // Slot returns the slot number that corresponds to Next. Slot() phase0.Slot } @@ -75,9 +75,9 @@ func newWithCustomTimer(logger *zap.Logger, cfg Config, timerProvider TimerProvi } } -// Next implements SlotTicker.Next. +// Advance implements SlotTicker.Advance. // Note, this method is not thread-safe. -func (s *slotTicker) Next() <-chan time.Time { +func (s *slotTicker) Advance() <-chan time.Time { timeSinceGenesis := time.Since(s.genesisTime) if timeSinceGenesis < 0 { // we are waiting for slotTicker to tick at s.genesisTime (signaling 0th slot start) diff --git a/operator/slotticker/slotticker_test.go b/operator/slotticker/slotticker_test.go index e6408e1faf..995f006e50 100644 --- a/operator/slotticker/slotticker_test.go +++ b/operator/slotticker/slotticker_test.go @@ -25,7 +25,7 @@ func TestSlotTicker(t *testing.T) { ticker := New(zap.NewNop(), Config{slotDuration, genesisTime}) for i := 0; i < numTicks; i++ { - <-ticker.Next() + <-ticker.Advance() slot := ticker.Slot() require.Equal(t, expectedSlot, slot) @@ -43,11 +43,11 @@ func TestSlotTicker2(t *testing.T) { //timeSinceGenesis := time.Since(genesisTime) //expectedSlot := phase0.Slot(timeSinceGenesis/slotDuration) + 1 ticker := New(zap.NewNop(), Config{slotDuration, genesisTime}) - <-ticker.Next() + <-ticker.Advance() firstSlot := ticker.Slot() require.Equal(t, phase0.Slot(1), firstSlot) - ch := ticker.Next() + ch := ticker.Advance() select { case <-ch: require.FailNowf(t, "unexpected tick", "expected to wait for dummyChan") @@ -66,7 +66,7 @@ func TestTickerInitialization(t *testing.T) { ticker := New(zap.NewNop(), Config{slotDuration, genesisTime}) start := time.Now() - <-ticker.Next() + <-ticker.Advance() slot := ticker.Slot() // Allow a small buffer (e.g., 10ms) due to code execution overhead @@ -85,7 +85,7 @@ func TestSlotNumberConsistency(t *testing.T) { var lastSlot phase0.Slot for i := 0; i < 10; i++ { - <-ticker.Next() + <-ticker.Advance() slot := ticker.Slot() require.Equal(t, lastSlot+1, slot) @@ -100,7 +100,7 @@ func TestGenesisInFuture(t *testing.T) { ticker := New(zap.NewNop(), Config{slotDuration, genesisTime}) start := time.Now() - <-ticker.Next() + <-ticker.Advance() // The first tick should occur after the genesis time expectedFirstTickDuration := genesisTime.Sub(start) @@ -121,7 +121,7 @@ func TestBoundedDrift(t *testing.T) { start := time.Now() for i := 0; i < ticks; i++ { - <-ticker.Next() + <-ticker.Advance() } expectedDuration := time.Duration(ticks) * slotDuration elapsed := time.Since(start) @@ -151,7 +151,7 @@ func TestMultipleSlotTickers(t *testing.T) { defer wg.Done() ticker := New(zap.NewNop(), Config{slotDuration, genesisTime}) for j := 0; j < ticksPerTimer; j++ { - <-ticker.Next() + <-ticker.Advance() } }() } @@ -180,7 +180,7 @@ func TestSlotSkipping(t *testing.T) { var lastSlot phase0.Slot for i := 1; i <= numTicks; i++ { // Starting loop from 1 for ease of skipInterval check select { - case <-ticker.Next(): + case <-ticker.Advance(): slot := ticker.Slot() // Ensure we never receive slots out of order or repeatedly @@ -193,7 +193,7 @@ func TestSlotSkipping(t *testing.T) { time.Sleep(slotDuration) // Ensure the next slot we receive is exactly 2 slots ahead of the previous slot - <-ticker.Next() + <-ticker.Advance() slotAfterDelay := ticker.Slot() require.Equal(t, lastSlot+2, slotAfterDelay, "Expected to skip a slot after introducing a delay") @@ -264,9 +264,9 @@ func TestDoubleTickWarning(t *testing.T) { mockTimerChan <- time.Now() // Call Next() twice to process the ticks - <-ticker.Next() + <-ticker.Advance() firstSlot := ticker.Slot() - <-ticker.Next() + <-ticker.Advance() secondSlot := ticker.Slot() require.NotEqual(t, firstSlot, secondSlot) @@ -300,20 +300,20 @@ func TestDoubleTickRealTimer(t *testing.T) { }, (&mockTimeProvider{timer: mockTimer}).NewTimer) // Wait for the first slot. - <-ticker.Next() + <-ticker.Advance() require.WithinDuration(t, firstSlotTime.Add(1*slotTime), time.Now(), 50*time.Millisecond, "Expected the first tick to occur after 1/10th of a slot") firstSlot := ticker.Slot() require.Equal(t, phase0.Slot(1), firstSlot) // Wait for the 2nd slot, but wake up early. mockTimer.fakeNextReset(slotTime / 2) - <-ticker.Next() + <-ticker.Advance() require.WithinDuration(t, firstSlotTime.Add(1*slotTime+slotTime/2), time.Now(), 50*time.Millisecond, "Expected the first tick to occur after 1/2th of a slot") secondSlot := ticker.Slot() require.Equal(t, phase0.Slot(2), secondSlot) // Expect the SlotTicker to realize it woke up early, and wait for the 3rd slot instead. - <-ticker.Next() + <-ticker.Advance() require.WithinDuration(t, firstSlotTime.Add(3*slotTime), time.Now(), 50*time.Millisecond, "Expected the first tick to occur after 1/10th of a slot") thirdSlot := ticker.Slot() require.Equal(t, phase0.Slot(3), thirdSlot) diff --git a/protocol/v2/ssv/runner/validator_registration.go b/protocol/v2/ssv/runner/validator_registration.go index 23985ced74..d967d1374f 100644 --- a/protocol/v2/ssv/runner/validator_registration.go +++ b/protocol/v2/ssv/runner/validator_registration.go @@ -378,7 +378,7 @@ func (s *VRSubmitter) start(ctx context.Context, ticker slotticker.SlotTicker) { select { case <-ctx.Done(): return - case <-ticker.Next(): + case <-ticker.Advance(): config := s.beaconConfig currentSlot := ticker.Slot()