amadan

carlos / platform Public

console: the self-serve rows, the cut and expiry predicates, and the reclaim rule's row evidence

Admission (atomic per instance), the open rows in the order the pass steps
them, per-box FIFO, the cut door's one write, the expiry decision bound to
its phase and deadline, abort, the finaliser (one terminal event), the
row's other columns, and MoveTerminalOrAbsent/HostHasIssuedRequest on the
store, which NewServer sets as selfServeRows.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude d66d055148cbbce4314a59db8cf203bbbb1f21eb parent 10133df
2 files changed, +381 −2
  • internal/console/selfserve_store.go +377 −0
  • internal/console/server.go +4 −2
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index a5d3ba9d2..856828622 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -21,8 +21,10 @@ import (
"errors"
"fmt"
"math"
+ "strings"
"time"
+ "github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
)
@@ -421,6 +423,165 @@ func ssWon(res sql.Result, err error) (bool, error) {
return n == 1, nil
}
+// ---- rows -------------------------------------------------------------------
+
+// AdmitSelfServeMove inserts m as the instance's open row, with its first
+// phase event, in one transaction. An instance with an open row admits
+// nothing: admitted is false and open is that row (the unique index on
+// (account, app, host) WHERE outcome = ” is the backstop, so two racing
+// admissions cannot both win).
+func (s *Store) AdmitSelfServeMove(m SelfServeMove) (admitted bool, open *SelfServeMove, err error) {
+ if m.ID == "" || m.Account == "" || m.App == "" || m.Host == "" {
+ return false, nil, fmt.Errorf("console: self-serve move admission needs an id, account, app and host")
+ }
+ if m.Phase == "" {
+ m.Phase = ssRequested
+ }
+ if m.CreatedAt.IsZero() {
+ m.CreatedAt = time.Now().UTC()
+ }
+ if m.UpdatedAt.IsZero() {
+ m.UpdatedAt = m.CreatedAt
+ }
+ if m.PhaseStartedAt.IsZero() {
+ m.PhaseStartedAt = m.CreatedAt
+ }
+ inc, err := ssUint("incarnation", m.Incarnation)
+ if err != nil {
+ return false, nil, err
+ }
+ participants, err := json.Marshal(m.Participants)
+ if err != nil {
+ return false, nil, err
+ }
+ tx, err := s.db.Begin()
+ if err != nil {
+ return false, nil, err
+ }
+ defer tx.Rollback()
+ res, err := tx.Exec(`INSERT INTO self_serve_moves
+ (id, account, app, host, incarnation, from_region, to_region, participants, display_name, requested_by,
+ phase, phase_started_at, step_deadline, window_deadline, phase_step, reason, created_at, updated_at)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT DO NOTHING`,
+ m.ID, m.Account, m.App, m.Host, inc, m.From, m.To, string(participants), m.DisplayName, m.RequestedBy,
+ m.Phase, ssNanos(m.PhaseStartedAt), ssNanos(m.StepDeadline), ssNanos(m.WindowDeadline), m.PhaseStep, m.Reason,
+ ssNanos(m.CreatedAt), ssNanos(m.UpdatedAt))
+ won, err := ssWon(res, err)
+ if err != nil {
+ return false, nil, err
+ }
+ if !won {
+ row := tx.QueryRow(`SELECT `+selfServeMoveCols+` FROM self_serve_moves WHERE account = ? AND app = ? AND host = ? AND outcome = ''`,
+ m.Account, m.App, m.Host)
+ open, err := scanSelfServeMove(row)
+ if errors.Is(err, sql.ErrNoRows) {
+ // The id is taken by a row that is not this instance's open
+ // one: an id collision, never an admission.
+ return false, nil, fmt.Errorf("console: self-serve move %s already exists", m.ID)
+ }
+ if err != nil {
+ return false, nil, err
+ }
+ return false, open, nil
+ }
+ if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: m.ID, Phase: m.Phase, At: m.CreatedAt, Reason: m.Reason}); err != nil {
+ return false, nil, err
+ }
+ return true, nil, tx.Commit()
+}
+
+// SelfServeMove is the row with id, or nil when there is none.
+func (s *Store) SelfServeMove(id string) (*SelfServeMove, error) {
+ m, err := scanSelfServeMove(s.db.QueryRow(`SELECT `+selfServeMoveCols+` FROM self_serve_moves WHERE id = ?`, id))
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ return m, err
+}
+
+// SelfServeMoveForHost is the instance's open row, and its newest terminal
+// row; either may be nil.
+func (s *Store) SelfServeMoveForHost(acct, app, host string) (open, terminal *SelfServeMove, err error) {
+ rows, err := s.selfServeMoves(`account = ? AND app = ? AND host = ? AND outcome = '' LIMIT 1`, acct, app, host)
+ if err != nil {
+ return nil, nil, err
+ }
+ if len(rows) > 0 {
+ open = &rows[0]
+ }
+ rows, err = s.selfServeMoves(`account = ? AND app = ? AND host = ? AND outcome <> '' ORDER BY created_at DESC, id DESC LIMIT 1`, acct, app, host)
+ if err != nil {
+ return nil, nil, err
+ }
+ if len(rows) > 0 {
+ terminal = &rows[0]
+ }
+ return open, terminal, nil
+}
+
+// selfServeOpenWhere is the rows the pass still has work on: open, or
+// terminal with a release, effects or the open record still owed.
+const selfServeOpenWhere = `(outcome = '' OR release_pending = 1 OR effects_done = 0 OR open_record_pending = 1)`
+
+// OpenSelfServeMoves are the rows the pass steps, ordered by created_at
+// then id: the order it steps them in, and the order per-box FIFO reads.
+func (s *Store) OpenSelfServeMoves() ([]SelfServeMove, error) {
+ return s.selfServeMoves(selfServeOpenWhere + ` ORDER BY created_at, id`)
+}
+
+// SelfServeRecoveryPending reports whether the journal holds anything the
+// pass still owes: an open row, a pending release, owed effects or open
+// record, or an undelivered event. Any of them puts the console in
+// recovery mode (selfServeMode).
+func (s *Store) SelfServeRecoveryPending() (bool, error) {
+ var n int
+ err := s.db.QueryRow(`SELECT (SELECT COUNT(*) FROM self_serve_moves WHERE ` + selfServeOpenWhere + `)
+ + (SELECT COUNT(*) FROM self_serve_events WHERE emitted = 0)`).Scan(&n)
+ return n > 0, err
+}
+
+// selfServeKeys are the boxes a row's participants reserve: their box ids,
+// or the sentinel alone when any participant is unnamed (the reservation
+// rule, offlineSelection.reservationKeys).
+func selfServeKeys(m *SelfServeMove) map[string]bool {
+ sel := offlineSelection{Source: m.Participants.Source, Target: m.Participants.Target}
+ keys := map[string]bool{}
+ for _, k := range sel.reservationKeys() {
+ keys[k] = true
+ }
+ return keys
+}
+
+// selfServeWaitsForEarlier is per-box FIFO: row may reserve only when no
+// older open row (by created_at, id) that shares any participant box is
+// still in requested or reserving. A sentinel row shares every box. Rows
+// with disjoint participants never wait on each other. open is
+// OpenSelfServeMoves' answer.
+func selfServeWaitsForEarlier(open []SelfServeMove, row *SelfServeMove) bool {
+ mine := selfServeKeys(row)
+ _, mineAll := mine[moves.OfflineSentinel]
+ for i := range open {
+ o := &open[i]
+ if o.ID == row.ID || o.Outcome != "" || (o.Phase != ssRequested && o.Phase != ssReserving) {
+ continue
+ }
+ if !o.CreatedAt.Before(row.CreatedAt) && !(o.CreatedAt.Equal(row.CreatedAt) && o.ID < row.ID) {
+ continue
+ }
+ theirs := selfServeKeys(o)
+ if _, all := theirs[moves.OfflineSentinel]; all || mineAll {
+ return true
+ }
+ for k := range theirs {
+ if mine[k] {
+ return true
+ }
+ }
+ }
+ return false
+}
+
// ---- the request lifecycle --------------------------------------------------
// selfServeRequest is the row's request with the cut state, read in one
@@ -598,6 +759,222 @@ WHERE id = ? AND phase = ? AND (abort_requested_at = 0 OR ? IN `+selfServeAbortA
return true, tx.Commit()
}
+// ---- deadlines, the cut and abort --------------------------------------------
+
+// SetSelfServeExpiry records (or clears) an expiry the worker saw, bound to
+// the phase and the exact deadline it read:
+//
+// WHERE phase = ? AND (window_deadline = ? for window | step_deadline = ? for step)
+// AND (kind <> 'window' OR cut_requested_at = 0)
+//
+// so a worker whose snapshot predates a successful cut cannot record the
+// obsolete confirmation expiry. False: the worker re-reads and decides
+// again. The expiry action later runs only while expiry_pending,
+// expiry_phase, expiry_kind and expiry_deadline still match.
+func (s *Store) SetSelfServeExpiry(id, phase string, kind string, sawDeadline time.Time, pending bool) (bool, error) {
+ var col string
+ switch kind {
+ case "window":
+ col = "window_deadline"
+ case "step":
+ col = "step_deadline"
+ default:
+ return false, fmt.Errorf("console: self-serve move %s: expiry kind %q is neither window nor step", id, kind)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ expiry_pending = ?, expiry_phase = ?, expiry_kind = ?, expiry_deadline = ?, updated_at = ?
+WHERE id = ? AND phase = ? AND `+col+` = ? AND (? <> 'window' OR cut_requested_at = 0)`,
+ ssBool(pending), phase, kind, ssNanos(sawDeadline), time.Now().UnixNano(),
+ id, phase, ssNanos(sawDeadline), kind))
+}
+
+// MarkSelfServeCut is the cut door's only write:
+//
+// WHERE phase = 'ready' AND abort_requested_at = 0 AND cut_requested_at = 0
+// AND expiry_pending = 0 AND request_expired = 0
+//
+// A confirmation expiry already committed wins (the door answers that the
+// move expired). The cut retires the confirmation deadline
+// (window_deadline = 0), starts the copy's wait (step_deadline), and gives
+// an unconsumed maintain request the same allowance, whether or not its
+// write has committed yet.
+func (s *Store) MarkSelfServeCut(id string, now time.Time, copyDeadline time.Time) (bool, error) {
+ if ssNanos(now) <= 0 {
+ return false, fmt.Errorf("console: self-serve move %s: a cut needs a time", id)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ cut_requested_at = ?, window_deadline = 0, step_deadline = ?,
+ request_deadline = CASE WHEN request_kind = 'maintain' AND request_issued <> 0 AND request_consumed = 0 THEN ? ELSE request_deadline END,
+ updated_at = ?
+WHERE id = ? AND phase = 'ready' AND abort_requested_at = 0 AND cut_requested_at = 0 AND expiry_pending = 0 AND request_expired = 0`,
+ ssNanos(now), ssNanos(copyDeadline), ssNanos(copyDeadline), ssNanos(now), id))
+}
+
+// RequestSelfServeAbort marks the abort requested while the row is in one
+// of phases and no abort is requested yet:
+//
+// WHERE phase IN (...) AND abort_requested_at = 0
+func (s *Store) RequestSelfServeAbort(id string, phases []string, now time.Time) (bool, error) {
+ if len(phases) == 0 {
+ return false, nil
+ }
+ if ssNanos(now) <= 0 {
+ return false, fmt.Errorf("console: self-serve move %s: an abort needs a time", id)
+ }
+ args := []any{ssNanos(now), ssNanos(now), id}
+ for _, p := range phases {
+ args = append(args, p)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET abort_requested_at = ?, updated_at = ?
+WHERE id = ? AND phase IN (?`+strings.Repeat(", ?", len(phases)-1)+`) AND abort_requested_at = 0`, args...))
+}
+
+// ---- the row's other columns ---------------------------------------------------
+
+func (s *Store) setSelfServe(what, id, set string, args ...any) error {
+ args = append(append(args, time.Now().UnixNano()), id)
+ res, err := s.db.Exec(`UPDATE self_serve_moves SET `+set+`, updated_at = ? WHERE id = ?`, args...)
+ return ssOne(res, err, what, id)
+}
+
+// SetSelfServeEstimate records the move's estimate: whole seconds and the
+// terms it was made from (JSON, Task 9).
+func (s *Store) SetSelfServeEstimate(id string, seconds int64, terms string) error {
+ return s.setSelfServe("estimate", id, `estimate_seconds = ?, estimate_terms = ?`, seconds, terms)
+}
+
+// RecordSelfServeHealth counts one health probe: attempts + 1, the
+// consecutive successes (reset by a failure), the first success kept once
+// set, the last probe's time and error.
+func (s *Store) RecordSelfServeHealth(id string, probeAt time.Time, ok bool, errText string) error {
+ return s.setSelfServe("health", id, `health_attempts = health_attempts + 1,
+ health_consecutive_ok = CASE WHEN ? THEN health_consecutive_ok + 1 ELSE 0 END,
+ health_first_ok_at = CASE WHEN ? AND health_first_ok_at = 0 THEN ? ELSE health_first_ok_at END,
+ health_last_probe_at = ?, health_last_error = ?`,
+ ssBool(ok), ssBool(ok), ssNanos(probeAt), ssNanos(probeAt), errText)
+}
+
+// SetSelfServeFinishGate records which finishing gate the move is at.
+func (s *Store) SetSelfServeFinishGate(id, gate string) error {
+ return s.setSelfServe("finish gate", id, `finish_gate = ?`, gate)
+}
+
+// SetSelfServeWarning records the move's warning (shown with the outcome).
+func (s *Store) SetSelfServeWarning(id, warning string) error {
+ return s.setSelfServe("warning", id, `warning = ?`, warning)
+}
+
+// SetSelfServeNotifiedSeq records the record_seq the serial was last bumped
+// for; a later tick bumps again while the record is ahead of it.
+func (s *Store) SetSelfServeNotifiedSeq(id string, seq uint64) error {
+ v, err := ssUint("notified_seq", seq)
+ if err != nil {
+ return err
+ }
+ return s.setSelfServe("notified seq", id, `notified_seq = ?`, v)
+}
+
+// SetSelfServeReopened records when the source reported reopened (abort).
+func (s *Store) SetSelfServeReopened(id string, at time.Time) error {
+ return s.setSelfServe("reopened", id, `source_reopened_at = ?`, ssNanos(at))
+}
+
+// SetSelfServeAbortFinished records when the abort's finish was written.
+func (s *Store) SetSelfServeAbortFinished(id string, at time.Time) error {
+ return s.setSelfServe("abort finished", id, `abort_finished_at = ?`, ssNanos(at))
+}
+
+// FinishSelfServeMove records the outcome, with release_pending = 1 and
+// effects_done = 0, consumes a confirmed request (the clearing write's),
+// and inserts the terminal event, in one transaction. Only an open row is
+// finished (WHERE outcome = ”): the finaliser run again after a crash
+// between its steps changes nothing and allocates no second terminal
+// event.
+func (s *Store) FinishSelfServeMove(id, outcome, reason, warning string, now time.Time) error {
+ if outcome == "" {
+ return fmt.Errorf("console: self-serve move %s: an outcome is required", id)
+ }
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ res, err := tx.Exec(`UPDATE self_serve_moves SET
+ outcome = ?, outcome_reason = ?, warning = CASE WHEN ? <> '' THEN ? ELSE warning END,
+ release_pending = 1, effects_done = 0,
+ request_consumed = CASE WHEN request_confirmed <> 0 THEN 1 ELSE request_consumed END,
+ updated_at = ?
+WHERE id = ? AND outcome = ''`, outcome, reason, warning, warning, ssNanos(now), id)
+ won, err := ssWon(res, err)
+ if err != nil {
+ return err
+ }
+ if !won {
+ var n int
+ if err := tx.QueryRow(`SELECT COUNT(*) FROM self_serve_moves WHERE id = ?`, id).Scan(&n); err != nil {
+ return err
+ }
+ if n == 0 {
+ return fmt.Errorf("console: finish: no self-serve move %q", id)
+ }
+ return nil // already finished
+ }
+ if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: id, Phase: ssDone, At: now, Reason: reason, Detail: outcome}); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// SetSelfServeEffectsDone records that the outcome's effects (the timings,
+// the member notice) are done.
+func (s *Store) SetSelfServeEffectsDone(id string) error {
+ return s.setSelfServe("effects done", id, `effects_done = 1`)
+}
+
+// ClearSelfServeRelease records that the move's reservations are released.
+func (s *Store) ClearSelfServeRelease(id string) error {
+ return s.setSelfServe("release", id, `release_pending = 0`)
+}
+
+// SetSelfServeOpenRecordPending records whether the open record
+// (moves.OpenRecord) is still owed.
+func (s *Store) SetSelfServeOpenRecordPending(id string, pending bool) error {
+ return s.setSelfServe("open record", id, `open_record_pending = ?`, ssBool(pending))
+}
+
+// ---- the reclaim rule's evidence (offlinereserve.go) ---------------------------
+
+var _ selfServeRowEvidence = (*Store)(nil)
+
+// MoveTerminalOrAbsent is reclaim rule (c): no row has id, or its row is
+// terminal with no release pending.
+func (s *Store) MoveTerminalOrAbsent(id string) (bool, error) {
+ var outcome string
+ var release int64
+ err := s.db.QueryRow(`SELECT outcome, release_pending FROM self_serve_moves WHERE id = ?`, id).Scan(&outcome, &release)
+ if errors.Is(err, sql.ErrNoRows) {
+ return true, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ return outcome != "" && release == 0, nil
+}
+
+// HostHasIssuedRequest is reclaim rule (d): a row for the host has an
+// issued request whose write may still be in flight. That is an open row
+// with request_issued set, or any row whose issued request was never
+// consumed (a terminal row's unconsumed request may still land). A
+// terminal row whose request was consumed issues nothing more, so it
+// cannot hold a reservation back for ever.
+func (s *Store) HostHasIssuedRequest(acct, app, host string) (bool, error) {
+ var n int
+ err := s.db.QueryRow(`SELECT COUNT(*) FROM self_serve_moves
+WHERE account = ? AND app = ? AND host = ? AND request_issued <> 0 AND (outcome = '' OR request_consumed = 0)`,
+ acct, app, host).Scan(&n)
+ return n > 0, err
+}
+
// ---- events -----------------------------------------------------------------
// InsertSelfServeEventTx inserts e inside tx, allocating its seq from the
diff --git a/internal/console/server.go b/internal/console/server.go
index d159aca2b..64ba53be2 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -882,8 +882,8 @@ type Server struct {
cutover *cutoverGate
appsMaterialized atomic.Bool
// The reclaim rule's evidence beyond the record (offlinereserve.go):
- // the self-serve rows (Task 4 sets it; nil is no rows) and the boxes'
- // cleanup reports (plan 4; nil settles no cleanup).
+ // the self-serve rows (NewServer sets the store; nil is no rows) and
+ // the boxes' cleanup reports (plan 4; nil settles no cleanup).
selfServeRows selfServeRowEvidence
cleanupReports cleanupReportEvidence
// The console's move pass (movesrecovery.go): its clock (nil is
@@ -1628,6 +1628,8 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
if s.lifetimesGoverned() {
s.readCutoverAtStart(context.Background())
}
+ // The reclaim rule reads the self-serve rows from this store.
+ s.selfServeRows = st
// The pre-accounts backfill: operators become owners of ops, every granted
// app name is claimed there, and non-operator grant holders are left out
// and REPORTED — a per-app grant was never authority over ops. Additive and