From 9ecacb8f93241911c9f54fdcce8fdac1bc374660 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Tue, 23 Jun 2026 20:08:30 -0400 Subject: [PATCH 1/2] feat: add fleet run-ledger reconcile gate Signed-off-by: Joshua Temple --- .github/actions/register-run/action.yaml | 128 +++++++++ .github/workflows/fleet-reconcile.yaml | 225 ++++++++++++++++ internal/fleetreconcile/cmd/main.go | 123 +++++++++ internal/fleetreconcile/fleetreconcile.go | 243 ++++++++++++++++++ .../fleetreconcile/fleetreconcile_test.go | 195 ++++++++++++++ 5 files changed, 914 insertions(+) create mode 100644 .github/actions/register-run/action.yaml create mode 100644 .github/workflows/fleet-reconcile.yaml create mode 100644 internal/fleetreconcile/cmd/main.go create mode 100644 internal/fleetreconcile/fleetreconcile.go create mode 100644 internal/fleetreconcile/fleetreconcile_test.go diff --git a/.github/actions/register-run/action.yaml b/.github/actions/register-run/action.yaml new file mode 100644 index 00000000..58af3e49 --- /dev/null +++ b/.github/actions/register-run/action.yaml @@ -0,0 +1,128 @@ +name: 'Register fleet run' +description: >- + Record a run a scenario suite gates - its id and the conclusion it expects - + into the fleet run ledger. The fleet-reconcile reusable workflow later + enumerates every run the repo produced in the scenario window and fails if any + run is in the window but not in this ledger, turning every fire-and-forget run + into a hard red. Call this right after a suite resolves a dispatched or + triggered run id, before (or alongside) the `gh run watch` that awaits it. + Safe to call many times across a suite: it appends one JSON line per call. + +inputs: + run-id: + description: 'The run id the suite is gating (the databaseId from gh run list).' + required: true + expected-conclusion: + description: >- + The conclusion this run must reach: "success" (the default) or "failure" + for a registered negative (a guard that must refuse, e.g. the + divergence-promote guard). Reconcile requires the actual conclusion to + equal this. + required: false + default: 'success' + reason: + description: >- + A short tag naming the scenario step that registered this run (e.g. + "hotfix-finalize" or "divergence-guard"). Appears in the reconcile report + so a gap is attributable. + required: true + ledger-path: + description: >- + Path to the ledger file. Defaults under $RUNNER_TEMP so it is per-runner + and survives across steps of the same job. For a MULTI-JOB suite, set + upload to "true" in every job that registers, and have the reconcile job + download every "cascade-run-ledger-*" artifact (see fleet-reconcile.yaml); + a single-job suite can leave upload off and pass this same path to + reconcile directly. + required: false + default: '' + upload: + description: >- + When "true", upload the ledger as a per-job artifact so it reaches the + reconcile job in a multi-job suite. Leave "false" (default) for a + single-job suite that passes the workspace ledger path straight to + reconcile. + required: false + default: 'false' + artifact-name: + description: >- + Artifact name when upload is "true". Must be unique per job so concurrent + jobs do not collide; default appends the job + a random suffix. The + reconcile job globs "cascade-run-ledger-*" to merge them. + required: false + default: '' + +outputs: + ledger-path: + description: 'The resolved ledger file path the entry was appended to.' + value: ${{ steps.append.outputs.ledger-path }} + +runs: + using: 'composite' + steps: + - name: Append run to the ledger + id: append + shell: bash + env: + RUN_ID: ${{ inputs.run-id }} + EXPECTED: ${{ inputs.expected-conclusion }} + REASON: ${{ inputs.reason }} + LEDGER_PATH_IN: ${{ inputs.ledger-path }} + run: | + set -euo pipefail + + # Validate the expectation up front so a typo cannot silently register a + # run that reconcile can never match. + case "$EXPECTED" in + success|failure) ;; + *) echo "::error::register-run: expected-conclusion must be 'success' or 'failure', got '$EXPECTED'"; exit 1 ;; + esac + if ! [[ "$RUN_ID" =~ ^[0-9]+$ ]]; then + echo "::error::register-run: run-id must be numeric, got '$RUN_ID'"; exit 1 + fi + if [ -z "${REASON:-}" ]; then + echo "::error::register-run: reason is required"; exit 1 + fi + + LEDGER="$LEDGER_PATH_IN" + if [ -z "$LEDGER" ]; then + LEDGER="${RUNNER_TEMP}/cascade-run-ledger.jsonl" + fi + mkdir -p "$(dirname "$LEDGER")" + + # Append one JSON line. jq -c guarantees valid JSON and correct escaping + # of the reason. Append (>>) is intentional: many calls build one ledger. + jq -cn \ + --argjson run_id "$RUN_ID" \ + --arg expected "$EXPECTED" \ + --arg reason "$REASON" \ + '{run_id: $run_id, expected: $expected, reason: $reason}' >> "$LEDGER" + + echo "registered run $RUN_ID (expected=$EXPECTED, reason=$REASON) -> $LEDGER" + echo "ledger-path=$LEDGER" >> "$GITHUB_OUTPUT" + + - name: Stage ledger for artifact upload + if: inputs.upload == 'true' + id: stage + shell: bash + env: + LEDGER: ${{ steps.append.outputs.ledger-path }} + ART_NAME_IN: ${{ inputs.artifact-name }} + run: | + set -euo pipefail + ART_NAME="$ART_NAME_IN" + if [ -z "$ART_NAME" ]; then + # Unique per job + run so concurrent registering jobs never collide. + ART_NAME="cascade-run-ledger-${GITHUB_JOB}-${GITHUB_RUN_ID}-${RANDOM}" + fi + echo "artifact-name=$ART_NAME" >> "$GITHUB_OUTPUT" + + - name: Upload ledger artifact + if: inputs.upload == 'true' + uses: actions/upload-artifact@de65e23aa2b7e23d713bb51fbfcb6d502f8067d6 # v4.6.2 + with: + name: ${{ steps.stage.outputs.artifact-name }} + path: ${{ steps.append.outputs.ledger-path }} + if-no-files-found: error + retention-days: 7 + overwrite: true diff --git a/.github/workflows/fleet-reconcile.yaml b/.github/workflows/fleet-reconcile.yaml new file mode 100644 index 00000000..495d49f2 --- /dev/null +++ b/.github/workflows/fleet-reconcile.yaml @@ -0,0 +1,225 @@ +# Fleet Reconcile - the structural coverage gate for the cascade example fleet. +# +# This is maintainer fleet infra: hand-written tooling that lives in cascade's +# repo and is CALLED by each example repo's scenario-suite.yaml as its final +# job. It is NOT a cascade product feature and NOT part of cascade's generated +# output. +# +# Why it exists: a scenario suite only verifies the runs it remembers to wait +# on. A single scenario action causes SECONDARY runs (a PR-close hotfix Finalize +# run, a seed-PR preview run, an incidental push-orchestrate) that suites +# routinely forget to gate, so a suite can stay green over a red run it caused. +# This gate makes "no unasserted run in the scenario window" a structural +# invariant: it enumerates EVERY run the repo produced since the scenario began +# and fails if any run is in the window but not in the suite's run ledger, or +# concluded other than the suite registered. +# +# How a suite uses it: +# 1. At scenario start, record an ISO-8601 timestamp (window-start). +# 2. For every run the suite gates, call the register-run composite action +# (stablekernel/cascade/.github/actions/register-run@) with the run +# id and its expected conclusion. For a MULTI-JOB suite, pass upload: true +# so each job's ledger reaches this job as an artifact; for a single-job +# suite, pass the workspace ledger path via `ledger-path`. +# 3. As the suite's final job (needs: [], if: always()), +# call this workflow with window-start and the ledger artifact name. +# +# The reconcile logic lives in cascade's own Go (internal/fleetreconcile), so it +# is unit-tested with synthetic run lists and cannot silently regress. +name: Fleet Reconcile + +on: + workflow_call: + inputs: + window-start: + description: >- + ISO-8601 timestamp (UTC, e.g. 2026-06-23T14:00:00Z) of when the + scenario began. Every run created at or after this in this repo is + reconciled against the ledger. + required: true + type: string + ledger-artifact: + description: >- + Name (or glob) of the run-ledger artifact(s) uploaded by register-run. + Defaults to the per-job pattern register-run uses. Downloaded and + merged before reconcile. Leave empty when the suite passed an + in-workspace ledger via ledger-path instead. + required: false + type: string + default: 'cascade-run-ledger-*' + ledger-path: + description: >- + Path to an in-workspace ledger (single-job suites). Used only when no + artifact is found. Empty by default. + required: false + type: string + default: '' + cascade-ref: + description: >- + The cascade ref to check out for the reconcile core (the rc tag under + test, or a branch/sha). When empty, falls back to github.workflow_sha, + the SHA of this reusable workflow as resolved for the caller, so the + Go core always matches the gate version the caller pinned. + required: false + type: string + default: '' + allow-workflows: + description: >- + Comma-separated workflow names to reconcile. Empty (default) means all + cascade-generated workflows in the window. Set this to scope out an + unrelated sibling workflow that shares the repo. + required: false + type: string + default: '' + run-list-limit: + description: 'Max runs to fetch per page from gh run list (paginated to capture all).' + required: false + type: number + default: 200 + +permissions: + contents: read + +jobs: + reconcile: + name: Reconcile scenario-window runs + runs-on: ubuntu-latest + permissions: + contents: read + actions: read + steps: + # Check out cascade itself for the reconcile core. The reusable workflow + # is referenced at a ref by the caller; we re-check-out cascade at that + # same ref (or an override) so the Go core matches the gate version. + - name: Check out cascade (reconcile core) + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + repository: stablekernel/cascade + ref: ${{ inputs.cascade-ref || github.workflow_sha }} + + - name: Set up Go + uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.4.0 + with: + go-version-file: go.mod + cache: true + + # Merge every per-job ledger artifact into one JSONL file. merge-multiple + # concatenates same-named files across artifacts; the glob captures every + # registering job's upload. Missing artifacts are tolerated (a suite that + # gated nothing, or one using the in-workspace ledger path instead). + - name: Download run-ledger artifacts + id: ledger + continue-on-error: true + uses: actions/download-artifact@634f93cb2916e3fdff6788551b99b062d0335ce0 # v5.0.0 + with: + pattern: ${{ inputs.ledger-artifact }} + path: ${{ runner.temp }}/ledger-artifacts + merge-multiple: true + + - name: Assemble the ledger + id: assemble + env: + ART_DIR: ${{ runner.temp }}/ledger-artifacts + LEDGER_PATH_IN: ${{ inputs.ledger-path }} + run: | + set -euo pipefail + LEDGER="${RUNNER_TEMP}/cascade-run-ledger.jsonl" + : > "$LEDGER" + + # Prefer downloaded artifacts; merge-multiple may have produced one or + # more .jsonl files. Concatenate them all. + if [ -d "$ART_DIR" ]; then + found=$(find "$ART_DIR" -type f -name '*.jsonl' | wc -l | tr -d ' ') + if [ "$found" -gt 0 ]; then + find "$ART_DIR" -type f -name '*.jsonl' -exec cat {} + >> "$LEDGER" + fi + fi + + # Fall back to an in-workspace ledger (single-job suite path). + if [ ! -s "$LEDGER" ] && [ -n "$LEDGER_PATH_IN" ] && [ -f "$LEDGER_PATH_IN" ]; then + cat "$LEDGER_PATH_IN" >> "$LEDGER" + fi + + lines=$(grep -c . "$LEDGER" || true) + echo "ledger has ${lines:-0} registered run(s)" + echo "ledger-path=$LEDGER" >> "$GITHUB_OUTPUT" + + # Enumerate EVERY run created in this repo since window-start, paginating + # so a busy scenario window is fully captured. gh run list paginates with + # --limit; we loop, advancing nothing (the API has no cursor on this + # endpoint), so we request a high limit and de-dup defensively. + - name: Enumerate scenario-window runs + id: runs + env: + GH_TOKEN: ${{ github.token }} + REPO: ${{ github.repository }} + WINDOW_START: ${{ inputs.window-start }} + LIMIT: ${{ inputs.run-list-limit }} + run: | + set -euo pipefail + RUNS_JSON="${RUNNER_TEMP}/scenario-runs.json" + + # gh run list --created ">=" filters server-side to the window. + # --limit caps results; we set it high and page by walking creation + # time: fetch a page, and if it is full, narrow to runs strictly older + # than the oldest seen and fetch again, accumulating. Run ids dedupe + # the boundary overlap. + fetch_page() { + local since="$1" + gh run list --repo "$REPO" \ + --created ">=${since}" \ + --limit "$LIMIT" \ + --json databaseId,workflowName,event,conclusion,status,headBranch,createdAt + } + + acc="[]" + since="$WINDOW_START" + for _ in $(seq 1 20); do + page=$(fetch_page "$since") + count=$(echo "$page" | jq 'length') + acc=$(jq -s '.[0] + .[1] | unique_by(.databaseId)' \ + <(echo "$acc") <(echo "$page")) + if [ "$count" -lt "$LIMIT" ]; then + break + fi + # Page was full: advance the window to the oldest createdAt in this + # page so the next request walks further back. (>= is inclusive, so + # the oldest run reappears and is de-duped by databaseId.) + since=$(echo "$page" | jq -r 'min_by(.createdAt) | .createdAt') + done + + echo "$acc" > "$RUNS_JSON" + total=$(jq 'length' "$RUNS_JSON") + echo "captured ${total} run(s) in the scenario window" + echo "runs-path=$RUNS_JSON" >> "$GITHUB_OUTPUT" + + - name: Reconcile + env: + LEDGER: ${{ steps.assemble.outputs.ledger-path }} + RUNS: ${{ steps.runs.outputs.runs-path }} + SELF_RUN_ID: ${{ github.run_id }} + ALLOW: ${{ inputs.allow-workflows }} + run: | + set -euo pipefail + # Exit 0 = every run accounted for; 1 = a coverage gap reds the gate; + # 2 = tool/input error. The report is printed to the job log and the + # step summary so a red gate names the unaccounted run. + set +e + OUT=$(go run ./internal/fleetreconcile/cmd \ + --runs "$RUNS" \ + --ledger "$LEDGER" \ + --self-run-id "$SELF_RUN_ID" \ + --allow-workflows "$ALLOW") + code=$? + set -e + + echo "$OUT" + { + echo "## Fleet Reconcile" + echo "" + echo '```' + echo "$OUT" + echo '```' + } >> "$GITHUB_STEP_SUMMARY" + + exit "$code" diff --git a/internal/fleetreconcile/cmd/main.go b/internal/fleetreconcile/cmd/main.go new file mode 100644 index 00000000..a3879331 --- /dev/null +++ b/internal/fleetreconcile/cmd/main.go @@ -0,0 +1,123 @@ +// Command fleet-reconcile is the runnable wrapper around the fleetreconcile +// core. The fleet-reconcile reusable workflow invokes it with `go run` after a +// scenario suite finishes: it reads the run ledger (JSONL, one LedgerEntry per +// line) and the `gh run list` JSON the workflow captured for the scenario +// window, classifies every run, prints the report, and exits non-zero if any +// run is unaccounted for. +// +// This is fleet maintainer tooling, not part of cascade's generated output. +package main + +import ( + "bufio" + "encoding/json" + "flag" + "fmt" + "os" + "strings" + + "github.com/stablekernel/cascade/internal/fleetreconcile" +) + +func main() { + if err := run(os.Args[1:], os.Stdout); err != nil { + fmt.Fprintf(os.Stderr, "fleet-reconcile: %v\n", err) + os.Exit(2) // 2 = tool error (bad input); 1 = gate failed; 0 = gate passed + } +} + +func run(args []string, out *os.File) error { + fs := flag.NewFlagSet("fleet-reconcile", flag.ContinueOnError) + ledgerPath := fs.String("ledger", "", "path to the run-ledger JSONL file (empty = no registered runs)") + runsPath := fs.String("runs", "", "path to the gh-run-list JSON array for the scenario window (required)") + selfRunID := fs.Int64("self-run-id", 0, "this reconcile run's own id, excluded from reconciliation") + allow := fs.String("allow-workflows", "", "comma-separated workflow names to reconcile; empty = all") + if err := fs.Parse(args); err != nil { + return err + } + if *runsPath == "" { + return fmt.Errorf("--runs is required") + } + + runs, err := readRuns(*runsPath) + if err != nil { + return fmt.Errorf("reading runs: %w", err) + } + ledger, err := readLedger(*ledgerPath) + if err != nil { + return fmt.Errorf("reading ledger: %w", err) + } + + opts := fleetreconcile.Options{SelfRunID: *selfRunID} + if names := splitNonEmpty(*allow); len(names) > 0 { + opts.AllowWorkflows = make(map[string]struct{}, len(names)) + for _, n := range names { + opts.AllowWorkflows[n] = struct{}{} + } + } + + rep := fleetreconcile.Reconcile(runs, ledger, opts) + if _, err := fmt.Fprint(out, fleetreconcile.FormatReport(rep)); err != nil { + return fmt.Errorf("writing report: %w", err) + } + if !rep.Passed() { + os.Exit(1) + } + return nil +} + +func readRuns(path string) ([]fleetreconcile.Run, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var runs []fleetreconcile.Run + if err := json.Unmarshal(data, &runs); err != nil { + return nil, fmt.Errorf("parsing run-list JSON: %w", err) + } + return runs, nil +} + +// readLedger parses a JSONL ledger: one JSON LedgerEntry per non-blank line. +// A missing or empty path yields no entries (a suite that gated nothing). +func readLedger(path string) ([]fleetreconcile.LedgerEntry, error) { + if path == "" { + return nil, nil + } + f, err := os.Open(path) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + defer func() { _ = f.Close() }() + + var entries []fleetreconcile.LedgerEntry + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + line := 0 + for sc.Scan() { + line++ + text := strings.TrimSpace(sc.Text()) + if text == "" { + continue + } + var e fleetreconcile.LedgerEntry + if err := json.Unmarshal([]byte(text), &e); err != nil { + return nil, fmt.Errorf("ledger line %d: %w", line, err) + } + entries = append(entries, e) + } + return entries, sc.Err() +} + +func splitNonEmpty(s string) []string { + var out []string + for _, p := range strings.Split(s, ",") { + if p = strings.TrimSpace(p); p != "" { + out = append(out, p) + } + } + return out +} diff --git a/internal/fleetreconcile/fleetreconcile.go b/internal/fleetreconcile/fleetreconcile.go new file mode 100644 index 00000000..236cd9da --- /dev/null +++ b/internal/fleetreconcile/fleetreconcile.go @@ -0,0 +1,243 @@ +// Package fleetreconcile holds the run-ledger reconcile core that backs the +// fleet-reconcile reusable workflow. It is fleet maintainer tooling, not part +// of cascade's generated output: a scenario suite registers every run it gates +// (its id and the conclusion it expects) into a ledger, and after the suite +// finishes this core enumerates every run the repo produced in the scenario +// window and fails if any run is unaccounted for. +// +// Keeping the decision logic here (rather than inline in YAML) makes the gate +// unit-testable against synthetic run lists with no live GitHub, so the core +// guarantee - an unregistered failing run reds the gate - is proven by a table +// test rather than asserted by hand. +package fleetreconcile + +import ( + "fmt" + "sort" + "strings" +) + +// LedgerEntry is one line a suite appended via the register-run action: the +// run it gated, the conclusion it expects that run to reach, and a short tag +// describing which scenario step registered it. +type LedgerEntry struct { + RunID int64 `json:"run_id"` + Expected string `json:"expected"` // "success" or "failure" + Reason string `json:"reason"` +} + +// Run is the subset of `gh run list --json ...` fields the reconcile needs. +// Field names mirror the gh JSON keys so the workflow can pass the raw list +// straight through. +type Run struct { + DatabaseID int64 `json:"databaseId"` + WorkflowName string `json:"workflowName"` + Event string `json:"event"` + Conclusion string `json:"conclusion"` // success|failure|cancelled|skipped|"" (in-flight) + Status string `json:"status"` // completed|in_progress|queued + HeadBranch string `json:"headBranch"` +} + +// Verdict classifies one run against the ledger. Outcome buckets are mutually +// exclusive so the report can never double-count a run. +type Verdict struct { + Run Run + Bucket Bucket + Detail string // human-readable why, for the report + Expects string // for accounted runs, the registered expectation +} + +// Bucket is the reconcile classification of a single run. +type Bucket int + +const ( + // Accounted: the run is in the ledger and its actual conclusion matched + // the registered expectation (success-matched or failure-matched). + Accounted Bucket = iota + // BenignUnregistered: not in the ledger but allowed - a success or a + // skipped run needs no explicit registration. + BenignUnregistered + // SupersededCancelled: a cancelled/skipped run provably superseded by a + // later run of the same workflow + concurrency lane. Never a failure. + SupersededCancelled + // InFlight: the run has not concluded yet (e.g. the suite's own run, or a + // late sibling). Not counted as a gap; reported for transparency. + InFlight + // FailGap: an unregistered non-success (the fire-and-forget the gate + // exists to catch), or a registered run whose conclusion did not match + // its expectation. These fail the gate. + FailGap +) + +// Report is the full reconcile outcome over a window of runs. +type Report struct { + Verdicts []Verdict + // Failing is the subset of Verdicts in the FailGap bucket, in input order. + Failing []Verdict +} + +// Passed reports whether the gate should exit zero (no failing runs). +func (r Report) Passed() bool { return len(r.Failing) == 0 } + +// Options tune the classification without weakening it. +type Options struct { + // SelfRunID is this reconcile run's own id; it is always excluded so the + // gate never reconciles against itself. Zero means "no self to exclude". + SelfRunID int64 + // AllowWorkflows, when non-empty, scopes reconcile to runs whose + // WorkflowName is in the set. A run of any other workflow is ignored + // (an unrelated sibling workflow's runs are out of scope). Empty means + // "reconcile every workflow's runs in the window". + AllowWorkflows map[string]struct{} +} + +// Reconcile classifies every run against the ledger and returns the report. +// It never lets a `failure` conclusion be skipped as benign or superseded: +// only `cancelled`/`skipped` runs are eligible for the superseded path, and +// only when a strictly-later registered (accounted) run shares the same +// (workflow, concurrency lane) so the cancellation is provably a supersede, +// not a masked failure. +func Reconcile(runs []Run, ledger []LedgerEntry, opts Options) Report { + byID := make(map[int64]LedgerEntry, len(ledger)) + for _, e := range ledger { + byID[e.RunID] = e + } + + rep := Report{} + for _, r := range runs { + if opts.SelfRunID != 0 && r.DatabaseID == opts.SelfRunID { + continue + } + if len(opts.AllowWorkflows) > 0 { + if _, ok := opts.AllowWorkflows[r.WorkflowName]; !ok { + continue + } + } + + v := classify(r, byID, runs) + rep.Verdicts = append(rep.Verdicts, v) + if v.Bucket == FailGap { + rep.Failing = append(rep.Failing, v) + } + } + return rep +} + +func classify(r Run, ledger map[int64]LedgerEntry, all []Run) Verdict { + // Registered runs: the conclusion MUST match the expectation. A + // registered expected:failure that actually failed is accounted; a + // registered expected:success that failed is a hard gap. + if e, ok := ledger[r.DatabaseID]; ok { + if r.Status != "completed" || r.Conclusion == "" { + return Verdict{Run: r, Bucket: InFlight, Expects: e.Expected, + Detail: fmt.Sprintf("registered (%s) but still %s", e.Reason, statusLabel(r))} + } + if r.Conclusion == e.Expected { + return Verdict{Run: r, Bucket: Accounted, Expects: e.Expected, + Detail: fmt.Sprintf("registered %s, got %s (%s)", e.Expected, r.Conclusion, e.Reason)} + } + return Verdict{Run: r, Bucket: FailGap, Expects: e.Expected, + Detail: fmt.Sprintf("registered run expected %s but concluded %s (%s)", e.Expected, r.Conclusion, e.Reason)} + } + + // Unregistered runs. + if r.Status != "completed" || r.Conclusion == "" { + return Verdict{Run: r, Bucket: InFlight, + Detail: fmt.Sprintf("unregistered run still %s", statusLabel(r))} + } + + switch r.Conclusion { + case "success", "skipped": + return Verdict{Run: r, Bucket: BenignUnregistered, + Detail: fmt.Sprintf("unregistered %s (benign)", r.Conclusion)} + case "cancelled": + if supersededBy := findSupersedingRun(r, ledger, all); supersededBy != 0 { + return Verdict{Run: r, Bucket: SupersededCancelled, + Detail: fmt.Sprintf("cancelled, superseded by registered run %d (same %s lane)", supersededBy, r.WorkflowName)} + } + return Verdict{Run: r, Bucket: FailGap, + Detail: "cancelled but no superseding registered run in the same lane (could mask a real failure)"} + default: // failure, timed_out, action_required, neutral, startup_failure, ... + return Verdict{Run: r, Bucket: FailGap, + Detail: fmt.Sprintf("unregistered non-success run (%s) created in the scenario window but never registered", r.Conclusion)} + } +} + +// findSupersedingRun returns the id of a registered (in-ledger) run that +// provably supersedes the cancelled run r: same workflow and the same +// concurrency lane (head branch), with a strictly greater run id (run ids are +// monotonic, so a larger id is a later run). It only consults registered runs +// so a cancelled run cannot be excused by another stray cancelled run - the +// superseder must itself be an asserted run. Returns 0 when none qualifies. +func findSupersedingRun(r Run, ledger map[int64]LedgerEntry, all []Run) int64 { + byID := make(map[int64]Run, len(all)) + for _, x := range all { + byID[x.DatabaseID] = x + } + var best int64 + for id := range ledger { + later, ok := byID[id] + if !ok { + continue + } + if later.WorkflowName != r.WorkflowName { + continue + } + if !sameLane(later, r) { + continue + } + if later.DatabaseID > r.DatabaseID && later.DatabaseID > best { + best = later.DatabaseID + } + } + return best +} + +// sameLane reports whether two runs share a concurrency lane. GitHub keys a +// generated workflow's concurrency on the head ref (branch), so a later run on +// the same branch+workflow is the run that cancelled the earlier queued one. +func sameLane(a, b Run) bool { return a.HeadBranch == b.HeadBranch } + +func statusLabel(r Run) string { + if r.Status != "" { + return r.Status + } + return "in-flight" +} + +// FormatReport renders a deterministic, human-readable reconcile report. The +// failing set is printed last and most loudly so a red gate is unambiguous. +func FormatReport(rep Report) string { + var b strings.Builder + + groups := map[Bucket][]Verdict{} + for _, v := range rep.Verdicts { + groups[v.Bucket] = append(groups[v.Bucket], v) + } + + section := func(title string, bucket Bucket) { + vs := groups[bucket] + fmt.Fprintf(&b, "%s (%d):\n", title, len(vs)) + sort.Slice(vs, func(i, j int) bool { return vs[i].Run.DatabaseID < vs[j].Run.DatabaseID }) + for _, v := range vs { + fmt.Fprintf(&b, " - run %d [%s] %s\n", v.Run.DatabaseID, v.Run.WorkflowName, v.Detail) + } + } + + section("Accounted (registered + matched)", Accounted) + section("Benign unregistered (success/skipped)", BenignUnregistered) + section("Superseded-cancelled (concurrency)", SupersededCancelled) + section("In-flight (not yet concluded)", InFlight) + + fmt.Fprintf(&b, "FAILING - coverage gaps (%d):\n", len(rep.Failing)) + for _, v := range rep.Failing { + fmt.Fprintf(&b, " - run %d [%s] %s\n", v.Run.DatabaseID, v.Run.WorkflowName, v.Detail) + } + + if rep.Passed() { + b.WriteString("RESULT: PASS - every run in the scenario window is accounted for.\n") + } else { + fmt.Fprintf(&b, "RESULT: FAIL - %d run(s) unaccounted for. See FAILING above.\n", len(rep.Failing)) + } + return b.String() +} diff --git a/internal/fleetreconcile/fleetreconcile_test.go b/internal/fleetreconcile/fleetreconcile_test.go new file mode 100644 index 00000000..8c519503 --- /dev/null +++ b/internal/fleetreconcile/fleetreconcile_test.go @@ -0,0 +1,195 @@ +package fleetreconcile + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +// completed is a helper for a finished run. +func completed(id int64, wf, conclusion, branch string) Run { + return Run{DatabaseID: id, WorkflowName: wf, Event: "push", Status: "completed", Conclusion: conclusion, HeadBranch: branch} +} + +func TestReconcile(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + runs []Run + ledger []LedgerEntry + opts Options + wantPass bool + wantFailIDs []int64 + }{ + { + // (a) every run registered and matching -> PASS. + name: "all registered and matched passes", + runs: []Run{ + completed(1, "Orchestrate", "success", "main"), + completed(2, "Promote", "success", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 1, Expected: "success", Reason: "push-orchestrate"}, + {RunID: 2, Expected: "success", Reason: "promote-hop"}, + }, + wantPass: true, + }, + { + // (b) THE CORE GUARANTEE: an unregistered failure in the window + // fails the gate. This is the fire-and-forget the gate exists to + // catch (e.g. the 3env hotfix Finalize run). + name: "unregistered failure fails the gate", + runs: []Run{ + completed(1, "Orchestrate", "success", "main"), + completed(99, "Cascade Hotfix", "failure", "main"), // never registered + }, + ledger: []LedgerEntry{ + {RunID: 1, Expected: "success", Reason: "push-orchestrate"}, + }, + wantPass: false, + wantFailIDs: []int64{99}, + }, + { + // (c1) a registered expected:failure that actually failed -> PASS + // (negative guard, e.g. divergence-promote guard). + name: "registered expected-failure that failed passes", + runs: []Run{ + completed(7, "Promote", "failure", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 7, Expected: "failure", Reason: "divergence-guard"}, + }, + wantPass: true, + }, + { + // (c2) a registered expected:success that actually failed -> FAIL. + name: "registered expected-success that failed fails", + runs: []Run{ + completed(8, "Release", "failure", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 8, Expected: "success", Reason: "release-publish"}, + }, + wantPass: false, + wantFailIDs: []int64{8}, + }, + { + // A registered expected:failure that unexpectedly SUCCEEDED is also + // a gap: the negative guard did not fire (never let a guard pass by + // going green when it should have refused). + name: "registered expected-failure that succeeded fails", + runs: []Run{ + completed(9, "Promote", "success", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 9, Expected: "failure", Reason: "downgrade-guard"}, + }, + wantPass: false, + wantFailIDs: []int64{9}, + }, + { + // Unregistered success and skipped runs are benign. + name: "unregistered success and skipped are benign", + runs: []Run{ + completed(10, "Orchestrate", "success", "main"), + completed(11, "Cascade PR Preview", "skipped", "feature"), + }, + wantPass: true, + }, + { + // A cancelled run provably superseded by a later REGISTERED run of + // the same workflow + lane is skipped (GHA concurrency cancels the + // older queued run). Never a false positive. + name: "cancelled superseded by later registered run is benign", + runs: []Run{ + completed(20, "Orchestrate", "cancelled", "main"), + completed(21, "Orchestrate", "success", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 21, Expected: "success", Reason: "push-orchestrate-latest"}, + }, + wantPass: true, + }, + { + // A cancelled run with NO superseding registered run in its lane is + // a gap: it could be masking a real cancellation we never asserted. + name: "cancelled with no superseder fails", + runs: []Run{ + completed(30, "Orchestrate", "cancelled", "main"), + }, + wantPass: false, + wantFailIDs: []int64{30}, + }, + { + // A failure is NEVER excused as superseded, even if a later + // registered run shares the lane. Failures always red the gate + // unless explicitly registered as expected:failure. + name: "failure is never treated as superseded", + runs: []Run{ + completed(40, "Orchestrate", "failure", "main"), + completed(41, "Orchestrate", "success", "main"), + }, + ledger: []LedgerEntry{ + {RunID: 41, Expected: "success", Reason: "push-orchestrate-latest"}, + }, + wantPass: false, + wantFailIDs: []int64{40}, + }, + { + // The reconcile run excludes itself and ignores out-of-scope + // workflows (an unrelated sibling's runs do not red the gate). + name: "self-run excluded and out-of-scope workflow ignored", + runs: []Run{ + completed(50, "Fleet Reconcile", "failure", "main"), // self (in-flight in reality; failure here proves exclusion) + completed(51, "Some Unrelated CI", "failure", "main"), // not in allowlist + completed(52, "Orchestrate", "success", "main"), + }, + ledger: []LedgerEntry{{RunID: 52, Expected: "success", Reason: "orchestrate"}}, + opts: Options{ + SelfRunID: 50, + AllowWorkflows: map[string]struct{}{"Orchestrate": {}}, + }, + wantPass: true, + }, + { + // An unregistered run still in-flight is not a gap (reported only). + name: "in-flight unregistered run is not a gap", + runs: []Run{ + {DatabaseID: 60, WorkflowName: "Orchestrate", Status: "in_progress", Conclusion: "", HeadBranch: "main"}, + }, + wantPass: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + rep := Reconcile(tt.runs, tt.ledger, tt.opts) + require.Equal(t, tt.wantPass, rep.Passed(), "pass/fail mismatch\n%s", FormatReport(rep)) + + var gotFail []int64 + for _, v := range rep.Failing { + gotFail = append(gotFail, v.Run.DatabaseID) + } + require.ElementsMatch(t, tt.wantFailIDs, gotFail, "failing set mismatch\n%s", FormatReport(rep)) + }) + } +} + +// TestFormatReport_FailIsLoud proves the rendered report names the failing run +// and states a FAIL result so a red gate is never ambiguous in the log. +func TestFormatReport_FailIsLoud(t *testing.T) { + t.Parallel() + rep := Reconcile( + []Run{completed(99, "Cascade Hotfix", "failure", "main")}, + nil, + Options{}, + ) + out := FormatReport(rep) + require.False(t, rep.Passed()) + require.Contains(t, out, "run 99") + require.Contains(t, out, "RESULT: FAIL") + require.True(t, strings.Contains(out, "never registered")) +} From 05289ab0cb7282be66a012abcae703f78ac79349 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Tue, 23 Jun 2026 20:21:15 -0400 Subject: [PATCH 2/2] fix(fleet): fail closed on truncated run enumeration and missing ledger Signed-off-by: Joshua Temple --- .github/workflows/fleet-reconcile.yaml | 121 +++++----- internal/fleetreconcile/cmd/main.go | 97 +++++++- internal/fleetreconcile/enumerate.go | 123 ++++++++++ internal/fleetreconcile/enumerate_test.go | 212 ++++++++++++++++++ internal/fleetreconcile/fleetreconcile.go | 5 + .../fleetreconcile/fleetreconcile_test.go | 17 ++ 6 files changed, 516 insertions(+), 59 deletions(-) create mode 100644 internal/fleetreconcile/enumerate.go create mode 100644 internal/fleetreconcile/enumerate_test.go diff --git a/.github/workflows/fleet-reconcile.yaml b/.github/workflows/fleet-reconcile.yaml index 495d49f2..c63a260a 100644 --- a/.github/workflows/fleet-reconcile.yaml +++ b/.github/workflows/fleet-reconcile.yaml @@ -54,6 +54,20 @@ on: required: false type: string default: '' + require-ledger: + description: >- + When true (the default), the gate requires a ledger to be present: a + missing ledger artifact (or, with ledger-path, a missing/empty file) + fails the gate instead of reconciling against an empty ledger. This is + fail-closed - a ledger that merely failed to download must never look + like "no registered runs" and let an expected:failure run that + actually succeeded pass as benign. Set false ONLY for a suite that + intentionally registers nothing (every run in its window is meant to + be benign-unregistered); such a suite has no expected:failure runs to + mis-pass. + required: false + type: boolean + default: true cascade-ref: description: >- The cascade ref to check out for the reconcile core (the rc tag under @@ -72,7 +86,11 @@ on: type: string default: '' run-list-limit: - description: 'Max runs to fetch per page from gh run list (paginated to capture all).' + description: >- + Page size for the strict-backward run enumeration (the gh run list + --limit per page). The Go core pages until the window is exhausted and + fails closed on truncation, so this is a per-page size, not a hard cap + on total runs reconciled. required: false type: number default: 200 @@ -121,13 +139,17 @@ jobs: env: ART_DIR: ${{ runner.temp }}/ledger-artifacts LEDGER_PATH_IN: ${{ inputs.ledger-path }} + REQUIRE_LEDGER: ${{ inputs.require-ledger }} + DOWNLOAD_OUTCOME: ${{ steps.ledger.outcome }} run: | set -euo pipefail LEDGER="${RUNNER_TEMP}/cascade-run-ledger.jsonl" : > "$LEDGER" # Prefer downloaded artifacts; merge-multiple may have produced one or - # more .jsonl files. Concatenate them all. + # more .jsonl files. Concatenate them all. Count what we found so the + # empty case can be distinguished from a present-but-empty ledger. + found=0 if [ -d "$ART_DIR" ]; then found=$(find "$ART_DIR" -type f -name '*.jsonl' | wc -l | tr -d ' ') if [ "$found" -gt 0 ]; then @@ -136,80 +158,71 @@ jobs: fi # Fall back to an in-workspace ledger (single-job suite path). + used_path=false if [ ! -s "$LEDGER" ] && [ -n "$LEDGER_PATH_IN" ] && [ -f "$LEDGER_PATH_IN" ]; then cat "$LEDGER_PATH_IN" >> "$LEDGER" + used_path=true fi lines=$(grep -c . "$LEDGER" || true) - echo "ledger has ${lines:-0} registered run(s)" + echo "ledger has ${lines:-0} registered run(s) (artifact files: ${found}, download outcome: ${DOWNLOAD_OUTCOME})" + + # Fail-closed: when a ledger is required (default), a download that + # failed or produced no ledger must NOT pass as "no registered runs". + # An expected:failure run whose ledger entry merely failed to download + # would otherwise look benign-unregistered and wrongly pass the gate. + if [ "$REQUIRE_LEDGER" = "true" ]; then + if [ -n "$LEDGER_PATH_IN" ]; then + # In-workspace mode: the named ledger must exist and be non-empty. + if [ "$used_path" != "true" ] || [ ! -s "$LEDGER" ]; then + echo "::error::fleet-reconcile: require-ledger is set but no ledger was found at ledger-path '${LEDGER_PATH_IN}'. Refusing to reconcile against an empty ledger (an expected:failure run that succeeded could pass as benign). Set require-ledger:false only for a suite that registers nothing." + exit 1 + fi + else + # Artifact mode: the download must have succeeded and yielded at + # least one ledger file. A missing artifact is treated as an error + # (a suite that registers runs must always upload its ledger). + if [ "$DOWNLOAD_OUTCOME" != "success" ] || [ "$found" -eq 0 ]; then + echo "::error::fleet-reconcile: require-ledger is set but no ledger artifact was downloaded (outcome: ${DOWNLOAD_OUTCOME}, files: ${found}). Refusing to reconcile against an empty ledger (an expected:failure run that succeeded could pass as benign). Ensure the suite uploads its ledger, or set require-ledger:false for a suite that registers nothing." + exit 1 + fi + fi + fi + echo "ledger-path=$LEDGER" >> "$GITHUB_OUTPUT" - # Enumerate EVERY run created in this repo since window-start, paginating - # so a busy scenario window is fully captured. gh run list paginates with - # --limit; we loop, advancing nothing (the API has no cursor on this - # endpoint), so we request a high limit and de-dup defensively. - - name: Enumerate scenario-window runs - id: runs + # Enumerate EVERY run created in this repo since window-start AND reconcile + # in one binary call. The enumeration/pagination lives in the unit-tested + # Go core (internal/fleetreconcile.EnumerateRuns): it pages strictly + # backward by (createdAt, run id) so a boundary-timestamp cluster cannot + # stall the walk, dedupes by run id, and - critically - FAILS CLOSED if the + # page cap is reached on a full page or a same-timestamp cluster cannot be + # paged. It never reconciles a truncated window. The binary shells out to + # `gh run list --created ">=window-start" ...` for each page. + - name: Enumerate and reconcile scenario-window runs env: GH_TOKEN: ${{ github.token }} REPO: ${{ github.repository }} WINDOW_START: ${{ inputs.window-start }} LIMIT: ${{ inputs.run-list-limit }} - run: | - set -euo pipefail - RUNS_JSON="${RUNNER_TEMP}/scenario-runs.json" - - # gh run list --created ">=" filters server-side to the window. - # --limit caps results; we set it high and page by walking creation - # time: fetch a page, and if it is full, narrow to runs strictly older - # than the oldest seen and fetch again, accumulating. Run ids dedupe - # the boundary overlap. - fetch_page() { - local since="$1" - gh run list --repo "$REPO" \ - --created ">=${since}" \ - --limit "$LIMIT" \ - --json databaseId,workflowName,event,conclusion,status,headBranch,createdAt - } - - acc="[]" - since="$WINDOW_START" - for _ in $(seq 1 20); do - page=$(fetch_page "$since") - count=$(echo "$page" | jq 'length') - acc=$(jq -s '.[0] + .[1] | unique_by(.databaseId)' \ - <(echo "$acc") <(echo "$page")) - if [ "$count" -lt "$LIMIT" ]; then - break - fi - # Page was full: advance the window to the oldest createdAt in this - # page so the next request walks further back. (>= is inclusive, so - # the oldest run reappears and is de-duped by databaseId.) - since=$(echo "$page" | jq -r 'min_by(.createdAt) | .createdAt') - done - - echo "$acc" > "$RUNS_JSON" - total=$(jq 'length' "$RUNS_JSON") - echo "captured ${total} run(s) in the scenario window" - echo "runs-path=$RUNS_JSON" >> "$GITHUB_OUTPUT" - - - name: Reconcile - env: LEDGER: ${{ steps.assemble.outputs.ledger-path }} - RUNS: ${{ steps.runs.outputs.runs-path }} SELF_RUN_ID: ${{ github.run_id }} ALLOW: ${{ inputs.allow-workflows }} run: | set -euo pipefail # Exit 0 = every run accounted for; 1 = a coverage gap reds the gate; - # 2 = tool/input error. The report is printed to the job log and the - # step summary so a red gate names the unaccounted run. + # 2 = tool/input error (which includes a truncated/stalled enumeration, + # so a window we could not fully enumerate also reds the gate). The + # report is printed to the job log and the step summary so a red gate + # names the unaccounted run. set +e OUT=$(go run ./internal/fleetreconcile/cmd \ - --runs "$RUNS" \ + --window-start "$WINDOW_START" \ + --repo "$REPO" \ + --page-size "$LIMIT" \ --ledger "$LEDGER" \ --self-run-id "$SELF_RUN_ID" \ - --allow-workflows "$ALLOW") + --allow-workflows "$ALLOW" 2>&1) code=$? set -e diff --git a/internal/fleetreconcile/cmd/main.go b/internal/fleetreconcile/cmd/main.go index a3879331..90157104 100644 --- a/internal/fleetreconcile/cmd/main.go +++ b/internal/fleetreconcile/cmd/main.go @@ -14,6 +14,8 @@ import ( "flag" "fmt" "os" + "os/exec" + "sort" "strings" "github.com/stablekernel/cascade/internal/fleetreconcile" @@ -29,19 +31,23 @@ func main() { func run(args []string, out *os.File) error { fs := flag.NewFlagSet("fleet-reconcile", flag.ContinueOnError) ledgerPath := fs.String("ledger", "", "path to the run-ledger JSONL file (empty = no registered runs)") - runsPath := fs.String("runs", "", "path to the gh-run-list JSON array for the scenario window (required)") + runsPath := fs.String("runs", "", "path to a pre-fetched gh-run-list JSON array (mutually exclusive with --window-start)") + windowStart := fs.String("window-start", "", "ISO-8601 scenario window-start; enumerate runs via gh from here (mutually exclusive with --runs)") + repo := fs.String("repo", "", "owner/name repo to enumerate (required with --window-start)") + pageSize := fs.Int("page-size", 200, "gh run list page size for enumeration") + maxPages := fs.Int("max-pages", 50, "safety cap on enumeration pages; reaching it on a full page fails closed") selfRunID := fs.Int64("self-run-id", 0, "this reconcile run's own id, excluded from reconciliation") allow := fs.String("allow-workflows", "", "comma-separated workflow names to reconcile; empty = all") if err := fs.Parse(args); err != nil { return err } - if *runsPath == "" { - return fmt.Errorf("--runs is required") + if (*runsPath == "") == (*windowStart == "") { + return fmt.Errorf("exactly one of --runs or --window-start is required") } - runs, err := readRuns(*runsPath) + runs, err := loadRuns(*runsPath, *windowStart, *repo, *pageSize, *maxPages) if err != nil { - return fmt.Errorf("reading runs: %w", err) + return fmt.Errorf("loading runs: %w", err) } ledger, err := readLedger(*ledgerPath) if err != nil { @@ -66,6 +72,21 @@ func run(args []string, out *os.File) error { return nil } +// loadRuns returns the runs to reconcile: either a pre-fetched JSON array +// (runsPath, used by tests and callers that captured the list themselves) or, +// when windowStart is set, the full set enumerated from gh by paging strictly +// backward. Paging in Go (behind the unit-tested EnumerateRuns) means the gate +// fails closed on a truncated window instead of silently dropping runs. +func loadRuns(runsPath, windowStart, repo string, pageSize, maxPages int) ([]fleetreconcile.Run, error) { + if runsPath != "" { + return readRuns(runsPath) + } + if repo == "" { + return nil, fmt.Errorf("--repo is required with --window-start") + } + return fleetreconcile.EnumerateRuns(windowStart, pageSize, maxPages, ghFetcher(repo, windowStart, pageSize)) +} + func readRuns(path string) ([]fleetreconcile.Run, error) { data, err := os.ReadFile(path) if err != nil { @@ -78,6 +99,72 @@ func readRuns(path string) ([]fleetreconcile.Run, error) { return runs, nil } +// ghFetcher returns a PageFetcher backed by `gh run list`. gh filters +// server-side to the scenario window (--created ">=windowStart") and returns +// runs newest-first; the fetcher slices off everything at or newer than the +// cursor so each call yields a strictly-older page. Slicing locally (rather +// than relying on a server-side run-id bound gh does not offer) keeps the +// strict-backward cursor - including its id half - working through a +// tied-timestamp boundary cluster. +func ghFetcher(repo, windowStart string, pageSize int) fleetreconcile.PageFetcher { + return func(cursor fleetreconcile.PageCursor) ([]fleetreconcile.Run, error) { + // Narrow the server-side date range to the window's older end at the + // cursor's timestamp so each gh call returns a genuinely smaller slice + // and the --limit page boundary advances. The range is inclusive on + // both ends; the cursor's id half (applied below) drops the runs at the + // boundary timestamp we have already consumed, so a tied-timestamp + // boundary cluster still pages through cleanly. + created := ">=" + windowStart + if !cursor.IsZero() { + created = windowStart + ".." + cursor.CreatedAt + } + // #nosec G204 - repo, windowStart, and the cursor timestamp come from + // the trusted reusable workflow inputs and gh itself, not arbitrary + // user data; every gh arg is a fixed flag. + cmd := exec.Command("gh", "run", "list", + "--repo", repo, + "--created", created, + "--limit", fmt.Sprintf("%d", pageSize), + "--json", "databaseId,workflowName,event,conclusion,status,headBranch,createdAt") + out, err := cmd.Output() + if err != nil { + return nil, fmt.Errorf("gh run list: %w", err) + } + var all []fleetreconcile.Run + if err := json.Unmarshal(out, &all); err != nil { + return nil, fmt.Errorf("parsing gh run list JSON: %w", err) + } + // Newest-first by (createdAt, id) so the cursor slice is well defined. + sort.SliceStable(all, func(i, j int) bool { + if all[i].CreatedAt != all[j].CreatedAt { + return all[i].CreatedAt > all[j].CreatedAt + } + return all[i].DatabaseID > all[j].DatabaseID + }) + page := make([]fleetreconcile.Run, 0, pageSize) + for _, r := range all { + if !cursor.IsZero() && !runOlderThanCursor(r, cursor) { + continue + } + page = append(page, r) + if len(page) == pageSize { + break + } + } + return page, nil + } +} + +// runOlderThanCursor reports whether r sorts strictly older than the cursor in +// newest-first (createdAt, id) order. It mirrors the enumerator's ordering so +// the gh fetcher pages exactly as EnumerateRuns expects. +func runOlderThanCursor(r fleetreconcile.Run, c fleetreconcile.PageCursor) bool { + if r.CreatedAt != c.CreatedAt { + return r.CreatedAt < c.CreatedAt + } + return r.DatabaseID < c.DatabaseID +} + // readLedger parses a JSONL ledger: one JSON LedgerEntry per non-blank line. // A missing or empty path yields no entries (a suite that gated nothing). func readLedger(path string) ([]fleetreconcile.LedgerEntry, error) { diff --git a/internal/fleetreconcile/enumerate.go b/internal/fleetreconcile/enumerate.go new file mode 100644 index 00000000..fe04bb84 --- /dev/null +++ b/internal/fleetreconcile/enumerate.go @@ -0,0 +1,123 @@ +package fleetreconcile + +import "fmt" + +// PageCursor is a strict-backward paging position over runs ordered newest +// first by (CreatedAt, DatabaseID). A page fetched at a cursor returns only +// runs that sort strictly OLDER than the cursor: an earlier CreatedAt, or the +// same CreatedAt with a smaller DatabaseID. Using the (timestamp, id) pair as +// the cursor - rather than the timestamp alone - is what lets the walk advance +// through a cluster of runs that share one CreatedAt: a shared-timestamp +// boundary can never stall the loop, because the id half of the cursor still +// strictly decreases. +type PageCursor struct { + CreatedAt string + // DatabaseID is the run id at the cursor. Together with CreatedAt it forms + // the strict lower bound: the next page is everything ordered after this + // run in newest-first order. + DatabaseID int64 +} + +// IsZero reports whether the cursor is the initial "no upper bound" position, +// i.e. the first page (newest runs). +func (c PageCursor) IsZero() bool { return c.CreatedAt == "" && c.DatabaseID == 0 } + +// PageFetcher fetches one newest-first page of runs strictly older than the +// cursor. A zero cursor (PageCursor{}) means "the newest page, no upper bound". +// It returns at most a page-sized slice; a returned page shorter than the page +// size signals the source is exhausted. The real fetcher (in the cmd binary) +// shells out to `gh run list --created ">=" --json ...` and +// slices the result by the cursor; tests inject a synthetic fetcher. +type PageFetcher func(cursor PageCursor) ([]Run, error) + +// EnumerateRuns walks every run created at or after windowStart, strictly +// backward in (CreatedAt, DatabaseID) order, deduping by run id, and returns +// the full set. It is the fail-CLOSED replacement for the old inline 20-page +// `break` loop: it never returns a truncated window. +// +// windowStart is an ISO-8601 timestamp; runs with CreatedAt < windowStart are +// outside the scenario window and are excluded. pageSize is the fetcher's page +// limit (used to tell a final short page from a full one). maxPages caps the +// walk so a misbehaving source cannot loop forever. +// +// Fail-closed guarantee: if maxPages is reached while the last fetched page was +// still FULL (so more in-window runs may remain unenumerated), EnumerateRuns +// returns an error rather than a partial set. A truncated window could drop an +// unaccounted failing run and silently pass the gate, so truncation is always +// an error, never a quiet partial result. +func EnumerateRuns(windowStart string, pageSize, maxPages int, fetch PageFetcher) ([]Run, error) { + if pageSize <= 0 { + return nil, fmt.Errorf("pageSize must be positive, got %d", pageSize) + } + if maxPages <= 0 { + return nil, fmt.Errorf("maxPages must be positive, got %d", maxPages) + } + + seen := make(map[int64]struct{}) + var out []Run + cursor := PageCursor{} // start at the newest page + + for page := 0; page < maxPages; page++ { + runs, err := fetch(cursor) + if err != nil { + return nil, fmt.Errorf("fetching run page %d: %w", page+1, err) + } + + reachedOlder := false // saw a run older than the window: walk is complete + added := 0 // new in-window runs this page contributed + var oldest *Run // oldest (last, newest-first) run on this page, for the next cursor + for i := range runs { + r := runs[i] + oldest = &runs[i] + if r.CreatedAt < windowStart { + // Older than the window: everything from here back is out of + // scope, and the source is newest-first, so we are done. + reachedOlder = true + continue + } + if _, dup := seen[r.DatabaseID]; dup { + continue + } + seen[r.DatabaseID] = struct{}{} + out = append(out, r) + added++ + } + + // The window is fully enumerated once we have either seen a run older + // than window-start (the source is newest-first, so nothing older + // remains in window) or received a short page (the source is exhausted). + if reachedOlder || len(runs) < pageSize { + return out, nil + } + + // A full page that contributed no new in-window runs means the source + // cannot advance past this position - a tied-timestamp cluster larger + // than one page that the cursor's id half could not page through (the + // source returned only runs we have already consumed). We cannot prove + // the window is fully enumerated, so fail closed rather than stop early. + if added == 0 { + return nil, fmt.Errorf( + "run enumeration stalled: a full page yielded no new runs, so a "+ + "same-timestamp run cluster exceeds the %d-run page size and cannot "+ + "be fully paged (refusing to reconcile a partial window since %s)", + pageSize, windowStart) + } + + // The page was full and entirely within the window: more runs may + // remain. Advance the cursor strictly past the oldest run on this page. + // Because the cursor carries the run id, a boundary cluster sharing one + // CreatedAt cannot stall the walk - the id strictly decreases. + if oldest == nil { // defensive: a full page is never empty + return out, nil + } + cursor = PageCursor{CreatedAt: oldest.CreatedAt, DatabaseID: oldest.DatabaseID} + } + + // Reached the page cap with the last page still full: the window may be + // truncated. Fail closed - never reconcile a window we did not fully + // enumerate. + return nil, fmt.Errorf( + "run enumeration truncated: hit the %d-page safety cap with a full final page; "+ + "more runs may remain in the window since %s (refusing to reconcile a partial window)", + maxPages, windowStart) +} diff --git a/internal/fleetreconcile/enumerate_test.go b/internal/fleetreconcile/enumerate_test.go new file mode 100644 index 00000000..dc9f092a --- /dev/null +++ b/internal/fleetreconcile/enumerate_test.go @@ -0,0 +1,212 @@ +package fleetreconcile + +import ( + "sort" + "testing" + + "github.com/stretchr/testify/require" +) + +// fakeStore is an in-memory run store that mimics `gh run list`: it holds runs +// (each with a databaseId and createdAt) and serves pages newest-first, +// restricted to runs whose createdAt is <= the inclusive upper bound the +// enumerator asks for. It records every bound it was queried with so a test can +// prove the enumerator advanced (and did not stall re-fetching one page). +type fakeStore struct { + runs []Run + pageSize int + calls []string + fetches int +} + +func newFakeStore(pageSize int, runs []Run) *fakeStore { + cp := make([]Run, len(runs)) + copy(cp, runs) + // Newest-first by createdAt, then by id, so equal timestamps have a stable + // order the way the real API returns them. + sort.SliceStable(cp, func(i, j int) bool { + if cp[i].CreatedAt != cp[j].CreatedAt { + return cp[i].CreatedAt > cp[j].CreatedAt + } + return cp[i].DatabaseID > cp[j].DatabaseID + }) + return &fakeStore{runs: cp, pageSize: pageSize} +} + +// fetch serves one newest-first page strictly older than the given cursor: +// runs whose (createdAt, databaseId) sorts strictly before the cursor. A zero +// cursor means no upper bound (newest runs). This mirrors a real fetcher that +// requests `gh run list --created ">=windowStart"` and slices by a descending +// (createdAt, id) cursor, which advances even through a tied-timestamp cluster. +func (s *fakeStore) fetch(cursor PageCursor) ([]Run, error) { + s.fetches++ + s.calls = append(s.calls, cursor.CreatedAt) + var page []Run + for _, r := range s.runs { + if !cursor.IsZero() && !olderThan(r, cursor) { + continue + } + page = append(page, r) + if len(page) == s.pageSize { + break + } + } + return page, nil +} + +// olderThan reports whether run r sorts strictly after the cursor in +// newest-first order: an earlier createdAt, or the same createdAt with a +// smaller id. It is the test mirror of the enumerator's own ordering. +func olderThan(r Run, c PageCursor) bool { + if r.CreatedAt != c.CreatedAt { + return r.CreatedAt < c.CreatedAt + } + return r.DatabaseID < c.DatabaseID +} + +func ids(runs []Run) []int64 { + out := make([]int64, 0, len(runs)) + for _, r := range runs { + out = append(out, r.DatabaseID) + } + return out +} + +// TestEnumerateRuns_BoundaryClusterFullyCaptured proves the strict-backward +// enumerator captures a boundary-timestamp CLUSTER larger than one page: many +// runs share the exact createdAt at the page boundary. The old `>=` page loop +// stalled here (since never advanced past the shared timestamp); the new +// enumerator must page through the whole cluster and return every run. +func TestEnumerateRuns_BoundaryClusterFullyCaptured(t *testing.T) { + t.Parallel() + + // pageSize 3, but five runs share the boundary timestamp T2, which is the + // oldest timestamp on the first full page. A timestamp-only advance with an + // inclusive bound would keep re-serving the same cluster; the enumerator + // must use the id cursor within a tied-timestamp cluster to make progress. + const ( + windowStart = "2026-06-23T10:00:00Z" + t1 = "2026-06-23T12:00:00Z" + t2 = "2026-06-23T11:00:00Z" // the boundary cluster timestamp + ) + runs := []Run{ + completedAt(1, "Orchestrate", "success", "main", t1), + completedAt(2, "Orchestrate", "success", "main", t1), + completedAt(10, "Orchestrate", "failure", "main", t2), + completedAt(11, "Orchestrate", "failure", "main", t2), + completedAt(12, "Orchestrate", "failure", "main", t2), + completedAt(13, "Orchestrate", "failure", "main", t2), + completedAt(14, "Orchestrate", "failure", "main", t2), + } + store := newFakeStore(3, runs) + + got, err := EnumerateRuns(windowStart, 3, 50, store.fetch) + require.NoError(t, err) + + want := []int64{1, 2, 10, 11, 12, 13, 14} + gotIDs := ids(got) + sort.Slice(gotIDs, func(i, j int) bool { return gotIDs[i] < gotIDs[j] }) + require.Equal(t, want, gotIDs, "boundary cluster must be fully captured, not truncated") +} + +// TestEnumerateRuns_CapOverflowFailsClosed proves that when the safety page cap +// is reached while the last page was still FULL (more runs may remain), the +// enumerator returns an ERROR instead of a truncated set. Never reconcile a +// window we did not fully enumerate. +func TestEnumerateRuns_CapOverflowFailsClosed(t *testing.T) { + t.Parallel() + + // Far more runs than maxPages*pageSize can reach, all strictly inside the + // window and each with a distinct descending timestamp so paging always + // advances but never finishes within the cap. + const windowStart = "2026-06-23T00:00:00Z" + var runs []Run + for i := 0; i < 100; i++ { + // createdAt descends as i grows; all are > windowStart. + ts := descendingTS(i) + runs = append(runs, completedAt(int64(1000+i), "Orchestrate", "failure", "main", ts)) + } + store := newFakeStore(5, runs) + + _, err := EnumerateRuns(windowStart, 5, 3, store.fetch) // cap of 3 pages = 15 runs max, window has 100 + require.Error(t, err, "reaching the page cap on a full page must fail closed") + require.Contains(t, err.Error(), "truncat") +} + +// TestEnumerateRuns_DedupesAcrossPages proves a run that appears on two adjacent +// pages (the inclusive boundary re-serves it) is counted once. +func TestEnumerateRuns_DedupesAcrossPages(t *testing.T) { + t.Parallel() + + const ( + windowStart = "2026-06-23T00:00:00Z" + t1 = "2026-06-23T12:00:00Z" + t2 = "2026-06-23T11:00:00Z" + t3 = "2026-06-23T10:00:00Z" + ) + runs := []Run{ + completedAt(1, "Orchestrate", "success", "main", t1), + completedAt(2, "Orchestrate", "success", "main", t2), + completedAt(3, "Orchestrate", "failure", "main", t3), + } + store := newFakeStore(2, runs) // pageSize 2 forces a re-fetch with overlap + + got, err := EnumerateRuns(windowStart, 2, 50, store.fetch) + require.NoError(t, err) + + gotIDs := ids(got) + sort.Slice(gotIDs, func(i, j int) bool { return gotIDs[i] < gotIDs[j] }) + require.Equal(t, []int64{1, 2, 3}, gotIDs, "each run counted exactly once") +} + +// stalledStore models a source that cannot page past a tied-timestamp cluster +// larger than one page: it always returns the newest pageSize runs at the +// boundary timestamp, ignoring the cursor's id half (as `gh run list --created` +// would when a single timestamp has more runs than --limit). The enumerator +// must fail closed rather than stop early and miss the older cluster runs. +type stalledStore struct { + page []Run +} + +func (s *stalledStore) fetch(_ PageCursor) ([]Run, error) { return s.page, nil } + +// TestEnumerateRuns_UnpageableClusterFailsClosed proves that when a full page +// contributes no new runs (an oversized same-timestamp cluster the source +// cannot page through), the enumerator fails closed instead of silently +// returning a partial window. +func TestEnumerateRuns_UnpageableClusterFailsClosed(t *testing.T) { + t.Parallel() + + const t1 = "2026-06-23T11:00:00Z" + page := []Run{ + completedAt(3, "Orchestrate", "failure", "main", t1), + completedAt(2, "Orchestrate", "failure", "main", t1), + } + store := &stalledStore{page: page} // pageSize 2, always returns the same 2 + + _, err := EnumerateRuns("2026-06-23T00:00:00Z", 2, 50, store.fetch) + require.Error(t, err, "an unpageable same-timestamp cluster must fail closed") + require.Contains(t, err.Error(), "stall") +} + +// TestEnumerateRuns_StopsAtWindowStart proves the enumerator stops once it has +// reached runs older than the window: out-of-window runs are excluded and the +// walk does not page forever. +func TestEnumerateRuns_StopsAtWindowStart(t *testing.T) { + t.Parallel() + + const ( + windowStart = "2026-06-23T11:00:00Z" + inWindow = "2026-06-23T12:00:00Z" + preWindow = "2026-06-23T09:00:00Z" + ) + runs := []Run{ + completedAt(1, "Orchestrate", "success", "main", inWindow), + completedAt(2, "Orchestrate", "failure", "main", preWindow), // before window-start + } + store := newFakeStore(5, runs) + + got, err := EnumerateRuns(windowStart, 5, 50, store.fetch) + require.NoError(t, err) + require.Equal(t, []int64{1}, ids(got), "runs older than window-start are excluded") +} diff --git a/internal/fleetreconcile/fleetreconcile.go b/internal/fleetreconcile/fleetreconcile.go index 236cd9da..27443bac 100644 --- a/internal/fleetreconcile/fleetreconcile.go +++ b/internal/fleetreconcile/fleetreconcile.go @@ -36,6 +36,11 @@ type Run struct { Conclusion string `json:"conclusion"` // success|failure|cancelled|skipped|"" (in-flight) Status string `json:"status"` // completed|in_progress|queued HeadBranch string `json:"headBranch"` + // CreatedAt is the run's ISO-8601 creation timestamp (the gh `createdAt` + // key). It is the cursor the enumerator pages backward on. ISO-8601 UTC + // timestamps in the gh output sort lexicographically in time order, so the + // enumerator compares them as strings. + CreatedAt string `json:"createdAt"` } // Verdict classifies one run against the ledger. Outcome buckets are mutually diff --git a/internal/fleetreconcile/fleetreconcile_test.go b/internal/fleetreconcile/fleetreconcile_test.go index 8c519503..0f369a6f 100644 --- a/internal/fleetreconcile/fleetreconcile_test.go +++ b/internal/fleetreconcile/fleetreconcile_test.go @@ -1,6 +1,7 @@ package fleetreconcile import ( + "fmt" "strings" "testing" @@ -12,6 +13,22 @@ func completed(id int64, wf, conclusion, branch string) Run { return Run{DatabaseID: id, WorkflowName: wf, Event: "push", Status: "completed", Conclusion: conclusion, HeadBranch: branch} } +// completedAt is completed with an explicit createdAt for enumerator tests. +func completedAt(id int64, wf, conclusion, branch, createdAt string) Run { + r := completed(id, wf, conclusion, branch) + r.CreatedAt = createdAt + return r +} + +// descendingTS yields strictly descending ISO-8601 timestamps as i grows, so a +// set of runs built from increasing i has distinct, always-advancing createdAt +// values inside a single day. +func descendingTS(i int) string { + h := 23 - (i / 60) + m := 59 - (i % 60) + return fmt.Sprintf("2026-06-23T%02d:%02d:00Z", h, m) +} + func TestReconcile(t *testing.T) { t.Parallel()