diff --git a/docs/design.md b/docs/design.md index 78165b9..f38bf04 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 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/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..d097a96 100644 --- a/internal/engine/checkpoint.go +++ b/internal/engine/checkpoint.go @@ -3,6 +3,8 @@ package engine import ( "context" "fmt" + "sort" + "strings" "github.com/rbtr/shunt/internal/checkpoint" ) @@ -33,7 +35,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 +89,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 +103,38 @@ 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 +// 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 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 { 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 (e *Engine) ownsStagingBranch(branch string) bool { + return e.cfg.StagingBranch != "" && strings.HasPrefix(branch, e.cfg.StagingBranch+"-") } func clonePending(in [][]int) [][]int { diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index ccaeeb4..acb0de8 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,56 @@ 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 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{} 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) }