fix(exporter): populate dutyStore in all modes — stop blackholing exit/proposer/sync-contribution messages

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,32 @@ 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, 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,
-		NewAttesterHandler(dutyStore.Attester, opts.ExporterMode),
 		NewProposerHandler(dutyStore.Proposer, opts.ExporterMode),
 		NewSyncCommitteeHandler(dutyStore.SyncCommittee, opts.ExporterMode),
+		NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh, 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,
 			NewCommitteeHandler(dutyStore.Attester, dutyStore.SyncCommittee),
 			NewValidatorRegistrationHandler(opts.ValidatorRegistrationCh),
-			NewVoluntaryExitHandler(dutyStore.VoluntaryExit, opts.ValidatorExitCh),
 		)
 	}
 	return s

operator/duties/scheduler_test.go

@@ -693,3 +693,80 @@ 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)
+				},
+			})
+
+			got := make([]string, 0, len(s.dutyHandlers))
+			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")
+				}
+			}
+		})
+	}
+}

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

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"
 )
 
@@ -32,7 +34,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 +172,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)
 
@@ -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
 

operator/node.go

@@ -78,10 +78,6 @@ type Node struct {
 	exporterRead *exporter2.Exporter
 }
 
-func shouldRunDutyScheduler(exporterOpts exporter.Options) bool {
-	return !exporterOpts.Enabled || exporterOpts.Mode == exporter.ModeArchive
-}
-
 // 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,38 +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 archive mode we 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,
-		})
+	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,
@@ -174,13 +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)
-		}
-	} else {
-		n.logger.Info("exporter standard mode: skipping duty scheduler")
+	if err := n.dutyScheduler.Start(ctx); err != nil {
+		return fmt.Errorf("failed to run duty scheduler: %w", err)
 	}
 
 	n.validatorsCtrl.StartNetworkHandlers()
@@ -248,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.