amadan

carlos / platform Public

console: the self-serve journal's schema and its gate labels

self_serve_moves (one open row per instance by a partial unique index),
self_serve_events, self_serve_timings and move_timings (Task 9's schema)
in one migration. The three app-scoped tables are gated by account sqid
and app; move_timings names neither. The claim invariant names the new
writers no door's open run reaches.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 29cee09f591881d543eb63440b6db74e12123a7c parent 27c4f7f
4 files changed, +438 −1
  • internal/console/claim_invariant_test.go +6 −0
  • internal/console/selfserve_store.go +420 −0
  • internal/console/store.go +5 −1
  • internal/console/store_gate.go +7 −0
diff --git a/internal/console/claim_invariant_test.go b/internal/console/claim_invariant_test.go
index 3703ca8e7..77aaf7bf7 100644
--- a/internal/console/claim_invariant_test.go
+++ b/internal/console/claim_invariant_test.go
@@ -175,6 +175,12 @@ var claimWriterReasons = map[string]string{
"which no door reaches until plan 1 Task 3 adds the create door; an armed (claims-on) store refuses these writes without a permit, which Task 2's tests pin",
"sql/record_retirements": "written only by the claims-off instance-lifetime protocol's persisted intents (BeginAllocation, BeginRetirement and BeginMigration in lifetimes_store.go), " +
"which no door reaches until plan 1 Task 3 adds the create door; an armed (claims-on) store refuses these writes without a permit, which Task 2's tests pin",
+ "sql/self_serve_moves": "the self-serve moves journal (selfserve_store.go), written only by a claims-off console's self-serve door and its pass (StartSelfServeMovesPass), " +
+ "both off on a host-claim console (the switch needs claims off and the pass logs that and returns); an armed (claims-on) store refuses these writes without a permit",
+ "sql/self_serve_events": "the self-serve moves journal's events (selfserve_store.go), inserted with the row changes above and marked emitted only by the claims-off pass's delivery; " +
+ "off on a host-claim console like the journal, and refused without a permit by an armed store",
+ "sql/self_serve_timings": "the self-serve moves journal's timing marks (Task 9), written only by the claims-off pass; " +
+ "off on a host-claim console like the journal, and refused without a permit by an armed store",
}
// claimCoordinators are boundary methods that hold the gate for a caller and
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
new file mode 100644
index 000000000..d7b955abf
--- /dev/null
+++ b/internal/console/selfserve_store.go
@@ -0,0 +1,420 @@
+package console
+
+// The self-serve journal (self-serve moves plan 2, Task 4; spec §7): one
+// self_serve_moves row per move, keyed by the move id, the per-row event
+// sequence in self_serve_events, and the timing tables Task 9 fills.
+//
+// Every record write the job makes is a persisted request: the row holds
+// one request at a time (its kind, the phase that issued it, a nonce, the
+// incarnation and record_seq it is bound to and the stamp it proposes)
+// written before the write, and every lifecycle UPDATE below is predicated
+// on the nonce and the request's flags. Each predicate answers a named race
+// (a stale worker, a lost reply, a void after an expiry, a cut against a
+// stale snapshot), so none of them may be loosened.
+//
+// Timestamps are Unix nanoseconds; zero is "not yet".
+
+import (
+ "database/sql"
+ "encoding/json"
+ "fmt"
+ "math"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The phases a row moves through (spec §7).
+const (
+ ssRequested = "requested"
+ ssReserving = "reserving"
+ ssQueued = "queued"
+ ssReports = "reports"
+ ssStaging = "staging"
+ ssPrechecks = "prechecks"
+ ssReady = "ready"
+ ssRestage = "restage"
+ ssBarrier = "barrier"
+ ssStopping = "stopping"
+ ssSealing = "sealing"
+ ssLanding = "landing"
+ ssLanded = "landed"
+ ssOpenUncertain = "open-uncertain"
+ ssOpened = "opened"
+ ssHealth = "health"
+ ssFinishing = "finishing"
+ ssReverting = "reverting"
+ ssDone = "done"
+)
+
+// selfServeFirstSubstep is the substep a phase entered by the job's own
+// progression (AdvanceSelfServe) starts at: the phases that write more
+// than once dispatch on phase_step (Task 7). Every other phase has none.
+var selfServeFirstSubstep = map[string]string{
+ ssStopping: "start",
+ ssFinishing: "gate1",
+ ssPrechecks: "precheck",
+ ssReverting: "abort",
+ ssStaging: "round",
+ ssReady: "round",
+}
+
+// selfServeAbortExpiring are the request kinds whose expiry action is
+// abort: their request carries the issuing phase's step deadline. Every
+// other kind (gate1, gate2, finish, abort, the abort path's finish,
+// estimate) belongs to a phase that pages and retains, so its write must
+// always be allowed to complete and its request carries no deadline.
+var selfServeAbortExpiring = map[string]bool{
+ "queue": true, "stage": true, "maintain": true, "precheck": true,
+ "start": true, "seal": true, "flip": true, "open": true,
+}
+
+// selfServeAbortAdvances are the phases an advance may enter while an
+// abort is requested: the abort's own path, and the phases past open,
+// where the abort flag is obsolete (an open that committed is never
+// aborted).
+const selfServeAbortAdvances = `('reverting','done','opened','health','finishing')`
+
+// msgSelfServeWaitingEarlier is a reserving row's reason while an older
+// row sharing a box has not reserved yet (per-box FIFO).
+const msgSelfServeWaitingEarlier = "waiting for an earlier move on this box"
+
+// persistedRequest is the row's one request, loaded by one SELECT of the
+// request columns together with cut_requested_at (Task 7 compares the cut
+// state read with the request it is deciding about).
+type persistedRequest struct {
+ Kind, Phase, Nonce string
+ ExpectInc, ExpectSeq uint64
+ Proposed, Deadline, CutRequestedAt time.Time
+ Confirmed, Consumed, Voided bool
+ Expired bool
+}
+
+// SelfServeMove is one self_serve_moves row.
+type SelfServeMove struct {
+ ID, Account, App, Host string
+ Incarnation uint64
+ From, To string
+ Participants provision.Participants
+ DisplayName string
+ RequestedBy string
+
+ Phase string
+ PhaseStartedAt time.Time
+ StepDeadline, WindowDeadline time.Time
+ ExpiryPending bool
+ ExpiryPhase, ExpiryKind string
+ ExpiryDeadline time.Time
+ StagingStartedAt time.Time
+
+ // The outstanding (or last) request. RequestIssued and
+ // RequestConfirmedAt are the times behind Request's flags.
+ RequestIssued time.Time
+ Request persistedRequest
+ RequestConfirmedAt time.Time
+ RequestAbandoned bool
+
+ NextMaintenanceAt time.Time
+ CutRequestedAt time.Time
+ PhaseStep string
+ StageRequest time.Time
+ PrecheckRequest time.Time
+ Rounds int
+ EstimateSeconds int64
+ EstimateTerms string
+ Reason string
+
+ HealthAttempts int
+ HealthConsecutiveOK int
+ HealthFirstOKAt time.Time
+ HealthLastProbeAt time.Time
+ HealthLastError string
+ FinishGate string
+
+ NotifiedSeq uint64
+ EventSeq int64
+
+ Outcome, OutcomeReason, Warning string
+ AbortRequestedAt time.Time
+ SourceReopenedAt time.Time
+ AbortFinishedAt time.Time
+ EffectsDone bool
+ ReleasePending bool
+ OpenRecordPending bool
+
+ CreatedAt, UpdatedAt time.Time
+}
+
+// SelfServeEvent is one self_serve_events row: a phase change, refusal,
+// fact decision, lane or health probe or timing, sequenced per row.
+type SelfServeEvent struct {
+ ID string
+ Seq int64
+ Phase string
+ Account, App, Host string
+ At time.Time
+ Reason, Detail string
+ Emitted bool
+}
+
+// migrateSelfServe creates the journal's tables. Additive and idempotent,
+// called from Store.migrate. One migration owns all four: move_timings has
+// Task 9's schema, and Task 9 adds the methods on both timing tables.
+func (s *Store) migrateSelfServe() error {
+ for _, stmt := range []string{
+ `CREATE TABLE IF NOT EXISTS self_serve_moves (
+ id TEXT PRIMARY KEY,
+ account TEXT NOT NULL DEFAULT '',
+ app TEXT NOT NULL DEFAULT '',
+ host TEXT NOT NULL DEFAULT '',
+ incarnation INTEGER NOT NULL DEFAULT 0,
+ from_region TEXT NOT NULL DEFAULT '',
+ to_region TEXT NOT NULL DEFAULT '',
+ participants TEXT NOT NULL DEFAULT '',
+ display_name TEXT NOT NULL DEFAULT '',
+ requested_by TEXT NOT NULL DEFAULT '',
+ phase TEXT NOT NULL DEFAULT '',
+ phase_started_at INTEGER NOT NULL DEFAULT 0,
+ step_deadline INTEGER NOT NULL DEFAULT 0,
+ window_deadline INTEGER NOT NULL DEFAULT 0,
+ expiry_pending INTEGER NOT NULL DEFAULT 0,
+ expiry_phase TEXT NOT NULL DEFAULT '',
+ expiry_kind TEXT NOT NULL DEFAULT '',
+ expiry_deadline INTEGER NOT NULL DEFAULT 0,
+ staging_started_at INTEGER NOT NULL DEFAULT 0,
+ request_issued INTEGER NOT NULL DEFAULT 0,
+ request_kind TEXT NOT NULL DEFAULT '',
+ request_phase TEXT NOT NULL DEFAULT '',
+ request_nonce TEXT NOT NULL DEFAULT '',
+ request_expect_seq INTEGER NOT NULL DEFAULT 0,
+ request_proposed INTEGER NOT NULL DEFAULT 0,
+ request_confirmed INTEGER NOT NULL DEFAULT 0,
+ request_consumed INTEGER NOT NULL DEFAULT 0,
+ request_voided INTEGER NOT NULL DEFAULT 0,
+ request_abandoned INTEGER NOT NULL DEFAULT 0,
+ request_expect_inc INTEGER NOT NULL DEFAULT 0,
+ request_deadline INTEGER NOT NULL DEFAULT 0,
+ request_expired INTEGER NOT NULL DEFAULT 0,
+ next_maintenance_at INTEGER NOT NULL DEFAULT 0,
+ cut_requested_at INTEGER NOT NULL DEFAULT 0,
+ phase_step TEXT NOT NULL DEFAULT '',
+ stage_request INTEGER NOT NULL DEFAULT 0,
+ precheck_request INTEGER NOT NULL DEFAULT 0,
+ rounds INTEGER NOT NULL DEFAULT 0,
+ estimate_seconds INTEGER NOT NULL DEFAULT 0,
+ estimate_terms TEXT NOT NULL DEFAULT '',
+ reason TEXT NOT NULL DEFAULT '',
+ health_attempts INTEGER NOT NULL DEFAULT 0,
+ health_consecutive_ok INTEGER NOT NULL DEFAULT 0,
+ health_first_ok_at INTEGER NOT NULL DEFAULT 0,
+ health_last_probe_at INTEGER NOT NULL DEFAULT 0,
+ health_last_error TEXT NOT NULL DEFAULT '',
+ finish_gate TEXT NOT NULL DEFAULT '',
+ notified_seq INTEGER NOT NULL DEFAULT 0,
+ event_seq INTEGER NOT NULL DEFAULT 0,
+ outcome TEXT NOT NULL DEFAULT '',
+ outcome_reason TEXT NOT NULL DEFAULT '',
+ warning TEXT NOT NULL DEFAULT '',
+ abort_requested_at INTEGER NOT NULL DEFAULT 0,
+ source_reopened_at INTEGER NOT NULL DEFAULT 0,
+ abort_finished_at INTEGER NOT NULL DEFAULT 0,
+ effects_done INTEGER NOT NULL DEFAULT 0,
+ release_pending INTEGER NOT NULL DEFAULT 0,
+ open_record_pending INTEGER NOT NULL DEFAULT 0,
+ created_at INTEGER NOT NULL DEFAULT 0,
+ updated_at INTEGER NOT NULL DEFAULT 0
+);`,
+ // One open row per instance: admission's atomicity.
+ `CREATE UNIQUE INDEX IF NOT EXISTS self_serve_moves_open ON self_serve_moves(account, app, host) WHERE outcome = ''`,
+ `CREATE TABLE IF NOT EXISTS self_serve_events (
+ id TEXT NOT NULL,
+ seq INTEGER NOT NULL,
+ phase TEXT NOT NULL DEFAULT '',
+ account TEXT NOT NULL DEFAULT '',
+ app TEXT NOT NULL DEFAULT '',
+ host TEXT NOT NULL DEFAULT '',
+ at INTEGER NOT NULL DEFAULT 0,
+ reason TEXT NOT NULL DEFAULT '',
+ detail TEXT NOT NULL DEFAULT '',
+ emitted INTEGER NOT NULL DEFAULT 0,
+ PRIMARY KEY (id, seq)
+);`,
+ `CREATE TABLE IF NOT EXISTS self_serve_timings (
+ id TEXT NOT NULL,
+ mark TEXT NOT NULL,
+ account TEXT NOT NULL DEFAULT '',
+ app TEXT NOT NULL DEFAULT '',
+ host TEXT NOT NULL DEFAULT '',
+ at INTEGER NOT NULL DEFAULT 0,
+ bytes INTEGER NOT NULL DEFAULT 0,
+ PRIMARY KEY (id, mark)
+);`,
+ // The deployment's move timings (Task 9): one row per finished
+ // move and phase, keyed by move id; the estimate's medians read it.
+ `CREATE TABLE IF NOT EXISTS move_timings (
+ id TEXT NOT NULL,
+ phase TEXT NOT NULL,
+ from_region TEXT NOT NULL DEFAULT '',
+ to_region TEXT NOT NULL DEFAULT '',
+ bytes INTEGER NOT NULL DEFAULT 0,
+ seconds REAL NOT NULL DEFAULT 0,
+ at INTEGER NOT NULL DEFAULT 0,
+ PRIMARY KEY (id, phase)
+);`,
+ } {
+ if _, err := s.db.Exec(stmt); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// ---- encoding ---------------------------------------------------------------
+
+// ssNanos is t as the column holds it: Unix nanoseconds, 0 for zero.
+func ssNanos(t time.Time) int64 {
+ if t.IsZero() {
+ return 0
+ }
+ return t.UnixNano()
+}
+
+// ssTime reads a nanosecond column back; 0 is the zero time.
+func ssTime(n int64) time.Time {
+ if n == 0 {
+ return time.Time{}
+ }
+ return time.Unix(0, n).UTC()
+}
+
+// ssUint stores a counter, refusing one SQLite cannot hold.
+func ssUint(what string, v uint64) (int64, error) {
+ if v > math.MaxInt64 {
+ return 0, fmt.Errorf("console: self-serve move %s %d does not fit the console store", what, v)
+ }
+ return int64(v), nil
+}
+
+func ssStoredUint(what string, v int64) (uint64, error) {
+ if v < 0 {
+ return 0, fmt.Errorf("console: stored self-serve move %s %d is negative", what, v)
+ }
+ return uint64(v), nil
+}
+
+func ssBool(b bool) int64 {
+ if b {
+ return 1
+ }
+ return 0
+}
+
+const selfServeMoveCols = `id, account, app, host, incarnation, from_region, to_region, participants, display_name, requested_by,
+ phase, phase_started_at, step_deadline, window_deadline, expiry_pending, expiry_phase, expiry_kind, expiry_deadline, staging_started_at,
+ request_issued, request_kind, request_phase, request_nonce, request_expect_seq, request_proposed, request_confirmed, request_consumed,
+ request_voided, request_abandoned, request_expect_inc, request_deadline, request_expired, next_maintenance_at, cut_requested_at,
+ phase_step, stage_request, precheck_request, rounds, estimate_seconds, estimate_terms, reason,
+ health_attempts, health_consecutive_ok, health_first_ok_at, health_last_probe_at, health_last_error, finish_gate,
+ notified_seq, event_seq, outcome, outcome_reason, warning, abort_requested_at, source_reopened_at, abort_finished_at,
+ effects_done, release_pending, open_record_pending, created_at, updated_at`
+
+type ssScanner interface{ Scan(dest ...any) error }
+
+func scanSelfServeMove(r ssScanner) (*SelfServeMove, error) {
+ var (
+ m SelfServeMove
+ inc, expSeq, expInc, notified int64
+ participants string
+ phaseStarted, step, window, expiryDeadline, stagingStart int64
+ expiryPending, issued, proposed, confirmed, consumed int64
+ voided, abandoned, reqDeadline, expired, nextMaint, cut int64
+ stageReq, precheckReq, firstOK, lastProbe int64
+ abortAt, reopenedAt, abortDone, effects, release, openRec int64
+ created, updated int64
+ )
+ if err := r.Scan(&m.ID, &m.Account, &m.App, &m.Host, &inc, &m.From, &m.To, &participants, &m.DisplayName, &m.RequestedBy,
+ &m.Phase, &phaseStarted, &step, &window, &expiryPending, &m.ExpiryPhase, &m.ExpiryKind, &expiryDeadline, &stagingStart,
+ &issued, &m.Request.Kind, &m.Request.Phase, &m.Request.Nonce, &expSeq, &proposed, &confirmed, &consumed,
+ &voided, &abandoned, &expInc, &reqDeadline, &expired, &nextMaint, &cut,
+ &m.PhaseStep, &stageReq, &precheckReq, &m.Rounds, &m.EstimateSeconds, &m.EstimateTerms, &m.Reason,
+ &m.HealthAttempts, &m.HealthConsecutiveOK, &firstOK, &lastProbe, &m.HealthLastError, &m.FinishGate,
+ &notified, &m.EventSeq, &m.Outcome, &m.OutcomeReason, &m.Warning, &abortAt, &reopenedAt, &abortDone,
+ &effects, &release, &openRec, &created, &updated); err != nil {
+ return nil, err
+ }
+ var err error
+ if m.Incarnation, err = ssStoredUint("incarnation", inc); err != nil {
+ return nil, err
+ }
+ if m.Request.ExpectInc, err = ssStoredUint("expected incarnation", expInc); err != nil {
+ return nil, err
+ }
+ if m.Request.ExpectSeq, err = ssStoredUint("expected record_seq", expSeq); err != nil {
+ return nil, err
+ }
+ if m.NotifiedSeq, err = ssStoredUint("notified_seq", notified); err != nil {
+ return nil, err
+ }
+ if participants != "" {
+ if err := json.Unmarshal([]byte(participants), &m.Participants); err != nil {
+ return nil, fmt.Errorf("console: self-serve move %s participants: %w", m.ID, err)
+ }
+ }
+ m.PhaseStartedAt, m.StepDeadline, m.WindowDeadline = ssTime(phaseStarted), ssTime(step), ssTime(window)
+ m.ExpiryPending, m.ExpiryDeadline, m.StagingStartedAt = expiryPending != 0, ssTime(expiryDeadline), ssTime(stagingStart)
+ m.RequestIssued, m.RequestConfirmedAt, m.RequestAbandoned = ssTime(issued), ssTime(confirmed), abandoned != 0
+ m.Request.Proposed, m.Request.Deadline, m.Request.CutRequestedAt = ssTime(proposed), ssTime(reqDeadline), ssTime(cut)
+ m.Request.Confirmed, m.Request.Consumed, m.Request.Voided, m.Request.Expired = confirmed != 0, consumed != 0, voided != 0, expired != 0
+ m.NextMaintenanceAt, m.CutRequestedAt = ssTime(nextMaint), ssTime(cut)
+ m.StageRequest, m.PrecheckRequest = ssTime(stageReq), ssTime(precheckReq)
+ m.HealthFirstOKAt, m.HealthLastProbeAt = ssTime(firstOK), ssTime(lastProbe)
+ m.AbortRequestedAt, m.SourceReopenedAt, m.AbortFinishedAt = ssTime(abortAt), ssTime(reopenedAt), ssTime(abortDone)
+ m.EffectsDone, m.ReleasePending, m.OpenRecordPending = effects != 0, release != 0, openRec != 0
+ m.CreatedAt, m.UpdatedAt = ssTime(created), ssTime(updated)
+ return &m, nil
+}
+
+func (s *Store) selfServeMoves(where string, args ...any) ([]SelfServeMove, error) {
+ rows, err := s.db.Query(`SELECT `+selfServeMoveCols+` FROM self_serve_moves WHERE `+where, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []SelfServeMove
+ for rows.Next() {
+ m, err := scanSelfServeMove(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, *m)
+ }
+ return out, rows.Err()
+}
+
+// ssOne requires an UPDATE to have changed exactly one row: a setter on a
+// row that does not exist is an error, never a silent no-op.
+func ssOne(res sql.Result, err error, what, id string) error {
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n != 1 {
+ return fmt.Errorf("console: %s: no self-serve move %q", what, id)
+ }
+ return nil
+}
+
+// ssWon reports whether a predicated UPDATE matched its row.
+func ssWon(res sql.Result, err error) (bool, error) {
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return false, err
+ }
+ return n == 1, nil
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index c985833dc..ec6f0c5b1 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -572,7 +572,11 @@ func (s *Store) migrate() error {
return err
}
// Door keys' own tables (doorkeys_store.go), for the same reason.
- return s.migrateDoorKeys()
+ if err := s.migrateDoorKeys(); err != nil {
+ return err
+ }
+ // The self-serve moves journal (selfserve_store.go), for the same reason.
+ return s.migrateSelfServe()
}
// mailCountsBackfillVersion is the user_version this migration stamps. The
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go
index 8395d0f8c..d51976fcb 100644
--- a/internal/console/store_gate.go
+++ b/internal/console/store_gate.go
@@ -212,6 +212,12 @@ var claimSQLTables = map[string]claimSQLTable{
"host_allocations": {Sqid: "account", App: "app", Labels: []string{"app_lifetime"}},
"record_retirements": {Sqid: "account", App: "app"},
"record_migrations": {Sqid: "account", App: "app"},
+ // The self-serve moves journal (selfserve_store.go): a claims-off
+ // console's moves, their events and timing marks, keyed by the account
+ // sqid and app name.
+ "self_serve_moves": {Sqid: "account", App: "app"},
+ "self_serve_events": {Sqid: "account", App: "app"},
+ "self_serve_timings": {Sqid: "account", App: "app"},
// ---- not app-scoped: each with its reason ----
"sessions": {NonApp: "browser sessions, keyed by token hash and address"},
@@ -233,6 +239,7 @@ var claimSQLTables = map[string]claimSQLTable{
"claim_sql_identity": {NonApp: "private receipt database identity"},
"claim_sql_receipts": {NonApp: "private SQL effect receipts and fences"},
"pending_actions": {NonApp: "replay-by-token actions, keyed by token hash; carries no app column and re-enters the ordinary handlers"},
+ "move_timings": {NonApp: "the deployment's finished-move phase durations by region pair, keyed by move id and phase; names no account or app, read only for the move estimate's medians"},
}
// claimSQLReservedName is the TEMP permit table. A main-schema table by that