From 230d598986876100c7dc0b1c28f091c58277d56f Mon Sep 17 00:00:00 2001 From: Jin Wang Date: Mon, 13 Jul 2026 15:55:58 -0700 Subject: [PATCH] refactor(flow): Track current operation run phase - Track current operation run phase index in operation run table so that the dipatcher needs to lock the targets in the current phase only. --- .../flow/internal/converter/dao/converter.go | 2 + ...operation_run_current_phase_index.down.sql | 6 + ...0_operation_run_current_phase_index.up.sql | 21 ++ .../flow/internal/db/model/operation_run.go | 1 + .../manager/dispatcher/decision.go | 36 ++-- .../manager/dispatcher/dependencies.go | 3 +- .../manager/dispatcher/dispatcher_test.go | 59 +++++- .../operationrun/manager/dispatcher/phase.go | 35 ++-- .../manager/dispatcher/preparation.go | 27 ++- .../manager/dispatcher/reconciliation.go | 58 +----- .../operationrun/manager/manager_test.go | 38 +++- .../operationrun/manager/manual_controls.go | 45 ++++- .../internal/operationrun/manager/store.go | 10 +- .../operationrun/manager/store/dispatch.go | 69 ++++++- .../operationrun/manager/store/store.go | 11 +- .../operationrun/manager/store/store_test.go | 19 +- .../internal/operationrun/operationrun.go | 1 + .../flow/internal/operationrun/progress.go | 148 +++++--------- .../internal/operationrun/progress_test.go | 185 ++++++------------ rest-api/flow/internal/operationrun/stats.go | 34 +++- 20 files changed, 454 insertions(+), 354 deletions(-) create mode 100644 rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.down.sql create mode 100644 rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.up.sql diff --git a/rest-api/flow/internal/converter/dao/converter.go b/rest-api/flow/internal/converter/dao/converter.go index d70d652428..aadf37de92 100644 --- a/rest-api/flow/internal/converter/dao/converter.go +++ b/rest-api/flow/internal/converter/dao/converter.go @@ -337,6 +337,7 @@ func OperationRunFrom(dao *model.OperationRun) *operationrun.OperationRun { Status: dao.Status, StatusReason: dao.StatusReason, StatusMessage: dao.StatusMessage, + CurrentPhaseIndex: dao.CurrentPhaseIndex, Selector: dao.Selector, Options: dao.Options, OperationTemplate: dao.OperationTemplate, @@ -362,6 +363,7 @@ func OperationRunTo(run *operationrun.OperationRun) *model.OperationRun { Status: run.Status, StatusReason: run.StatusReason, StatusMessage: run.StatusMessage, + CurrentPhaseIndex: run.CurrentPhaseIndex, Selector: run.Selector, Options: run.Options, OperationTemplate: run.OperationTemplate, diff --git a/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.down.sql b/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.down.sql new file mode 100644 index 0000000000..31c39994a7 --- /dev/null +++ b/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.down.sql @@ -0,0 +1,6 @@ +-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +-- SPDX-License-Identifier: Apache-2.0 + +ALTER TABLE operation_run + DROP COLUMN IF EXISTS current_phase_index; + diff --git a/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.up.sql b/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.up.sql new file mode 100644 index 0000000000..ea66980309 --- /dev/null +++ b/rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.up.sql @@ -0,0 +1,21 @@ +-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +-- SPDX-License-Identifier: Apache-2.0 + +ALTER TABLE operation_run + ADD COLUMN current_phase_index INTEGER NOT NULL DEFAULT 0 CHECK (current_phase_index >= 0); + +UPDATE operation_run AS orun +SET current_phase_index = COALESCE( + ( + SELECT MIN(ort.phase_index) + FROM operation_run_target AS ort + WHERE ort.operation_run_id = orun.id + AND ort.status NOT IN ('completed', 'failed', 'terminated', 'skipped') + ), + ( + SELECT MAX(ort.phase_index) + FROM operation_run_target AS ort + WHERE ort.operation_run_id = orun.id + ), + 0 +); diff --git a/rest-api/flow/internal/db/model/operation_run.go b/rest-api/flow/internal/db/model/operation_run.go index 39b66a3913..e4ad8b9c72 100644 --- a/rest-api/flow/internal/db/model/operation_run.go +++ b/rest-api/flow/internal/db/model/operation_run.go @@ -25,6 +25,7 @@ type OperationRun struct { Status operationrun.OperationRunStatus `bun:"status,type:varchar(32),notnull"` //nolint:lll StatusReason operationrun.OperationRunStatusReason `bun:"status_reason,type:varchar(64),notnull"` //nolint:lll StatusMessage string `bun:"status_message,nullzero"` + CurrentPhaseIndex int32 `bun:"current_phase_index,notnull"` Selector json.RawMessage `bun:"selector,type:jsonb,notnull"` Options json.RawMessage `bun:"options,type:jsonb,notnull"` OperationTemplate json.RawMessage `bun:"operation_template,type:jsonb,notnull"` diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/decision.go b/rest-api/flow/internal/operationrun/manager/dispatcher/decision.go index a6fab10e44..edc6e4de04 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/decision.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/decision.go @@ -139,7 +139,7 @@ func (d *Dispatcher) decide( return newStopDecision(dispatchRunTransition{}), nil } - if prep.summary.TargetCount == 0 { + if prep.summary.SelectedTargetCount() == 0 { return newStopDecision(failRunTransition("operation run has no targets")), nil } @@ -176,7 +176,7 @@ func (d *Dispatcher) decide( } safetyDecision := prep.safetyPolicy.evaluate( - prep.summary.TargetPhaseSummary, + prep.summary, ) if safetyDecision.pause { return newStopDecision( @@ -185,20 +185,30 @@ func (d *Dispatcher) decide( } phaseDecision := prep.phasePolicy.evaluate( - prep.summary.TargetPhaseSummary, - prep.summary.previousPhaseTerminalChanged(), + prep.summary, ) - if phaseDecision.pause { + switch phaseDecision.action { + case phaseDecisionActionAdvance: + // The next phase's targets were not locked during this transaction. + // Persist the phase pointer first; the next pass will lock and submit them. + prep.run.CurrentPhaseIndex++ + return newStopDecision(startRunTransition(phaseDecision.message)), nil + case phaseDecisionActionPause: return newStopDecision( pauseRunTransition(phaseDecision.reason, phaseDecision.message), ), nil + case phaseDecisionActionClaim: + return newClaimDecision( + startRunTransition(phaseDecision.message), + prep.options, + prep.op, + prep.conflictPolicy, + targets, + ), nil + default: + return dispatchDecision{}, fmt.Errorf( + "unknown phase decision action %d", + phaseDecision.action, + ) } - - return newClaimDecision( - startRunTransition(phaseDecision.message), - prep.options, - prep.op, - prep.conflictPolicy, - targets, - ), nil } diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/dependencies.go b/rest-api/flow/internal/operationrun/manager/dispatcher/dependencies.go index 5f9e7233bb..e265185a75 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/dependencies.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/dependencies.go @@ -19,7 +19,8 @@ type Store interface { RunInTransaction(ctx context.Context, fn func(ctx context.Context) error) error FetchRunnableIDs(ctx context.Context, limit int) ([]uuid.UUID, error) LockRunnable(ctx context.Context, id uuid.UUID) (*operationrun.OperationRun, error) - LockOperationRunTargets(ctx context.Context, runID uuid.UUID) ([]*operationrun.OperationRunTarget, error) + LockOperationRunTargets(ctx context.Context, runID uuid.UUID, phaseIndex int32) ([]*operationrun.OperationRunTarget, error) + GetTargetPhaseAggregate(ctx context.Context, runID uuid.UUID, currentPhaseIndex int32) (operationrun.TargetPhaseAggregate, error) UpdateRunState(ctx context.Context, run *operationrun.OperationRun) error UpdateTargetState(ctx context.Context, target *operationrun.OperationRunTarget) error } diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/dispatcher_test.go b/rest-api/flow/internal/operationrun/manager/dispatcher/dispatcher_test.go index 50f3155ddf..7b6043df94 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/dispatcher_test.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/dispatcher_test.go @@ -224,7 +224,7 @@ func TestDispatchRunReconcilesCompletedTaskAndCompletesRun(t *testing.T) { require.Equal(t, now, *store.run.FinishedAt) } -func TestDispatchRunFailsWhenTerminalChangesSpanMultiplePhases(t *testing.T) { +func TestDispatchRunLocksOnlyCurrentPhaseTargets(t *testing.T) { runID := uuid.New() firstTaskID := uuid.New() secondTaskID := uuid.New() @@ -261,8 +261,11 @@ func TestDispatchRunFailsWhenTerminalChangesSpanMultiplePhases(t *testing.T) { }, now) err := dispatcher.dispatchRun(context.Background(), runID) - require.ErrorContains(t, err, "terminal target changes span multiple phases") - require.Equal(t, operationrun.OperationRunStatusRunning, store.run.Status) + require.NoError(t, err) + require.Equal(t, operationrun.OperationRunTargetStatusCompleted, first.Status) + require.Equal(t, operationrun.OperationRunTargetStatusSubmitted, second.Status) + require.Equal(t, operationrun.OperationRunStatusPaused, store.run.Status) + require.Equal(t, operationrun.OperationRunStatusReasonPhaseGate, store.run.StatusReason) } func TestDispatchRunPausesWhenSafetyGateTrips(t *testing.T) { @@ -624,13 +627,11 @@ func TestDecideSetsTerminalStatusFromTargetOutcomes(t *testing.T) { run := &operationrun.OperationRun{ Status: operationrun.OperationRunStatusRunning, } - summary := newReconciliationSummary() - summary.CurrentPhaseIndex = -1 - summary.TargetCount = len(tt.statuses) + summary := operationrun.TargetPhaseSummary{} + summary.TotalPhases = 1 + summary.CurrentPhaseStats.SelectedTargets = len(tt.statuses) for _, status := range tt.statuses { - if status.IsFailedOrTerminated() { - summary.FailedOrTerminatedTargetCount++ - } + summary.CurrentPhaseStats.StatusCounts.Add(status) } decision, err := (&Dispatcher{}).decide(&preparedDispatch{ @@ -696,8 +697,37 @@ func (s *fakeStore) LockRunnable( func (s *fakeStore) LockOperationRunTargets( ctx context.Context, runID uuid.UUID, + phaseIndex int32, ) ([]*operationrun.OperationRunTarget, error) { - return s.targets, nil + targets := make([]*operationrun.OperationRunTarget, 0) + for _, target := range s.targets { + if target.PhaseIndex == phaseIndex { + targets = append(targets, target) + } + } + return targets, nil +} + +func (s *fakeStore) GetTargetPhaseAggregate( + ctx context.Context, + runID uuid.UUID, + currentPhaseIndex int32, +) (operationrun.TargetPhaseAggregate, error) { + var totalPhases int32 + completedStats := operationrun.PhaseStats{PhaseIndex: max(currentPhaseIndex-1, 0)} + for _, target := range s.targets { + if target.PhaseIndex+1 > totalPhases { + totalPhases = target.PhaseIndex + 1 + } + if target.PhaseIndex < currentPhaseIndex { + completedStats.AddTarget(target) + } + } + + return operationrun.TargetPhaseAggregate{ + TotalPhases: totalPhases, + CompletedPhaseStats: completedStats, + }, nil } func (s *fakeStore) UpdateRunState( @@ -759,10 +789,19 @@ func (s *dispatchOnceStore) LockRunnable( func (s *dispatchOnceStore) LockOperationRunTargets( ctx context.Context, runID uuid.UUID, + phaseIndex int32, ) ([]*operationrun.OperationRunTarget, error) { return nil, nil } +func (s *dispatchOnceStore) GetTargetPhaseAggregate( + ctx context.Context, + runID uuid.UUID, + currentPhaseIndex int32, +) (operationrun.TargetPhaseAggregate, error) { + return operationrun.TargetPhaseAggregate{}, nil +} + func (s *dispatchOnceStore) UpdateRunState( ctx context.Context, run *operationrun.OperationRun, diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/phase.go b/rest-api/flow/internal/operationrun/manager/dispatcher/phase.go index a0330d23ab..79ba720d16 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/phase.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/phase.go @@ -13,6 +13,20 @@ type phasePolicyRuntime struct { autoAdvance bool } +type phaseDecisionAction int + +const ( + phaseDecisionActionClaim phaseDecisionAction = iota + phaseDecisionActionAdvance + phaseDecisionActionPause +) + +type phaseDecision struct { + action phaseDecisionAction + reason operationrun.OperationRunStatusReason + message string +} + func newPhasePolicy(options *operationrun.Options) (*phasePolicyRuntime, error) { if err := options.PhasePolicy.Validate(); err != nil { return nil, fmt.Errorf("phase policy: %w", err) @@ -25,27 +39,22 @@ func newPhasePolicy(options *operationrun.Options) (*phasePolicyRuntime, error) func (p phasePolicyRuntime) evaluate( summary operationrun.TargetPhaseSummary, - previousPhaseTerminalChanged bool, -) pauseDecision { - if summary.CurrentPhaseIndex == 0 || - !previousPhaseTerminalChanged || - !summary.CurrentPhaseNotStarted() { - // No phase boundary was crossed into a fresh phase, so there is - // nothing for phase policy to pause or report. - return pauseDecision{ - pause: false, +) phaseDecision { + if !summary.CurrentPhaseTerminal() || !summary.HasNextPhase() { + return phaseDecision{ + action: phaseDecisionActionClaim, } } if p.autoAdvance { - return pauseDecision{ - pause: false, + return phaseDecision{ + action: phaseDecisionActionAdvance, message: "advanced to next phase", } } - return pauseDecision{ - pause: true, + return phaseDecision{ + action: phaseDecisionActionPause, reason: operationrun.OperationRunStatusReasonPhaseGate, message: "waiting for phase advance", } diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/preparation.go b/rest-api/flow/internal/operationrun/manager/dispatcher/preparation.go index a6424fbe65..60f603ed46 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/preparation.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/preparation.go @@ -22,7 +22,7 @@ type preparedDispatch struct { safetyPolicy *safetyPolicyRuntime phasePolicy *phasePolicyRuntime prepareErr error - summary reconciliationSummary + summary operationrun.TargetPhaseSummary changed map[uuid.UUID]*operationrun.OperationRunTarget } @@ -82,18 +82,31 @@ func (d *Dispatcher) prepare( return prep, nil } - // TODO: After operation_run.current_phase_index is persisted, lock only the - // current phase targets here and use SQL aggregates for cumulative safety - // gates and final completion instead of loading every target for the run. - targets, err := d.deps.Store.LockOperationRunTargets(ctx, run.ID) + targets, err := d.deps.Store.LockOperationRunTargets( + ctx, + run.ID, + run.CurrentPhaseIndex, + ) if err != nil { return nil, fmt.Errorf("lock operation run targets: %w", err) } - prep.summary, err = d.reconcileTargets(ctx, targets, prep.changed) - if err != nil { + if err := d.reconcileTargets(ctx, targets, prep.changed); err != nil { return nil, err } + aggregate, err := d.deps.Store.GetTargetPhaseAggregate( + ctx, + run.ID, + run.CurrentPhaseIndex, + ) + if err != nil { + return nil, fmt.Errorf("summarize operation run targets: %w", err) + } + prep.summary = operationrun.NewTargetPhaseSummary( + run.CurrentPhaseIndex, + aggregate, + targets, + ) return prep, nil } diff --git a/rest-api/flow/internal/operationrun/manager/dispatcher/reconciliation.go b/rest-api/flow/internal/operationrun/manager/dispatcher/reconciliation.go index 72c865a572..d29a0ccd11 100644 --- a/rest-api/flow/internal/operationrun/manager/dispatcher/reconciliation.go +++ b/rest-api/flow/internal/operationrun/manager/dispatcher/reconciliation.go @@ -12,61 +12,18 @@ import ( operationrun "github.com/NVIDIA/infra-controller/rest-api/flow/internal/operationrun" ) -// reconciliationSummary summarizes the target state observed while reconciling -// child task statuses. -type reconciliationSummary struct { - operationrun.TargetPhaseSummary - terminalChangedPhase int32 -} - -// newReconciliationSummary initializes fields used by the reconciliation -// helpers. -func newReconciliationSummary() reconciliationSummary { - return reconciliationSummary{ - TargetPhaseSummary: operationrun.TargetPhaseSummary{CurrentPhaseIndex: -1}, - terminalChangedPhase: -1, - } -} - -// previousPhaseTerminalChanged reports whether this dispatch moved the -// immediately preceding phase forward to terminal state. -func (s reconciliationSummary) previousPhaseTerminalChanged() bool { - return s.terminalChangedPhase >= 0 && - s.CurrentPhaseIndex > 0 && - s.terminalChangedPhase == s.CurrentPhaseIndex-1 -} - -func (s *reconciliationSummary) recordTerminalChangedPhase(phase int32) error { - if s.terminalChangedPhase < 0 { - s.terminalChangedPhase = phase - return nil - } - - if s.terminalChangedPhase == phase { - return nil - } - - return fmt.Errorf( - "terminal target changes span multiple phases: %d and %d", - s.terminalChangedPhase, - phase, - ) -} - // reconcileTargets copies child task status back into operation-run targets, -// then summarizes the post-reconciliation target state. +// recording any target changes for persistence. func (d *Dispatcher) reconcileTargets( ctx context.Context, targets []*operationrun.OperationRunTarget, changed map[uuid.UUID]*operationrun.OperationRunTarget, -) (reconciliationSummary, error) { - result := newReconciliationSummary() - +) error { for _, target := range targets { if target.TaskID != nil && !target.Status.IsTerminal() { task, err := d.deps.TaskStore.GetTask(ctx, *target.TaskID) if err != nil { - return reconciliationSummary{}, fmt.Errorf( + return fmt.Errorf( "get child task %s: %w", *target.TaskID, err, @@ -80,15 +37,8 @@ func (d *Dispatcher) reconcileTargets( changed[target.ID] = target } - if newStatus.IsTerminal() { - if err := result.recordTerminalChangedPhase(target.PhaseIndex); err != nil { - return reconciliationSummary{}, err - } - } } } - result.TargetPhaseSummary = operationrun.NewTargetPhaseSummary(targets) - - return result, nil + return nil } diff --git a/rest-api/flow/internal/operationrun/manager/manager_test.go b/rest-api/flow/internal/operationrun/manager/manager_test.go index 7559dd0f51..2d8558d505 100644 --- a/rest-api/flow/internal/operationrun/manager/manager_test.go +++ b/rest-api/flow/internal/operationrun/manager/manager_test.go @@ -325,7 +325,7 @@ func TestAdvancePhaseChecksExpectedPhase(t *testing.T) { require.Nil(t, got) require.ErrorIs(t, err, ErrOperationRunInvalidState) - require.ErrorContains(t, err, "expected phase 2, current phase is 1") + require.ErrorContains(t, err, "expected phase 2, next phase is 1") require.Zero(t, store.updateRunCalls) } @@ -343,8 +343,9 @@ func TestAdvancePhaseCompletesAllTerminalRun(t *testing.T) { testOperationRunTarget(runID, 1, operationrun.OperationRunTargetStatusCompleted), }, } + store.lockRun.CurrentPhaseIndex = 1 manager := newTestManager(t, store, planner.New(&mockTargetLookup{}, planner.Config{})) - expectedPhase := int32(1) + expectedPhase := int32(2) got, err := manager.AdvancePhase(context.Background(), runID, &expectedPhase) @@ -376,7 +377,7 @@ func TestAdvancePhaseRejectsInitialPhase(t *testing.T) { require.Nil(t, got) require.ErrorIs(t, err, ErrOperationRunInvalidState) - require.ErrorContains(t, err, "phase 0 is the initial phase") + require.ErrorContains(t, err, "phase 0 is not complete") require.Zero(t, store.updateRunCalls) } @@ -705,10 +706,39 @@ func (m *mockStore) ListTargets( func (m *mockStore) LockOperationRunTargets( ctx context.Context, runID uuid.UUID, + phaseIndex int32, ) ([]*operationrun.OperationRunTarget, error) { m.events = append(m.events, "lock_targets") m.lockTargetsCalls++ - return m.lockedTargets, nil + targets := make([]*operationrun.OperationRunTarget, 0) + for _, target := range m.lockedTargets { + if target.PhaseIndex == phaseIndex { + targets = append(targets, target) + } + } + return targets, nil +} + +func (m *mockStore) GetTargetPhaseAggregate( + ctx context.Context, + runID uuid.UUID, + currentPhaseIndex int32, +) (operationrun.TargetPhaseAggregate, error) { + var totalPhases int32 + completedStats := operationrun.PhaseStats{PhaseIndex: max(currentPhaseIndex-1, 0)} + for _, target := range m.lockedTargets { + if target.PhaseIndex+1 > totalPhases { + totalPhases = target.PhaseIndex + 1 + } + if target.PhaseIndex < currentPhaseIndex { + completedStats.AddTarget(target) + } + } + + return operationrun.TargetPhaseAggregate{ + TotalPhases: totalPhases, + CompletedPhaseStats: completedStats, + }, nil } func (m *mockStore) UpdateRunState( diff --git a/rest-api/flow/internal/operationrun/manager/manual_controls.go b/rest-api/flow/internal/operationrun/manager/manual_controls.go index acaae97c05..f1ed8259d8 100644 --- a/rest-api/flow/internal/operationrun/manager/manual_controls.go +++ b/rest-api/flow/internal/operationrun/manager/manual_controls.go @@ -299,11 +299,12 @@ func (m *ManagerImpl) continueOperationRun( case expectedPhaseIndex == resumeExpectedPhaseIndex: run.Start(now, "operation run resumed") case expectedPhaseIndex >= advanceWithoutExpectedPhaseIndex: + run.CurrentPhaseIndex++ run.Start( now, fmt.Sprintf( "advanced to phase %d", - summary.CurrentPhaseIndex, + run.CurrentPhaseIndex, ), ) default: @@ -380,17 +381,28 @@ func ensurePhaseCanAdvance( summary operationrun.TargetPhaseSummary, expectedPhaseIndex int32, ) error { - if err := summary.CheckExpectedNextPhase(expectedPhaseIndex); err != nil { - return invalidStateError("%w", err) + if !summary.CurrentPhaseTerminal() { + return invalidStateError( + "phase %d is not complete", + summary.CurrentPhaseStats.PhaseIndex, + ) } - if !summary.CurrentPhaseNotStarted() { + if !summary.HasNextPhase() { return invalidStateError( - "phase %d has already started", - summary.CurrentPhaseIndex, + "phase %d is the final phase", + summary.CurrentPhaseStats.PhaseIndex, ) } + nextPhaseIndex := summary.CurrentPhaseStats.PhaseIndex + 1 + if expectedPhaseIndex > 0 && expectedPhaseIndex != nextPhaseIndex { + return invalidStateError( + "expected phase %d, next phase is %d", + expectedPhaseIndex, + nextPhaseIndex, + ) + } return nil } @@ -401,12 +413,29 @@ func (m *ManagerImpl) targetPhaseSummaryForContinue( ) (operationrun.TargetPhaseSummary, error) { emptySummary := operationrun.TargetPhaseSummary{} - targets, err := m.store.LockOperationRunTargets(ctx, run.ID) + targets, err := m.store.LockOperationRunTargets( + ctx, + run.ID, + run.CurrentPhaseIndex, + ) if err != nil { return emptySummary, err } - summary := operationrun.NewTargetPhaseSummary(targets) + aggregate, err := m.store.GetTargetPhaseAggregate( + ctx, + run.ID, + run.CurrentPhaseIndex, + ) + if err != nil { + return emptySummary, err + } + + summary := operationrun.NewTargetPhaseSummary( + run.CurrentPhaseIndex, + aggregate, + targets, + ) if _, ok := summary.TerminalRunStatus(); ok { return summary, nil } diff --git a/rest-api/flow/internal/operationrun/manager/store.go b/rest-api/flow/internal/operationrun/manager/store.go index a3b919afca..b127abed0d 100644 --- a/rest-api/flow/internal/operationrun/manager/store.go +++ b/rest-api/flow/internal/operationrun/manager/store.go @@ -49,12 +49,20 @@ type Store interface { opts operationrun.TargetListOptions, ) ([]*operationrun.OperationRunTarget, int32, error) - // LockOperationRunTargets locks all materialized targets for one run. + // LockOperationRunTargets locks materialized targets for one run and phase. LockOperationRunTargets( ctx context.Context, runID uuid.UUID, + phaseIndex int32, ) ([]*operationrun.OperationRunTarget, error) + // GetTargetPhaseAggregate returns aggregate target facts for phase decisions. + GetTargetPhaseAggregate( + ctx context.Context, + runID uuid.UUID, + currentPhaseIndex int32, + ) (operationrun.TargetPhaseAggregate, error) + // UpdateRunState persists lifecycle fields owned by dispatch/manual // controls. UpdateRunState(ctx context.Context, run *operationrun.OperationRun) error diff --git a/rest-api/flow/internal/operationrun/manager/store/dispatch.go b/rest-api/flow/internal/operationrun/manager/store/dispatch.go index bb7cbcdd9e..b0cbab7e86 100644 --- a/rest-api/flow/internal/operationrun/manager/store/dispatch.go +++ b/rest-api/flow/internal/operationrun/manager/store/dispatch.go @@ -73,15 +73,17 @@ func (s *PostgresStore) LockRunnable( return dao.OperationRunFrom(rows[0]), nil } -// LockOperationRunTargets locks all materialized targets for one run. +// LockOperationRunTargets locks materialized targets for one run and phase. func (s *PostgresStore) LockOperationRunTargets( ctx context.Context, runID uuid.UUID, + phaseIndex int32, ) ([]*operationrun.OperationRunTarget, error) { var rows []dbmodel.OperationRunTarget err := s.idb(ctx).NewSelect(). Model(&rows). Where("ort.operation_run_id = ?", runID). + Where("ort.phase_index = ?", phaseIndex). OrderExpr("ort.phase_index ASC, ort.sequence_index ASC"). For("UPDATE"). Scan(ctx) @@ -96,6 +98,70 @@ func (s *PostgresStore) LockOperationRunTargets( return targets, nil } +// GetTargetPhaseAggregate returns aggregate target facts for phase decisions. +func (s *PostgresStore) GetTargetPhaseAggregate( + ctx context.Context, + runID uuid.UUID, + currentPhaseIndex int32, +) (operationrun.TargetPhaseAggregate, error) { + type aggregateRow struct { + TotalPhases int32 `bun:"total_phases"` + CompletedTargets int `bun:"completed_targets"` + CompletedCompleted int `bun:"completed_completed"` + CompletedFailed int `bun:"completed_failed"` + CompletedTerminated int `bun:"completed_terminated"` + CompletedSkipped int `bun:"completed_skipped"` + } + + var row aggregateRow + err := s.idb(ctx).NewSelect(). + TableExpr("operation_run_target AS ort"). + ColumnExpr("COALESCE(MAX(ort.phase_index) + 1, 0) AS total_phases"). + ColumnExpr( + "COUNT(*) FILTER (WHERE ort.phase_index < ?) AS completed_targets", + currentPhaseIndex, + ). + ColumnExpr( + "COUNT(*) FILTER (WHERE ort.phase_index < ? AND ort.status = ?) AS completed_completed", + currentPhaseIndex, + operationrun.OperationRunTargetStatusCompleted, + ). + ColumnExpr( + "COUNT(*) FILTER (WHERE ort.phase_index < ? AND ort.status = ?) AS completed_failed", + currentPhaseIndex, + operationrun.OperationRunTargetStatusFailed, + ). + ColumnExpr( + "COUNT(*) FILTER (WHERE ort.phase_index < ? AND ort.status = ?) AS completed_terminated", + currentPhaseIndex, + operationrun.OperationRunTargetStatusTerminated, + ). + ColumnExpr( + "COUNT(*) FILTER (WHERE ort.phase_index < ? AND ort.status = ?) AS completed_skipped", + currentPhaseIndex, + operationrun.OperationRunTargetStatusSkipped, + ). + Where("ort.operation_run_id = ?", runID). + Scan(ctx, &row) + if err != nil { + return operationrun.TargetPhaseAggregate{}, err + } + + return operationrun.TargetPhaseAggregate{ + TotalPhases: row.TotalPhases, + CompletedPhaseStats: operationrun.PhaseStats{ + PhaseIndex: max(currentPhaseIndex-1, 0), + SelectedTargets: row.CompletedTargets, + StatusCounts: operationrun.TargetStatusCounts{ + Completed: row.CompletedCompleted, + Failed: row.CompletedFailed, + Terminated: row.CompletedTerminated, + Skipped: row.CompletedSkipped, + }, + }, + }, nil +} + // UpdateRunState persists dispatcher-owned lifecycle fields. func (s *PostgresStore) UpdateRunState( ctx context.Context, @@ -110,6 +176,7 @@ func (s *PostgresStore) UpdateRunState( Set("status = ?", run.Status). Set("status_reason = ?", run.StatusReason). Set("status_message = ?", run.StatusMessage). + Set("current_phase_index = ?", run.CurrentPhaseIndex). Set("started_at = ?", run.StartedAt). Set("finished_at = ?", run.FinishedAt). Where("id = ?", run.ID). diff --git a/rest-api/flow/internal/operationrun/manager/store/store.go b/rest-api/flow/internal/operationrun/manager/store/store.go index 036c87c090..4525c56e3c 100644 --- a/rest-api/flow/internal/operationrun/manager/store/store.go +++ b/rest-api/flow/internal/operationrun/manager/store/store.go @@ -21,10 +21,9 @@ import ( ) const currentPhaseIndexSubquery = `( - SELECT MIN(phase_index) - FROM operation_run_target AS current_phase - WHERE current_phase.operation_run_id = ? - AND current_phase.status NOT IN (?) + SELECT current_phase_run.current_phase_index + FROM operation_run AS current_phase_run + WHERE current_phase_run.id = ? )` // txKeyType is an unexported type for the transaction context key, preventing @@ -167,6 +166,7 @@ func (s *PostgresStore) List( "status", "status_reason", "status_message", + "current_phase_index", "options", "operation_type", "operation_code", @@ -344,7 +344,6 @@ func applyTargetPhaseScope( return q.Where( "ort.phase_index = "+currentPhaseIndexSubquery, runID, - bun.In(operationrun.TerminalTargetStatuses()), ) } @@ -358,14 +357,12 @@ func applyTargetPhaseScope( return q.Where( "ort.phase_index < COALESCE("+currentPhaseIndexSubquery+", ort.phase_index + 1)", runID, - bun.In(operationrun.TerminalTargetStatuses()), ) case operationrun.TargetPhaseScopeCurrentAndCompletedPhases: // Same fallback rationale as CompletedPhases above. return q.Where( "ort.phase_index <= COALESCE("+currentPhaseIndexSubquery+", ort.phase_index)", runID, - bun.In(operationrun.TerminalTargetStatuses()), ) case operationrun.TargetPhaseScopeAllMaterializedTargets: return q diff --git a/rest-api/flow/internal/operationrun/manager/store/store_test.go b/rest-api/flow/internal/operationrun/manager/store/store_test.go index 4a39fed2d4..7a16204728 100644 --- a/rest-api/flow/internal/operationrun/manager/store/store_test.go +++ b/rest-api/flow/internal/operationrun/manager/store/store_test.go @@ -133,29 +133,30 @@ func TestApplyTargetPhaseScopeGeneratedSQL(t *testing.T) { wantNot []string }{ { - name: "current phase uses first non-terminal phase", + name: "current phase uses parent phase index", scope: operationrun.TargetPhaseScopeCurrentPhase, want: []string{ "ort.phase_index = (", - "SELECT MIN(phase_index)", - "current_phase.status NOT IN ('completed', 'failed', 'terminated', 'skipped')", + "SELECT current_phase_run.current_phase_index", + "FROM operation_run AS current_phase_run", }, wantNot: []string{ "MAX(phase_index)", "COALESCE(", + "current_phase.status NOT IN", }, }, { - name: "completed phases include all rows when no current phase exists", + name: "completed phases compare against parent phase index", scope: operationrun.TargetPhaseScopeCompletedPhases, want: []string{ "ort.phase_index < COALESCE(", - "SELECT MIN(phase_index)", - "current_phase.status NOT IN ('completed', 'failed', 'terminated', 'skipped')", + "SELECT current_phase_run.current_phase_index", "ort.phase_index + 1", }, wantNot: []string{ "MAX(phase_index)", + "current_phase.status NOT IN", }, }, { @@ -163,19 +164,19 @@ func TestApplyTargetPhaseScopeGeneratedSQL(t *testing.T) { scope: operationrun.TargetPhaseScopeCurrentAndCompletedPhases, want: []string{ "ort.phase_index <= COALESCE(", - "SELECT MIN(phase_index)", - "current_phase.status NOT IN ('completed', 'failed', 'terminated', 'skipped')", + "SELECT current_phase_run.current_phase_index", "ort.phase_index)", }, wantNot: []string{ "MAX(phase_index)", + "current_phase.status NOT IN", }, }, { name: "all materialized targets does not apply phase scope", scope: operationrun.TargetPhaseScopeAllMaterializedTargets, wantNot: []string{ - "SELECT MIN(phase_index)", + "SELECT current_phase_run.current_phase_index", "MAX(phase_index)", "COALESCE(", "current_phase.status NOT IN", diff --git a/rest-api/flow/internal/operationrun/operationrun.go b/rest-api/flow/internal/operationrun/operationrun.go index 744cdda6ed..57de9300e3 100644 --- a/rest-api/flow/internal/operationrun/operationrun.go +++ b/rest-api/flow/internal/operationrun/operationrun.go @@ -149,6 +149,7 @@ type OperationRun struct { Status OperationRunStatus StatusReason OperationRunStatusReason StatusMessage string + CurrentPhaseIndex int32 Selector json.RawMessage Options json.RawMessage OperationTemplate json.RawMessage diff --git a/rest-api/flow/internal/operationrun/progress.go b/rest-api/flow/internal/operationrun/progress.go index 43bb772124..e5066e680d 100644 --- a/rest-api/flow/internal/operationrun/progress.go +++ b/rest-api/flow/internal/operationrun/progress.go @@ -3,23 +3,18 @@ package operationrun -import ( - "errors" - "fmt" -) - // TargetPhaseSummary groups targets by the active rollout phase. The current // phase is the lowest phase index that still has non-terminal work. type TargetPhaseSummary struct { - TargetCount int - FailedOrTerminatedTargetCount int - CurrentPhaseIndex int32 - // CurrentPhaseTargets contains all targets in CurrentPhaseIndex, including - // terminal targets from that phase. + TotalPhases int32 CurrentPhaseTargets []*OperationRunTarget - // CompletedPhaseTargets contains all targets in phases before - // CurrentPhaseIndex. - CompletedPhaseTargets []*OperationRunTarget + CompletedPhaseStats PhaseStats + CurrentPhaseStats PhaseStats +} + +type TargetPhaseAggregate struct { + TotalPhases int32 + CompletedPhaseStats PhaseStats } // SafetyGateEvaluation reports whether a validated safety gate blocks progress. @@ -28,113 +23,65 @@ type SafetyGateEvaluation struct { Message string } -// NewTargetPhaseSummary finds the current rollout phase and the previous phase -// targets used for cumulative safety-gate evaluation. +// NewTargetPhaseSummary summarizes a persisted current phase from the current +// phase's target rows and SQL-derived aggregate stats for earlier phases. func NewTargetPhaseSummary( - targets []*OperationRunTarget, + currentPhaseIndex int32, + aggregate TargetPhaseAggregate, + currentPhaseTargets []*OperationRunTarget, ) TargetPhaseSummary { - summary := TargetPhaseSummary{CurrentPhaseIndex: -1} - - // First pass: count real targets and find the lowest phase that still has - // non-terminal work. That phase is the current phase for dispatch and - // manual phase advancement. - for _, target := range targets { - if target == nil { - continue - } - - summary.TargetCount++ - if target.Status.IsFailedOrTerminated() { - summary.FailedOrTerminatedTargetCount++ - } - - if target.Status.IsTerminal() { - continue - } - - if summary.CurrentPhaseIndex < 0 || - target.PhaseIndex < summary.CurrentPhaseIndex { - summary.CurrentPhaseIndex = target.PhaseIndex - } + summary := TargetPhaseSummary{ + TotalPhases: aggregate.TotalPhases, + CurrentPhaseTargets: currentPhaseTargets, + CompletedPhaseStats: aggregate.CompletedPhaseStats, + CurrentPhaseStats: PhaseStats{PhaseIndex: currentPhaseIndex}, } - if summary.IsAllTerminal() { - return summary - } - - // Second pass: now that the current phase is known, split targets into the - // current phase and already-completed prior phases for safety-gate checks. - for _, target := range targets { - if target == nil { - continue - } - - if target.PhaseIndex == summary.CurrentPhaseIndex { - summary.CurrentPhaseTargets = append( - summary.CurrentPhaseTargets, - target, - ) - } else if target.PhaseIndex < summary.CurrentPhaseIndex { - summary.CompletedPhaseTargets = append( - summary.CompletedPhaseTargets, - target, - ) - } - } + summary.CurrentPhaseStats.AddTargets(currentPhaseTargets) return summary } // IsAllTerminal reports whether there is no remaining active target work. func (s TargetPhaseSummary) IsAllTerminal() bool { - return s.CurrentPhaseIndex < 0 + return s.TotalPhases > 0 && + s.CurrentPhaseStats.PhaseIndex+1 >= s.TotalPhases && + s.CurrentPhaseStats.AllTargetsTerminal() +} + +func (s TargetPhaseSummary) CurrentPhaseTerminal() bool { + return s.CurrentPhaseStats.AllTargetsTerminal() +} + +func (s TargetPhaseSummary) HasNextPhase() bool { + return s.TotalPhases > 0 && s.CurrentPhaseStats.PhaseIndex+1 < s.TotalPhases } // TerminalRunStatus returns the terminal run status implied by an all-terminal // target set. The boolean is false when the run still has active work or has no // targets to summarize. func (s TargetPhaseSummary) TerminalRunStatus() (OperationRunStatus, bool) { - if !s.IsAllTerminal() || s.TargetCount == 0 { + targetCount := s.SelectedTargetCount() + if targetCount == 0 { + return "", false + } + + if !s.IsAllTerminal() { return "", false } - if s.FailedOrTerminatedTargetCount == s.TargetCount { + failedOrTerminatedCount := s.FailedOrTerminatedTargetCount() + if failedOrTerminatedCount == targetCount { return OperationRunStatusFailed, true } - if s.FailedOrTerminatedTargetCount > 0 { + if failedOrTerminatedCount > 0 { return OperationRunStatusCompletedWithFailures, true } return OperationRunStatusCompleted, true } -// CheckExpectedNextPhase verifies that the summary is waiting at a manually -// advanceable phase and that a positive expected phase matches the current -// phase. Phase 0 is the initial phase and starts without an advance gate; only -// phase 1 and later represent crossing a phase boundary. -func (s TargetPhaseSummary) CheckExpectedNextPhase( - expectedPhaseIndex int32, -) error { - if s.IsAllTerminal() { - return errors.New("not waiting at a next phase") - } - - if s.CurrentPhaseIndex == 0 { - return errors.New("phase 0 is the initial phase and cannot be advanced") - } - - if expectedPhaseIndex > 0 && expectedPhaseIndex != s.CurrentPhaseIndex { - return fmt.Errorf( - "expected phase %d, current phase is %d", - expectedPhaseIndex, - s.CurrentPhaseIndex, - ) - } - - return nil -} - // CurrentPhaseNotStarted reports whether the current phase still has only // untouched pending targets. func (s TargetPhaseSummary) CurrentPhaseNotStarted() bool { @@ -156,16 +103,23 @@ func (s TargetPhaseSummary) CurrentPhaseNotStarted() bool { // StatsForSafetyScope aggregates target outcomes over the safety-gate scope // selected by the user. func (s TargetPhaseSummary) StatsForSafetyScope(scope SafetyGateScope) PhaseStats { - stats := PhaseStats{} - stats.AddTargets(s.CurrentPhaseTargets) - + stats := s.CurrentPhaseStats if scope == SafetyGateScopeCumulativeRun { - stats.AddTargets(s.CompletedPhaseTargets) + stats.Add(s.CompletedPhaseStats) } - return stats } +func (s TargetPhaseSummary) SelectedTargetCount() int { + return s.CompletedPhaseStats.SelectedTargets + + s.CurrentPhaseStats.SelectedTargets +} + +func (s TargetPhaseSummary) FailedOrTerminatedTargetCount() int { + return s.CompletedPhaseStats.FailedOrTerminatedTargets() + + s.CurrentPhaseStats.FailedOrTerminatedTargets() +} + // EvaluateSafetyGates checks validated safety gates against the target summary. func (s TargetPhaseSummary) EvaluateSafetyGates( gates []SafetyGate, diff --git a/rest-api/flow/internal/operationrun/progress_test.go b/rest-api/flow/internal/operationrun/progress_test.go index 08265d0289..85a78214ce 100644 --- a/rest-api/flow/internal/operationrun/progress_test.go +++ b/rest-api/flow/internal/operationrun/progress_test.go @@ -10,51 +10,41 @@ import ( "github.com/stretchr/testify/require" ) -func TestNewTargetPhaseSummaryFindsLowestNonTerminalPhase(t *testing.T) { - phase0Completed := &OperationRunTarget{ - PhaseIndex: 0, - Status: OperationRunTargetStatusCompleted, - } - phase0Failed := &OperationRunTarget{ - PhaseIndex: 0, - Status: OperationRunTargetStatusFailed, - } - phase1Submitted := &OperationRunTarget{ +func TestNewTargetPhaseSummaryCombinesAggregateAndCurrentStats(t *testing.T) { + currentSubmitted := &OperationRunTarget{ PhaseIndex: 1, Status: OperationRunTargetStatusSubmitted, } - phase1Completed := &OperationRunTarget{ + currentCompleted := &OperationRunTarget{ PhaseIndex: 1, Status: OperationRunTargetStatusCompleted, } - phase2Pending := &OperationRunTarget{ - PhaseIndex: 2, - Status: OperationRunTargetStatusPending, - } - - summary := NewTargetPhaseSummary([]*OperationRunTarget{ - nil, - phase0Completed, - phase1Submitted, - phase2Pending, - phase1Completed, - phase0Failed, - }) + summary := NewTargetPhaseSummary( + 1, + TargetPhaseAggregate{ + TotalPhases: 3, + CompletedPhaseStats: PhaseStats{ + PhaseIndex: 0, + SelectedTargets: 2, + StatusCounts: TargetStatusCounts{ + Completed: 1, + Failed: 1, + }, + }, + }, + []*OperationRunTarget{currentSubmitted, currentCompleted}, + ) require.False(t, summary.IsAllTerminal()) - require.Equal(t, 5, summary.TargetCount) - require.Equal(t, 1, summary.FailedOrTerminatedTargetCount) - require.EqualValues(t, 1, summary.CurrentPhaseIndex) + require.True(t, summary.HasNextPhase()) + require.Equal(t, 4, summary.SelectedTargetCount()) + require.Equal(t, 1, summary.FailedOrTerminatedTargetCount()) + require.EqualValues(t, 1, summary.CurrentPhaseStats.PhaseIndex) require.Equal( t, - []*OperationRunTarget{phase1Submitted, phase1Completed}, + []*OperationRunTarget{currentSubmitted, currentCompleted}, summary.CurrentPhaseTargets, ) - require.Equal( - t, - []*OperationRunTarget{phase0Completed, phase0Failed}, - summary.CompletedPhaseTargets, - ) currentStats := summary.StatsForSafetyScope(SafetyGateScopeCurrentPhase) require.Equal(t, 2, currentStats.SelectedTargets) @@ -67,27 +57,6 @@ func TestNewTargetPhaseSummaryFindsLowestNonTerminalPhase(t *testing.T) { require.Equal(t, 1, cumulativeStats.StatusCounts.Failed) } -func TestNewTargetPhaseSummaryReportsAllTerminal(t *testing.T) { - summary := NewTargetPhaseSummary([]*OperationRunTarget{ - nil, - { - PhaseIndex: 0, - Status: OperationRunTargetStatusCompleted, - }, - { - PhaseIndex: 1, - Status: OperationRunTargetStatusFailed, - }, - }) - - require.True(t, summary.IsAllTerminal()) - require.Equal(t, 2, summary.TargetCount) - require.Equal(t, 1, summary.FailedOrTerminatedTargetCount) - require.EqualValues(t, -1, summary.CurrentPhaseIndex) - require.Empty(t, summary.CurrentPhaseTargets) - require.Empty(t, summary.CompletedPhaseTargets) -} - func TestTargetPhaseSummaryTerminalRunStatus(t *testing.T) { tests := []struct { name string @@ -98,21 +67,27 @@ func TestTargetPhaseSummaryTerminalRunStatus(t *testing.T) { { name: "still active", summary: TargetPhaseSummary{ - TargetCount: 1, - CurrentPhaseIndex: 1, + TotalPhases: 2, + CurrentPhaseStats: PhaseStats{ + PhaseIndex: 1, + SelectedTargets: 1, + }, }, }, { - name: "no targets", - summary: TargetPhaseSummary{ - CurrentPhaseIndex: -1, - }, + name: "no targets", + summary: TargetPhaseSummary{}, }, { name: "all completed", summary: TargetPhaseSummary{ - TargetCount: 2, - CurrentPhaseIndex: -1, + TotalPhases: 1, + CurrentPhaseStats: PhaseStats{ + SelectedTargets: 2, + StatusCounts: TargetStatusCounts{ + Completed: 2, + }, + }, }, wantStatus: OperationRunStatusCompleted, wantTerminalStatusFound: true, @@ -120,9 +95,14 @@ func TestTargetPhaseSummaryTerminalRunStatus(t *testing.T) { { name: "completed with failures", summary: TargetPhaseSummary{ - TargetCount: 2, - FailedOrTerminatedTargetCount: 1, - CurrentPhaseIndex: -1, + TotalPhases: 1, + CurrentPhaseStats: PhaseStats{ + SelectedTargets: 2, + StatusCounts: TargetStatusCounts{ + Completed: 1, + Failed: 1, + }, + }, }, wantStatus: OperationRunStatusCompletedWithFailures, wantTerminalStatusFound: true, @@ -130,9 +110,14 @@ func TestTargetPhaseSummaryTerminalRunStatus(t *testing.T) { { name: "all failed", summary: TargetPhaseSummary{ - TargetCount: 2, - FailedOrTerminatedTargetCount: 2, - CurrentPhaseIndex: -1, + TotalPhases: 1, + CurrentPhaseStats: PhaseStats{ + SelectedTargets: 2, + StatusCounts: TargetStatusCounts{ + Failed: 1, + Terminated: 1, + }, + }, }, wantStatus: OperationRunStatusFailed, wantTerminalStatusFound: true, @@ -149,62 +134,6 @@ func TestTargetPhaseSummaryTerminalRunStatus(t *testing.T) { } } -func TestTargetPhaseSummaryCheckExpectedNextPhase(t *testing.T) { - tests := []struct { - name string - currentPhaseIndex int32 - expectedPhase int32 - wantErr string - }{ - { - name: "all terminal", - currentPhaseIndex: -1, - wantErr: "not waiting at a next phase", - }, - { - name: "initial phase", - currentPhaseIndex: 0, - wantErr: "phase 0 is the initial phase and cannot be advanced", - }, - { - name: "matching phase", - currentPhaseIndex: 1, - expectedPhase: 1, - }, - { - name: "zero expectation", - currentPhaseIndex: 1, - }, - { - name: "negative expectation", - currentPhaseIndex: 1, - expectedPhase: -1, - }, - { - name: "mismatched phase", - currentPhaseIndex: 2, - expectedPhase: 1, - wantErr: "expected phase 1, current phase is 2", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - summary := TargetPhaseSummary{ - CurrentPhaseIndex: tt.currentPhaseIndex, - } - - err := summary.CheckExpectedNextPhase(tt.expectedPhase) - if tt.wantErr != "" { - require.ErrorContains(t, err, tt.wantErr) - return - } - - require.NoError(t, err) - }) - } -} - func TestTargetPhaseSummaryCurrentPhaseNotStarted(t *testing.T) { tests := []struct { name string @@ -262,8 +191,10 @@ func TestTargetPhaseSummaryCurrentPhaseNotStarted(t *testing.T) { } func TestEvaluateSafetyGatesReturnsFirstTrippedGate(t *testing.T) { - summary := TargetPhaseSummary{ - CurrentPhaseTargets: []*OperationRunTarget{ + summary := NewTargetPhaseSummary( + 1, + TargetPhaseAggregate{TotalPhases: 2}, + []*OperationRunTarget{ { PhaseIndex: 1, Status: OperationRunTargetStatusFailed, @@ -273,7 +204,7 @@ func TestEvaluateSafetyGatesReturnsFirstTrippedGate(t *testing.T) { Status: OperationRunTargetStatusCompleted, }, }, - } + ) evaluation := summary.EvaluateSafetyGates( []SafetyGate{ diff --git a/rest-api/flow/internal/operationrun/stats.go b/rest-api/flow/internal/operationrun/stats.go index 068c5a6713..6badc407e1 100644 --- a/rest-api/flow/internal/operationrun/stats.go +++ b/rest-api/flow/internal/operationrun/stats.go @@ -63,6 +63,21 @@ func (s *PhaseStats) AddTargets(targets []*OperationRunTarget) { } } +func (s *PhaseStats) Add(other PhaseStats) { + if s == nil { + return + } + + if other.PhaseIndex > s.PhaseIndex { + s.PhaseIndex = other.PhaseIndex + } + s.SelectedTargets += other.SelectedTargets + s.StatusCounts.Completed += other.StatusCounts.Completed + s.StatusCounts.Failed += other.StatusCounts.Failed + s.StatusCounts.Terminated += other.StatusCounts.Terminated + s.StatusCounts.Skipped += other.StatusCounts.Skipped +} + func (s *PhaseStats) AddTarget(target *OperationRunTarget) { if s == nil || target == nil { return @@ -72,7 +87,7 @@ func (s *PhaseStats) AddTarget(target *OperationRunTarget) { s.PhaseIndex = target.PhaseIndex } s.SelectedTargets++ - s.StatusCounts.AddStatus(target.Status) + s.StatusCounts.Add(target.Status) } // FailurePercent returns the percentage of selected targets that failed. @@ -83,6 +98,21 @@ func (s PhaseStats) FailurePercent() int { return s.StatusCounts.Failed * 100 / s.SelectedTargets } +func (s PhaseStats) TerminalTargets() int { + return s.StatusCounts.Completed + + s.StatusCounts.Failed + + s.StatusCounts.Terminated + + s.StatusCounts.Skipped +} + +func (s PhaseStats) AllTargetsTerminal() bool { + return s.SelectedTargets > 0 && s.TerminalTargets() == s.SelectedTargets +} + +func (s PhaseStats) FailedOrTerminatedTargets() int { + return s.StatusCounts.Failed + s.StatusCounts.Terminated +} + // SafetyGateTrippedMessage formats the pause message for a gate that tripped // against these stats. func (s PhaseStats) SafetyGateTrippedMessage(gate SafetyGate) string { @@ -122,7 +152,7 @@ func (s PhaseStats) SafetyGateTrippedMessage(gate SafetyGate) string { } } -func (s *TargetStatusCounts) AddStatus(status OperationRunTargetStatus) { +func (s *TargetStatusCounts) Add(status OperationRunTargetStatus) { if s == nil { return }