carlos / platform Public

console: one finaliser (selfserve_finalise.go): the outcome in phase done with any unconfirmed request resolved first, the timings hook, the release; a finished row is never advanced

FinishSelfServeMove sets phase done, clears open_record_pending and
refuses a row whose request is unconfirmed; AdvanceSelfServeToStep and
the new SetSelfServeStep require outcome = ''.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 1 day ago 017e77730bc4dda097eec7025eda2455e22283d8 parent 4925757

5 files changed, +230 −81

internal/console/selfserve_finalise.go +102 −0
internal/console/selfserve_finalise_test.go +79 −0
internal/console/selfserve_job.go +0 −64
internal/console/selfserve_store.go +37 −11
internal/console/selfserve_store_test.go +12 −6

diff --git a/internal/console/selfserve_finalise.go b/internal/console/selfserve_finalise.go

new file mode 100644

index 000000000..181e6acea

--- /dev/null

+++ b/internal/console/selfserve_finalise.go

@@ -0,0 +1,102 @@

+package console

+// The self-serve finaliser (self-serve moves plan 2 Task 8; spec §7.1's

+// release): one way a self-serve row ends, whoever cleared its record (the

+// job's own finish, an operator door, a recreated host). Four steps, each

+// durable before the next:

+//

+// 1. the outcome (FinishSelfServeMove: outcome, reason and warning, phase

+// done, release_pending = 1, effects_done = 0, open_record_pending = 0,

+// a confirmed clearing request consumed and the terminal event, one

+// transaction; refused while an unconfirmed request is outstanding);

+// 2. the terminal effects (the timings, Task 9's; the serial bump), then

+// effects_done;

+// 3. the reservations released (releaseBoxes under the lease);

+// 4. the release recorded (ClearSelfServeRelease).

+//

+// A crash between any two leaves the row in OpenSelfServeMoves and the

+// next tick resumes at the first unfinished step (selfServeStep's first

+// line). No row is finished with an unconfirmed request outstanding: its

+// write could still land, so it is resolved first (by the fence, or by the

+// record when the record has moved past it); a terminal row's request is

+// always consumed, so HostHasIssuedRequest never holds a released box for

+// ever.

+import (

+ "context"

+ "fmt"

+ "log"

+ "github.com/carlosframework/platform/internal/moves"

+ "github.com/carlosframework/platform/internal/provision"

+// selfServeFinalise ends a row with outcome, reason and warning (warning ""

+// keeps one the row already carries).

+func (s *Server) selfServeFinalise(ctx context.Context, t *leaseToken, row *SelfServeMove, outcome, reason, warning string) error {

+ if out, conf := selfServeOutstanding(row); out && !conf {

+ return fmt.Errorf("console: self-serve move %s: a %s request is outstanding; it is resolved before the move is finished", row.ID, row.Request.Kind)

+ }

+ if row.Outcome == "" {

+ now := s.selfServeNow()

+ if outcome == provision.OfflineOutcomeAborted && row.AbortFinishedAt.IsZero() && !row.AbortRequestedAt.IsZero() {

+ if err := s.store.SetSelfServeAbortFinished(row.ID, now); err != nil {

+ return err

+ }

+ TestHooks.point("finalise:outcome")

+ if err := s.store.FinishSelfServeMove(row.ID, outcome, reason, warning, now); err != nil {

+ return err

+ }

+ if cur, err := s.selfServeReload(row.ID); err != nil || cur == nil || cur.Outcome == "" {

+ // A request appeared meanwhile: the next tick decides from the

+ // row.

+ return err

+ }

+ log.Printf("console: self-serve move %s (%s): %s %s", row.ID, row.Host, outcome, reason)

+ }

+ return s.selfServeFinaliseRest(ctx, t, row.ID)

+// selfServeFinaliseRest is the finaliser from step (2) on, for a row that

+// has its outcome.

+func (s *Server) selfServeFinaliseRest(ctx context.Context, t *leaseToken, id string) error {

+ row, err := s.selfServeReload(id)

+ if err != nil || row == nil || row.Outcome == "" {

+ return err

+ }

+ if !row.EffectsDone {

+ TestHooks.point("finalise:effects")

+ if err := s.selfServeTimingsOnFinish(ctx, row); err != nil {

+ return fmt.Errorf("console: self-serve move %s: the timings: %w (retried next tick)", id, err)

+ }

+ if err := s.bumpSerialErr(ctx); err != nil {

+ return fmt.Errorf("console: self-serve move %s: the terminal bump: %w (retried next tick)", id, err)

+ }

+ if err := s.store.SetSelfServeEffectsDone(id); err != nil {

+ return err

+ }

+ if row.ReleasePending {

+ if !t.valid(s.selfServeNow()) {

+ return errSelfServeLeaseLost

+ }

+ TestHooks.point("finalise:release")

+ wctx, cancel := s.selfServeWriteCtx(ctx, t)

+ err := s.releaseBoxes(wctx, moves.OfflineKindMove, id)

+ cancel()

+ if err != nil {

+ return err

+ }

+ TestHooks.point("finalise:released")

+ if err := s.store.ClearSelfServeRelease(id); err != nil {

+ return err

+ }

+ return nil

+// selfServeTimingsOnFinish is the finaliser's timings effect: Task 9

+// derives the move_timings rows from the marks here. A no-op until then;

+// it must be idempotent (a crash before effects_done runs it again).

+func (s *Server) selfServeTimingsOnFinish(ctx context.Context, row *SelfServeMove) error { return nil }

diff --git a/internal/console/selfserve_finalise_test.go b/internal/console/selfserve_finalise_test.go

new file mode 100644

index 000000000..ea1b3f9ca

--- /dev/null

+++ b/internal/console/selfserve_finalise_test.go

@@ -0,0 +1,79 @@

+package console

+import (

+ "testing"

+ "time"

+// TestSelfServeFinishedRowsNeverAdvance: the finaliser's first step

+// (FinishSelfServeMove) ends the row in phase done, consumes its confirmed

+// clearing request and clears the open record owed, in one statement; it

+// refuses a row whose request is unconfirmed (its write may still land, so

+// a terminal row never holds an unconsumed request); and a stale worker

+// holding the row's last phase, or its consumed nonce, cannot advance a

+// finished row, or set a substep on it.

+func TestSelfServeFinishedRowsNeverAdvance(t *testing.T) {

+ st := ssStore(t)

+ now := ssT0.Add(time.Second)

+ t.Run("an unconfirmed request holds the outcome", func(t *testing.T) {

+ id := "mv_finunconfirmedfinunconfirm"

+ ssAdmit(t, st, ssMove(id, "a."+ssHost, ssT0, "ie-1", "uk-1"))

+ ssPhase(t, st, id, "reverting", ssT0.Add(time.Minute))

+ ssIssue(t, st, id, "finish", "reverting", "f1", now)

+ if err := st.FinishSelfServeMove(id, "aborted", "the move was cancelled", "", now); err != nil {

+ t.Fatal(err)

+ }

+ if r := ssLoad(t, st, id); r.Outcome != "" || r.Phase != "reverting" {

+ t.Fatalf("finished with an unconfirmed request: %+v", r)

+ }

+ if ok, err := st.ConfirmSelfServeRequest(id, "f1", now); err != nil || !ok {

+ t.Fatalf("confirm = %v, %v", ok, err)

+ }

+ if err := st.SetSelfServeOpenRecordPending(id, true); err != nil {

+ t.Fatal(err)

+ }

+ if err := st.FinishSelfServeMove(id, "aborted", "the move was cancelled", "", now); err != nil {

+ t.Fatal(err)

+ }

+ r := ssLoad(t, st, id)

+ switch {

+ case r.Outcome != "aborted" || r.Phase != "done" || !r.Request.Consumed || r.OpenRecordPending || !r.ReleasePending || r.EffectsDone:

+ t.Fatalf("finished row %+v", r)

+ }

+ if busy, err := st.HostHasIssuedRequest("bab", "shop", "a."+ssHost); err != nil || busy {

+ t.Fatalf("a finished row holds the host: %v, %v", busy, err)

+ }

+ })

+ t.Run("a stale worker cannot advance a finished row", func(t *testing.T) {

+ id := "mv_finstalefinstalefinstalefi"

+ ssAdmit(t, st, ssMove(id, "b."+ssHost, ssT0, "ie-1", "uk-2"))

+ ssPhase(t, st, id, "finishing", ssT0.Add(time.Minute))

+ ssIssue(t, st, id, "finish", "finishing", "g2", now)

+ if ok, err := st.ConfirmSelfServeRequest(id, "g2", now); err != nil || !ok {

+ t.Fatalf("confirm = %v, %v", ok, err)

+ }

+ // The worker read the row here, then the finaliser finished it.

+ if err := st.FinishSelfServeMove(id, "done", "", "", now); err != nil {

+ t.Fatal(err)

+ }

+ for _, to := range []string{"finishing", "reverting", "opened", "done"} {

+ if ok, err := st.AdvanceSelfServeToStep(id, "finishing", to, "", "", now.Add(time.Minute), time.Time{}, "", now); err != nil || ok {

+ t.Fatalf("advance finishing -> %s = %v, %v; want refused", to, ok, err)

+ }

+ if ok, err := st.AdvanceSelfServeToStep(id, "done", to, "", "", now.Add(time.Minute), time.Time{}, "", now); err != nil || ok {

+ t.Fatalf("advance done -> %s = %v, %v; want refused", to, ok, err)

+ }

+ if ok, err := st.AdvanceSelfServe(id, "finishing", "done", "", now, time.Time{}, "g2", now); err != nil || ok {

+ t.Fatalf("advance with the consumed nonce = %v, %v; want refused", ok, err)

+ }

+ if ok, err := st.SetSelfServeStep(id, "done", "", "gate2", now); err != nil || ok {

+ t.Fatalf("substep on a finished row = %v, %v; want refused", ok, err)

+ }

+ if r := ssLoad(t, st, id); r.Phase != "done" || r.Outcome != "done" {

+ t.Fatalf("the finished row moved: %+v", r)

+ }

+ })

diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go

index 03eaaf7c6..bbed8cd81 100644

--- a/internal/console/selfserve_job.go

+++ b/internal/console/selfserve_job.go

@@ -668,70 +668,6 @@ func (s *Server) selfServeAbortStep(ctx context.Context, t *leaseToken, row *Sel

return true, err

}

-// ---- the finaliser -----------------------------------------------------------

-// selfServeFinalise ends a row: (1) the outcome (FinishSelfServeMove, which

-// consumes a confirmed clearing request and inserts the terminal event in

-// one transaction); (2) the terminal effects (the serial bump; Task 9's

-// timings); (3) the reservations released; (4) the release recorded. A

-// crash between any two leaves the row in OpenSelfServeMoves and the next

-// tick resumes at the first unfinished step. A row with an unconfirmed

-// request is never finalised: the request is resolved first.

-func (s *Server) selfServeFinalise(ctx context.Context, t *leaseToken, row *SelfServeMove, outcome, reason, warning string) error {

- if out, conf := selfServeOutstanding(row); out && !conf {

- return fmt.Errorf("console: self-serve move %s: a %s request is outstanding; it is resolved before the move is finished", row.ID, row.Request.Kind)

- }

- if row.Outcome == "" {

- now := s.selfServeNow()

- if outcome == provision.OfflineOutcomeAborted && row.AbortFinishedAt.IsZero() && !row.AbortRequestedAt.IsZero() {

- if err := s.store.SetSelfServeAbortFinished(row.ID, now); err != nil {

- return err

- }

- TestHooks.point("finalise:outcome")

- if err := s.store.FinishSelfServeMove(row.ID, outcome, reason, warning, now); err != nil {

- return err

- }

- log.Printf("console: self-serve move %s (%s): %s %s", row.ID, row.Host, outcome, reason)

- }

- return s.selfServeFinaliseRest(ctx, t, row.ID)

-// selfServeFinaliseRest is the finaliser from step (2) on, for a row that

-// has its outcome.

-func (s *Server) selfServeFinaliseRest(ctx context.Context, t *leaseToken, id string) error {

- row, err := s.selfServeReload(id)

- if err != nil || row == nil || row.Outcome == "" {

- return err

- }

- if !row.EffectsDone {

- TestHooks.point("finalise:effects")

- if err := s.bumpSerialErr(ctx); err != nil {

- return fmt.Errorf("console: self-serve move %s: the terminal bump: %w (retried next tick)", id, err)

- }

- if err := s.store.SetSelfServeEffectsDone(id); err != nil {

- return err

- }

- if row.ReleasePending {

- if !t.valid(s.selfServeNow()) {

- return errSelfServeLeaseLost

- }

- TestHooks.point("finalise:release")

- wctx, cancel := s.selfServeWriteCtx(ctx, t)

- err := s.releaseBoxes(wctx, moves.OfflineKindMove, id)

- cancel()

- if err != nil {

- return err

- }

- TestHooks.point("finalise:released")

- if err := s.store.ClearSelfServeRelease(id); err != nil {

- return err

- }

- return nil

// ---- the phases --------------------------------------------------------------

// selfServePhase is step (6): the phase's own step.

diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go

index 8f372d8e5..e65abde9e 100644

--- a/internal/console/selfserve_store.go

+++ b/internal/console/selfserve_store.go

@@ -757,9 +757,12 @@ const selfServeMaintainEvery = 10 * time.Minute

// (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'))

+// phase = ? AND outcome = '' 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))

+// (a finished row is never advanced: a stale worker holding its last phase

+// cannot reopen it)

+//

// 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

@@ -788,7 +791,7 @@ func (s *Store) AdvanceSelfServeToStep(id, from, to, reason, substep string, ste

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+`)

+WHERE id = ? AND phase = ? AND outcome = '' 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), substep, reason,

to, to, to, to,

@@ -933,12 +936,20 @@ 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.

+// FinishSelfServeMove records the outcome, in phase done, with

+// release_pending = 1, effects_done = 0 and open_record_pending = 0 (an

+// open record is never owed by a finished move), consumes a confirmed

+// request (the clearing write's), and inserts the terminal event, in one

+// transaction. Only an open row with no unconfirmed request outstanding is

+// finished:

+//

+// WHERE outcome = '' AND NOT (request_issued <> 0 AND request_consumed = 0 AND request_confirmed = 0)

+//

+// so the finaliser run again after a crash between its steps changes

+// nothing and allocates no second terminal event, and a request whose

+// write may still land is never left unconsumed on a terminal row (it is

+// resolved first). Matching nothing is not an error: the row is finished

+// already, or a request is outstanding (the caller reads the row again).

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)

@@ -950,10 +961,11 @@ func (s *Store) FinishSelfServeMove(id, outcome, reason, warning string, now tim

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,

+ phase = 'done', phase_started_at = ?, step_deadline = 0, release_pending = 1, effects_done = 0, open_record_pending = 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)

+WHERE id = ? AND outcome = '' AND NOT (request_issued <> 0 AND request_consumed = 0 AND request_confirmed = 0)`,

+ outcome, reason, warning, warning, ssNanos(now), ssNanos(now), id)

won, err := ssWon(res, err)

if err != nil {

return err

@@ -966,7 +978,7 @@ WHERE id = ? AND outcome = ''`, outcome, reason, warning, warning, ssNanos(now),

if n == 0 {

return fmt.Errorf("console: finish: no self-serve move %q", id)

}

- return nil // already finished

+ return nil // already finished, or a request is outstanding

}

if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: id, Phase: ssDone, At: now, Reason: reason, Detail: outcome}); err != nil {

return err

@@ -1071,6 +1083,20 @@ func (s *Store) SetSelfServeReason(id, reason string) error {

return s.setSelfServe("reason", id, `reason = ?`, reason)

}

+// SetSelfServeStep moves an open row from one substep of its phase to

+// another with the substep's own deadline, when nothing is outstanding (an

+// operator's transition did the substep's write):

+//

+// WHERE phase = ? AND phase_step = ? AND outcome = '' AND (request_consumed <> 0 OR request_issued = 0)

+//

+// False: the row moved, or a request is outstanding; the step decides

+// again.

+func (s *Store) SetSelfServeStep(id, phase, from, to string, deadline time.Time) (bool, error) {

+ return ssWon(s.db.Exec(`UPDATE self_serve_moves SET phase_step = ?, step_deadline = ?, updated_at = ?

+WHERE id = ? AND phase = ? AND phase_step = ? AND outcome = '' AND (request_consumed <> 0 OR request_issued = 0)`,

+ to, ssNanos(deadline), time.Now().UnixNano(), id, phase, from))

// ---- 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 7221216a0..3bc3c34d1 100644

--- a/internal/console/selfserve_store_test.go

+++ b/internal/console/selfserve_store_test.go

@@ -256,21 +256,27 @@ func TestSelfServeRowEvidence(t *testing.T) {

ssIssue(t, st, ssID, "queue", "reserving", "n-queue", ssT0)

issued(ssHost, true)

issued(ssHost2, false)

+ // Task 8: a row is never finished with its request unconfirmed (the

+ // write may still land): the finish matches nothing until the

+ // request is resolved.

if err := st.FinishSelfServeMove(ssID, "aborted", "cancelled", "", ssT0.Add(time.Minute)); err != nil {

t.Fatal(err)

}

- done(ssID, false) // terminal, release pending

- // Still issued: the queue request was never confirmed or consumed.

+ done(ssID, false)

issued(ssHost, true)

- if err := st.ClearSelfServeRelease(ssID); err != nil {

+ if err := st.VoidSelfServeRequest(ssID, "n-queue"); err != nil {

t.Fatal(err)

}

- done(ssID, true)

- if _, err := st.db.Exec(`UPDATE self_serve_moves SET request_consumed = 1 WHERE id = ?`, ssID); err != nil {

+ if err := st.FinishSelfServeMove(ssID, "aborted", "cancelled", "", ssT0.Add(time.Minute)); err != nil {

t.Fatal(err)

}

- // A terminal row whose request was consumed issues nothing more.

+ done(ssID, false) // terminal, release pending

+ // A terminal row whose request was resolved issues nothing more.

issued(ssHost, false)

+ if err := st.ClearSelfServeRelease(ssID); err != nil {

+ t.Fatal(err)

+ }

+ done(ssID, true)

}

func TestAdmitIsAtomic(t *testing.T) {