fix(runner): conclude benign ErrNoValidDutiesToExecute as not_required, not failed

protocol/v2/ssv/runner/aggregator_committee.go

@@ -887,11 +887,18 @@ func (r *AggregatorCommitteeRunner) ProcessPostConsensus(
 		return fmt.Errorf("could not get expected post consensus roots and beacon objects: %w", err)
 	}
 	if len(beaconObjects) == 0 {
-		// Empty post-quorum (all beacon objects failed to build) is terminal and non-recoverable:
-		// committee_queue drops the message and terminates the runner on this error. Classify as
-		// failed (matching CommitteeRunner) rather than leaving the watcher to report a false stuck.
-		r.markDutyFailed(ErrNoValidDutiesToExecute)
-		return ErrNoValidDutiesToExecute
+		// NOT the benign terminal it is in the sibling CommitteeRunner. There, beaconObjects is built
+		// from LOCAL state, so divergent validator sets legitimately empty it on one operator. Here it
+		// is built purely from the DECIDED value with every error surfaced above, so an empty map means
+		// the decided data had zero aggregators and zero contributors — a value
+		// AggregatorCommitteeConsensusData.Validate() rejects and validateDecidedConsensusData enforces
+		// before it is ever stored as decided. Reaching this branch is an invariant violation (bypassed
+		// or regressed decided-value validation), a consensus-integrity signal that must stay loud:
+		// classify as failed. The sentinel is preserved in the chain so committee_queue still drops the
+		// message and terminates the runner.
+		err := fmt.Errorf("no beacon objects from decided data, decided-value validation should have rejected it: %w", ErrNoValidDutiesToExecute)
+		r.markDutyFailed(err)
+		return err
 	}
 
 	sort.Slice(roots, func(i, j int) bool {

protocol/v2/ssv/runner/aggregator_committee_test.go

@@ -172,6 +172,48 @@ func TestAggregatorCommitteeRunnerProcessPostConsensus_MarksFailedOnSubmitError(
 	require.False(t, env.runner.State.Succeeded, "a failed duty must not be marked succeeded")
 }
 
+// TestAggregatorCommitteeRunnerProcessPostConsensus_MarksFailedOnNoBeaconObjects pins the
+// empty-objects terminal as an invariant violation, NOT a benign no-op: beaconObjects is built purely
+// from the decided value (every construction error surfaced), and a decided value with zero
+// aggregators and zero contributors is rejected by AggregatorCommitteeConsensusData.Validate() before
+// it can ever be stored as decided. So reaching the branch means decided-value validation was
+// bypassed or regressed — it must conclude failed (loud, via the outcome watcher's warn) while still
+// carrying the sentinel for the queue's terminal-drop handling. The test necessarily hand-installs
+// State.DecidedValue to force the condition, because the normal flow cannot produce it.
+func TestAggregatorCommitteeRunnerProcessPostConsensus_MarksFailedOnNoBeaconObjects(t *testing.T) {
+	ctx := t.Context()
+	const version = spec.DataVersionElectra
+
+	base := protocoltesting.NewTestingBeaconNodeWrapped().(*protocoltesting.BeaconNodeWrapped)
+	env := newAggregatorCommitteeRunnerEnv(t, []int{1}, base)
+	duty := spectestingutils.TestingAggregatorCommitteeDutyForValidators([]int{1}, []int{}, version)
+
+	concluded := env.startAndFeedThroughConsensus(t, ctx, duty, version)
+
+	emptyDecided := &spectypes.AggregatorCommitteeConsensusData{Version: version}
+	encoded, err := emptyDecided.Encode()
+	require.NoError(t, err)
+	env.runner.State.DecidedValue = encoded
+
+	var postConsensusErr error
+	for _, psig := range postConsensusMsgsFromFixture(duty, env.keySetMap, version) {
+		if err := env.runner.ProcessPostConsensus(ctx, env.logger, psig); err != nil {
+			postConsensusErr = err
+		}
+	}
+
+	require.ErrorIs(t, postConsensusErr, ErrNoValidDutiesToExecute, "the sentinel must surface so the queue drops the message and terminates the runner")
+
+	select {
+	case c := <-concluded:
+		require.Equal(t, dutyOutcomeFailed, c.outcome, "empty beacon objects from decided data is an invariant violation and must conclude failed")
+		require.ErrorIs(t, c.reason, ErrNoValidDutiesToExecute)
+	default:
+		t.Fatal("expected a failed duty conclusion, got none")
+	}
+	require.False(t, env.runner.State.Succeeded, "an invariant violation must not be marked succeeded")
+}
+
 // TestAggregatorCommitteeRunnerProcessPostConsensus_DoesNotMarkFailedOnInvalidSigs asserts that the
 // recoverable reconstruct-invalid-signatures case is NOT concluded failed: the root can later re-cross
 // quorum on a subsequent message, so concluding here would mask a duty that still completes.

protocol/v2/ssv/runner/committee.go

@@ -379,6 +379,26 @@ listener:
 	)
 
 	if totalAttestations == 0 && totalSyncCommittee == 0 {
+		// A canceled context also lands here with zero counts: the duty feeder and the workers bail
+		// out on ctx.Err() before incrementing any counter. That is shutdown — the duty was abandoned,
+		// not completed-with-nothing-to-do — so return without concluding an outcome, mirroring
+		// markDutyFailed's context.Canceled filter (cancellation is never an outcome).
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		// Benign terminal: the committee decided but this operator ended up with zero valid duties to
+		// sign. Conclude as not_required so the watcher doesn't report a false "stuck"; the sentinel
+		// still tells committee_queue to drop the message and terminate the runner.
+		r.markDutyNotRequired()
+		r.measurements.EndDutyFlow()
+		recordTotalDutyDuration(ctx, r.measurements.TotalDutyTime(), spectypes.RoleCommittee, r.State.RunningInstance.State.Round)
+		const dutyFinishedNoValidDutiesEvent = "✔️successfully finished duty processing (no valid duties to sign)"
+		logger.Info(dutyFinishedNoValidDutiesEvent,
+			fields.ConsensusTime(r.measurements.ConsensusTime()),
+			fields.ConsensusRounds(uint64(r.State.RunningInstance.State.Round)),
+			fields.TotalDutyTime(r.measurements.TotalDutyTime()),
+		)
+		span.AddEvent(dutyFinishedNoValidDutiesEvent)
 		return ErrNoValidDutiesToExecute
 	}
 
@@ -514,6 +534,9 @@ func (r *CommitteeRunner) ProcessPostConsensus(ctx context.Context, logger *zap.
 	// are tagged recoverableReconstructError and must not be recorded as failed.
 	// Shutdown (context cancellation) needs no special-casing — markDutyFailed drops a context.Canceled
 	// reason, so a submission aborted by shutdown isn't recorded as a failure.
+	// The benign no-beacon-objects sentinel (ErrNoValidDutiesToExecute) pre-concludes the duty as
+	// not_required before returning, which makes this deferred markDutyFailed a no-op (concludeDuty
+	// is idempotent) — it must not be recorded as failed either.
 	defer func() {
 		if err != nil && !isRecoverableReconstructError(err) {
 			r.markDutyFailed(err)
@@ -529,6 +552,24 @@ func (r *CommitteeRunner) ProcessPostConsensus(ctx context.Context, logger *zap.
 		return fmt.Errorf("could not get expected post consensus roots and beacon objects: %w", err)
 	}
 	if len(beaconObjects) == 0 {
+		// Benign terminal: the committee reached consensus but this operator has no beacon objects to
+		// submit (divergent validator sets across the committee's operators — every duty was skipped
+		// as guard-invalid). An empty map here is guaranteed benign: an all-construction-failure empty
+		// result is surfaced as an error by expectedPostConsensusRootsAndBeaconObjects above and
+		// classified failed by the defer. Conclude as not_required before returning the sentinel;
+		// concludeDuty is idempotent, so the deferred markDutyFailed becomes a no-op. The sentinel
+		// still tells committee_queue to drop the message and terminate the runner.
+		r.markDutyNotRequired()
+		r.measurements.EndDutyFlow()
+		recordTotalDutyDuration(ctx, r.measurements.TotalDutyTime(), spectypes.RoleCommittee, r.State.RunningInstance.State.Round)
+		const dutyFinishedNoBeaconObjectsEvent = "✔️successfully finished duty processing (no beacon objects to submit)"
+		logger.Info(dutyFinishedNoBeaconObjectsEvent,
+			fields.ConsensusTime(r.measurements.ConsensusTime()),
+			fields.ConsensusRounds(uint64(r.State.RunningInstance.State.Round)),
+			fields.PostConsensusTime(r.measurements.PostConsensusTime()),
+			fields.TotalDutyTime(r.measurements.TotalDutyTime()),
+		)
+		span.AddEvent(dutyFinishedNoBeaconObjectsEvent)
 		return ErrNoValidDutiesToExecute
 	}
 
@@ -995,6 +1036,16 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context
 	epoch := r.NetworkConfig.EstimatedEpochAtSlot(slot)
 	dataVersion, _ := r.NetworkConfig.ForkAtEpoch(epoch)
 
+	// Skips fall into two classes: guard invalidations are benign (the #2903 divergent-validator-sets
+	// case — the duty is genuinely not this operator's to submit), while construction / domain-data /
+	// signing-root failures mean a submission was missed. The distinction only matters when NOTHING
+	// could be built: partial failures keep the per-validator debug-and-continue behavior so one
+	// validator's failure never blocks the others' submissions, but an all-failure empty result must
+	// surface as an error — otherwise the caller's len(beaconObjects)==0 branch would conclude the
+	// duty not_required, masking the miss (the sibling AggregatorCommitteeRunner surfaces these
+	// errors for the same reason).
+	var constructionErr error
+
 	for _, validatorDuty := range committeeDuty.ValidatorDuties {
 		if validatorDuty == nil {
 			continue
@@ -1013,19 +1064,22 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context
 			attestationResponse, err := specssv.ConstructVersionedAttestationWithoutSignature(attestationData, dataVersion, validatorDuty)
 			if err != nil {
 				logger.Debug("failed to construct attestation", zap.Error(err))
+				constructionErr = errors.Join(constructionErr, fmt.Errorf("construct attestation (validator %d): %w", validatorDuty.ValidatorIndex, err))
 				continue
 			}
 
 			// Root
 			domain, err := r.GetBeaconNode().DomainData(ctx, epoch, spectypes.DomainAttester)
 			if err != nil {
 				logger.Debug("failed to get attester domain", zap.Error(err))
+				constructionErr = errors.Join(constructionErr, fmt.Errorf("get attester domain (validator %d): %w", validatorDuty.ValidatorIndex, err))
 				continue
 			}
 
 			root, err := spectypes.ComputeETHSigningRoot(attestationData, domain)
 			if err != nil {
 				logger.Debug("failed to compute attester root", zap.Error(err))
+				constructionErr = errors.Join(constructionErr, fmt.Errorf("compute attester root (validator %d): %w", validatorDuty.ValidatorIndex, err))
 				continue
 			}
 
@@ -1047,13 +1101,15 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context
 			domain, err := r.GetBeaconNode().DomainData(ctx, epoch, spectypes.DomainSyncCommittee)
 			if err != nil {
 				logger.Debug("failed to get sync committee domain", zap.Error(err))
+				constructionErr = errors.Join(constructionErr, fmt.Errorf("get sync committee domain (validator %d): %w", validatorDuty.ValidatorIndex, err))
 				continue
 			}
 			// Eth root
 			blockRoot := spectypes.SSZBytes(beaconVote.BlockRoot[:])
 			root, err := spectypes.ComputeETHSigningRoot(blockRoot, domain)
 			if err != nil {
 				logger.Debug("failed to compute sync committee root", zap.Error(err))
+				constructionErr = errors.Join(constructionErr, fmt.Errorf("compute sync committee root (validator %d): %w", validatorDuty.ValidatorIndex, err))
 				continue
 			}
 
@@ -1067,6 +1123,9 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context
 			return nil, nil, nil, fmt.Errorf("invalid duty type: %s", validatorDuty.Type)
 		}
 	}
+	if len(beaconObjects) == 0 && constructionErr != nil {
+		return nil, nil, nil, fmt.Errorf("no beacon objects could be built: %w", constructionErr)
+	}
 	return attestationMap, syncCommitteeMap, beaconObjects, nil
 }