amadan

carlos / platform Public

console: the self-serve request lifecycle, each UPDATE predicated on its nonce

Issue (one request at a time, only from the owning phase, an abort-expiring
kind carrying the phase's step deadline), confirm, consume, abandon, void,
and the advance that consumes its confirmed request and inserts the phase
event in the same transaction.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 10133df267c26d4a7ba4f5e62b385187bf468e8d parent bc27b0e
1 file changed, +178 −0
  • internal/console/selfserve_store.go +178 −0
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index 02630c569..a5d3ba9d2 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -18,6 +18,7 @@ import (
"context"
"database/sql"
"encoding/json"
+ "errors"
"fmt"
"math"
"time"
@@ -420,6 +421,183 @@ func ssWon(res sql.Result, err error) (bool, error) {
return n == 1, nil
}
+// ---- the request lifecycle --------------------------------------------------
+
+// selfServeRequest is the row's request with the cut state, read in one
+// SELECT; nil when there is no row or no request was ever issued.
+func (s *Store) selfServeRequest(id string) (*persistedRequest, error) {
+ var (
+ pr persistedRequest
+ issued, expInc, expSeq, proposed, deadline, cut int64
+ confirmed, consumed, voided, expired int64
+ )
+ err := s.db.QueryRow(`SELECT request_issued, request_kind, request_phase, request_nonce, request_expect_inc, request_expect_seq,
+ request_proposed, request_deadline, cut_requested_at, request_confirmed, request_consumed, request_voided, request_expired
+FROM self_serve_moves WHERE id = ?`, id).Scan(&issued, &pr.Kind, &pr.Phase, &pr.Nonce, &expInc, &expSeq,
+ &proposed, &deadline, &cut, &confirmed, &consumed, &voided, &expired)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ if issued == 0 {
+ return nil, nil
+ }
+ if pr.ExpectInc, err = ssStoredUint("expected incarnation", expInc); err != nil {
+ return nil, err
+ }
+ if pr.ExpectSeq, err = ssStoredUint("expected record_seq", expSeq); err != nil {
+ return nil, err
+ }
+ pr.Proposed, pr.Deadline, pr.CutRequestedAt = ssTime(proposed), ssTime(deadline), ssTime(cut)
+ pr.Confirmed, pr.Consumed, pr.Voided, pr.Expired = confirmed != 0, consumed != 0, voided != 0, expired != 0
+ return &pr, nil
+}
+
+// IssueSelfServeRequest persists a new request on the row, before its
+// write, when the row is in phase, has no outcome and has no unconsumed
+// request:
+//
+// WHERE phase = ? AND outcome = '' AND (request_consumed <> 0 OR request_issued = 0)
+//
+// The request deadline is the row's step deadline for a kind whose expiry
+// action is abort, and 0 for every other kind. Not issued, existing is the
+// outstanding (issued, unconsumed) request whatever its kind and phase, or
+// nil when the row left the phase or ended.
+func (s *Store) IssueSelfServeRequest(id, kind, phase string, pr persistedRequest, now time.Time) (issued bool, existing *persistedRequest, err error) {
+ if pr.Nonce == "" || kind == "" || ssNanos(now) <= 0 {
+ return false, nil, fmt.Errorf("console: self-serve move %s: a request needs a nonce, a kind and a time", id)
+ }
+ inc, err := ssUint("expected incarnation", pr.ExpectInc)
+ if err != nil {
+ return false, nil, err
+ }
+ seq, err := ssUint("expected record_seq", pr.ExpectSeq)
+ if err != nil {
+ return false, nil, err
+ }
+ res, err := s.db.Exec(`UPDATE self_serve_moves SET
+ request_issued = ?, request_kind = ?, request_phase = ?, request_nonce = ?, request_expect_inc = ?, request_expect_seq = ?,
+ request_proposed = ?, request_deadline = CASE WHEN ? THEN step_deadline ELSE 0 END,
+ request_confirmed = 0, request_consumed = 0, request_voided = 0, request_abandoned = 0, request_expired = 0, updated_at = ?
+WHERE id = ? AND phase = ? AND outcome = '' AND (request_consumed <> 0 OR request_issued = 0)`,
+ ssNanos(now), kind, phase, pr.Nonce, inc, seq, ssNanos(pr.Proposed), ssBool(selfServeAbortExpiring[kind]), ssNanos(now),
+ id, phase)
+ won, err := ssWon(res, err)
+ if err != nil || won {
+ return won, nil, err
+ }
+ cur, err := s.selfServeRequest(id)
+ if err != nil || cur == nil || cur.Consumed {
+ return false, nil, err
+ }
+ return false, cur, nil
+}
+
+// ConfirmSelfServeRequest records that the record carries the request's
+// receipt:
+//
+// WHERE request_nonce = ? AND request_consumed = 0 AND request_voided = 0 AND request_confirmed = 0
+//
+// It also retires an expiry decision (request_expired = 0: a receipt that
+// arrives after a won expiry resolves the request successfully) and, for
+// stage and maintain, records stage_request, for precheck precheck_request,
+// as the proposed stamp. False: the caller re-reads the row; confirmed
+// already is a resume, voided or consumed is a stop.
+func (s *Store) ConfirmSelfServeRequest(id, nonce string, now time.Time) (bool, error) {
+ if ssNanos(now) <= 0 {
+ return false, fmt.Errorf("console: self-serve move %s: a confirmation needs a time", id)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ request_confirmed = ?, request_expired = 0,
+ stage_request = CASE WHEN request_kind IN ('stage','maintain') THEN request_proposed ELSE stage_request END,
+ precheck_request = CASE WHEN request_kind = 'precheck' THEN request_proposed ELSE precheck_request END,
+ updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_consumed = 0 AND request_voided = 0 AND request_confirmed = 0`,
+ ssNanos(now), ssNanos(now), id, nonce))
+}
+
+// ConsumeSelfServeRequest consumes a confirmed request whose wait is
+// satisfied, inside a phase that writes more than once, and in the same
+// statement moves to the next substep and gives its wait a fresh deadline:
+//
+// WHERE request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0
+func (s *Store) ConsumeSelfServeRequest(id, nonce, nextStep string, nextDeadline time.Time) (bool, error) {
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ request_consumed = 1, phase_step = ?, step_deadline = ?, updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0`,
+ nextStep, ssNanos(nextDeadline), time.Now().UnixNano(), id, nonce))
+}
+
+// AbandonSelfServeRequest gives up a confirmed request whose wait will
+// never be satisfied (abort, an aborting expiry, operator supersession),
+// and in the same statement gives the replacement wait its own deadline:
+//
+// WHERE request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0
+func (s *Store) AbandonSelfServeRequest(id, nonce string, nextDeadline time.Time) (bool, error) {
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ request_consumed = 1, request_abandoned = 1, step_deadline = ?, updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0`,
+ ssNanos(nextDeadline), time.Now().UnixNano(), id, nonce))
+}
+
+// VoidSelfServeRequest retires a request whose write did not land and
+// whose record moved on without its receipt. A confirmed request is never
+// voided:
+//
+// WHERE request_nonce = ? AND request_consumed = 0 AND request_confirmed = 0
+//
+// Matching nothing is not an error: the phase function decides again.
+func (s *Store) VoidSelfServeRequest(id, nonce string) error {
+ _, err := s.db.Exec(`UPDATE self_serve_moves SET
+ request_consumed = 1, request_voided = 1, updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_consumed = 0 AND request_confirmed = 0`,
+ time.Now().UnixNano(), id, nonce)
+ return err
+}
+
+// AdvanceSelfServe moves the row from one phase to the next and inserts
+// the phase event in the same transaction. One UPDATE, predicated
+//
+// phase = ? AND (abort_requested_at = 0 OR <to> IN ('reverting','done','opened','health','finishing'))
+// AND (<nonce> = '' OR (request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0))
+//
+// sets the phase, phase_started_at, the step and window deadlines, the new
+// phase's first substep and reason, clears an expiry bound to another
+// phase, and with a nonce consumes the request: consumption and the advance
+// are one statement, so a stale worker holding another nonce, or the same
+// nonce after a void, cannot advance.
+func (s *Store) AdvanceSelfServe(id, from, to, reason string, step, window time.Time, nonce string, now time.Time) (bool, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ res, err := tx.Exec(`UPDATE self_serve_moves SET
+ phase = ?, phase_started_at = ?, step_deadline = ?, window_deadline = ?, phase_step = ?, reason = ?,
+ expiry_pending = CASE WHEN expiry_phase <> ? THEN 0 ELSE expiry_pending END,
+ expiry_kind = CASE WHEN expiry_phase <> ? THEN '' ELSE expiry_kind END,
+ expiry_deadline = CASE WHEN expiry_phase <> ? THEN 0 ELSE expiry_deadline END,
+ expiry_phase = CASE WHEN expiry_phase <> ? THEN '' ELSE expiry_phase END,
+ request_consumed = CASE WHEN ? <> '' THEN 1 ELSE request_consumed END,
+ updated_at = ?
+WHERE id = ? AND phase = ? AND (abort_requested_at = 0 OR ? IN `+selfServeAbortAdvances+`)
+ AND (? = '' OR (request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0))`,
+ to, ssNanos(now), ssNanos(step), ssNanos(window), selfServeFirstSubstep[to], reason,
+ to, to, to, to,
+ nonce, ssNanos(now),
+ id, from, to, nonce, nonce)
+ won, err := ssWon(res, err)
+ if err != nil || !won {
+ return false, err
+ }
+ if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: id, Phase: to, At: now, Reason: reason}); err != nil {
+ return false, err
+ }
+ return true, tx.Commit()
+}
+
// ---- events -----------------------------------------------------------------
// InsertSelfServeEventTx inserts e inside tx, allocating its seq from the