amadan

carlos / platform Public

console: selfServeApply, the persisted request's bound write, confirmation by receipt and at-least-once notification

Every job record write is issued on the row first (nonce, expected
incarnation and record_seq, the proposed stamp), written once with
provision.SetRecordMoveAt under the lease's write context, confirmed only
by a judged read carrying the receipt, replayed as the same bound write,
voided when the record moved without it, and announced by a serial bump
recorded in notified_seq. An expired deadline is decided durably before
the write (ExpireSelfServeRequest); an expired request is never replayed.
console.TestHooks are the end-to-end witnesses' crash points and tick,
inert in production.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 9ef6e3204181ebb46108ab0bdcaeebc76df33ecb parent c2a059b
2 files changed, +432 −0
  • internal/console/selfserve_apply.go +358 −0
  • internal/console/testhooks.go +74 −0
diff --git a/internal/console/selfserve_apply.go b/internal/console/selfserve_apply.go
new file mode 100644
index 000000000..0f0608fb4
--- /dev/null
+++ b/internal/console/selfserve_apply.go
@@ -0,0 +1,358 @@
+package console
+
+// selfServeApply (self-serve moves plan 2 Task 7): every record transition
+// the self-serve job writes is a persisted request. The request (its kind,
+// the phase that issued it, a random nonce, the incarnation and record_seq
+// it is bound to and the exact stamp it proposes) is written to the row
+// before the record write; the write is provision.SetRecordMoveAt bound to
+// that incarnation and sequence, one CAS attempt with no fresh-read retry;
+// the mutation carries the proposed stamp verbatim and the nonce as its
+// receipt; the request is confirmed only by a judged read of the record
+// carrying the receipt; a replay is the same bound write again; and every
+// landed write is announced (the serial bump) at least once. A crash at any
+// point resumes, on the next tick, from the persisted request.
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// request is what a phase asks selfServeApply to write: the request's kind
+// (queue, stage, maintain, precheck, estimate, abort, finish, and Task 8's
+// start, seal, flip, open, finish1) and the phase issuing it. Deadline,
+// when set, is the wait's own allowance, written to the row's step deadline
+// with the request in one statement (the ready phase's maintenance round);
+// zero keeps the row's step deadline.
+type request struct {
+ Kind, Phase string
+ Deadline time.Time
+}
+
+// The ends of a step that selfServeApply can answer.
+var (
+ // errRequestRefused: the row left the phase or ended, or the request
+ // was consumed or voided under this worker. The step ends.
+ errRequestRefused = errors.New("console: self-serve move: the request was refused (the row moved on)")
+ // errRequestForeign: another request is outstanding. This caller never
+ // interprets it; reconciliation decides next tick.
+ errRequestForeign = errors.New("console: self-serve move: another request is outstanding")
+ // errRequestExpired: the request's deadline passed before it was
+ // confirmed and the expiry is decided. The step's fence-and-classify
+ // path takes over.
+ errRequestExpired = errors.New("console: self-serve move: the request expired before it was confirmed")
+ // errRequestVoided: the record advanced without the receipt (another
+ // writer); the request is voided.
+ errRequestVoided = errors.New("console: self-serve move: the record moved on without the request")
+ // errRecordGone: the record is absent, retired, stale or of another
+ // lifetime than the row's. No write bound to it can land.
+ errRecordGone = errors.New("console: self-serve move: the record is gone or was recreated")
+ // errReplayNeeded: reconciliation found the request's write did not
+ // land and the record is unchanged; the phase function replays it
+ // with its mutation.
+ errReplayNeeded = errors.New("console: self-serve move: the request's write has not landed")
+)
+
+// requestRefusal is the bound write's mutation refusing the very record it
+// was bound to (a provision.ErrMoveConflict or an open run): the replay
+// would refuse the same way, so it is not made. The request stays
+// outstanding; the phase decides (pre-offline: abort, which fences and
+// voids it).
+type requestRefusal struct{ err error }
+
+func (r *requestRefusal) Error() string {
+ return "console: self-serve move: the record refuses the request: " + r.err.Error()
+}
+func (r *requestRefusal) Unwrap() error { return r.err }
+
+// selfServeReplays bounds the replays of one bound write inside one
+// invocation; the next tick replays again, bounded by the lease and the
+// request's deadline.
+const selfServeReplays = 3
+
+// newRequestNonce is a request's identity: 16 random bytes.
+func newRequestNonce() (string, error) {
+ var b [16]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ return "", err
+ }
+ return "rq_" + hex.EncodeToString(b[:]), nil
+}
+
+// selfServeRecordOf is the row's record from a judged observation: an
+// unreadable or corrupt one is an error (the step waits); one that is
+// absent, retired, stale or of another incarnation than the row's is
+// errRecordGone.
+func selfServeRecordOf(obs provision.Observation, row *SelfServeMove) (*provision.Record, error) {
+ switch {
+ case obs.Verdict == provision.LifetimeUnreadable || obs.Verdict == provision.LifetimeCorrupt:
+ err := obs.Err
+ if err == nil {
+ err = provision.ErrLifetimeUnreadable
+ }
+ return nil, fmt.Errorf("console: self-serve move %s: the record: %w", row.ID, err)
+ case !obs.Present || obs.Verdict == provision.LifetimeStale || obs.Rec == nil || obs.Rec.Retired:
+ return nil, errRecordGone
+ case obs.Rec.Incarnation != row.Incarnation:
+ return obs.Rec, errRecordGone
+ }
+ return obs.Rec, nil
+}
+
+// selfServeHasReceipt reports whether rec carries the request's receipt:
+// the move's self_serve.last_request, or last_move.request for a write
+// that cleared the move.
+func selfServeHasReceipt(rec *provision.Record, nonce string) bool {
+ if rec == nil || nonce == "" {
+ return false
+ }
+ if m := rec.Move; m != nil && m.SelfServe != nil && m.SelfServe.LastRequest == nonce {
+ return true
+ }
+ return rec.LastMove != nil && rec.LastMove.Request == nonce
+}
+
+// selfServeMutationRefused reports whether a bound write's error is its
+// mutation refusing the record (the record does not admit the
+// transition), as against a lost race, a lost lease or a store failure.
+func selfServeMutationRefused(err error) bool {
+ var run *provision.MoveRunOpenError
+ return errors.Is(err, provision.ErrMoveConflict) || errors.As(err, &run) || errors.Is(err, provision.ErrLifetimeExhausted)
+}
+
+// selfServeApply writes one record transition for row as a persisted
+// request, in the plan's three steps:
+//
+// 1. the judged record gives the expected incarnation and record_seq; the
+// proposed stamp is *proposed when given, else ProposeStamp (max(now,
+// the record's latest stamp + 1 ns)); a fresh nonce; the request is
+// issued on the row. Not issued: no outstanding request is
+// errRequestRefused; another kind or phase is errRequestForeign; a
+// confirmed, unconsumed one of this kind is a resume (the record); any
+// other outstanding one of this kind is resumed at step 3;
+// 2. an expired deadline is decided durably first
+// (ExpireSelfServeRequest), and only its winner answers
+// errRequestExpired; then the one bound write, under the lease's write
+// context;
+// 3. an expired request is errRequestExpired at once; otherwise a fresh
+// judged read decides: the receipt confirms (and the serial is bumped),
+// an unchanged record is the write not landing (step 2 again, which
+// checks the deadline first), and a record that moved without the
+// receipt voids the request.
+//
+// A request with no deadline is never expired here. mut builds the
+// transition from the proposed stamp and the receipt; it must be the same
+// transition on every replay.
+func (s *Server) selfServeApply(ctx context.Context, t *leaseToken, row *SelfServeMove, req request, proposed *time.Time, mut func(at time.Time, rc provision.Receipt) func(*provision.Record) error) (*provision.Record, error) {
+ now := s.selfServeNow()
+ rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ if err != nil {
+ return nil, err
+ }
+ nonce, err := newRequestNonce()
+ if err != nil {
+ return nil, err
+ }
+ pr := persistedRequest{Kind: req.Kind, Phase: req.Phase, Nonce: nonce, ExpectInc: rec.Incarnation, ExpectSeq: rec.RecordSeq}
+ if proposed != nil {
+ pr.Proposed = proposed.UTC()
+ } else {
+ pr.Proposed = provision.ProposeStamp(rec, now)
+ }
+ TestHooks.point(req.Kind + ":issue")
+ var (
+ issued bool
+ existing *persistedRequest
+ )
+ if req.Deadline.IsZero() {
+ issued, existing, err = s.store.IssueSelfServeRequest(row.ID, req.Kind, req.Phase, pr, now)
+ } else {
+ issued, existing, err = s.store.IssueSelfServeRequestWithDeadline(row.ID, req.Kind, req.Phase, pr, req.Deadline, now)
+ }
+ if err != nil {
+ return nil, err
+ }
+ if !issued {
+ switch {
+ case existing == nil:
+ return nil, errRequestRefused
+ case existing.Kind != req.Kind || existing.Phase != req.Phase:
+ return nil, errRequestForeign
+ case existing.Confirmed && !existing.Consumed:
+ return rec, nil
+ }
+ return s.selfServeRun(ctx, t, row, *existing, mut, false)
+ }
+ // The issue wrote the request's deadline from the row in the same
+ // statement: read it back with the cut state, as every load does.
+ cur, err := s.store.selfServeRequest(row.ID)
+ if err != nil {
+ return nil, err
+ }
+ if cur == nil || cur.Nonce != nonce {
+ return nil, errRequestForeign
+ }
+ return s.selfServeRun(ctx, t, row, *cur, mut, true)
+}
+
+// selfServeResume is step 3 for the row's outstanding request, entered by
+// reconciliation before anything else: it confirms, voids or reports that
+// the write still has not landed (errReplayNeeded: the phase function
+// replays it with its mutation) or that the request expired.
+func (s *Server) selfServeResume(ctx context.Context, t *leaseToken, row *SelfServeMove) (*provision.Record, error) {
+ pr, err := s.store.selfServeRequest(row.ID)
+ if err != nil {
+ return nil, err
+ }
+ if pr == nil || pr.Consumed {
+ return nil, nil
+ }
+ if pr.Confirmed {
+ return nil, nil
+ }
+ return s.selfServeRun(ctx, t, row, *pr, nil, false)
+}
+
+// selfServeRun is steps 2 and 3 for pr, starting at the write (step 2) or
+// at the confirmation (step 3). mut nil is reconciliation's resume, which
+// never writes: an unchanged record answers errReplayNeeded.
+func (s *Server) selfServeRun(ctx context.Context, t *leaseToken, row *SelfServeMove, pr persistedRequest, mut func(at time.Time, rc provision.Receipt) func(*provision.Record) error, write bool) (*provision.Record, error) {
+ var lastErr error
+ refused := false
+ for attempt := 0; ; attempt++ {
+ if write {
+ // (2) The deadline first, decided durably.
+ now := s.selfServeNow()
+ if !pr.Deadline.IsZero() && !now.Before(pr.Deadline) {
+ TestHooks.point(pr.Kind + ":expire")
+ won, err := s.store.ExpireSelfServeRequest(row.ID, pr.Nonce, pr.Deadline, pr.CutRequestedAt)
+ if err != nil {
+ return nil, err
+ }
+ if won {
+ return nil, errRequestExpired
+ }
+ cur, err := s.store.selfServeRequest(row.ID)
+ switch {
+ case err != nil:
+ return nil, err
+ case cur == nil || cur.Nonce != pr.Nonce:
+ return nil, errRequestForeign
+ case cur.Consumed || cur.Voided:
+ return nil, errRequestRefused
+ case cur.Expired:
+ // A previous worker won the expiry and stopped before
+ // fencing: resume its path, never decide again.
+ return nil, errRequestExpired
+ }
+ pr = *cur // a cut extended the deadline
+ continue
+ }
+ if !t.valid(now) {
+ return nil, errSelfServeLeaseLost
+ }
+ TestHooks.point(pr.Kind + ":write")
+ wctx, cancel := s.selfServeWriteCtx(ctx, t)
+ _, werr := provision.SetRecordMoveAt(wctx, s.cfg.Releases, row.Account, row.App, row.Host,
+ provision.Expect{Incarnation: pr.ExpectInc, Seq: pr.ExpectSeq}, mut(pr.Proposed, provision.Receipt{Nonce: pr.Nonce}))
+ cancel()
+ lastErr, refused = werr, selfServeMutationRefused(werr)
+ if werr != nil && !refused {
+ log.Printf("console: self-serve move %s: %s write: %v (confirming from the record)", row.ID, pr.Kind, werr)
+ }
+ TestHooks.point(pr.Kind + ":written")
+ }
+ write = true
+
+ // (3) An expiry already decided is never replayed or voided here.
+ if pr.Expired {
+ return nil, errRequestExpired
+ }
+ rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ if errors.Is(err, errRecordGone) {
+ // Absent, retired or recreated: an If-Match write bound to the
+ // version this request read cannot land on it.
+ TestHooks.point(pr.Kind + ":void")
+ if err := s.store.VoidSelfServeRequest(row.ID, pr.Nonce); err != nil {
+ return nil, err
+ }
+ return nil, errRecordGone
+ }
+ if err != nil {
+ return nil, err
+ }
+ if selfServeHasReceipt(rec, pr.Nonce) {
+ TestHooks.point(pr.Kind + ":confirm")
+ ok, err := s.store.ConfirmSelfServeRequest(row.ID, pr.Nonce, s.selfServeNow())
+ if err != nil {
+ return nil, err
+ }
+ if !ok {
+ cur, err := s.store.selfServeRequest(row.ID)
+ if err != nil {
+ return nil, err
+ }
+ if cur != nil && cur.Nonce == pr.Nonce && cur.Confirmed && !cur.Consumed {
+ return rec, nil
+ }
+ return nil, errRequestRefused
+ }
+ s.selfServeNotify(ctx, row.ID, rec.RecordSeq, pr.Kind)
+ return rec, nil
+ }
+ if rec.Incarnation == pr.ExpectInc && rec.RecordSeq == pr.ExpectSeq {
+ // The write did not land.
+ switch {
+ case refused:
+ return nil, &requestRefusal{err: lastErr}
+ case mut == nil:
+ return nil, errReplayNeeded
+ case attempt >= selfServeReplays:
+ if lastErr == nil {
+ lastErr = errReplayNeeded
+ }
+ return nil, lastErr
+ }
+ continue
+ }
+ // The record moved on without the receipt. The bound write cannot
+ // commit on a record past the sequence it was bound to.
+ TestHooks.point(pr.Kind + ":void")
+ if err := s.store.VoidSelfServeRequest(row.ID, pr.Nonce); err != nil {
+ return nil, err
+ }
+ return nil, errRequestVoided
+ }
+}
+
+// selfServeNotify announces a landed write (the serial bump) and, once the
+// bump succeeded, records the record_seq it announced. A failed bump
+// leaves notified_seq behind, and the step bumps again on every later tick
+// until one succeeds (selfServeCatchUp).
+func (s *Server) selfServeNotify(ctx context.Context, id string, seq uint64, kind string) {
+ TestHooks.point(kind + ":bump")
+ if err := s.bumpSerialErr(ctx); err != nil {
+ log.Printf("console: self-serve move %s: serial bump: %v (bumped again next tick)", id, err)
+ return
+ }
+ TestHooks.point(kind + ":notified")
+ if err := s.store.SetSelfServeNotifiedSeq(id, seq); err != nil {
+ log.Printf("console: self-serve move %s: notified_seq: %v (bumped again next tick)", id, err)
+ }
+}
+
+// selfServeCatchUp is the at-least-once half of the notification: a row
+// whose confirmed request's landed sequence (expect_seq + 1) was never
+// announced is bumped now.
+func (s *Server) selfServeCatchUp(ctx context.Context, row *SelfServeMove, rec *provision.Record) {
+ if rec == nil || !row.Request.Confirmed || row.NotifiedSeq >= row.Request.ExpectSeq+1 {
+ return
+ }
+ s.selfServeNotify(ctx, row.ID, rec.RecordSeq, row.Request.Kind)
+}
diff --git a/internal/console/testhooks.go b/internal/console/testhooks.go
new file mode 100644
index 000000000..6a7d1ef5e
--- /dev/null
+++ b/internal/console/testhooks.go
@@ -0,0 +1,74 @@
+package console
+
+import (
+ "context"
+ "sync"
+ "time"
+)
+
+// TestHooks are the self-serve job's seams for the end-to-end witnesses
+// (internal/e2e/selfserve, self-serve moves plan 2 Tasks 7 and 8), which
+// drive a real console beside the real boxes from outside this package.
+// Nothing in production sets a fault or calls a hook: with no fault set,
+// every crash point is a nil check and nothing else.
+var TestHooks = &testHooks{}
+
+type testHooks struct {
+ mu sync.Mutex
+ fault func(point string)
+}
+
+// testCrash is what Crash panics with: the console process dying at a
+// crash point. Tick recovers it and nothing else.
+type testCrash struct{ point string }
+
+// Fault runs f at every crash point of the self-serve job from now on (nil
+// clears it). f is told the point's name ("<kind>:<step>" around a record
+// write, "advance:<from>-><to>" before a phase change, and the like) and
+// calls Crash to stop the tick there.
+func (h *testHooks) Fault(f func(point string)) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.fault = f
+}
+
+// Crash, called from a Fault function, ends the tick at that point as a
+// console that died there would: nothing after it runs.
+func (h *testHooks) Crash(point string) { panic(testCrash{point: point}) }
+
+// point is one crash point: a nil check in production.
+func (h *testHooks) point(name string) {
+ h.mu.Lock()
+ f := h.fault
+ h.mu.Unlock()
+ if f != nil {
+ f(name)
+ }
+}
+
+// Tick runs one self-serve moves pass on s with its clock at now, as the
+// pass's ticker would (the lease, the events, one step per open row), and
+// reports whether the tick crashed at a point, and which.
+func (h *testHooks) Tick(ctx context.Context, s *Server, now time.Time) (crashed string) {
+ s.selfServeClock = func() time.Time { return now }
+ defer func() {
+ s.selfServeClock = nil
+ if r := recover(); r != nil {
+ c, ok := r.(testCrash)
+ if !ok {
+ panic(r)
+ }
+ crashed = c.point
+ }
+ }()
+ s.selfServeIteration(ctx)
+ return ""
+}
+
+// NameConsole names s in the moves lease as id: a console restarted on the
+// same host and database takes back the lease its predecessor held, as
+// the next acquire of a lease that still names this console does.
+func (h *testHooks) NameConsole(s *Server, id string) {
+ s.selfServeIDOnce.Do(func() {})
+ s.selfServeConsole = id
+}