From 06f6f59b6b30483e1b8ba0178721abe2a7c47cb5 Mon Sep 17 00:00:00 2001 From: Evan Baker Date: Fri, 4 Sep 2026 23:57:19 -0500 Subject: [PATCH 1/2] engine: discard stale staging evidence on restart --- docs/design.md | 7 ++--- internal/checkpoint/bolt/bolt_test.go | 7 ++++- internal/engine/checkpoint.go | 20 +++++++++++++-- internal/engine/engine_test.go | 37 +++++++++++++++++++++++++++ mq/checkpoint/checkpoint.go | 11 +++++--- 5 files changed, 73 insertions(+), 9 deletions(-) diff --git a/docs/design.md b/docs/design.md index 78165b9..3c232f8 100644 --- a/docs/design.md +++ b/docs/design.md @@ -289,9 +289,10 @@ bad PR, bounce it, and land the good PRs. the queue from open auto-merge PRs. It may repeat a staging run; a PR already released to the forge may finish while shunt restarts. With `SHUNT_STATE_PATH`, shunt persists the pending frontier, linger state, - bisection counters, and active batch metadata in bbolt. Active batches are - re-staged after restore so no additional PR is released from a pre-restart - result that may have been invalidated by a base change. + bisection counters, and active batch metadata in bbolt. On restore, shunt + removes each in-flight staging branch and re-stages its PRs, so no additional + PR is released from a pre-restart result that may have been invalidated by a + base or source-head change. ## Observability diff --git a/internal/checkpoint/bolt/bolt_test.go b/internal/checkpoint/bolt/bolt_test.go index 393d01a..aea509b 100644 --- a/internal/checkpoint/bolt/bolt_test.go +++ b/internal/checkpoint/bolt/bolt_test.go @@ -32,6 +32,8 @@ func TestStoreSavesLoadsAndDeletesQueue(t *testing.T) { StagingSHA: "stage-4", BaseGeneration: 2, Outcome: "failure", + Phase: "bisecting", + PhaseSince: time.Date(2026, 6, 24, 10, 1, 0, 0, time.UTC), }}, LingerSince: time.Date(2026, 6, 24, 10, 0, 0, 0, time.UTC), BaseGeneration: 2, @@ -102,7 +104,7 @@ func TestQueueSnapshotJSONReadsLegacyFieldNames(t *testing.T) { data := []byte(`{ "Key":{"Owner":"o","Repo":"r","Base":"main"}, "Pending":[[1]], - "Active":[{"PRs":[{"Number":1,"HeadSHA":"head-1"}],"StagingBranch":"mq/main/staging","StagingSHA":"stage-1","BaseGeneration":2,"Outcome":"failure"}], + "Active":[{"PRs":[{"Number":1,"HeadSHA":"head-1"}],"StagingBranch":"mq/main/staging","StagingSHA":"stage-1","BaseGeneration":2,"Outcome":"failure","Phase":"bisecting","PhaseSince":"2026-06-24T10:01:00Z"}], "LingerSince":"0001-01-01T00:00:00Z", "BaseGeneration":2, "StagingSequence":7 @@ -120,4 +122,7 @@ func TestQueueSnapshotJSONReadsLegacyFieldNames(t *testing.T) { if got := snapshot.StagingSequence; got != 7 { t.Fatalf("staging sequence = %d, want 7", got) } + if got := snapshot.Active[0].Phase; got != "bisecting" { + t.Fatalf("phase = %q, want bisecting", got) + } } diff --git a/internal/engine/checkpoint.go b/internal/engine/checkpoint.go index 4406dcc..0aa50b9 100644 --- a/internal/engine/checkpoint.go +++ b/internal/engine/checkpoint.go @@ -3,6 +3,7 @@ package engine import ( "context" "fmt" + "sort" "github.com/rbtr/shunt/internal/checkpoint" ) @@ -33,7 +34,9 @@ func (e *Engine) loadCheckpoint(ctx context.Context) error { if snapshot.Key != e.queueKey() { return fmt.Errorf("queue checkpoint key mismatch: got %s/%s@%s", snapshot.Key.Owner, snapshot.Key.Repo, snapshot.Key.Base) } - e.applySnapshot(snapshot) + if err := e.applySnapshot(ctx, snapshot); err != nil { + return err + } e.checkpointLoaded = true e.checkpointExists = true return nil @@ -85,6 +88,8 @@ func (e *Engine) snapshot() checkpoint.QueueSnapshot { StagingSHA: a.stagingSHA, BaseGeneration: a.baseGen, Outcome: a.outcome, + Phase: a.phase, + PhaseSince: a.phaseSince, } } return checkpoint.QueueSnapshot{ @@ -97,19 +102,30 @@ func (e *Engine) snapshot() checkpoint.QueueSnapshot { } } -func (e *Engine) applySnapshot(snapshot checkpoint.QueueSnapshot) { +// applySnapshot restores unresolved candidates. In-flight staging attempts are +// deliberately not resumed: after a restart, Shunt cannot prove their gate +// evidence still includes the current base and source heads. It removes each +// old staging branch and re-queues its PRs for fresh staging instead. +func (e *Engine) applySnapshot(ctx context.Context, snapshot checkpoint.QueueSnapshot) error { e.pending = clonePending(snapshot.Pending) for _, active := range snapshot.Active { + if err := e.fc.DeleteBranch(ctx, e.cfg.Owner, e.cfg.Repo, active.StagingBranch); err != nil { + e.logger.Warn("failed to delete restored staging branch", "branch", active.StagingBranch, "error", err) + } nums := make([]int, len(active.PRs)) for i, pr := range active.PRs { nums[i] = pr.Number } e.pending = append(e.pending, nums) } + sort.SliceStable(e.pending, func(i, j int) bool { + return e.pending[i][0] < e.pending[j][0] + }) e.active = nil e.lingerSince = snapshot.LingerSince e.baseGen = snapshot.BaseGeneration e.stagingSeq = snapshot.StagingSequence + return nil } func clonePending(in [][]int) [][]int { diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index ccaeeb4..adb3055 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -840,6 +840,13 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) { if store.saved == nil || len(store.saved.Active) != 1 { t.Fatalf("checkpoint active batches = %v, want 1 active batch", store.saved) } + if got := store.saved.Active[0].Phase; got != "waiting_gate" { + t.Fatalf("checkpoint phase = %q, want waiting_gate", got) + } + if store.saved.Active[0].PhaseSince.IsZero() { + t.Fatal("checkpoint phase timestamp is zero") + } + staleBranch := m.stagingBranches[0] restarted := New(cfg, m, m) if err := restarted.Reconcile(context.Background()); err != nil { @@ -848,6 +855,9 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) { if got := len(m.staged); got != 2 { t.Fatalf("restored active batch should be restaged; staged = %d, want 2", got) } + if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:"+staleBranch) { + t.Fatalf("calls = %s, want stale staging branch deleted", got) + } if got := fmt.Sprint(m.merged); got != "[]" { t.Fatalf("merged after restage = %s, want []", got) } @@ -860,6 +870,33 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) { } } +func TestCheckpointRestoreKeepsActiveCandidateBeforeLaterPendingWork(t *testing.T) { + m := newMock(-1, 1, 2, 3) + store := &memoryCheckpointStore{saved: &checkpoint.QueueSnapshot{ + Key: checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"}, + Pending: [][]int{{3}}, + Active: []checkpoint.ActiveBatchSnapshot{{ + PRs: []checkpoint.PullRequestSnapshot{ + {Number: 1, HeadSHA: "head-1"}, + {Number: 2, HeadSHA: "head-2"}, + }, + StagingBranch: "mq/main/staging-old", + StagingSHA: "stage-old", + }}, + }} + cfg := Config{Owner: "o", Repo: "r", Base: "main", StatusCtx: "merge-queue", StagingBranch: "mq/main/staging", Checkpoint: store} + + if err := New(cfg, m, m).Reconcile(context.Background()); err != nil { + t.Fatalf("restore queue: %v", err) + } + if got := fmt.Sprint(m.staged); got != "[[1 2]]" { + t.Fatalf("staged = %s, want the restored active candidate before later work", got) + } + if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:mq/main/staging-old") { + t.Fatalf("calls = %s, want restored staging branch deleted", got) + } +} + func TestCheckpointRestartRestagesRemainderAfterReleasedPRMerges(t *testing.T) { m := newMock(-1, 1, 2) store := &memoryCheckpointStore{} diff --git a/mq/checkpoint/checkpoint.go b/mq/checkpoint/checkpoint.go index d90f32c..2afb22e 100644 --- a/mq/checkpoint/checkpoint.go +++ b/mq/checkpoint/checkpoint.go @@ -29,15 +29,17 @@ type QueueSnapshot struct { StagingSequence int `json:"StagingSequence"` } -// ActiveBatchSnapshot records a staging branch currently waiting on its gate. -// On restore the engine re-queues these PRs for fresh staging (applySnapshot -// sets active = nil) — it does not resume the staged branch. +// ActiveBatchSnapshot records a staging attempt that was in flight. On +// restore the engine removes its staging branch and re-queues its PRs for fresh +// staging, rather than trusting evidence from before the restart. type ActiveBatchSnapshot struct { PRs []PullRequestSnapshot `json:"PRs"` StagingBranch string `json:"StagingBranch"` StagingSHA string `json:"StagingSHA"` BaseGeneration int `json:"BaseGeneration"` Outcome string `json:"Outcome"` + Phase string `json:"Phase"` + PhaseSince time.Time `json:"PhaseSince"` } // PullRequestSnapshot is the PR identity needed to re-queue a batch. @@ -92,6 +94,9 @@ func (s QueueSnapshot) Validate() error { if active.Outcome != "" && active.Outcome != "success" && active.Outcome != "failure" && active.Outcome != "cancelled" && active.Outcome != "error" { return fmt.Errorf("queue checkpoint active batch %d has invalid outcome %q", i, active.Outcome) } + if active.Phase != "" && active.Phase != "waiting_gate" && active.Phase != "waiting_merge" && active.Phase != "bisecting" { + return fmt.Errorf("queue checkpoint active batch %d has invalid phase %q", i, active.Phase) + } if len(active.PRs) == 0 { return fmt.Errorf("queue checkpoint active batch %d has no PRs", i) } From 47246a7f688019b7a61d743af7209d545a8d4b9e Mon Sep 17 00:00:00 2001 From: Evan Baker Date: Sat, 5 Sep 2026 00:01:21 -0500 Subject: [PATCH 2/2] engine: limit restored staging cleanup to shunt branches --- docs/design.md | 6 +++--- internal/engine/checkpoint.go | 15 ++++++++++++--- internal/engine/engine_test.go | 23 +++++++++++++++++++++++ 3 files changed, 38 insertions(+), 6 deletions(-) diff --git a/docs/design.md b/docs/design.md index 3c232f8..f38bf04 100644 --- a/docs/design.md +++ b/docs/design.md @@ -290,9 +290,9 @@ bad PR, bounce it, and land the good PRs. released to the forge may finish while shunt restarts. With `SHUNT_STATE_PATH`, shunt persists the pending frontier, linger state, bisection counters, and active batch metadata in bbolt. On restore, shunt - removes each in-flight staging branch and re-stages its PRs, so no additional - PR is released from a pre-restart result that may have been invalidated by a - base or source-head change. + removes each known staging branch and re-stages its PRs, so no additional PR + is released from a pre-restart result that may have been invalidated by a base + or source-head change. ## Observability diff --git a/internal/engine/checkpoint.go b/internal/engine/checkpoint.go index 0aa50b9..d097a96 100644 --- a/internal/engine/checkpoint.go +++ b/internal/engine/checkpoint.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sort" + "strings" "github.com/rbtr/shunt/internal/checkpoint" ) @@ -105,12 +106,16 @@ func (e *Engine) snapshot() checkpoint.QueueSnapshot { // applySnapshot restores unresolved candidates. In-flight staging attempts are // deliberately not resumed: after a restart, Shunt cannot prove their gate // evidence still includes the current base and source heads. It removes each -// old staging branch and re-queues its PRs for fresh staging instead. +// Shunt-owned staging branch and re-queues its PRs for fresh staging instead. func (e *Engine) applySnapshot(ctx context.Context, snapshot checkpoint.QueueSnapshot) error { e.pending = clonePending(snapshot.Pending) for _, active := range snapshot.Active { - if err := e.fc.DeleteBranch(ctx, e.cfg.Owner, e.cfg.Repo, active.StagingBranch); err != nil { - e.logger.Warn("failed to delete restored staging branch", "branch", active.StagingBranch, "error", err) + if e.ownsStagingBranch(active.StagingBranch) { + if err := e.fc.DeleteBranch(ctx, e.cfg.Owner, e.cfg.Repo, active.StagingBranch); err != nil { + e.logger.Warn("failed to delete restored staging branch", "branch", active.StagingBranch, "error", err) + } + } else { + e.logger.Warn("not deleting restored branch outside staging namespace", "branch", active.StagingBranch) } nums := make([]int, len(active.PRs)) for i, pr := range active.PRs { @@ -128,6 +133,10 @@ func (e *Engine) applySnapshot(ctx context.Context, snapshot checkpoint.QueueSna return nil } +func (e *Engine) ownsStagingBranch(branch string) bool { + return e.cfg.StagingBranch != "" && strings.HasPrefix(branch, e.cfg.StagingBranch+"-") +} + func clonePending(in [][]int) [][]int { if in == nil { return nil diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index adb3055..acb0de8 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -897,6 +897,29 @@ func TestCheckpointRestoreKeepsActiveCandidateBeforeLaterPendingWork(t *testing. } } +func TestCheckpointRestoreDoesNotDeleteBranchOutsideStagingNamespace(t *testing.T) { + m := newMock(-1, 1) + store := &memoryCheckpointStore{saved: &checkpoint.QueueSnapshot{ + Key: checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"}, + Active: []checkpoint.ActiveBatchSnapshot{{ + PRs: []checkpoint.PullRequestSnapshot{{Number: 1, HeadSHA: "head-1"}}, + StagingBranch: "main", + StagingSHA: "stage-old", + }}, + }} + cfg := Config{Owner: "o", Repo: "r", Base: "main", StatusCtx: "merge-queue", StagingBranch: "mq/main/staging", Checkpoint: store} + + if err := New(cfg, m, m).Reconcile(context.Background()); err != nil { + t.Fatalf("restore queue: %v", err) + } + if got := fmt.Sprint(m.calls); strings.Contains(got, "delete:main") { + t.Fatalf("calls = %s, must not delete a branch outside the staging namespace", got) + } + if got := fmt.Sprint(m.staged); got != "[[1]]" { + t.Fatalf("staged = %s, want PR requeued for fresh staging", got) + } +} + func TestCheckpointRestartRestagesRemainderAfterReleasedPRMerges(t *testing.T) { m := newMock(-1, 1, 2) store := &memoryCheckpointStore{}