diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 5e6fa79..4c3879a 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -763,6 +763,7 @@ func (e *Engine) checkActive(ctx context.Context) (bool, error) { } prs := a.prs + e.markBatchSuperseded(a, "staging gate produced no result") e.cleanupBatch(ctx, a) refs := make([]gitops.MergedRef, len(prs)) for i, pr := range prs { @@ -943,6 +944,7 @@ func (e *Engine) discardIneligibleActive(ctx context.Context, a *activeBatch, pr return } remaining := removeNum(numbersOf(a.prs), pr) + e.markBatchSuperseded(a, "a candidate became ineligible") e.cleanupBatch(ctx, a) e.enqueueWithState("requeued after another PR became ineligible", remaining) e.observeQueueExit(pr, "ineligible") @@ -996,6 +998,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge if current.State != "open" { e.skipLand(ctx, staged, current, fmt.Sprintf("state changed to %q", current.State), len(a.prs) > 1, a.debugURL, false) e.requeueActiveRemainder("requeued after an earlier PR left the batch", a.prs[1:]) + e.markBatchSuperseded(a, "a PR left the batch before landing") e.cleanupBatch(ctx, a) return true, merged, nil } @@ -1013,6 +1016,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge true, ) e.requeueActiveRemainder("requeued after PR head changed", a.prs) + e.markBatchSuperseded(a, "a PR head changed before landing") e.cleanupBatch(ctx, a) return true, merged, nil } @@ -1031,6 +1035,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge } e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true) e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs) + e.markBatchSuperseded(a, "auto-merge was cancelled before landing") e.cleanupBatch(ctx, a) return true, merged, nil } @@ -1067,6 +1072,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge } e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true) e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs) + e.markBatchSuperseded(a, "auto-merge was cancelled during recovery") e.cleanupBatch(ctx, a) return true, merged, nil } @@ -1125,6 +1131,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge return true, merged, nil } e.requeueActiveRemainder("retrying after forge merge recovery", a.prs) + e.markBatchSuperseded(a, "native auto-merge timed out") e.cleanupBatch(ctx, a) if err := e.scheduleAutomerge(ctx, current); err != nil { return false, merged, fmt.Errorf("restore auto-merge for PR #%d: %w", staged.Number, err) @@ -1179,6 +1186,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge } e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true) e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs) + e.markBatchSuperseded(a, "auto-merge was cancelled before release") e.cleanupBatch(ctx, a) return true, merged, nil } @@ -1791,6 +1799,7 @@ func (e *Engine) freeSlotForEarlierPending(ctx context.Context) { return } a := e.active[idx] + e.markBatchSuperseded(a, "an earlier queue entry took the staging slot") e.cleanupBatch(ctx, a) e.enqueueWithState("requeued; waiting for an earlier queue entry", numbersOf(a.prs)) e.logger.Info("speculative batch requeued for earlier candidate", "prs", numbersOf(a.prs), "earlier", e.pending[0]) @@ -1800,6 +1809,7 @@ func (e *Engine) requeueStaleActive(ctx context.Context, a *activeBatch) { if e.requeuedTreeNode(ctx, a, "base branch advanced", "re-queued: base branch advanced mid-test") { return } + e.markBatchSuperseded(a, "base branch advanced") e.cleanupBatch(ctx, a) e.enqueueWithState("requeued after base branch advanced", numbersOf(a.prs)) e.logger.Info("stale speculative batch requeued after base advanced", "prs", numbersOf(a.prs)) @@ -1809,11 +1819,26 @@ func (e *Engine) requeueChangedActive(ctx context.Context, a *activeBatch) { if e.requeuedTreeNode(ctx, a, "a pinned candidate changed", "re-queued: a pinned candidate changed mid-test") { return } + e.markBatchSuperseded(a, "a PR head changed") e.cleanupBatch(ctx, a) e.enqueueWithState("requeued after PR head changed", numbersOf(a.prs)) e.logger.Info("active batch requeued after PR head changed", "prs", numbersOf(a.prs)) } +// markBatchSuperseded records that a staging attempt ended without a source +// decision because its candidates will be tested again. The transition lets +// callers retire the old staging row instead of leaving it as running. +// `superseded` means "this attempt was replaced", not "a PR failed". +func (e *Engine) markBatchSuperseded(a *activeBatch, reason string) { + if a.stagingBranch == "" { + return + } + e.recordTransition(Transition{ + Kind: "node_superseded", PRs: numbersOf(a.prs), StagingBranch: a.stagingBranch, + RunID: a.runID, LineagePath: a.lineagePath, Reason: reason, + }) +} + // supersedeSpeculative discards a fanout-staged bisection node whose gate ran // against an accumulator the resolved frontier no longer matches, and re-queues // the same node — keeping its run id and lineage path — so it re-stages on the diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index 5091385..76969a4 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -1796,6 +1796,15 @@ func TestMissingGateRetriesWithBackoffThenBounces(t *testing.T) { if got := e.active[0].missingGateRetries; got != retry { t.Fatalf("missing-gate retries = %d, want %d", got, retry) } + found := false + for _, tr := range e.Transitions() { + if tr.Kind == "node_superseded" && tr.StagingBranch == m.stagingBranches[retry-1] { + found = true + } + } + if !found { + t.Fatalf("retry %d did not report the replaced staging attempt", retry) + } } if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:mq/main/staging") { t.Fatalf("calls = %s, want stale staging branch deleted", got) @@ -2191,6 +2200,16 @@ func TestNativeMergeTimeoutBlocksRestoresAndRequeues(t *testing.T) { if got := fmt.Sprint(e.pending); got != "[[1]]" { t.Fatalf("pending = %s, want timed-out PR requeued", got) } + var superseded bool + for _, transition := range e.Transitions() { + if transition.Kind == "node_superseded" && transition.StagingBranch == m.stagingBranches[0] { + superseded = true + break + } + } + if !superseded { + t.Fatalf("transitions = %v, want timed-out staging attempt marked superseded", e.Transitions()) + } if got := strings.Join(m.comments[1], "\n"); !strings.Contains(got, "Merge did not complete") { t.Fatalf("outcome comment = %q, want timeout outcome", got) } diff --git a/internal/engine/transitions.go b/internal/engine/transitions.go index 2bf461c..3cef5e5 100644 --- a/internal/engine/transitions.go +++ b/internal/engine/transitions.go @@ -8,8 +8,9 @@ import ( // Transition is a structured record of one merge-queue lifecycle event that // happened during a single Reconcile call: a batch was staged, its gate -// passed, it was bisected, a PR bounced, or a PR landed. It exists so a -// caller can persist what actually happened without parsing log text. +// passed, it was bisected, a staging attempt was superseded, a PR bounced, or +// a PR landed. It exists so a caller can persist what actually happened without +// parsing log text. // // Additive to the engine's existing logger.Info/Warn calls at the same // sites — recording a Transition never changes what gets logged. @@ -21,13 +22,12 @@ type Transition struct { // whole bisection root was torn down after its base branch advanced or a // pinned candidate changed mid-test; its candidates are re-queued as a // fresh root and no source decision is published), or "node_superseded" (a - // speculatively-staged bisection node whose gate ran against an accumulator - // the resolved frontier no longer matches; re-staged on the correct - // baseline, no source decision). + // staging attempt was replaced without a source decision, including a + // speculative node whose accumulator no longer matches). Kind string `json:"kind"` - // PRs is the batch's PR set for "staged"/"gate_success"/"bisected"/"held", - // or the single terminated/landed PR (as a length-1 slice) for - // "bounced"/"landed". + // PRs is the batch's PR set for "staged"/"gate_success"/"bisected"/ + // "held"/"node_superseded", or the single terminated/landed PR (as a + // length-1 slice) for "bounced"/"landed". PRs []int `json:"prs"` StagingBranch string `json:"staging_branch"` // RunID and LineagePath identify this batch's position in a bisection @@ -37,8 +37,9 @@ type Transition struct { // name strings from scratch. RunID string `json:"run_id"` LineagePath string `json:"lineage_path"` - // Reason is set for "bounced" (why the PR was rejected) and for "held" - // (the held gate outcome: "success", "failure", or "error"). + // Reason is set for "bounced" (why the PR was rejected), "held" (the held + // gate outcome: "success", "failure", or "error"), and + // "node_superseded" (why the staging attempt was replaced). Reason string `json:"reason,omitempty"` // EventID is a deterministic key for transitions that drive an // irreversible side effect (a merge counter, a bounce notification): the diff --git a/mq/mq.go b/mq/mq.go index 8c391b6..3861bbb 100644 --- a/mq/mq.go +++ b/mq/mq.go @@ -197,8 +197,9 @@ func (eng *Engine) Reconcile(ctx context.Context) error { } // Transition is a structured record of one merge-queue lifecycle event -// (a batch staged, its gate passed, it bisected, a PR bounced, or a PR -// landed) recorded during a Reconcile call. See LastTransitions. +// (a batch staged, its gate passed, it bisected, a staging attempt was +// superseded, a PR bounced, or a PR landed) recorded during a Reconcile call. +// See LastTransitions. type Transition = engine.Transition // LastTransitions returns the transitions recorded during the most