diff --git a/protocol/v2/ssv/runner/aggregator_committee.go b/protocol/v2/ssv/runner/aggregator_committee.go index 1d6a34499f..75c0bac288 100644 --- a/protocol/v2/ssv/runner/aggregator_committee.go +++ b/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 { diff --git a/protocol/v2/ssv/runner/aggregator_committee_test.go b/protocol/v2/ssv/runner/aggregator_committee_test.go index 818beb5a60..1edbaf5ea3 100644 --- a/protocol/v2/ssv/runner/aggregator_committee_test.go +++ b/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. diff --git a/protocol/v2/ssv/runner/committee.go b/protocol/v2/ssv/runner/committee.go index 47f26d8f50..6026ce025f 100644 --- a/protocol/v2/ssv/runner/committee.go +++ b/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,6 +1064,7 @@ 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 } @@ -1020,12 +1072,14 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context 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,6 +1101,7 @@ 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 @@ -1054,6 +1109,7 @@ func (r *CommitteeRunner) expectedPostConsensusRootsAndBeaconObjects(ctx context 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 } diff --git a/protocol/v2/ssv/runner/committee_postconsensus_classification_test.go b/protocol/v2/ssv/runner/committee_postconsensus_classification_test.go index f3464ecee6..1c0f54c26d 100644 --- a/protocol/v2/ssv/runner/committee_postconsensus_classification_test.go +++ b/protocol/v2/ssv/runner/committee_postconsensus_classification_test.go @@ -4,9 +4,11 @@ import ( "bytes" "context" "errors" + "sync/atomic" "testing" "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/go-eth2-client/spec/phase0" spectypes "github.com/ssvlabs/ssv-spec/types" spectestingutils "github.com/ssvlabs/ssv-spec/types/testingutils" "github.com/stretchr/testify/require" @@ -122,3 +124,170 @@ func TestCommitteeRunnerProcessPostConsensus_RecoverableInvalidSigsThenSucceeds( t.Fatal("expected a succeeded duty conclusion after recovery, got none") } } + +// invalidateDutiesInGuard marks every validator duty of the committee duty invalid in the guard +// stub, so expectedPostConsensusRootsAndBeaconObjects (and the ProcessConsensus signing loop) skips +// them all. +func invalidateDutiesInGuard(guard *committeeDutyGuardStub, duty *spectypes.CommitteeDuty) { + guard.validErrs = make(map[string]error) + for _, vd := range duty.ValidatorDuties { + key := guard.validKey(vd.Type, spectypes.ValidatorPK(vd.PubKey), vd.DutySlot()) + guard.validErrs[key] = errors.New("duty no longer valid") + } +} + +// TestCommitteeRunnerProcessPostConsensus_MarksNotRequiredOnNoBeaconObjects is the regression test +// for #2903: a post-consensus quorum where this operator ends up with no beacon objects to submit +// (e.g. divergent validator sets across the committee's operators — modeled here by invalidating +// the duties in the guard after consensus) is a benign terminal. It must conclude not_required — +// not failed (the previous behavior, surfacing as a spurious "⚠️ duty failed") and not a silent +// stall — while still surfacing the sentinel for the queue's terminal-drop handling. +func TestCommitteeRunnerProcessPostConsensus_MarksNotRequiredOnNoBeaconObjects(t *testing.T) { + guard := &committeeDutyGuardStub{} + env := newCommitteeRunnerEnv(t, []int{1}, guard, &doppelgangerStub{}) + duty := spectestingutils.TestingCommitteeDuty([]int{1}, nil, spec.DataVersionElectra) + + env.startAndDecideCommitteeDuty(t, duty) + concluded := observeConclusion(env) + + invalidateDutiesInGuard(guard, duty) + + var postConsensusErr error + for id := spectypes.OperatorID(1); id <= 3; id++ { + msg := spectestingutils.PostConsensusCommitteeMsgForDuty(duty, env.keySetMap, id) + if err := env.runner.ProcessPostConsensus(context.Background(), env.logger, msg); err != nil { + postConsensusErr = err + } + } + + require.ErrorIs(t, postConsensusErr, ErrNoValidDutiesToExecute, "the benign sentinel must surface to the queue") + + select { + case c := <-concluded: + require.Equal(t, dutyOutcomeNotRequired, c.outcome, "no beacon objects to submit must conclude not_required, not failed") + require.NoError(t, c.reason) + default: + t.Fatal("expected a not_required duty conclusion, got none") + } + require.True(t, env.runner.State.Succeeded, "not_required is a correct completion") + require.Empty(t, env.beacon.GetBroadcastedRoots(), "nothing should have been submitted") +} + +// faultyDomainDataBeacon wraps the testing beacon node with a switchable DomainData failure, so a +// test can let consensus-phase signing succeed and then fail every post-consensus beacon-object +// construction. All other behavior is inherited via the embedded *BeaconNodeWrapped. +type faultyDomainDataBeacon struct { + *protocoltesting.BeaconNodeWrapped + domainErr error + fail atomic.Bool +} + +func (b *faultyDomainDataBeacon) DomainData(ctx context.Context, epoch phase0.Epoch, domain phase0.DomainType) (phase0.Domain, error) { + if b.fail.Load() { + return phase0.Domain{}, b.domainErr + } + return b.BeaconNodeWrapped.DomainData(ctx, epoch, domain) +} + +// TestCommitteeRunnerProcessPostConsensus_MarksFailedOnAllConstructionFailures guards the boundary of +// the not_required classification: an empty beacon-objects map caused by every per-validator +// construction failing (modeled by DomainData failing after consensus) is a MISSED submission, not a +// benign no-op — expectedPostConsensusRootsAndBeaconObjects must surface the error so the duty +// concludes failed instead of not_required. +func TestCommitteeRunnerProcessPostConsensus_MarksFailedOnAllConstructionFailures(t *testing.T) { + base := protocoltesting.NewTestingBeaconNodeWrapped().(*protocoltesting.BeaconNodeWrapped) + domainErr := errors.New("domain data unavailable") + faulty := &faultyDomainDataBeacon{BeaconNodeWrapped: base, domainErr: domainErr} + + env := newCommitteeRunnerEnvWithBeacon(t, []int{1}, faulty) + duty := spectestingutils.TestingCommitteeDuty([]int{1}, nil, spec.DataVersionElectra) + + env.startAndDecideCommitteeDuty(t, duty) + concluded := observeConclusion(env) + + // Consensus-phase signing has already fetched domain data successfully; from here on every + // post-consensus object construction fails, emptying the beacon-objects map for a duty this + // operator WAS supposed to submit. + faulty.fail.Store(true) + + var postConsensusErr error + for id := spectypes.OperatorID(1); id <= 3; id++ { + msg := spectestingutils.PostConsensusCommitteeMsgForDuty(duty, env.keySetMap, id) + if err := env.runner.ProcessPostConsensus(context.Background(), env.logger, msg); err != nil { + postConsensusErr = err + } + } + + require.ErrorIs(t, postConsensusErr, domainErr, "the construction failure must surface, not be swallowed into an empty map") + require.NotErrorIs(t, postConsensusErr, ErrNoValidDutiesToExecute, "an all-construction-failure empty map must not be classified as the benign sentinel") + + select { + case c := <-concluded: + require.Equal(t, dutyOutcomeFailed, c.outcome, "a missed submission must conclude failed, not not_required") + require.ErrorIs(t, c.reason, domainErr) + default: + t.Fatal("expected a failed duty conclusion, got none") + } + require.False(t, env.runner.State.Succeeded, "a missed submission must not be marked succeeded") +} + +// TestCommitteeRunnerProcessConsensus_MarksNotRequiredOnNoValidDuties covers the consensus-phase +// sibling of the #2903 sentinel: a committee that decides while this operator has zero valid duties +// to sign (all invalidated in the guard before consensus) previously concluded via no marker at +// all, surfacing as a false "stuck". It must conclude not_required and surface the sentinel. +func TestCommitteeRunnerProcessConsensus_MarksNotRequiredOnNoValidDuties(t *testing.T) { + guard := &committeeDutyGuardStub{} + env := newCommitteeRunnerEnv(t, []int{1}, guard, &doppelgangerStub{}) + duty := spectestingutils.TestingCommitteeDuty([]int{1}, nil, spec.DataVersionElectra) + + ctx := t.Context() + require.NoError(t, env.runner.StartNewDuty(ctx, env.logger, duty, env.sampleKey.Threshold)) + concluded := observeConclusion(env) + + invalidateDutiesInGuard(guard, duty) + + var consensusErr error + for _, msg := range spectestingutils.CommitteeInputForDuty(duty, duty.Slot, env.keySetMap, false) { + if err := env.runner.ProcessConsensus(ctx, env.logger, msg); err != nil { + consensusErr = err + } + } + + require.ErrorIs(t, consensusErr, ErrNoValidDutiesToExecute, "the benign sentinel must surface to the queue") + + select { + case c := <-concluded: + require.Equal(t, dutyOutcomeNotRequired, c.outcome, "deciding with zero valid duties must conclude not_required") + require.NoError(t, c.reason) + default: + t.Fatal("expected a not_required duty conclusion, got none") + } + require.True(t, env.runner.State.Succeeded, "not_required is a correct completion") +} + +// TestCommitteeRunnerProcessConsensus_CancelledContextDoesNotConcludeNotRequired guards the +// consensus-phase zero-duties branch against shutdown: a canceled context also reaches it with zero +// counts (the duty feeder and workers bail out before counting), but an abandoned duty must not be +// recorded as a not_required completion — cancellation is never an outcome, mirroring +// markDutyFailed's context.Canceled filter. +func TestCommitteeRunnerProcessConsensus_CancelledContextDoesNotConcludeNotRequired(t *testing.T) { + env := newCommitteeRunnerEnv(t, []int{1}, &committeeDutyGuardStub{}, &doppelgangerStub{}) + duty := spectestingutils.TestingCommitteeDuty([]int{1}, nil, spec.DataVersionElectra) + + require.NoError(t, env.runner.StartNewDuty(t.Context(), env.logger, duty, env.sampleKey.Threshold)) + concluded := observeConclusion(env) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + var consensusErr error + for _, msg := range spectestingutils.CommitteeInputForDuty(duty, duty.Slot, env.keySetMap, false) { + if err := env.runner.ProcessConsensus(ctx, env.logger, msg); err != nil { + consensusErr = err + } + } + + require.ErrorIs(t, consensusErr, context.Canceled, "shutdown must surface the cancellation, not the benign sentinel") + require.Empty(t, concluded, "an abandoned duty must not conclude any outcome") + require.False(t, env.runner.State.Succeeded, "an abandoned duty is not a completion") +} diff --git a/protocol/v2/ssv/validator/committee_queue.go b/protocol/v2/ssv/validator/committee_queue.go index 8b217ea34b..19759037cd 100644 --- a/protocol/v2/ssv/validator/committee_queue.go +++ b/protocol/v2/ssv/validator/committee_queue.go @@ -228,13 +228,19 @@ func (c *Committee) ConsumeQueue( const couldNotHandleMsgLogPrefix = "could not handle message, " switch { case errors.Is(err, runner.ErrNoValidDutiesToExecute): - const droppingMsgDueToNoValidDutiesToExecuteEvent = "❗ " + couldNotHandleMsgLogPrefix + "dropping message and terminating committee-runner" - msgLogger.Error(droppingMsgDueToNoValidDutiesToExecuteEvent, zap.Error(err)) + // Terminal, not a handling failure: the committee decided but this operator has no + // duties to execute, so the message is dropped and the runner terminated without + // error-level noise. The runner concluded the duty outcome before returning the + // sentinel (not_required for the benign zero-duties cases; failed for + // AggregatorCommitteeRunner's empty-decided-data invariant violation, which also + // warns via the outcome watcher), so nothing is lost by the quiet drop here. + const droppingMsgDueToNoValidDutiesToExecuteEvent = "no valid duties to execute, dropping message and terminating committee-runner" + msgLogger.Debug(droppingMsgDueToNoValidDutiesToExecuteEvent, zap.Error(err)) msgState.span.AddEvent(droppingMsgDueToNoValidDutiesToExecuteEvent, trace.WithAttributes( attribute.String("drop_reason", err.Error()), attribute.Int64("attempt", currentAttempt), )) - msgState.span.SetStatus(codes.Error, droppingMsgDueToNoValidDutiesToExecuteEvent) + msgState.span.SetStatus(codes.Ok, "") msgState.span.End() msgStates.Delete(msgKey) return