From c50df3739de0fcd5ad78bbec7455398a55f2c223 Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 11:25:43 +0300 Subject: [PATCH 1/7] fix(exporter): populate dutyStore in all modes so message validation doesn't blackhole exits, proposer & sync-contribution messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exporter standard mode skipped the duty scheduler entirely (#2711), leaving dutyStore.Proposer, dutyStore.SyncCommittee, and dutyStore.VoluntaryExit empty. Message validation consults all three stores for every incoming message; with empty stores it returned ErrNoDuty / ErrTooManyDutiesPerEpoch, causing gossipsub to mark messages as Ignored (not forwarded to mesh peers, not re-validated). Archive mode had the same gap for VoluntaryExit: VoluntaryExitHandler was excluded from the exporter-mode scheduler under the comment "only executes duties", which missed that it also populates the validation store. Fix (two parts): 1. Always run the duty scheduler (standard and archive exporter both get AllShares provider, PrefetchingBeacon, NoopExecutor — same as archive before). 2. Restructure scheduler handler registration: - Proposer, SyncCommittee, VoluntaryExit: all modes (needed for validation). In exporter mode VoluntaryExit descriptors have OwnValidator=false so no exit is ever executed; the handler only populates dutyStore.VoluntaryExit. - Attester: operator + archive-exporter only (duty tracing needs it; standard exporter does not, and fetching duties for all network validators is expensive without the tracing benefit). - Committee, ValidatorRegistration: operator only (execution-only). Fixes the 2-slot channel-send timeout logged as "failed to schedule voluntary exit duty!" in both exporter modes (VoluntaryExitHandler now consumes the channel). Also removes the dead "exporter standard mode: skipping duty scheduler" log line. Closes #2886. Co-Authored-By: Claude Sonnet 4.6 --- operator/duties/scheduler.go | 23 +++++++++++++++++++---- operator/node.go | 11 +++++------ operator/node_test.go | 2 +- 3 files changed, 25 insertions(+), 11 deletions(-) diff --git a/operator/duties/scheduler.go b/operator/duties/scheduler.go index 2bf09c53a3..eaf8c2ad86 100644 --- a/operator/duties/scheduler.go +++ b/operator/duties/scheduler.go @@ -95,6 +95,10 @@ type SchedulerOptions struct { // executing duties (e.g., validator registration). When true, scheduler // still fetches/stores duties for all validators but does not execute them. ExporterMode bool + // ArchiveMode indicates this is an archive-mode exporter that needs the Attester handler + // for duty tracing. Only meaningful when ExporterMode is true; standard-mode exporters skip + // the Attester handler to avoid fetching duties for all network validators. + ArchiveMode bool } type Scheduler struct { @@ -165,19 +169,30 @@ func NewScheduler(logger *zap.Logger, opts *SchedulerOptions) *Scheduler { s.exporterMode = opts.ExporterMode - // These handlers fetch & record duties from the beacon node and are needed in both operator & exporter modes. - // When adding a new handler here, ensure it supports both modes. + // Proposer, SyncCommittee, and VoluntaryExit are needed in all modes. + // Proposer and SyncCommittee populate dutyStore entries consulted by message validation on all + // node types. VoluntaryExit populates the per-(slot,pk) duty count also consulted by validation; + // in exporter mode all exit descriptors have OwnValidator=false so no exit is ever executed. s.dutyHandlers = append(s.dutyHandlers, - NewAttesterHandler(dutyStore.Attester, opts.ExporterMode), NewProposerHandler(dutyStore.Proposer, opts.ExporterMode), NewSyncCommitteeHandler(dutyStore.SyncCommittee, opts.ExporterMode), + NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh), ) + + // Attester is needed for duty execution (operator) and duty tracing (archive exporter). + // Standard-mode exporter skips it: the attester store is not checked by message validation, + // and fetching duties for all network validators is expensive without the tracing benefit. + if !opts.ExporterMode || opts.ArchiveMode { + s.dutyHandlers = append(s.dutyHandlers, + NewAttesterHandler(dutyStore.Attester, opts.ExporterMode), + ) + } + // These handlers only execute duties and are not needed in exporter mode. if !opts.ExporterMode { s.dutyHandlers = append(s.dutyHandlers, NewCommitteeHandler(dutyStore.Attester, dutyStore.SyncCommittee), NewValidatorRegistrationHandler(opts.ValidatorRegistrationCh), - NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh), ) } return s diff --git a/operator/node.go b/operator/node.go index 50278f7ff1..18f76c9354 100644 --- a/operator/node.go +++ b/operator/node.go @@ -78,8 +78,8 @@ type Node struct { exporterRead *exporter2.Exporter } -func shouldRunDutyScheduler(exporterOpts exporter.Options) bool { - return !exporterOpts.Enabled || exporterOpts.Mode == exporter.ModeArchive +func shouldRunDutyScheduler(_ exporter.Options) bool { + return true } // New is the constructor of Node @@ -105,8 +105,8 @@ func New(logger *zap.Logger, opts Options, exporterOpts exporter.Options, slotTi var dutyScheduler *duties.Scheduler if shouldRunDutyScheduler(exporterOpts) { - // Prepare scheduler wiring; in exporter archive mode we swap to AllShares provider, - // a prefetching beacon adapter, and a no-op executor. + // Prepare scheduler wiring; in exporter mode (both standard and archive) swap to AllShares + // provider, a prefetching beacon adapter, and a no-op executor. schedulerBeacon := duties.BeaconNode(opts.BeaconNode) validatorProvider := duties.ValidatorProvider(selfValidatorStore) dutyExecutor := duties.DutyExecutor(opts.ValidatorController) @@ -132,6 +132,7 @@ func New(logger *zap.Logger, opts Options, exporterOpts exporter.Options, slotTi SlotTickerProvider: slotTickerProvider, P2PNetwork: opts.P2PNetwork, ExporterMode: exporterOpts.Enabled, + ArchiveMode: exporterOpts.Mode == exporter.ModeArchive, }) } @@ -179,8 +180,6 @@ func (n *Node) Start(ctx context.Context) error { if err := n.dutyScheduler.Start(ctx); err != nil { return fmt.Errorf("failed to run duty scheduler: %w", err) } - } else { - n.logger.Info("exporter standard mode: skipping duty scheduler") } n.validatorsCtrl.StartNetworkHandlers() diff --git a/operator/node_test.go b/operator/node_test.go index b994f9dcc4..a1f77480b2 100644 --- a/operator/node_test.go +++ b/operator/node_test.go @@ -27,7 +27,7 @@ func TestShouldRunDutyScheduler(t *testing.T) { Enabled: true, Mode: exporter.ModeStandard, }, - expected: false, + expected: true, }, { name: "exporter archive", From d104430480eeea81fb8f900e9334af2d3b496dbc Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 11:42:57 +0300 Subject: [PATCH 2/7] test(duties): verify NewScheduler registers correct handlers per mode Adds a table-driven test confirming that operator, standard-exporter, and archive-exporter modes each get exactly the expected set of duty handlers. This directly validates the fix for the exporter dutyStore blackholing bug. Co-Authored-By: Claude Sonnet 4.6 --- operator/duties/scheduler_test.go | 58 +++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/operator/duties/scheduler_test.go b/operator/duties/scheduler_test.go index e0b064f837..de6437c5fb 100644 --- a/operator/duties/scheduler_test.go +++ b/operator/duties/scheduler_test.go @@ -693,3 +693,61 @@ func TestScheduler_HandleHeadEvent_DoesNotBlockWithoutReorgConsumer(t *testing.T PreviousDutyDependentRootChanged: true, }}, drainReorgEvents(s.reorgCh)) } + +func TestNewScheduler_HandlerRegistration(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + exporterMode bool + archiveMode bool + wantHandlers []string + }{ + { + name: "operator mode", + exporterMode: false, + archiveMode: false, + wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT", "ATTESTER", "CLUSTER", "VALIDATOR_REGISTRATION"}, + }, + { + name: "standard exporter", + exporterMode: true, + archiveMode: false, + // VoluntaryExit is included so the dutyStore is populated for message validation; + // Attester is excluded because the standard exporter does not trace duties. + wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT"}, + }, + { + name: "archive exporter", + exporterMode: true, + archiveMode: true, + // Archive exporter needs Attester for full duty tracing but still skips + // execution-only handlers (CLUSTER, VALIDATOR_REGISTRATION). + wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT", "ATTESTER"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + s := NewScheduler(zap.NewNop(), &SchedulerOptions{ + ExporterMode: tc.exporterMode, + ArchiveMode: tc.archiveMode, + SlotTickerProvider: func() slotticker.SlotTicker { + return NewMockSlotTicker(ctx) + }, + }) + + var got []string + for _, h := range s.dutyHandlers { + got = append(got, h.Name()) + } + + require.ElementsMatch(t, tc.wantHandlers, got) + }) + } +} From 9527b0526587614b9621b4fa0df42409c013f83b Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 13:43:04 +0300 Subject: [PATCH 3/7] fix(lint): preallocate got slice in TestNewScheduler_HandlerRegistration (prealloc) Co-Authored-By: Claude Sonnet 4.6 --- operator/duties/scheduler_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/operator/duties/scheduler_test.go b/operator/duties/scheduler_test.go index de6437c5fb..697ff4245e 100644 --- a/operator/duties/scheduler_test.go +++ b/operator/duties/scheduler_test.go @@ -742,7 +742,7 @@ func TestNewScheduler_HandlerRegistration(t *testing.T) { }, }) - var got []string + got := make([]string, 0, len(s.dutyHandlers)) for _, h := range s.dutyHandlers { got = append(got, h.Name()) } From 863bc1116cc8f1514373185085f385e22e0574f4 Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 16:26:15 +0300 Subject: [PATCH 4/7] fix(duties): harden exporter safety + remove vestigial shouldRunDutyScheduler - Add exporterMode field to VoluntaryExitHandler and pass it from NewScheduler. processExecution now guards explicitly against execution in exporter mode, making the OwnValidator=false invariant a two-layer defense rather than an implicit contract. - Extend TestNewScheduler_HandlerRegistration to assert that exporterMode propagates into Proposer, SyncCommittee, and VoluntaryExit handlers, and that Scheduler.exporterMode (which gates ExecuteDuties) is wired correctly. - Remove shouldRunDutyScheduler (always returned true) and inline its body in New(). Remove the unreachable else branch in Start() that could never fire once the scheduler was always created. Co-Authored-By: Claude Sonnet 4.6 --- operator/duties/scheduler.go | 8 ++- operator/duties/scheduler_test.go | 21 ++++++- operator/duties/voluntary_exit.go | 12 +++- operator/duties/voluntary_exit_test.go | 4 +- operator/node.go | 79 ++++++++++---------------- operator/node_test.go | 47 --------------- 6 files changed, 69 insertions(+), 102 deletions(-) diff --git a/operator/duties/scheduler.go b/operator/duties/scheduler.go index eaf8c2ad86..efb8da04cc 100644 --- a/operator/duties/scheduler.go +++ b/operator/duties/scheduler.go @@ -171,12 +171,14 @@ func NewScheduler(logger *zap.Logger, opts *SchedulerOptions) *Scheduler { // Proposer, SyncCommittee, and VoluntaryExit are needed in all modes. // Proposer and SyncCommittee populate dutyStore entries consulted by message validation on all - // node types. VoluntaryExit populates the per-(slot,pk) duty count also consulted by validation; - // in exporter mode all exit descriptors have OwnValidator=false so no exit is ever executed. + // node types. VoluntaryExit populates the per-(slot,pk) duty count also consulted by validation. + // In exporter mode, exit descriptors always have OwnValidator=false (no owned validators), so + // the handler never queues exits for execution; VoluntaryExitHandler.processExecution also + // guards explicitly against execution in exporter mode as a second line of defense. s.dutyHandlers = append(s.dutyHandlers, NewProposerHandler(dutyStore.Proposer, opts.ExporterMode), NewSyncCommitteeHandler(dutyStore.SyncCommittee, opts.ExporterMode), - NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh), + NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh, opts.ExporterMode), ) // Attester is needed for duty execution (operator) and duty tracing (archive exporter). diff --git a/operator/duties/scheduler_test.go b/operator/duties/scheduler_test.go index 697ff4245e..e5ef28d6af 100644 --- a/operator/duties/scheduler_test.go +++ b/operator/duties/scheduler_test.go @@ -746,8 +746,27 @@ func TestNewScheduler_HandlerRegistration(t *testing.T) { for _, h := range s.dutyHandlers { got = append(got, h.Name()) } - require.ElementsMatch(t, tc.wantHandlers, got) + + // Verify exporterMode propagates into each handler that carries the field. + // This locks down the wiring so that a future change cannot accidentally give + // exporter handlers a real executor path. + // + // Scheduler.ExecuteDuties already guards against exporter-mode execution, but + // individual handlers with their own exporterMode field provide a second line of + // defense (e.g. VoluntaryExitHandler.processExecution's explicit guard). + require.Equal(t, tc.exporterMode, s.exporterMode) + + for _, h := range s.dutyHandlers { + switch h := h.(type) { + case *ProposerHandler: + require.Equal(t, tc.exporterMode, h.exporterMode, "ProposerHandler.exporterMode") + case *SyncCommitteeHandler: + require.Equal(t, tc.exporterMode, h.exporterMode, "SyncCommitteeHandler.exporterMode") + case *VoluntaryExitHandler: + require.Equal(t, tc.exporterMode, h.exporterMode, "VoluntaryExitHandler.exporterMode") + } + } }) } } diff --git a/operator/duties/voluntary_exit.go b/operator/duties/voluntary_exit.go index 6d296dc3c0..021e543a4a 100644 --- a/operator/duties/voluntary_exit.go +++ b/operator/duties/voluntary_exit.go @@ -90,14 +90,16 @@ type VoluntaryExitHandler struct { validatorExitCh <-chan ExitDescriptor dutyQueue []*queuedExit blockSlots map[uint64]phase0.Slot + exporterMode bool } -func NewVoluntaryExitHandler(duties *dutystore.VoluntaryExitDuties, validatorExitCh <-chan ExitDescriptor) *VoluntaryExitHandler { +func NewVoluntaryExitHandler(duties *dutystore.VoluntaryExitDuties, validatorExitCh <-chan ExitDescriptor, exporterMode bool) *VoluntaryExitHandler { return &VoluntaryExitHandler{ duties: duties, validatorExitCh: validatorExitCh, dutyQueue: make([]*queuedExit, 0), blockSlots: map[uint64]phase0.Slot{}, + exporterMode: exporterMode, } } @@ -226,6 +228,14 @@ func (h *VoluntaryExitHandler) processExecution(ctx context.Context, slot phase0 span.SetAttributes(observability.DutyCountAttribute(len(dutiesForExecution))) if dutyCount := len(dutiesForExecution); dutyCount != 0 { + // Exporter nodes populate the duty store for p2p message-validation only and must + // never execute exits. In practice the queue is always empty because HandleDuties + // skips OwnValidator=false descriptors, and exporter mode never produces + // OwnValidator=true ones. This guard catches any future upstream invariant break. + if h.exporterMode { + h.logger.Error("BUG: voluntary exit execution attempted in exporter mode; OwnValidator invariant broken") + return + } h.dutiesExecutor.ExecuteDuties(ctx, dutiesForExecution, h.dutyExecutionDeadline(slot)) h.logger.Debug("executed voluntary exit duties", fields.Slot(slot), diff --git a/operator/duties/voluntary_exit_test.go b/operator/duties/voluntary_exit_test.go index a276c16c9d..9ed125f058 100644 --- a/operator/duties/voluntary_exit_test.go +++ b/operator/duties/voluntary_exit_test.go @@ -32,7 +32,7 @@ func TestVoluntaryExitHandler_HandleDuties(t *testing.T) { t.Parallel() exitCh := make(chan ExitDescriptor) - handler := NewVoluntaryExitHandler(dutystore.NewVoluntaryExit(), exitCh) + handler := NewVoluntaryExitHandler(dutystore.NewVoluntaryExit(), exitCh, false) // Duty executor expects deadline to be set on the parent context (see "parent-context has no deadline set"). // This deadline needs to be large enough to not prevent tests from executing their intended flow. @@ -170,7 +170,7 @@ func TestVoluntaryExitHandler_HandleDuties_LateObservedExitWaitsPastFollowDistan t.Parallel() exitCh := make(chan ExitDescriptor) - handler := NewVoluntaryExitHandler(dutystore.NewVoluntaryExit(), exitCh) + handler := NewVoluntaryExitHandler(dutystore.NewVoluntaryExit(), exitCh, false) ctx, cancel := context.WithTimeout(t.Context(), 5*time.Minute) diff --git a/operator/node.go b/operator/node.go index 18f76c9354..d785d26fa8 100644 --- a/operator/node.go +++ b/operator/node.go @@ -78,10 +78,6 @@ type Node struct { exporterRead *exporter2.Exporter } -func shouldRunDutyScheduler(_ exporter.Options) bool { - return true -} - // New is the constructor of Node func New(logger *zap.Logger, opts Options, exporterOpts exporter.Options, slotTickerProvider slotticker.Provider, qbftStorage *qbftstorage.ParticipantStores) *Node { selfValidatorStore := opts.ValidatorStore.WithOperatorID(opts.ValidatorOptions.OperatorDataStore.GetOperatorID) @@ -103,39 +99,36 @@ func New(logger *zap.Logger, opts Options, exporterOpts exporter.Options, slotTi proposalPreparationsFn = feeRecipientController.GetProposalPreparations } - var dutyScheduler *duties.Scheduler - if shouldRunDutyScheduler(exporterOpts) { - // Prepare scheduler wiring; in exporter mode (both standard and archive) swap to AllShares - // provider, a prefetching beacon adapter, and a no-op executor. - schedulerBeacon := duties.BeaconNode(opts.BeaconNode) - validatorProvider := duties.ValidatorProvider(selfValidatorStore) - dutyExecutor := duties.DutyExecutor(opts.ValidatorController) - - if exporterOpts.Enabled { - validatorProvider = duties.NewAllSharesProvider(opts.ValidatorStore) - dutyExecutor = duties.NewNoopExecutor() - schedulerBeacon = duties.NewPrefetchingBeacon(logger, opts.BeaconNode, opts.NetworkConfig.Beacon, opts.ValidatorStore) - } + // All node modes run the duty scheduler. In exporter mode swap to AllShares provider, + // a prefetching beacon adapter, and a no-op executor. + schedulerBeacon := duties.BeaconNode(opts.BeaconNode) + validatorProvider := duties.ValidatorProvider(selfValidatorStore) + dutyExecutor := duties.DutyExecutor(opts.ValidatorController) - dutyScheduler = duties.NewScheduler(logger, &duties.SchedulerOptions{ - Ctx: opts.Context, - BeaconNode: schedulerBeacon, - ExecutionClient: opts.ExecutionClient, - BeaconConfig: opts.NetworkConfig.Beacon, - ValidatorProvider: validatorProvider, - ValidatorController: opts.ValidatorController, - DutyExecutor: dutyExecutor, - IndicesChgCh: opts.ValidatorController.IndicesChangeChan(), - ValidatorRegistrationCh: opts.ValidatorController.ValidatorRegistrationChan(), - ValidatorExitCh: opts.ValidatorController.ValidatorExitChan(), - DutyStore: opts.DutyStore, - SlotTickerProvider: slotTickerProvider, - P2PNetwork: opts.P2PNetwork, - ExporterMode: exporterOpts.Enabled, - ArchiveMode: exporterOpts.Mode == exporter.ModeArchive, - }) + if exporterOpts.Enabled { + validatorProvider = duties.NewAllSharesProvider(opts.ValidatorStore) + dutyExecutor = duties.NewNoopExecutor() + schedulerBeacon = duties.NewPrefetchingBeacon(logger, opts.BeaconNode, opts.NetworkConfig.Beacon, opts.ValidatorStore) } + dutyScheduler := duties.NewScheduler(logger, &duties.SchedulerOptions{ + Ctx: opts.Context, + BeaconNode: schedulerBeacon, + ExecutionClient: opts.ExecutionClient, + BeaconConfig: opts.NetworkConfig.Beacon, + ValidatorProvider: validatorProvider, + ValidatorController: opts.ValidatorController, + DutyExecutor: dutyExecutor, + IndicesChgCh: opts.ValidatorController.IndicesChangeChan(), + ValidatorRegistrationCh: opts.ValidatorController.ValidatorRegistrationChan(), + ValidatorExitCh: opts.ValidatorController.ValidatorExitChan(), + DutyStore: opts.DutyStore, + SlotTickerProvider: slotTickerProvider, + P2PNetwork: opts.P2PNetwork, + ExporterMode: exporterOpts.Enabled, + ArchiveMode: exporterOpts.Mode == exporter.ModeArchive, + }) + node := &Node{ logger: logger.Named(log.NameOperator), validatorsCtrl: opts.ValidatorController, @@ -175,11 +168,8 @@ func (n *Node) Start(ctx context.Context) error { return fmt.Errorf("start WS server: %w", err) } - // Start the duty scheduler in modes that use it. - if n.dutyScheduler != nil { - if err := n.dutyScheduler.Start(ctx); err != nil { - return fmt.Errorf("failed to run duty scheduler: %w", err) - } + if err := n.dutyScheduler.Start(ctx); err != nil { + return fmt.Errorf("failed to run duty scheduler: %w", err) } n.validatorsCtrl.StartNetworkHandlers() @@ -247,15 +237,8 @@ func (n *Node) Start(ctx context.Context) error { n.logger.Info("operator node has been started", fields.OperatorID(n.validatorOptions.OperatorDataStore.GetOperatorID())) - if n.dutyScheduler != nil { - if err := n.dutyScheduler.Wait(); err != nil { - n.logger.Fatal("duty scheduler exited with error", zap.Error(err)) - } - } else { - if !n.exporterOptions.Enabled || n.exporterOptions.Mode != exporter.ModeStandard { - n.logger.Fatal("duty scheduler is nil for non-exporter-standard node") - } - <-ctx.Done() + if err := n.dutyScheduler.Wait(); err != nil { + n.logger.Fatal("duty scheduler exited with error", zap.Error(err)) } // The p2p network is owned by its creator (cli/operator), which closes it via defer. diff --git a/operator/node_test.go b/operator/node_test.go index a1f77480b2..18b97f2b56 100644 --- a/operator/node_test.go +++ b/operator/node_test.go @@ -1,48 +1 @@ package operator - -import ( - "testing" - - "github.com/stretchr/testify/require" - - "github.com/ssvlabs/ssv/exporter" -) - -func TestShouldRunDutyScheduler(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - exporterOpts exporter.Options - expected bool - }{ - { - name: "regular operator", - exporterOpts: exporter.Options{}, - expected: true, - }, - { - name: "exporter standard", - exporterOpts: exporter.Options{ - Enabled: true, - Mode: exporter.ModeStandard, - }, - expected: true, - }, - { - name: "exporter archive", - exporterOpts: exporter.Options{ - Enabled: true, - Mode: exporter.ModeArchive, - }, - expected: true, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - require.Equal(t, tc.expected, shouldRunDutyScheduler(tc.exporterOpts)) - }) - } -} From 0eae766064deea59b4f34a7eeedd6f02ac2d2b58 Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 17:04:34 +0300 Subject: [PATCH 5/7] test(duties): cover exporter guard + scheduler wiring to fix codecov/patch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add two tests that bring the patch coverage above the 61.4% target: - TestVoluntaryExitHandler_ExporterModeGuard: exercises the explicit guard added in processExecution (exporterMode=true path). Directly pushes an item onto dutyQueue and asserts ExecuteDuties is never called — gomock strict-mode fails the test if the guard is removed. - TestNew_ExporterMode_SchedulerWiring: calls operator.New() in exporter mode with a minimal stub set, covering the three restructured scheduler-wiring lines (schedulerBeacon / validatorProvider / dutyExecutor) that appeared as new lines in the diff after shouldRunDutyScheduler was removed. Co-Authored-By: Claude Sonnet 4.6 --- operator/duties/voluntary_exit_test.go | 33 ++++++++++++++++ operator/node_test.go | 53 ++++++++++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/operator/duties/voluntary_exit_test.go b/operator/duties/voluntary_exit_test.go index 9ed125f058..3f79cf0a1c 100644 --- a/operator/duties/voluntary_exit_test.go +++ b/operator/duties/voluntary_exit_test.go @@ -11,10 +11,12 @@ import ( ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" + "go.uber.org/zap" spectypes "github.com/ssvlabs/ssv-spec/types" "github.com/ssvlabs/ssv/eth/executionclient" + "github.com/ssvlabs/ssv/networkconfig" "github.com/ssvlabs/ssv/operator/duties/dutystore" ) @@ -223,6 +225,37 @@ func TestVoluntaryExitHandler_HandleDuties_LateObservedExitWaitsPastFollowDistan ticker.WaitShutdown() } +// TestVoluntaryExitHandler_ExporterModeGuard verifies the explicit guard in processExecution: +// even when an item reaches the execution gate, an exporter-mode handler must never call +// ExecuteDuties. This guards against an upstream invariant break (OwnValidator=true in exporter mode). +func TestVoluntaryExitHandler_ExporterModeGuard(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + + exitCh := make(chan ExitDescriptor) + handler := NewVoluntaryExitHandler(dutystore.NewVoluntaryExit(), exitCh, true) + + mockExecutor := NewMockDutiesExecutor(ctrl) + + handler.logger = zap.NewNop() + handler.beaconConfig = networkconfig.TestNetwork.Beacon + handler.dutiesExecutor = mockExecutor + + // Push a queued exit whose earliestExecutionSlot has already passed. + const slot = phase0.Slot(100) + handler.dutyQueue = []*queuedExit{{ + duty: &spectypes.ValidatorDuty{ + Type: spectypes.BNRoleVoluntaryExit, + }, + earliestExecutionSlot: slot - 1, + }} + + // If the guard is absent, processExecution calls ExecuteDuties and gomock fails the test + // with an unexpected call. The guard's explicit return makes this a no-op for exporters. + handler.processExecution(context.Background(), slot) +} + func create1to1BlockSlotMapping(scheduler *Scheduler) *atomic.Uint64 { var headerByNumberCalls atomic.Uint64 diff --git a/operator/node_test.go b/operator/node_test.go index 18b97f2b56..6b78a45ac1 100644 --- a/operator/node_test.go +++ b/operator/node_test.go @@ -1 +1,54 @@ package operator + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "go.uber.org/zap" + + "github.com/ssvlabs/ssv/exporter" + "github.com/ssvlabs/ssv/networkconfig" + operatordatastore "github.com/ssvlabs/ssv/operator/datastore" + "github.com/ssvlabs/ssv/operator/duties/dutystore" + "github.com/ssvlabs/ssv/operator/slotticker" + mockslotticker "github.com/ssvlabs/ssv/operator/slotticker/mocks" + "github.com/ssvlabs/ssv/operator/validator" + registrymocks "github.com/ssvlabs/ssv/registry/storage/mocks" +) + +// TestNew_ExporterMode_SchedulerWiring verifies that operator.New() completes successfully in exporter +// mode and wires the duty scheduler with the AllShares provider path (no fee-recipient controller). +// This covers the restructured scheduler-wiring lines that were previously gated by shouldRunDutyScheduler. +func TestNew_ExporterMode_SchedulerWiring(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + + mockVS := registrymocks.NewMockValidatorStore(ctrl) + mockVS.EXPECT().WithOperatorID(gomock.Any()).Return(nil) + + mockTicker := mockslotticker.NewMockSlotTicker(ctrl) + + opts := Options{ + NetworkConfig: networkconfig.TestNetwork, + Context: context.Background(), + ValidatorStore: mockVS, + ValidatorController: new(validator.Controller), + ValidatorOptions: validator.ControllerOptions{ + OperatorDataStore: operatordatastore.New(nil), + }, + DutyStore: dutystore.New(), + } + + node := New( + zap.NewNop(), + opts, + exporter.Options{Enabled: true}, + func() slotticker.SlotTicker { return mockTicker }, + nil, + ) + require.NotNil(t, node) + require.NotNil(t, node.dutyScheduler) +} From f927ae027995a202f872d57353949169decf50ce Mon Sep 17 00:00:00 2001 From: Roy-blox Date: Mon, 15 Jun 2026 17:09:11 +0300 Subject: [PATCH 6/7] fix(test): gofmt struct field alignment in node_test.go Co-Authored-By: Claude Sonnet 4.6 --- operator/node_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/operator/node_test.go b/operator/node_test.go index 6b78a45ac1..622abddba2 100644 --- a/operator/node_test.go +++ b/operator/node_test.go @@ -32,9 +32,9 @@ func TestNew_ExporterMode_SchedulerWiring(t *testing.T) { mockTicker := mockslotticker.NewMockSlotTicker(ctrl) opts := Options{ - NetworkConfig: networkconfig.TestNetwork, - Context: context.Background(), - ValidatorStore: mockVS, + NetworkConfig: networkconfig.TestNetwork, + Context: context.Background(), + ValidatorStore: mockVS, ValidatorController: new(validator.Controller), ValidatorOptions: validator.ControllerOptions{ OperatorDataStore: operatordatastore.New(nil), From 651052beb1ca1e722e4b6cc8f1540f2f9dc4e3f8 Mon Sep 17 00:00:00 2001 From: rehs0y Date: Tue, 23 Jun 2026 11:19:22 +0300 Subject: [PATCH 7/7] feat(exporter): remove standard mode, add optional duty-trace retention (#2895) --- cli/operator/config.go | 45 ++++--- cli/operator/config_test.go | 67 +++------- cli/operator/node.go | 54 ++------ cli/operator/node_test.go | 24 +--- exporter/opts.go | 10 +- exporter/store/store.go | 37 ++++++ exporter/store/store_test.go | 81 ++++++++++++ operator/duties/scheduler.go | 28 ++-- operator/duties/scheduler_test.go | 19 +-- operator/dutytracer/collector.go | 117 ++++++++++++++++- operator/dutytracer/collector_test.go | 176 ++++++++++++++++++++++++++ operator/dutytracer/store.go | 3 + operator/dutytracer/store_metrics.go | 8 ++ operator/node.go | 1 - operator/validator/controller.go | 7 +- 15 files changed, 500 insertions(+), 177 deletions(-) diff --git a/cli/operator/config.go b/cli/operator/config.go index b82c16b026..5fcec9331b 100644 --- a/cli/operator/config.go +++ b/cli/operator/config.go @@ -3,6 +3,7 @@ package operator import ( "errors" "fmt" + "os" "time" "github.com/ilyakaznacheev/cleanenv" @@ -64,13 +65,12 @@ const maxSafeProposerDelay = 1000 * time.Millisecond // nodeMode is the resolved operating mode of the node, derived once from ExporterOptions by // resolveAndValidate so startup can dispatch on a typed value instead of re-deriving the mode -// from ExporterOptions.Enabled / .Mode at each site. +// from ExporterOptions.Enabled at each site. type nodeMode int const ( - modeOperator nodeMode = iota // not an exporter - modeExporterStandard // exporter, standard tracing - modeExporterArchive // exporter, archive tracing (pre-consensus + consensus) + modeOperator nodeMode = iota // not an exporter + modeExporter // exporter (full duty tracing: pre-consensus + consensus + post-consensus) ) // resolved carries config-derived state computed by resolveAndValidate (not operator-provided): @@ -136,11 +136,10 @@ func (c *config) resolveAndValidate(logger *zap.Logger) (resolved, error) { // Resolve the operating mode last so a doubly-misconfigured node still surfaces the signing // or proposer-delay error first. - m, err := resolveMode(c.ExporterOptions) - if err != nil { - return resolved{}, err + res.mode = resolveMode(c.ExporterOptions) + if res.mode != modeOperator { + warnDeprecatedExporterEnv(logger) } - res.mode = m return res, nil } @@ -218,20 +217,26 @@ func (c *config) resolveSigning() (resolved, error) { return res, nil } -// resolveMode derives the node's operating mode from the exporter options, rejecting an -// unrecognized EXPORTER_MODE up front (fail-fast). A non-exporter node is always modeOperator, -// regardless of the (then-irrelevant) EXPORTER_MODE. -func resolveMode(opts exporter.Options) (nodeMode, error) { +// resolveMode derives the node's operating mode from the exporter options. An enabled exporter +// always runs in full duty-tracing mode; a non-exporter node is modeOperator. (The legacy +// standard/archive EXPORTER_MODE distinction was removed — exporters are always full tracers.) +func resolveMode(opts exporter.Options) nodeMode { if !opts.Enabled { - return modeOperator, nil + return modeOperator } - switch opts.Mode { - case exporter.ModeStandard: - return modeExporterStandard, nil - case exporter.ModeArchive: - return modeExporterArchive, nil - default: - return modeOperator, fmt.Errorf("invalid exporter mode %q (must be %q or %q)", opts.Mode, exporter.ModeStandard, exporter.ModeArchive) + return modeExporter +} + +// warnDeprecatedExporterEnv flags removed exporter env vars (set via a pre-existing deployment) so +// operators notice a stale value is now ignored: exporters always run full duty tracing, and +// retention moved from slots to EXPORTER_RETAIN_EPOCHS. Best-effort — only env vars are checked, +// not equivalent YAML keys (cleanenv silently drops unknown fields). +func warnDeprecatedExporterEnv(logger *zap.Logger) { + for _, env := range []string{"EXPORTER_MODE", "EXPORTER_RETAIN_SLOTS"} { + if v, ok := os.LookupEnv(env); ok { + logger.Warn("ignoring removed exporter config option; exporters always run full duty tracing — use EXPORTER_RETAIN_EPOCHS for retention", + zap.String("env", env), zap.String("value", v)) + } } } diff --git a/cli/operator/config_test.go b/cli/operator/config_test.go index 64012a59c7..9d67a931f9 100644 --- a/cli/operator/config_test.go +++ b/cli/operator/config_test.go @@ -22,10 +22,6 @@ import ( const ( testSignerEndpoint = "http://signer:9000" testOperatorKey = "super-secret-operator-key" - - // substring of resolveMode's error for an unrecognized EXPORTER_MODE (kept as a const so - // the repeated assertion doesn't trip goconst). - msgInvalidExporterMode = "invalid exporter mode" ) func Test_config_load(t *testing.T) { @@ -122,8 +118,9 @@ func Test_resolveAndValidate_signingErrorContext(t *testing.T) { } } -// Test_resolveAndValidate_mode verifies the operating mode is resolved into the result and that -// an invalid EXPORTER_MODE fails validation up front. +// Test_resolveAndValidate_mode verifies the operating mode is resolved into the result: a +// non-exporter is modeOperator, and an enabled exporter is modeExporter (full duty tracing). The +// legacy standard/archive EXPORTER_MODE validation was removed along with the mode itself. func Test_resolveAndValidate_mode(t *testing.T) { t.Run("non-exporter -> modeOperator", func(t *testing.T) { c := config{} @@ -133,22 +130,12 @@ func Test_resolveAndValidate_mode(t *testing.T) { require.Equal(t, modeOperator, res.mode) }) - t.Run("exporter archive -> modeExporterArchive", func(t *testing.T) { + t.Run("exporter -> modeExporter", func(t *testing.T) { c := config{} c.ExporterOptions.Enabled = true - c.ExporterOptions.Mode = exporter.ModeArchive res, err := c.resolveAndValidate(zap.NewNop()) require.NoError(t, err) - require.Equal(t, modeExporterArchive, res.mode) - }) - - t.Run("invalid exporter mode -> error", func(t *testing.T) { - c := config{} - c.ExporterOptions.Enabled = true - c.ExporterOptions.Mode = "bogus" - _, err := c.resolveAndValidate(zap.NewNop()) - require.Error(t, err) - require.Contains(t, err.Error(), msgInvalidExporterMode) + require.Equal(t, modeExporter, res.mode) }) } @@ -425,36 +412,24 @@ func Test_resolveSigning(t *testing.T) { } } -// Test_resolveMode covers operating-mode resolution and the fail-fast rejection of an -// unrecognized EXPORTER_MODE. +// Test_resolveMode covers operating-mode resolution: an enabled exporter resolves to modeExporter +// (full duty tracing), everything else to modeOperator. func Test_resolveMode(t *testing.T) { - tests := []struct { - name string - enabled bool - mode string - want nodeMode - wantErr string - }{ - {name: "not exporter -> operator", enabled: false, mode: "", want: modeOperator}, - {name: "not exporter ignores mode -> operator", enabled: false, mode: exporter.ModeArchive, want: modeOperator}, - {name: "exporter standard", enabled: true, mode: exporter.ModeStandard, want: modeExporterStandard}, - {name: "exporter archive", enabled: true, mode: exporter.ModeArchive, want: modeExporterArchive}, - {name: "exporter invalid -> error", enabled: true, mode: "bogus", wantErr: msgInvalidExporterMode}, - {name: "exporter empty mode -> error", enabled: true, mode: "", wantErr: msgInvalidExporterMode}, - } + require.Equal(t, modeOperator, resolveMode(exporter.Options{Enabled: false})) + require.Equal(t, modeExporter, resolveMode(exporter.Options{Enabled: true})) +} - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got, err := resolveMode(exporter.Options{Enabled: tt.enabled, Mode: tt.mode}) - if tt.wantErr != "" { - require.Error(t, err) - require.Contains(t, err.Error(), tt.wantErr) - return - } - require.NoError(t, err) - require.Equal(t, tt.want, got) - }) - } +// Test_warnDeprecatedExporterEnv verifies a stale, now-removed exporter env var is flagged rather +// than silently ignored. +func Test_warnDeprecatedExporterEnv(t *testing.T) { + t.Setenv("EXPORTER_MODE", "archive") + t.Setenv("EXPORTER_RETAIN_SLOTS", "50400") + + core, logs := observer.New(zapcore.WarnLevel) + warnDeprecatedExporterEnv(zap.New(core)) + + require.Equal(t, 1, logs.FilterField(zap.String("env", "EXPORTER_MODE")).Len()) + require.Equal(t, 1, logs.FilterField(zap.String("env", "EXPORTER_RETAIN_SLOTS")).Len()) } func Test_warnIfSSVAPIAddressUnset(t *testing.T) { diff --git a/cli/operator/node.go b/cli/operator/node.go index 704eb20c1c..bebee44ca5 100644 --- a/cli/operator/node.go +++ b/cli/operator/node.go @@ -8,10 +8,8 @@ import ( "net/http" "strconv" "strings" - "sync" "time" - "github.com/attestantio/go-eth2-client/spec/phase0" spectypes "github.com/ssvlabs/ssv-spec/types" "go.uber.org/zap" @@ -359,12 +357,6 @@ func newNode(ctx context.Context, cfg *config, logger *zap.Logger, res resolved, }) } - if res.mode == modeExporterStandard { - retain := cfg.ExporterOptions.RetainSlots - threshold := networkConfig.EstimatedCurrentSlot() - initSlotPruning(ctx, storageMap, slotTickerProvider, threshold, retain) - } - fixedSubnets, err := networkcommons.SubnetsFromString(cfg.P2pNetworkConfig.Subnets) if err != nil { return nil, fmt.Errorf("failed to parse fixed subnets: %w", err) @@ -383,12 +375,16 @@ func newNode(ctx context.Context, cfg *config, logger *zap.Logger, res resolved, metadata.WithSyncInterval(cfg.SSVOptions.ValidatorOptions.MetadataUpdateInterval), ) - // Exporter duty tracing. An invalid EXPORTER_MODE is rejected up front by resolveAndValidate, - // so res.mode here is always one of the known modes. + // Exporter duty tracing. Exporters always run full duty tracing — collecting pre-consensus, + // consensus, and post-consensus steps and serving them via the read API. (The legacy + // standard/archive EXPORTER_MODE distinction was removed; "standard" was a strict, and broken, + // subset.) RetainEpochs bounds on-disk trace history; 0 (default) retains indefinitely. var collector *dutytracer.Collector - switch res.mode { - case modeExporterArchive: - logger.Info("exporter mode: archive") + if res.isExporter() { + retainSlots := cfg.ExporterOptions.RetainEpochs * networkConfig.SlotsPerEpoch + logger.Info("exporter enabled (full duty tracing)", + zap.Uint64("retain_epochs", cfg.ExporterOptions.RetainEpochs), + zap.Uint64("retain_slots", retainSlots)) dstore := &dutytracer.DutyTraceStoreMetrics{ Store: dutytracestore.New(db), } @@ -397,12 +393,8 @@ func newNode(ctx context.Context, cfg *config, logger *zap.Logger, res resolved, dstore, networkConfig.Beacon, decidedStreamPublisherFn, dutyStore) - go collector.Start(ctx, slotTickerProvider) + go collector.Start(ctx, slotTickerProvider, retainSlots) cfg.SSVOptions.ExporterRead = exporter2.NewExporter(logger, storageMap, collector, nodeStorage.ValidatorStore()) - case modeExporterStandard: - logger.Info("exporter mode: standard") - case modeOperator: - // not an exporter: no duty-trace collector } doppelgangerHandler := buildDoppelganger(logger, cfg, res, networkConfig.Beacon, consensusClient, validatorProvider, slotTickerProvider) @@ -597,7 +589,7 @@ func (n *node) start() error { Shares: n.nodeStorage.Shares(), }, hexporter.NewExporter(n.logger, n.storageMap, n.collector, n.nodeStorage.ValidatorStore()), - n.mode == modeExporterArchive, + n.mode == modeExporter, ) _, apiServeErr, err := apiServer.Start(n.ctx) if err != nil { @@ -664,30 +656,6 @@ func setupOperatorDataStore( return operatordatastore.New(operatorData), nil } -func initSlotPruning(ctx context.Context, stores *ibftstorage.ParticipantStores, slotTickerProvider slotticker.Provider, slot phase0.Slot, retain uint64) { - var wg sync.WaitGroup - - threshold := slot - phase0.Slot(retain) - - // async perform initial slot gc - _ = stores.Each(func(_ spectypes.BeaconRole, store ibftstorage.ParticipantStore) error { - wg.Add(1) - go func() { - defer wg.Done() - store.Prune(ctx, threshold) - }() - return nil - }) - - wg.Wait() - - // start background job for removing old slots on every tick - _ = stores.Each(func(_ spectypes.BeaconRole, store ibftstorage.ParticipantStore) error { - go store.PruneContinuously(ctx, slotTickerProvider, phase0.Slot(retain)) - return nil - }) -} - // buildDoppelganger returns the node's doppelganger-protection provider: a no-op for exporter nodes // (and for operator nodes with protection disabled), or a real handler when protection is enabled. func buildDoppelganger( diff --git a/cli/operator/node_test.go b/cli/operator/node_test.go index b8f8078862..56ff118797 100644 --- a/cli/operator/node_test.go +++ b/cli/operator/node_test.go @@ -12,7 +12,6 @@ import ( "github.com/ssvlabs/ssv/doppelganger" "github.com/ssvlabs/ssv/eth/executionclient" - "github.com/ssvlabs/ssv/exporter" "github.com/ssvlabs/ssv/hprobe" "github.com/ssvlabs/ssv/network" "github.com/ssvlabs/ssv/networkconfig" @@ -126,19 +125,15 @@ func Test_newNode_wiresOperatorNode(t *testing.T) { } } -// Test_newNode_wiresExporterNode mirrors the operator smoke test for the exporter paths: with no -// signing identity, it asserts newNode() wires the graph for both exporter modes and that the -// mode-specific divergences hold — no key manager in either, and a duty-trace collector only in -// archive mode. +// Test_newNode_wiresExporterNode mirrors the operator smoke test for the exporter path: with no +// signing identity, it asserts newNode() wires the graph for an exporter node — no key manager, +// and a duty-trace collector (exporters always run full duty tracing). func Test_newNode_wiresExporterNode(t *testing.T) { for _, tc := range []struct { - name string - mode nodeMode - exporterMode string - wantCollector bool + name string + mode nodeMode }{ - {name: "standard", mode: modeExporterStandard, exporterMode: exporter.ModeStandard, wantCollector: false}, - {name: "archive", mode: modeExporterArchive, exporterMode: exporter.ModeArchive, wantCollector: true}, + {name: "exporter", mode: modeExporter}, } { t.Run(tc.name, func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) @@ -152,7 +147,6 @@ func Test_newNode_wiresExporterNode(t *testing.T) { cfg := &config{} cfg.DBOptions.Path = t.TempDir() cfg.ExporterOptions.Enabled = true - cfg.ExporterOptions.Mode = tc.exporterMode cfg.MetricsAPIPort = 0 cfg.SSVAPIPort = 0 cfg.WsAPIPort = 0 @@ -170,11 +164,7 @@ func Test_newNode_wiresExporterNode(t *testing.T) { require.NotNil(t, a.operatorNode) require.Nil(t, a.keyManager, "exporter nodes have no key manager") - if tc.wantCollector { - require.NotNil(t, a.collector, "archive mode wires a duty-trace collector") - } else { - require.Nil(t, a.collector, "standard mode has no duty-trace collector") - } + require.NotNil(t, a.collector, "exporter wires a duty-trace collector") require.NoError(t, a.Close()) }) diff --git a/exporter/opts.go b/exporter/opts.go index e34e2a12cb..c1d53e0112 100644 --- a/exporter/opts.go +++ b/exporter/opts.go @@ -1,12 +1,6 @@ package exporter type Options struct { - Enabled bool `yaml:"Enabled" env:"EXPORTER" env-default:"false" env-description:"Enable exporter mode to track post-consensus participations"` - Mode string `yaml:"Mode" env:"EXPORTER_MODE" env-default:"standard" env-description:"Set to 'archive' to also track pre-consensus and consensus steps. Defaults to 'standard'"` - RetainSlots uint64 `yaml:"RetainSlots" env:"EXPORTER_RETAIN_SLOTS" env-default:"50400" env-description:"Number of slots to retain in export data"` + Enabled bool `yaml:"Enabled" env:"EXPORTER" env-default:"false" env-description:"Enable exporter mode to track validator duties and network consensus participation (full duty tracing)"` + RetainEpochs uint64 `yaml:"RetainEpochs" env:"EXPORTER_RETAIN_EPOCHS" env-default:"0" env-description:"Best-effort retention: prune on-disk duty traces older than this many epochs. 0 (default) retains indefinitely. Enforced forward from process start — history that ages out while the node is down is not reclaimed after a restart"` } - -const ( - ModeArchive = "archive" - ModeStandard = "standard" -) diff --git a/exporter/store/store.go b/exporter/store/store.go index a5e7410449..fd465031e5 100644 --- a/exporter/store/store.go +++ b/exporter/store/store.go @@ -354,6 +354,27 @@ func (s *DutyTraceStore) DeleteScheduledSlot(slot phase0.Slot) error { return errs.ErrorOrNil() } +// PruneSlot removes all duty-trace data for a single slot: validator duties (all roles/indices), +// committee duties (all committees), validator->committee links, and scheduled-role bitmaps. It +// backs the exporter's optional retention window (EXPORTER_RETAIN_EPOCHS). Keys are slot-prefixed, +// so each group is removed with a single prefix drop. +func (s *DutyTraceStore) PruneSlot(slot phase0.Slot) error { + var errs *multierror.Error + if err := s.db.DropPrefix(s.makeValidatorSlotPrefix(slot)); err != nil { + errs = multierror.Append(errs, fmt.Errorf("prune validator duties (slot=%d): %w", slot, err)) + } + if err := s.db.DropPrefix(s.makeCommitteeSlotPrefix(slot)); err != nil { + errs = multierror.Append(errs, fmt.Errorf("prune committee duties (slot=%d): %w", slot, err)) + } + if err := s.db.DropPrefix(s.makeValidatorCommitteePrefix(slot)); err != nil { + errs = multierror.Append(errs, fmt.Errorf("prune validator-committee links (slot=%d): %w", slot, err)) + } + if err := s.db.DropPrefix(s.makeScheduledSlotPrefix(slot)); err != nil { + errs = multierror.Append(errs, fmt.Errorf("prune scheduled duties (slot=%d): %w", slot, err)) + } + return errs.ErrorOrNil() +} + // SaveScheduled stores a compact map (validator index -> role mask) for a slot. func (s *DutyTraceStore) SaveScheduled(slot phase0.Slot, schedule map[phase0.ValidatorIndex]rolemask.Mask) error { if len(schedule) == 0 { @@ -431,6 +452,14 @@ func (s *DutyTraceStore) makeValidatorPrefix(slot phase0.Slot, role spectypes.Be return prefix } +// makeValidatorSlotPrefix returns the "vd"+slot prefix covering all roles and validator indices +// for a slot (used to drop an entire slot's validator duties at once during pruning). +func (s *DutyTraceStore) makeValidatorSlotPrefix(slot phase0.Slot) []byte { + prefix := make([]byte, 0, len(validatorDutyTraceKey)+slotKeyLen) + prefix = append(prefix, []byte(validatorDutyTraceKey)...) + return append(prefix, slotToByteSlice(slot)...) +} + func (s *DutyTraceStore) makeCommitteeSlotPrefix(slot phase0.Slot) []byte { prefix := make([]byte, 0, len(committeeDutyTraceKey)+4) prefix = append(prefix, []byte(committeeDutyTraceKey)...) @@ -452,6 +481,14 @@ func (s *DutyTraceStore) makeValidatorCommitteePrefix(slot phase0.Slot) []byte { return append(prefix, slotToByteSlice(slot)...) } +// makeScheduledSlotPrefix returns the "sd"+slot prefix covering every role's scheduled bitmap for a +// slot, so a slot's scheduled data is dropped in a single prefix delete during pruning. +func (s *DutyTraceStore) makeScheduledSlotPrefix(slot phase0.Slot) []byte { + prefix := make([]byte, 0, len(scheduledDutyKey)+slotKeyLen) + prefix = append(prefix, []byte(scheduledDutyKey)...) + return append(prefix, slotToByteSlice(slot)...) +} + func (s *DutyTraceStore) makeScheduledRolePrefix(slot phase0.Slot, role spectypes.BeaconRole) []byte { prefix := make([]byte, 0, len(scheduledDutyKey)+slotKeyLen+1) prefix = append(prefix, []byte(scheduledDutyKey)...) diff --git a/exporter/store/store_test.go b/exporter/store/store_test.go index abcf5bf11b..d4fc8ce93e 100644 --- a/exporter/store/store_test.go +++ b/exporter/store/store_test.go @@ -229,6 +229,87 @@ func TestSaveScheduledDuties(t *testing.T) { assert.ElementsMatch(t, []phase0.ValidatorIndex{2}, syncers) } +func TestPruneSlot(t *testing.T) { + logger := zap.NewNop() + db, err := kv.NewInMemory(logger, basedb.Options{}) + require.NoError(t, err) + defer db.Close() + + s := store.New(db) + + const prune = phase0.Slot(1) + const keep = phase0.Slot(2) + + // Validator duties span multiple roles/indices so we prove the whole "vd"+slot keyspace is + // dropped, not just the attester example. + roles := []spectypes.BeaconRole{spectypes.BNRoleAttester, spectypes.BNRoleProposer, spectypes.BNRoleSyncCommittee} + + // Populate every trace kind at both the slot to prune and an adjacent slot to keep. + for _, slot := range []phase0.Slot{prune, keep} { + for i, role := range roles { + require.NoError(t, s.SaveValidatorDuty(&exporter.ValidatorDutyTrace{ + Slot: slot, Role: role, Validator: phase0.ValidatorIndex(100 + i), + })) + } + require.NoError(t, s.SaveCommitteeDuty(makeCTrace(slot, 'a'))) + require.NoError(t, s.SaveCommitteeDuty(makeCTrace(slot, 'b'))) + require.NoError(t, s.SaveCommitteeDutyLinks(slot, map[phase0.ValidatorIndex]spectypes.CommitteeID{1: {1, 1, 1}})) + require.NoError(t, s.SaveScheduled(slot, map[phase0.ValidatorIndex]rolemask.Mask{1: rolemask.BitAttester | rolemask.BitProposer})) + } + + require.NoError(t, s.PruneSlot(prune)) + + // Pruned slot: every trace kind is gone, across all validator roles. + for _, role := range roles { + vds, err := s.GetValidatorDuties(role, prune) + require.NoError(t, err) + require.Empty(t, vds, "validator duties for role %v should be pruned", role) + } + + cds, err := s.GetCommitteeDuties(prune) + require.NoError(t, err) + require.Empty(t, cds, "committee duties should be pruned") + + links, err := s.GetCommitteeDutyLinks(prune) + require.NoError(t, err) + require.Empty(t, links, "committee links should be pruned") + + sched, err := s.GetScheduled(prune) + require.NoError(t, err) + require.Empty(t, sched, "scheduled duties should be pruned") + + // Adjacent slot is untouched, across all validator roles. + for _, role := range roles { + vds, err := s.GetValidatorDuties(role, keep) + require.NoError(t, err) + require.Len(t, vds, 1, "validator duties for role %v should be retained", role) + } + + cds, err = s.GetCommitteeDuties(keep) + require.NoError(t, err) + require.Len(t, cds, 2) + + links, err = s.GetCommitteeDutyLinks(keep) + require.NoError(t, err) + require.Len(t, links, 1) + + sched, err = s.GetScheduled(keep) + require.NoError(t, err) + require.Len(t, sched, 1) +} + +// TestPruneSlot_DBError exercises PruneSlot's error aggregation: a closed DB makes every prefix +// drop fail, and PruneSlot must surface a (combined) error rather than silently succeeding. +func TestPruneSlot_DBError(t *testing.T) { + logger := zap.NewNop() + db, err := kv.NewInMemory(logger, basedb.Options{}) + require.NoError(t, err) + require.NoError(t, db.Close()) // subsequent DropPrefix/Delete calls now fail + + s := store.New(db) + require.Error(t, s.PruneSlot(1)) +} + func TestAddScheduledRole_UnionsIndices(t *testing.T) { logger := zap.NewNop() db, err := kv.NewInMemory(logger, basedb.Options{}) diff --git a/operator/duties/scheduler.go b/operator/duties/scheduler.go index efb8da04cc..d32d4cfc35 100644 --- a/operator/duties/scheduler.go +++ b/operator/duties/scheduler.go @@ -95,10 +95,6 @@ type SchedulerOptions struct { // executing duties (e.g., validator registration). When true, scheduler // still fetches/stores duties for all validators but does not execute them. ExporterMode bool - // ArchiveMode indicates this is an archive-mode exporter that needs the Attester handler - // for duty tracing. Only meaningful when ExporterMode is true; standard-mode exporters skip - // the Attester handler to avoid fetching duties for all network validators. - ArchiveMode bool } type Scheduler struct { @@ -169,27 +165,21 @@ func NewScheduler(logger *zap.Logger, opts *SchedulerOptions) *Scheduler { s.exporterMode = opts.ExporterMode - // Proposer, SyncCommittee, and VoluntaryExit are needed in all modes. - // Proposer and SyncCommittee populate dutyStore entries consulted by message validation on all - // node types. VoluntaryExit populates the per-(slot,pk) duty count also consulted by validation. - // In exporter mode, exit descriptors always have OwnValidator=false (no owned validators), so - // the handler never queues exits for execution; VoluntaryExitHandler.processExecution also - // guards explicitly against execution in exporter mode as a second line of defense. + // Attester, Proposer, SyncCommittee, and VoluntaryExit are needed in all modes (operator and + // exporter). Proposer and SyncCommittee populate dutyStore entries consulted by message + // validation on all node types; VoluntaryExit populates the per-(slot,pk) duty count also + // consulted by validation; Attester is needed for duty execution (operator) and full duty + // tracing (exporter). In exporter mode, exit descriptors always have OwnValidator=false (no + // owned validators), so the handler never queues exits for execution; VoluntaryExitHandler. + // processExecution also guards explicitly against execution in exporter mode as a second line + // of defense. s.dutyHandlers = append(s.dutyHandlers, NewProposerHandler(dutyStore.Proposer, opts.ExporterMode), NewSyncCommitteeHandler(dutyStore.SyncCommittee, opts.ExporterMode), NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh, opts.ExporterMode), + NewAttesterHandler(dutyStore.Attester, opts.ExporterMode), ) - // Attester is needed for duty execution (operator) and duty tracing (archive exporter). - // Standard-mode exporter skips it: the attester store is not checked by message validation, - // and fetching duties for all network validators is expensive without the tracing benefit. - if !opts.ExporterMode || opts.ArchiveMode { - s.dutyHandlers = append(s.dutyHandlers, - NewAttesterHandler(dutyStore.Attester, opts.ExporterMode), - ) - } - // These handlers only execute duties and are not needed in exporter mode. if !opts.ExporterMode { s.dutyHandlers = append(s.dutyHandlers, diff --git a/operator/duties/scheduler_test.go b/operator/duties/scheduler_test.go index e5ef28d6af..5e2bf4cd37 100644 --- a/operator/duties/scheduler_test.go +++ b/operator/duties/scheduler_test.go @@ -700,29 +700,19 @@ func TestNewScheduler_HandlerRegistration(t *testing.T) { tests := []struct { name string exporterMode bool - archiveMode bool wantHandlers []string }{ { name: "operator mode", exporterMode: false, - archiveMode: false, wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT", "ATTESTER", "CLUSTER", "VALIDATOR_REGISTRATION"}, }, { - name: "standard exporter", + name: "exporter mode", exporterMode: true, - archiveMode: false, - // VoluntaryExit is included so the dutyStore is populated for message validation; - // Attester is excluded because the standard exporter does not trace duties. - wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT"}, - }, - { - name: "archive exporter", - exporterMode: true, - archiveMode: true, - // Archive exporter needs Attester for full duty tracing but still skips - // execution-only handlers (CLUSTER, VALIDATOR_REGISTRATION). + // Exporters run full duty tracing: Proposer/SyncCommittee/VoluntaryExit populate the + // stores message validation consults, and Attester is needed for duty tracing. Only the + // execution-only handlers (CLUSTER, VALIDATOR_REGISTRATION) are skipped. wantHandlers: []string{"PROPOSER", "SYNC_COMMITTEE", "VOLUNTARY_EXIT", "ATTESTER"}, }, } @@ -736,7 +726,6 @@ func TestNewScheduler_HandlerRegistration(t *testing.T) { s := NewScheduler(zap.NewNop(), &SchedulerOptions{ ExporterMode: tc.exporterMode, - ArchiveMode: tc.archiveMode, SlotTickerProvider: func() slotticker.SlotTicker { return NewMockSlotTicker(ctx) }, diff --git a/operator/dutytracer/collector.go b/operator/dutytracer/collector.go index 73d9a48a8e..abe70a106b 100644 --- a/operator/dutytracer/collector.go +++ b/operator/dutytracer/collector.go @@ -73,6 +73,10 @@ type Collector struct { // scheduleJobs is a bounded queue for async schedule computation to avoid // blocking the hot path and DB with synchronous writes. scheduleJobs chan phase0.Slot + + // retention enforces the optional duty-trace retention window (EXPORTER_RETAIN_EPOCHS) and is + // the single source of truth for the retained-slot floor consulted by the collection paths. + retention *retentionState } type DomainDataProvider interface { @@ -106,6 +110,7 @@ func New( duties: duties, scheduleJobs: make(chan phase0.Slot, 32), } + collector.retention = &retentionState{store: store, logger: collector.logger} return collector } @@ -116,13 +121,17 @@ type scRootKey struct { blockRoot phase0.Root } -func (c *Collector) Start(ctx context.Context, tickerProvider slotticker.Provider) { - c.logger.Info("start duty tracer cache to disk evictor") +// Start runs the duty-tracer background loop: it evicts in-memory traces to disk each slot and, +// when retainSlots > 0, prunes on-disk trace history older than the retention window. retainSlots +// of 0 disables pruning (retain indefinitely — the default). +func (c *Collector) Start(ctx context.Context, tickerProvider slotticker.Provider, retainSlots uint64) { + c.logger.Info("start duty tracer cache to disk evictor", zap.Uint64("retain_slots", retainSlots)) ticker := tickerProvider() // Start schedule filler in a separate goroutine to avoid blocking eviction. go c.startScheduleFiller(ctx, tickerProvider) // Start a single worker to process schedule writes asynchronously. go c.runScheduleWorker(ctx) + for { select { case <-ctx.Done(): @@ -130,7 +139,74 @@ func (c *Collector) Start(ctx context.Context, tickerProvider slotticker.Provide case <-ticker.Next(): currentSlot := ticker.Slot() c.evict(currentSlot) + c.retention.advance(currentSlot, phase0.Slot(retainSlots)) + } + } +} + +// maxPrunePerTick bounds the slots a single tick may prune, so a large jump in the window floor +// (e.g. after a processing pause) cannot stall the eviction loop; the remainder is reclaimed on +// subsequent ticks. +const maxPrunePerTick = 64 + +// retentionState is the single source of truth for the duty-trace retention window. The eviction +// loop advances it once per slot; the message-collection and schedule-write paths consult expired() +// so they never (re)create trace data for a slot that has already aged out of the window — without +// it, a late message or schedule job could resurrect a slot just deleted from disk. +type retentionState struct { + store DutyTraceStore + logger *zap.Logger + + // floor is the lowest slot still inside the retention window; slots strictly below it are + // expired. 0 means retention is disabled (retain indefinitely — the default). Read concurrently + // from collection goroutines, written only from the eviction loop. + floor atomic.Uint64 + + // cursor is the next slot to delete from disk; meaningful once seeded. Accessed only from the + // single eviction-loop goroutine. + cursor phase0.Slot + seeded bool +} + +// expired reports whether slot has aged out of the retention window and must not be (re)written. +func (r *retentionState) expired(slot phase0.Slot) bool { + floor := r.floor.Load() + return floor > 0 && uint64(slot) < floor +} + +// advance moves the retention window to currentSlot and deletes a bounded batch of newly-expired +// slots from disk. Safeguards: +// - retainSlots == 0 disables retention (retain indefinitely — the default); +// - the currentSlot <= retainSlots guard avoids unsigned slot underflow early in the chain's life; +// - on first activation the cursor seeds at the window floor, so enabling retention does not +// trigger an unbounded backfill of the chain's entire pre-existing history; +// - per-tick work is capped (maxPrunePerTick) so a large floor jump cannot stall eviction; +// - the cursor lets a coalesced/late tick reclaim every skipped slot rather than leaking it. +// +// The cursor is in-memory, so retention is best-effort and forward-only across restarts: slots that +// expire while the node is down are not reclaimed afterwards (see EXPORTER_RETAIN_EPOCHS docs). +// Trace reads never depend on pruned data. +func (r *retentionState) advance(currentSlot, retainSlots phase0.Slot) { + if retainSlots == 0 || currentSlot <= retainSlots { + return + } + floor := currentSlot - retainSlots // slots strictly below floor are expired + r.floor.Store(uint64(floor)) + if !r.seeded { + r.cursor, r.seeded = floor, true + return + } + + var pruned int + for ; r.cursor < floor && pruned < maxPrunePerTick; pruned++ { + if err := r.store.PruneSlot(r.cursor); err != nil { + r.logger.Warn("failed to prune expired duty traces", fields.Slot(r.cursor), zap.Error(err)) } + r.cursor++ + } + if pruned > 0 { + r.logger.Debug("pruned expired duty traces", + zap.Uint64("through_slot", uint64(r.cursor-1)), zap.Int("count", pruned)) } } @@ -164,6 +240,11 @@ func (c *Collector) evict(currentSlot phase0.Slot) { } func (c *Collector) getOrCreateValidatorTrace(slot phase0.Slot, role spectypes.BeaconRole, index phase0.ValidatorIndex) (*validatorDutyTrace, bool, error) { + // Drop traces for slots already pruned out of the retention window so a late message cannot + // resurrect them on disk. + if c.retention.expired(slot) { + return nil, false, errExpiredSlot + } // check late arrival if uint64(slot) <= c.lastEvictedSlot.Load() { if _, found := c.inFlightValidator.GetOrSet(index, struct{}{}); found { @@ -217,7 +298,17 @@ func (c *Collector) getOrCreateValidatorTrace(slot phase0.Slot, role spectypes.B var errInFlight = errors.New("in flight") +// errExpiredSlot signals that a message is for a slot already outside the retention window. The +// collection path drops such messages silently so retention can't be defeated by a late write +// resurrecting a slot that was just pruned from disk. +var errExpiredSlot = errors.New("slot outside retention window") + func (c *Collector) getOrCreateCommitteeTrace(slot phase0.Slot, committeeID spectypes.CommitteeID) (*committeeDutyTrace, bool, error) { + // Drop traces for slots already pruned out of the retention window so a late message cannot + // resurrect them on disk. + if c.retention.expired(slot) { + return nil, false, errExpiredSlot + } // check late arrival if uint64(slot) <= c.lastEvictedSlot.Load() { if _, found := c.inFlightCommittee.GetOrSet(committeeID, struct{}{}); found { @@ -523,6 +614,9 @@ func (c *Collector) Collect(ctx context.Context, msg *queue.SSVMessage, verifySi go c.collectLateMessage(ctx, msg, verifySig) return nil } + if errors.Is(err, errExpiredSlot) { + return nil // message is for a slot outside the retention window; drop silently + } return err } @@ -543,6 +637,10 @@ func (c *Collector) collectLateMessage(ctx context.Context, msg *queue.SSVMessag // if another late message is in flight (for the same ID) - try `maxRetryCount` times before giving up for tries < maxRetryCount { err = c.collect(ctx, msg, verifySig) + if errors.Is(err, errExpiredSlot) { + err = nil // slot aged out of the retention window while retrying; drop silently + return + } if !errors.Is(err, errInFlight) { return } @@ -1396,9 +1494,18 @@ func (c *Collector) runScheduleWorker(ctx context.Context) { case <-ctx.Done(): return case s := <-c.scheduleJobs: - if err := c.computeAndPersistScheduleForSlot(s); err != nil { - c.logger.Debug("schedule worker compute/persist", fields.Slot(s), zap.Error(err)) - } + c.processScheduleJob(s) } } } + +// processScheduleJob persists the per-slot scheduled-role map, unless the slot has already aged out +// of the retention window — recreating its scheduled data would defeat pruning. +func (c *Collector) processScheduleJob(s phase0.Slot) { + if c.retention.expired(s) { + return // slot outside the retention window; don't recreate its scheduled data + } + if err := c.computeAndPersistScheduleForSlot(s); err != nil { + c.logger.Debug("schedule worker compute/persist", fields.Slot(s), zap.Error(err)) + } +} diff --git a/operator/dutytracer/collector_test.go b/operator/dutytracer/collector_test.go index 373c9fb3bf..141e6827c7 100644 --- a/operator/dutytracer/collector_test.go +++ b/operator/dutytracer/collector_test.go @@ -23,6 +23,7 @@ import ( "github.com/ssvlabs/ssv/exporter/rolemask" "github.com/ssvlabs/ssv/exporter/store" "github.com/ssvlabs/ssv/networkconfig" + "github.com/ssvlabs/ssv/operator/slotticker" "github.com/ssvlabs/ssv/protocol/v2/ssv/queue" ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types" "github.com/ssvlabs/ssv/registry/storage" @@ -1817,12 +1818,187 @@ type mockDutyTraceStore struct { committeeDutyTrace *exporter.CommitteeDutyTrace saveCommitteeDutyLinkFn func(slot phase0.Slot, index phase0.ValidatorIndex, id spectypes.CommitteeID) error scheduled map[phase0.Slot]map[phase0.ValidatorIndex]rolemask.Mask + prunedSlots []phase0.Slot } func (m *mockDutyTraceStore) SaveCommitteeDuties(slot phase0.Slot, duties []*exporter.CommitteeDutyTrace) error { return m.err } +func (m *mockDutyTraceStore) PruneSlot(slot phase0.Slot) error { + m.prunedSlots = append(m.prunedSlots, slot) + return m.err +} + +func TestRetentionState(t *testing.T) { + const retain = phase0.Slot(100) + + t.Run("disabled when retainSlots is zero", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(1_000_000, 0) + require.Empty(t, st.prunedSlots) + require.False(t, r.seeded) + require.False(t, r.expired(0), "nothing is expired when retention is disabled") + }) + + t.Run("no underflow before the window fills", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(50, retain) // currentSlot < retain + require.Empty(t, st.prunedSlots) + require.False(t, r.seeded) + require.False(t, r.expired(0)) + }) + + t.Run("first activation seeds the floor forward without backfilling old history", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(1_000_000, retain) // floor is huge; must NOT prune from slot 0 + require.Empty(t, st.prunedSlots) + require.True(t, r.seeded) + require.Equal(t, phase0.Slot(1_000_000-100), r.cursor) + // floor guards the collection path immediately, even before the cursor prunes anything + require.True(t, r.expired(1_000_000-101)) + require.False(t, r.expired(1_000_000-100)) + }) + + t.Run("prunes exactly the slot that ages out each advancing tick", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(150, retain) // seed: floor/cursor=50 + r.advance(151, retain) // prune 50 + r.advance(152, retain) // prune 51 + require.Equal(t, []phase0.Slot{50, 51}, st.prunedSlots) + require.Equal(t, phase0.Slot(52), r.cursor) + }) + + t.Run("catches up skipped slots but caps work per tick", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(150, retain) // seed: floor/cursor=50 + st.prunedSlots = nil + r.advance(150+1000, retain) // floor jumps to 1050; must cap, not stall + require.Len(t, st.prunedSlots, maxPrunePerTick) + require.Equal(t, phase0.Slot(50), st.prunedSlots[0]) + require.Equal(t, phase0.Slot(50+maxPrunePerTick), r.cursor) + }) + + t.Run("expired tracks the current floor", func(t *testing.T) { + st := &mockDutyTraceStore{} + r := &retentionState{store: st, logger: zap.NewNop()} + r.advance(150, retain) // floor=50 + require.True(t, r.expired(49)) + require.False(t, r.expired(50)) + require.False(t, r.expired(51)) + }) +} + +// TestCollector_dropsExpiredSlot verifies the collection hot path refuses to (re)create traces for +// a slot below the retention floor, so a late message cannot resurrect a slot already pruned. +func TestCollector_dropsExpiredSlot(t *testing.T) { + c := New(zap.NewNop(), nil, nil, &mockDutyTraceStore{}, networkconfig.TestNetwork.Beacon, nil, nil) + c.retention.floor.Store(100) // slots < 100 are expired + + _, _, err := c.getOrCreateValidatorTrace(50, spectypes.BNRoleAttester, 1) + require.ErrorIs(t, err, errExpiredSlot) + + _, _, err = c.getOrCreateCommitteeTrace(50, spectypes.CommitteeID{}) + require.ErrorIs(t, err, errExpiredSlot) + + // A slot inside the window is not dropped by the guard. + _, _, err = c.getOrCreateValidatorTrace(150, spectypes.BNRoleAttester, 1) + require.NotErrorIs(t, err, errExpiredSlot) +} + +// fakeSlotTicker fires a single pre-loaded tick and reports a fixed slot. +type fakeSlotTicker struct { + ch chan time.Time + slot phase0.Slot +} + +func (f *fakeSlotTicker) Next() <-chan time.Time { return f.ch } +func (f *fakeSlotTicker) Slot() phase0.Slot { return f.slot } + +// TestCollector_Start_advancesRetention drives the eviction loop one tick and asserts it advances +// the retention window (sets the floor). Only the main loop's ticker fires; the schedule filler's +// ticker stays silent so the worker doesn't run with the test's empty deps. +func TestCollector_Start_advancesRetention(t *testing.T) { + c := New(zap.NewNop(), nil, nil, &mockDutyTraceStore{}, networkconfig.TestNetwork.Beacon, nil, nil) + + var calls int + provider := func() slotticker.SlotTicker { + calls++ + ch := make(chan time.Time, 1) + if calls == 1 { // the main eviction loop is the first to ask for a ticker + ch <- time.Now() + } + return &fakeSlotTicker{ch: ch, slot: 1_000_000} + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { c.Start(ctx, provider, 100); close(done) }() + + require.Eventually(t, func() bool { return c.retention.floor.Load() > 0 }, + 2*time.Second, 5*time.Millisecond, "Start should advance the retention floor on a tick") + require.Equal(t, uint64(1_000_000-100), c.retention.floor.Load()) + + cancel() + <-done +} + +// TestCollector_processScheduleJob_skipsExpired verifies the schedule worker drops jobs for slots +// outside the retention window (so they cannot recreate pruned scheduled data). With nil duties, +// only the guard's early return keeps computeAndPersistScheduleForSlot from being reached. +func TestCollector_processScheduleJob_skipsExpired(t *testing.T) { + c := New(zap.NewNop(), nil, nil, &mockDutyTraceStore{}, networkconfig.TestNetwork.Beacon, nil, nil) + c.retention.floor.Store(100) + require.NotPanics(t, func() { c.processScheduleJob(50) }, "expired slot must be skipped before persist") +} + +// TestCollector_Collect_dropsExpiredSlot drives a real message for a slot below the retention floor +// through Collect end-to-end and asserts it is dropped silently — proving a late message cannot +// resurrect a pruned slot. +func TestCollector_Collect_dropsExpiredSlot(t *testing.T) { + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + + const slot = phase0.Slot(1) + vIndex := phase0.ValidatorIndex(55) + identifier := spectypes.NewMsgID([4]byte{}, []byte("pk"), spectypes.RoleAggregator) + + validators := registrystoragemocks.NewMockValidatorStore(ctrl) + validators.EXPECT().ValidatorIndex(gomock.Any()).Return(vIndex, true).AnyTimes() + + c := New(zap.NewNop(), validators, nil, &mockDutyTraceStore{}, networkconfig.TestNetwork.Beacon, nil, nil) + c.retention.floor.Store(100) // slot 1 is far outside the retention window + + psm := spectypes.PartialSignatureMessages{ + Type: spectypes.PostConsensusPartialSig, + Slot: slot, + Messages: []*spectypes.PartialSignatureMessage{ + {ValidatorIndex: vIndex, Signer: 99, PartialSignature: make([]byte, 96), SigningRoot: [32]byte{1}}, + }, + } + data, err := psm.Encode() + require.NoError(t, err) + + require.NoError(t, c.Collect(t.Context(), buildPartialSigMessage(identifier, data), dummyVerify)) +} + +// TestDutyTraceStoreMetrics_PruneSlot covers the metrics wrapper used in production (node.go wraps +// the store in DutyTraceStoreMetrics). +func TestDutyTraceStoreMetrics_PruneSlot(t *testing.T) { + db, err := kv.NewInMemory(zap.NewNop(), basedb.Options{}) + require.NoError(t, err) + defer db.Close() + + m := &DutyTraceStoreMetrics{Store: store.New(db)} + require.NoError(t, m.PruneSlot(123)) +} + func (m *mockDutyTraceStore) SaveCommitteeDutyLink(slot phase0.Slot, index phase0.ValidatorIndex, id spectypes.CommitteeID) error { if m.saveCommitteeDutyLinkFn != nil { return m.saveCommitteeDutyLinkFn(slot, index, id) diff --git a/operator/dutytracer/store.go b/operator/dutytracer/store.go index 35bb5f1029..7499b0057b 100644 --- a/operator/dutytracer/store.go +++ b/operator/dutytracer/store.go @@ -42,6 +42,9 @@ type DutyTraceStore interface { // Compact scheduled duties I/O SaveScheduled(slot phase0.Slot, schedule map[phase0.ValidatorIndex]rolemask.Mask) error GetScheduled(slot phase0.Slot) (map[phase0.ValidatorIndex]rolemask.Mask, error) + + // PruneSlot removes all trace data for a slot, backing the optional retention window. + PruneSlot(slot phase0.Slot) error } func (c *Collector) GetCommitteeID(slot phase0.Slot, index phase0.ValidatorIndex) (spectypes.CommitteeID, error) { diff --git a/operator/dutytracer/store_metrics.go b/operator/dutytracer/store_metrics.go index 2af831a1d8..f03a8e862e 100644 --- a/operator/dutytracer/store_metrics.go +++ b/operator/dutytracer/store_metrics.go @@ -47,6 +47,14 @@ func (d *DutyTraceStoreMetrics) SaveCommitteeDutyLink(slot phase0.Slot, index ph return d.Store.SaveCommitteeDutyLink(slot, index, id) } +func (d *DutyTraceStoreMetrics) PruneSlot(slot phase0.Slot) error { + start := time.Now() + defer func() { + record("prune", "prune_slot", start) + }() + return d.Store.PruneSlot(slot) +} + func (d *DutyTraceStoreMetrics) GetCommitteeDutyLink(slot phase0.Slot, index phase0.ValidatorIndex) (spectypes.CommitteeID, error) { start := time.Now() defer func() { diff --git a/operator/node.go b/operator/node.go index d785d26fa8..0f6832fda2 100644 --- a/operator/node.go +++ b/operator/node.go @@ -126,7 +126,6 @@ func New(logger *zap.Logger, opts Options, exporterOpts exporter.Options, slotTi SlotTickerProvider: slotTickerProvider, P2PNetwork: opts.P2PNetwork, ExporterMode: exporterOpts.Enabled, - ArchiveMode: exporterOpts.Mode == exporter.ModeArchive, }) node := &Node{ diff --git a/operator/validator/controller.go b/operator/validator/controller.go index 170bb927ee..537f7e9c5c 100644 --- a/operator/validator/controller.go +++ b/operator/validator/controller.go @@ -388,12 +388,13 @@ func (c *Controller) handleWorkerMessages(ctx context.Context, msg network.Decod ncv = item.Value() } - if c.validatorCommonOpts.ExporterOptions.Mode == exporter.ModeArchive { - // use new exporter functionality + if c.traceCollector != nil { + // Exporter nodes run full duty tracing via the collector. The collector is wired only for + // exporter nodes (see cli/operator/node.go), so a non-nil collector means this is an exporter. return c.traceCollector.Collect(c.ctx, ssvMsg, ncv.VerifySig) } - // use old exporter functionality + // Operator nodes record post-consensus participation via the legacy participant store. return c.handleNonCommitteeMessages(ctx, ssvMsg, ncv) }