From df5400eb2715e66f6671b14aa3b91c270e5a06b5 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Mon, 29 Jun 2026 02:34:06 -0300 Subject: [PATCH] =?UTF-8?q?feat(state):=20C1=20=E2=80=94=20saga=5Fphase=20?= =?UTF-8?q?v2=20migration=20+=20resumability=20CRUD=20(spec=2008/09)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the up-saga resumability foundation that gates the whole orchestrator (C5): - state schema v2: the `saga_phase` table keyed by (ctx, workspace, scope, phase) with status/fingerprint/timestamps/error; forward-only, backup-before-migrate. - CRUD: StartPhase (upsert, clears prior state), SatisfyPhase, FailPhase, GetPhase, PhaseSatisfied (the fingerprint-matched skip check a re-run uses), PhasesFor, ClearPhase (compensation). All scoped by Docker context, intended under the flock. Race-clean; migration v1->v2 + phase lifecycle + fingerprint re-arm tested. First chunk of the nightly M2->M7 build (see PROGRESS.md). Co-Authored-By: Claude Opus 4.8 (1M context) --- PROGRESS.md | 2 +- internal/state/migrations.go | 20 ++++++ internal/state/saga.go | 126 +++++++++++++++++++++++++++++++++++ internal/state/saga_test.go | 90 +++++++++++++++++++++++++ 4 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 internal/state/saga.go create mode 100644 internal/state/saga_test.go diff --git a/PROGRESS.md b/PROGRESS.md index c10f63e..dee7e39 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -19,7 +19,7 @@ and verification. **Merge model:** PR-per-chunk, auto-merge on green CI. ## Frontier ### M2-remainder (core saga) -- [ ] C1 state v2 `saga_phase` table + CRUD — TODO *(kickoff)* +- [x] C1 state v2 `saga_phase` table + CRUD — REVIEW (PR `nightly/C1`) - [ ] C2 docker `ContainerInspect`/`ContainerLogs` + mock — TODO *(kickoff)* - [ ] C3a config `healthcheck:`/`hooks:` structs — TODO *(kickoff)* - [ ] C3b `internal/health` thin poller — BLOCKED (C2, C3a) diff --git a/internal/state/migrations.go b/internal/state/migrations.go index d824a16..4fb8e9d 100644 --- a/internal/state/migrations.go +++ b/internal/state/migrations.go @@ -13,6 +13,7 @@ type migration struct { // or remove a released one. var migrations = []migration{ {version: 1, stmt: schemaV1}, + {version: 2, stmt: schemaV2}, } // schemaV1 is the initial ledger (spec 08 §Tables). Every mutable row is scoped @@ -94,6 +95,25 @@ CREATE TABLE IF NOT EXISTS schema_version ( ); ` +// schemaV2 (spec 08 §saga_phase) adds the up-saga resumability table: one row per +// (workspace, scope, phase) recording its status + an input fingerprint, so a +// re-run skips satisfied phases and a crash mid-saga resumes deterministically. +const schemaV2 = ` +CREATE TABLE IF NOT EXISTS saga_phase ( + ctx TEXT NOT NULL, + workspace TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT '', -- '' = workspace-wide, else project name + phase TEXT NOT NULL, -- preflight|clone|network|shared|provision|secrets|generate|compose-up|hooks|... + status TEXT NOT NULL, -- pending|started|satisfied|failed + fingerprint TEXT NOT NULL DEFAULT '', -- SHA-256 of the phase's inputs + started_at TEXT, + satisfied_at TEXT, + error TEXT, + PRIMARY KEY (ctx, workspace, scope, phase), + FOREIGN KEY (ctx) REFERENCES docker_context(name) ON DELETE CASCADE +); +` + // migrate applies any pending migrations inside a transaction per step, backing // up the DB file before the first mutating step. Forward-only. func (db *DB) migrate() error { diff --git a/internal/state/saga.go b/internal/state/saga.go new file mode 100644 index 0000000..75a7b38 --- /dev/null +++ b/internal/state/saga.go @@ -0,0 +1,126 @@ +package state + +import ( + "database/sql" + "fmt" +) + +// Saga phase statuses. +const ( + PhasePending = "pending" + PhaseStarted = "started" + PhaseSatisfied = "satisfied" + PhaseFailed = "failed" +) + +// SagaPhase is one row of the saga_phase resumability table (spec 08/09). A phase +// is keyed by (workspace, scope, phase); scope is "" for workspace-wide phases or +// a project name for per-project ones. +type SagaPhase struct { + Workspace string + Scope string + Phase string + Status string + Fingerprint string + StartedAt string + SatisfiedAt string + Error string +} + +// StartPhase marks a phase started with the given input fingerprint (upsert). +// Clears any prior satisfied/failed/error state for a fresh attempt. Hold the lock. +func (db *DB) StartPhase(workspace, scope, phase, fingerprint string) error { + _, err := db.Exec(` + INSERT INTO saga_phase (ctx, workspace, scope, phase, status, fingerprint, started_at) + VALUES (?,?,?,?,?,?,datetime('now')) + ON CONFLICT(ctx, workspace, scope, phase) + DO UPDATE SET status=excluded.status, fingerprint=excluded.fingerprint, + started_at=excluded.started_at, satisfied_at=NULL, error=NULL`, + db.Ctx, workspace, scope, phase, PhaseStarted, fingerprint) + if err != nil { + return fmt.Errorf("start phase %s/%s/%s: %w", workspace, scope, phase, err) + } + return nil +} + +// SatisfyPhase marks a phase satisfied (completed). Hold the lock. +func (db *DB) SatisfyPhase(workspace, scope, phase string) error { + _, err := db.Exec(`UPDATE saga_phase SET status=?, satisfied_at=datetime('now'), error=NULL + WHERE ctx=? AND workspace=? AND scope=? AND phase=?`, + PhaseSatisfied, db.Ctx, workspace, scope, phase) + if err != nil { + return fmt.Errorf("satisfy phase %s/%s/%s: %w", workspace, scope, phase, err) + } + return nil +} + +// FailPhase marks a phase failed with an error message. Hold the lock. +func (db *DB) FailPhase(workspace, scope, phase, errMsg string) error { + _, err := db.Exec(`UPDATE saga_phase SET status=?, error=? + WHERE ctx=? AND workspace=? AND scope=? AND phase=?`, + PhaseFailed, errMsg, db.Ctx, workspace, scope, phase) + if err != nil { + return fmt.Errorf("fail phase %s/%s/%s: %w", workspace, scope, phase, err) + } + return nil +} + +// GetPhase returns a phase row, ok=false if it has never been recorded. +func (db *DB) GetPhase(workspace, scope, phase string) (SagaPhase, bool, error) { + var p SagaPhase + var started, satisfied, errMsg sql.NullString + err := db.QueryRow(`SELECT workspace, scope, phase, status, fingerprint, started_at, satisfied_at, error + FROM saga_phase WHERE ctx=? AND workspace=? AND scope=? AND phase=?`, + db.Ctx, workspace, scope, phase). + Scan(&p.Workspace, &p.Scope, &p.Phase, &p.Status, &p.Fingerprint, &started, &satisfied, &errMsg) + if err == sql.ErrNoRows { + return SagaPhase{}, false, nil + } + if err != nil { + return SagaPhase{}, false, fmt.Errorf("get phase: %w", err) + } + p.StartedAt, p.SatisfiedAt, p.Error = started.String, satisfied.String, errMsg.String + return p, true, nil +} + +// PhaseSatisfied reports whether a phase is already satisfied for the SAME input +// fingerprint — the skip check a re-run uses to avoid redoing unchanged work. A +// changed fingerprint (config/template/param edit) re-arms the phase. +func (db *DB) PhaseSatisfied(workspace, scope, phase, fingerprint string) (bool, error) { + p, ok, err := db.GetPhase(workspace, scope, phase) + if err != nil || !ok { + return false, err + } + return p.Status == PhaseSatisfied && p.Fingerprint == fingerprint, nil +} + +// PhasesFor returns all recorded phases for a workspace, ordered, for `status`. +func (db *DB) PhasesFor(workspace string) ([]SagaPhase, error) { + rows, err := db.Query(`SELECT workspace, scope, phase, status, fingerprint, started_at, satisfied_at, error + FROM saga_phase WHERE ctx=? AND workspace=? ORDER BY scope, phase`, db.Ctx, workspace) + if err != nil { + return nil, fmt.Errorf("phases for %s: %w", workspace, err) + } + defer rows.Close() + var out []SagaPhase + for rows.Next() { + var p SagaPhase + var started, satisfied, errMsg sql.NullString + if err := rows.Scan(&p.Workspace, &p.Scope, &p.Phase, &p.Status, &p.Fingerprint, &started, &satisfied, &errMsg); err != nil { + return nil, err + } + p.StartedAt, p.SatisfiedAt, p.Error = started.String, satisfied.String, errMsg.String + out = append(out, p) + } + return out, rows.Err() +} + +// ClearPhase removes a phase row (compensation / forced re-run). Hold the lock. +func (db *DB) ClearPhase(workspace, scope, phase string) error { + _, err := db.Exec(`DELETE FROM saga_phase WHERE ctx=? AND workspace=? AND scope=? AND phase=?`, + db.Ctx, workspace, scope, phase) + if err != nil { + return fmt.Errorf("clear phase %s/%s/%s: %w", workspace, scope, phase, err) + } + return nil +} diff --git a/internal/state/saga_test.go b/internal/state/saga_test.go new file mode 100644 index 0000000..c9dd82a --- /dev/null +++ b/internal/state/saga_test.go @@ -0,0 +1,90 @@ +package state + +import "testing" + +func TestSagaPhaseLifecycle(t *testing.T) { + db := openTestDB(t) + + // Unrecorded phase: not satisfied, GetPhase ok=false. + if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp1"); ok { + t.Fatal("unrecorded phase should not be satisfied") + } + if _, ok, _ := db.GetPhase("acme", "", "network"); ok { + t.Fatal("unrecorded phase should report ok=false") + } + + // Start → satisfied for fingerprint fp1. + if err := db.StartPhase("acme", "", "network", "fp1"); err != nil { + t.Fatal(err) + } + if err := db.SatisfyPhase("acme", "", "network"); err != nil { + t.Fatal(err) + } + if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp1"); !ok { + t.Error("phase should be satisfied for the same fingerprint") + } + // A changed fingerprint re-arms the phase (skip check fails). + if ok, _ := db.PhaseSatisfied("acme", "", "network", "fp2"); ok { + t.Error("a changed fingerprint must NOT count as satisfied") + } + + p, ok, err := db.GetPhase("acme", "", "network") + if err != nil || !ok { + t.Fatalf("get: ok=%v err=%v", ok, err) + } + if p.Status != PhaseSatisfied || p.SatisfiedAt == "" || p.StartedAt == "" { + t.Errorf("phase row = %+v", p) + } +} + +func TestSagaPhaseRestartClearsState(t *testing.T) { + db := openTestDB(t) + _ = db.StartPhase("acme", "api", "provision", "fpA") + _ = db.FailPhase("acme", "api", "provision", "boom") + + p, _, _ := db.GetPhase("acme", "api", "provision") + if p.Status != PhaseFailed || p.Error != "boom" { + t.Fatalf("expected failed with error, got %+v", p) + } + // Re-start clears the prior error + satisfied_at. + if err := db.StartPhase("acme", "api", "provision", "fpB"); err != nil { + t.Fatal(err) + } + p, _, _ = db.GetPhase("acme", "api", "provision") + if p.Status != PhaseStarted || p.Error != "" || p.Fingerprint != "fpB" { + t.Errorf("restart did not reset state: %+v", p) + } +} + +func TestSagaPhasesForAndClear(t *testing.T) { + db := openTestDB(t) + _ = db.StartPhase("acme", "", "network", "f") + _ = db.SatisfyPhase("acme", "", "network") + _ = db.StartPhase("acme", "api", "compose-up", "f") + + got, err := db.PhasesFor("acme") + if err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("phases = %d, want 2", len(got)) + } + + if err := db.ClearPhase("acme", "", "network"); err != nil { + t.Fatal(err) + } + if got, _ := db.PhasesFor("acme"); len(got) != 1 { + t.Fatalf("phases after clear = %d, want 1", len(got)) + } +} + +func TestSchemaVersionIsTwo(t *testing.T) { + db := openTestDB(t) + v, err := db.SchemaVersion() + if err != nil { + t.Fatal(err) + } + if v != 2 { + t.Errorf("schema version = %d, want 2 (saga_phase migration applied)", v) + } +}