amadan

carlos / platform Public

console: the self-serve job's store methods: the expiry decision, the expired void, the maintenance consume, the advance into a substep

ExpireSelfServeRequest, VoidExpiredSelfServeRequest, ConsumeSelfServeMaintenance,
AdvanceSelfServeToStep (AdvanceSelfServe is it at the first substep; staging
and ready stamp their start and idle interval in the same statement),
IssueSelfServeRequestWithDeadline, AbortSelfServe and SetSelfServeReason,
each predicated as the plan states it (self-serve moves plan 2 Task 7).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude c2a059b76765c0474e2925aaf8e4ff385f34612f parent 611ba4a
2 files changed, +307 −13
  • internal/console/selfserve_store.go +140 −13
  • internal/console/selfserve_store_test.go +167 −0
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index 7f2b71f8c..8f372d8e5 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -628,6 +628,19 @@ FROM self_serve_moves WHERE id = ?`, id).Scan(&issued, &pr.Kind, &pr.Phase, &pr.
// 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) {
+ return s.issueSelfServeRequest(id, kind, phase, pr, nil, now)
+}
+
+// IssueSelfServeRequestWithDeadline is IssueSelfServeRequest for a wait
+// that starts with its own allowance (the ready phase's maintenance round):
+// under the same predicate it sets the row's step deadline to stepDeadline
+// and the request deadline from it, in the one statement, so a crash can
+// never leave the round's deadline set without its request.
+func (s *Store) IssueSelfServeRequestWithDeadline(id, kind, phase string, pr persistedRequest, stepDeadline, now time.Time) (issued bool, existing *persistedRequest, err error) {
+ return s.issueSelfServeRequest(id, kind, phase, pr, &stepDeadline, now)
+}
+
+func (s *Store) issueSelfServeRequest(id, kind, phase string, pr persistedRequest, stepDeadline *time.Time, 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)
}
@@ -639,13 +652,21 @@ func (s *Store) IssueSelfServeRequest(id, kind, phase string, pr persistedReques
if err != nil {
return false, nil, err
}
+ // With stepDeadline the row's step deadline is set first in the SET
+ // list; SQLite evaluates every expression against the row as it was,
+ // so request_deadline names the new deadline explicitly.
+ set, args := `step_deadline = step_deadline, request_deadline = CASE WHEN ? THEN step_deadline ELSE 0 END`, []any{ssBool(selfServeAbortExpiring[kind])}
+ if stepDeadline != nil {
+ set = `step_deadline = ?, request_deadline = CASE WHEN ? THEN ? ELSE 0 END`
+ args = []any{ssNanos(*stepDeadline), ssBool(selfServeAbortExpiring[kind]), ssNanos(*stepDeadline)}
+ }
+ args = append([]any{ssNanos(now), kind, phase, pr.Nonce, inc, seq, ssNanos(pr.Proposed)}, args...)
+ args = append(args, ssNanos(now), id, phase)
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_proposed = ?, `+set+`,
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)
+WHERE id = ? AND phase = ? AND outcome = '' AND (request_consumed <> 0 OR request_issued = 0)`, args...)
won, err := ssWon(res, err)
if err != nil || won {
return won, nil, err
@@ -720,17 +741,38 @@ WHERE id = ? AND request_nonce = ? AND request_consumed = 0 AND request_confirme
}
// AdvanceSelfServe moves the row from one phase to the next and inserts
-// the phase event in the same transaction. One UPDATE, predicated
+// the phase event in the same transaction, entering the new phase at its
+// first substep (the job's own progression). It is
+// AdvanceSelfServeToStep with selfServeFirstSubstep[to].
+func (s *Store) AdvanceSelfServe(id, from, to, reason string, step, window time.Time, nonce string, now time.Time) (bool, error) {
+ return s.AdvanceSelfServeToStep(id, from, to, reason, selfServeFirstSubstep[to], step, window, nonce, now)
+}
+
+// selfServeMaintainEvery is the ready phase's idle interval: a maintenance
+// restage is due this long after the last one ended (spec §6.2).
+const selfServeMaintainEvery = 10 * time.Minute
+
+// AdvanceSelfServeToStep moves the row from one phase to the next, entering
+// it at substep: the first substep for the job's own progression
+// (AdvanceSelfServe), the substep the record dictates for reconciliation,
+// and restage's adopted or issue for the cut handoff. 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) {
+// sets the phase, phase_started_at, the step and window deadlines, the
+// 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. Entering staging stamps staging_started_at
+// (once), and entering ready starts the idle interval
+// (next_maintenance_at = now + 10 min), in the same statement. The phase
+// event goes in the same transaction.
+func (s *Store) AdvanceSelfServeToStep(id, from, to, reason, substep string, step, window time.Time, nonce string, now time.Time) (bool, error) {
+ var nextMaint int64
+ if to == ssReady {
+ nextMaint = ssNanos(now.Add(selfServeMaintainEvery))
+ }
tx, err := s.db.Begin()
if err != nil {
return false, err
@@ -743,12 +785,17 @@ func (s *Store) AdvanceSelfServe(id, from, to, reason string, step, window time.
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,
+ staging_started_at = CASE WHEN ? = 'staging' AND staging_started_at = 0 THEN ? ELSE staging_started_at END,
+ next_maintenance_at = CASE WHEN ? <> 0 THEN ? ELSE next_maintenance_at 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, ssNanos(now), ssNanos(step), ssNanos(window), substep, reason,
to, to, to, to,
- nonce, ssNanos(now),
+ nonce,
+ to, ssNanos(now),
+ nextMaint, nextMaint,
+ ssNanos(now),
id, from, to, nonce, nonce)
won, err := ssWon(res, err)
if err != nil || !won {
@@ -944,6 +991,86 @@ func (s *Store) SetSelfServeOpenRecordPending(id string, pending bool) error {
return s.setSelfServe("open record", id, `open_record_pending = ?`, ssBool(pending))
}
+// ---- the job's own decisions (self-serve moves plan 2 Task 7) ---------------
+
+// ExpireSelfServeRequest decides, durably, that the outstanding request's
+// deadline passed before it was confirmed: one UPDATE predicated on the
+// nonce, the deadline the worker saw and the cut state it read with the
+// request,
+//
+// WHERE request_nonce = ? AND request_deadline = ? AND cut_requested_at = ?
+// AND request_consumed = 0 AND request_expired = 0
+//
+// so a cut that landed since (and extended the deadline) makes the decision
+// lose, and only one worker ever wins it. Only the winner may fence, void
+// and abort (selfServeApply step 2).
+func (s *Store) ExpireSelfServeRequest(id, nonce string, deadline, sawCutRequestedAt time.Time) (bool, error) {
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET request_expired = 1, updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_deadline = ? AND cut_requested_at = ? AND request_consumed = 0 AND request_expired = 0`,
+ time.Now().UnixNano(), id, nonce, ssNanos(deadline), ssNanos(sawCutRequestedAt)))
+}
+
+// VoidExpiredSelfServeRequest retires a won expiry whose fenced observation
+// carries no receipt, and records the abort it obliges, in one UPDATE:
+//
+// WHERE request_nonce = ? AND request_expired = 1 AND request_confirmed = 0 AND request_consumed = 0
+//
+// It sets request_consumed and request_voided, and abort_requested_at only
+// when it is zero, so an abort a cancellation already requested keeps its
+// own time (and its reason: outcome_reason is the expiry's only when this
+// void is what requests the abort). Zero rows: the request changed under
+// the worker, which reloads it and dispatches again.
+func (s *Store) VoidExpiredSelfServeRequest(id, nonce string, now time.Time, reason string) (bool, error) {
+ if ssNanos(now) <= 0 {
+ return false, fmt.Errorf("console: self-serve move %s: a void needs a time", id)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ request_consumed = 1, request_voided = 1,
+ outcome_reason = CASE WHEN abort_requested_at = 0 THEN ? ELSE outcome_reason END,
+ abort_requested_at = CASE WHEN abort_requested_at = 0 THEN ? ELSE abort_requested_at END,
+ updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_expired = 1 AND request_confirmed = 0 AND request_consumed = 0`,
+ reason, ssNanos(now), ssNanos(now), id, nonce))
+}
+
+// ConsumeSelfServeMaintenance is the ready phase's maintenance-to-idle
+// consume: ConsumeSelfServeRequest's predicate with the cut's,
+//
+// WHERE request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0 AND cut_requested_at = 0
+//
+// so it never runs after a cut won (the stale worker's UPDATE matches
+// nothing; it re-reads the row and takes the cut handoff). It leaves the
+// row idle: substep round, no step deadline, the next round due at
+// nextMaintenance. window_deadline is not touched.
+func (s *Store) ConsumeSelfServeMaintenance(id, nonce string, nextMaintenance time.Time) (bool, error) {
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET
+ request_consumed = 1, phase_step = 'round', step_deadline = 0, next_maintenance_at = ?, updated_at = ?
+WHERE id = ? AND request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0 AND cut_requested_at = 0`,
+ ssNanos(nextMaintenance), time.Now().UnixNano(), id, nonce))
+}
+
+// AbortSelfServe is the job's own abort decision (a failed fact, a replaced
+// participant, an expiry whose action is abort): abort_requested_at and the
+// outcome's reason, once, on an open row that has no abort yet:
+//
+// WHERE outcome = '' AND abort_requested_at = 0
+//
+// reason is also the row's reason, so the view says why the move is going
+// back. False: an abort (a cancellation, an earlier decision) was already
+// requested, and it keeps its own time and reason.
+func (s *Store) AbortSelfServe(id, reason string, now time.Time) (bool, error) {
+ if ssNanos(now) <= 0 {
+ return false, fmt.Errorf("console: self-serve move %s: an abort needs a time", id)
+ }
+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET abort_requested_at = ?, outcome_reason = ?, reason = ?, updated_at = ?
+WHERE id = ? AND outcome = '' AND abort_requested_at = 0`, ssNanos(now), reason, reason, ssNanos(now), id))
+}
+
+// SetSelfServeReason records why an open row waits (the view's reason).
+func (s *Store) SetSelfServeReason(id, reason string) error {
+ return s.setSelfServe("reason", id, `reason = ?`, reason)
+}
+
// ---- the reclaim rule's evidence (offlinereserve.go) ---------------------------
var _ selfServeRowEvidence = (*Store)(nil)
diff --git a/internal/console/selfserve_store_test.go b/internal/console/selfserve_store_test.go
index a67c60684..7221216a0 100644
--- a/internal/console/selfserve_store_test.go
+++ b/internal/console/selfserve_store_test.go
@@ -1019,3 +1019,170 @@ func TestClaimStoreOpensWithEveryNewTable(t *testing.T) {
}
}
}
+
+// TestSelfServeJobStorePredicates: the job's own store methods (Task 7),
+// each predicated as the plan states it: the expiry decision against the
+// cut state read with the request, the expired void with its abort in one
+// statement, the maintenance consume that never runs after a cut, the
+// advance into a named substep, the job's abort decision and the issue
+// that sets its round's own deadline.
+func TestSelfServeJobStorePredicates(t *testing.T) {
+ st := ssStore(t)
+ now := ssT0.Add(time.Second)
+
+ t.Run("expire", func(t *testing.T) {
+ ssAdmit(t, st, ssMove("mv_expireexpireexpireexpireab", "a."+ssHost, ssT0, "ie-1", "uk-1"))
+ id := "mv_expireexpireexpireexpireab"
+ step := ssT0.Add(time.Minute)
+ ssPhase(t, st, id, "staging", step)
+ ssIssue(t, st, id, "stage", "staging", "e1", now)
+ if ok, err := st.ExpireSelfServeRequest(id, "e1", step, ssT0); err != nil || ok {
+ t.Fatalf("expire against a cut state not read = %v, %v; want lost", ok, err)
+ }
+ if ok, err := st.ExpireSelfServeRequest(id, "e1", step.Add(time.Second), time.Time{}); err != nil || ok {
+ t.Fatalf("expire against another deadline = %v, %v; want lost", ok, err)
+ }
+ if ok, err := st.ExpireSelfServeRequest(id, "e-other", step, time.Time{}); err != nil || ok {
+ t.Fatalf("expire with another nonce = %v, %v; want lost", ok, err)
+ }
+ if ok, err := st.ExpireSelfServeRequest(id, "e1", step, time.Time{}); err != nil || !ok {
+ t.Fatalf("expire = %v, %v; want won", ok, err)
+ }
+ if ok, err := st.ExpireSelfServeRequest(id, "e1", step, time.Time{}); err != nil || ok {
+ t.Fatalf("second expire = %v, %v; want lost (decided once)", ok, err)
+ }
+ pr, _ := st.selfServeRequest(id)
+ if !pr.Expired {
+ t.Fatalf("request after expiry = %+v", pr)
+ }
+ })
+
+ t.Run("void expired", func(t *testing.T) {
+ id := "mv_voidexpvoidexpvoidexpvoidx"
+ ssAdmit(t, st, ssMove(id, "b."+ssHost, ssT0, "ie-1", "uk-2"))
+ step := ssT0.Add(time.Minute)
+ ssPhase(t, st, id, "staging", step)
+ ssIssue(t, st, id, "stage", "staging", "v1", now)
+ if ok, err := st.VoidExpiredSelfServeRequest(id, "v1", now, "r"); err != nil || ok {
+ t.Fatalf("void before the expiry = %v, %v; want nothing", ok, err)
+ }
+ if ok, _ := st.ExpireSelfServeRequest(id, "v1", step, time.Time{}); !ok {
+ t.Fatal("expire lost")
+ }
+ at := now.Add(time.Minute)
+ if ok, err := st.VoidExpiredSelfServeRequest(id, "v1", at, "the reason"); err != nil || !ok {
+ t.Fatalf("void = %v, %v", ok, err)
+ }
+ m := ssLoad(t, st, id)
+ if !m.Request.Voided || !m.Request.Consumed || !m.AbortRequestedAt.Equal(at) || m.OutcomeReason != "the reason" {
+ t.Fatalf("after the void: %+v abort %v reason %q", m.Request, m.AbortRequestedAt, m.OutcomeReason)
+ }
+ if ok, _ := st.VoidExpiredSelfServeRequest(id, "v1", at.Add(time.Hour), "again"); ok {
+ t.Fatal("second void matched")
+ }
+
+ // A cancellation already requested keeps its own time and reason.
+ id2 := "mv_voidcancelvoidcancelvoidcx"
+ ssAdmit(t, st, ssMove(id2, "c."+ssHost, ssT0, "ie-1", "uk-3"))
+ ssPhase(t, st, id2, "staging", step)
+ ssIssue(t, st, id2, "stage", "staging", "v2", now)
+ _, _ = st.ExpireSelfServeRequest(id2, "v2", step, time.Time{})
+ if ok, _ := st.RequestSelfServeAbort(id2, []string{"staging"}, now); !ok {
+ t.Fatal("cancel lost")
+ }
+ if ok, _ := st.VoidExpiredSelfServeRequest(id2, "v2", at, "expiry reason"); !ok {
+ t.Fatal("void lost")
+ }
+ m = ssLoad(t, st, id2)
+ if !m.AbortRequestedAt.Equal(now) || m.OutcomeReason != "" {
+ t.Fatalf("a cancel's abort was replaced: abort %v reason %q", m.AbortRequestedAt, m.OutcomeReason)
+ }
+
+ // A confirmed request is never an expired void.
+ id3 := "mv_voidconfvoidconfvoidconfvx"
+ ssAdmit(t, st, ssMove(id3, "d."+ssHost, ssT0, "ie-1", "uk-4"))
+ ssPhase(t, st, id3, "staging", step)
+ ssIssue(t, st, id3, "stage", "staging", "v3", now)
+ _, _ = st.ExpireSelfServeRequest(id3, "v3", step, time.Time{})
+ _, _ = st.ConfirmSelfServeRequest(id3, "v3", now)
+ if ok, _ := st.VoidExpiredSelfServeRequest(id3, "v3", at, "r"); ok {
+ t.Fatal("a confirmed request was voided")
+ }
+ })
+
+ t.Run("maintenance consume", func(t *testing.T) {
+ id := "mv_maintmaintmaintmaintmainta"
+ ssAdmit(t, st, ssMove(id, "e."+ssHost, ssT0, "ie-1", "uk-5"))
+ ssPhase(t, st, id, "ready", time.Time{})
+ round := now.Add(5 * time.Minute)
+ ok, _, err := st.IssueSelfServeRequestWithDeadline(id, "maintain", "ready", persistedRequest{Nonce: "m1", Proposed: now}, round, now)
+ if err != nil || !ok {
+ t.Fatalf("issue = %v, %v", ok, err)
+ }
+ m := ssLoad(t, st, id)
+ if !m.StepDeadline.Equal(round) || !m.Request.Deadline.Equal(round) {
+ t.Fatalf("the round's deadline: step %v request %v; want %v", m.StepDeadline, m.Request.Deadline, round)
+ }
+ _, _ = st.ConfirmSelfServeRequest(id, "m1", now)
+ next := now.Add(10 * time.Minute)
+ if ok, err := st.ConsumeSelfServeMaintenance(id, "m1", next); err != nil || !ok {
+ t.Fatalf("consume = %v, %v", ok, err)
+ }
+ m = ssLoad(t, st, id)
+ if !m.StepDeadline.IsZero() || !m.NextMaintenanceAt.Equal(next) || m.PhaseStep != "round" || !m.Request.Consumed {
+ t.Fatalf("idle after the round: %+v", m)
+ }
+
+ // After a cut the consume matches nothing.
+ ok, _, _ = st.IssueSelfServeRequestWithDeadline(id, "maintain", "ready", persistedRequest{Nonce: "m2", Proposed: now}, round, now)
+ _, _ = st.ConfirmSelfServeRequest(id, "m2", now)
+ if ok, _ := st.MarkSelfServeCut(id, now, now.Add(5*time.Minute)); !ok {
+ t.Fatal("cut lost")
+ }
+ if ok, err := st.ConsumeSelfServeMaintenance(id, "m2", next); err != nil || ok {
+ t.Fatalf("consume after the cut = %v, %v; want nothing", ok, err)
+ }
+ })
+
+ t.Run("advance to a substep", func(t *testing.T) {
+ id := "mv_substepsubstepsubstepsubst"
+ ssAdmit(t, st, ssMove(id, "f."+ssHost, ssT0, "ie-1", "uk-6"))
+ ssPhase(t, st, id, "staging", time.Time{})
+ if ok, err := st.AdvanceSelfServeToStep(id, "staging", "stopping", "", "seal", now.Add(time.Minute), time.Time{}, "", now); err != nil || !ok {
+ t.Fatalf("advance = %v, %v", ok, err)
+ }
+ if m := ssLoad(t, st, id); m.Phase != "stopping" || m.PhaseStep != "seal" {
+ t.Fatalf("after the advance %s/%s", m.Phase, m.PhaseStep)
+ }
+ id2 := "mv_stagestartstagestartstagex"
+ ssAdmit(t, st, ssMove(id2, "g."+ssHost, ssT0, "ie-1", "uk-7"))
+ ssPhase(t, st, id2, "reports", time.Time{})
+ if ok, _ := st.AdvanceSelfServe(id2, "reports", "staging", "", now.Add(time.Minute), time.Time{}, "", now); !ok {
+ t.Fatal("advance lost")
+ }
+ if m := ssLoad(t, st, id2); !m.StagingStartedAt.Equal(now) || m.PhaseStep != "round" {
+ t.Fatalf("staging entered: started %v step %q", m.StagingStartedAt, m.PhaseStep)
+ }
+ ssPhase(t, st, id2, "prechecks", time.Time{})
+ if ok, _ := st.AdvanceSelfServe(id2, "prechecks", "ready", "", time.Time{}, now.Add(24*time.Hour), "", now); !ok {
+ t.Fatal("advance lost")
+ }
+ if m := ssLoad(t, st, id2); !m.NextMaintenanceAt.Equal(now.Add(10*time.Minute)) || !m.StagingStartedAt.Equal(now) {
+ t.Fatalf("ready entered: next maintenance %v, staging started %v", m.NextMaintenanceAt, m.StagingStartedAt)
+ }
+ })
+
+ t.Run("abort decision", func(t *testing.T) {
+ id := "mv_abortdecideabortdecideabor"
+ ssAdmit(t, st, ssMove(id, "h."+ssHost, ssT0, "ie-1", "uk-8"))
+ if ok, err := st.AbortSelfServe(id, "first", now); err != nil || !ok {
+ t.Fatalf("abort = %v, %v", ok, err)
+ }
+ if ok, _ := st.AbortSelfServe(id, "second", now.Add(time.Second)); ok {
+ t.Fatal("a second abort decision matched")
+ }
+ if m := ssLoad(t, st, id); m.OutcomeReason != "first" || m.Reason != "first" || !m.AbortRequestedAt.Equal(now) {
+ t.Fatalf("after the abort: %q %q %v", m.OutcomeReason, m.Reason, m.AbortRequestedAt)
+ }
+ })
+}