carlos / platform Public

self-serve job: a landed write stays owed its serial bump on the row (final review F9)

The confirmation records the landed record_seq as announce_seq in the same
statement (ConfirmSelfServeRequestLanded); the bump that announces it clears
it (SetSelfServeNotifiedSeq, now monotonic). Catch-up runs at the start of
every step, before the outstanding request is resumed or fenced, keyed to
announce_seq whatever the current request's state; a row confirmed before
the column existed keeps the earlier rule. The column is additive.

Witness: TestLandedWriteIsAnnouncedBehindTheNextRequest (e2e), same console
and after a restart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 day ago 0f46b21758750bf7810b82e193c3c80cc8c7cdf2 parent e971dae

5 files changed, +132 −17

internal/console/selfserve_apply.go +15 −7
internal/console/selfserve_job.go +5 −4
internal/console/selfserve_store.go +34 −6
internal/e2e/selfserve/catch_up_test.go +72 −0
internal/e2e/selfserve/job_world_test.go +6 −0

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

index 734e9cbe3..391a1ffc4 100644

--- a/internal/console/selfserve_apply.go

+++ b/internal/console/selfserve_apply.go

@@ -332,7 +332,7 @@ func (s *Server) selfServeRun(ctx context.Context, t *leaseToken, row *SelfServe

}

if selfServeHasReceipt(rec, pr.Nonce) {

TestHooks.point(pr.Kind + ":confirm")

- ok, err := s.store.ConfirmSelfServeRequest(row.ID, pr.Nonce, s.selfServeNow())

+ ok, err := s.store.ConfirmSelfServeRequestLanded(row.ID, pr.Nonce, rec.RecordSeq, s.selfServeNow())

if err != nil {

return nil, err

}

@@ -375,9 +375,10 @@ func (s *Server) selfServeRun(ctx context.Context, t *leaseToken, row *SelfServe

}

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

+// bump succeeded, records the record_seq it announced, clearing the owed

+// announcement (announce_seq) it covers. A failed bump leaves it owed, 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 {

@@ -390,10 +391,17 @@ func (s *Server) selfServeNotify(ctx context.Context, id string, seq uint64, kin

}

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

+// selfServeCatchUp is the at-least-once half of the notification, run at

+// the start of every step before any record transition: a landed write

+// still owed its bump (announce_seq, persisted with its confirmation) is

+// bumped now, whatever the current request's state (final review F9). A

+// row confirmed before announce_seq existed keeps the earlier rule: its

+// confirmed request's landed sequence (expect_seq + 1) not yet announced.

func (s *Server) selfServeCatchUp(ctx context.Context, row *SelfServeMove, rec *provision.Record) {

+ if row.AnnounceSeq != 0 {

+ s.selfServeNotify(ctx, row.ID, row.AnnounceSeq, row.Request.Kind)

+ return

+ }

if rec == nil || !row.Request.Confirmed || row.NotifiedSeq >= row.Request.ExpectSeq+1 {

return

}

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

index a07a5dbc1..e8cf4db7b 100644

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

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

@@ -237,6 +237,10 @@ func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServ

s.selfServePageRow(ctx, row, ssPageOpenRecord, ssPageWhyOpenRecord(row.ID), now)

}

+ // A landed write still owed its announcement is announced before

+ // anything else writes (F9).

+ s.selfServeCatchUp(ctx, row, rec)

// (3) The outstanding request first.

if out, conf := selfServeOutstanding(row); (out && !conf) || row.Request.Expired && !row.Request.Consumed {

_, rerr := s.selfServeResume(ctx, t, row)

@@ -266,9 +270,6 @@ func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServ

return rerr

}

- if !gone {

- s.selfServeCatchUp(ctx, row, rec)

- }

if done, err := s.selfServeReconcile(ctx, t, row, rec, gone, now); done || err != nil {

return err

}

@@ -565,7 +566,7 @@ func (s *Server) selfServeResolveFenced(ctx context.Context, t *leaseToken, row

// observation that carries its receipt, and announces it.

func (s *Server) selfServeConfirmFenced(ctx context.Context, row *SelfServeMove, rec *provision.Record) error {

TestHooks.point(row.Request.Kind + ":confirm")

- ok, err := s.store.ConfirmSelfServeRequest(row.ID, row.Request.Nonce, s.selfServeNow())

+ ok, err := s.store.ConfirmSelfServeRequestLanded(row.ID, row.Request.Nonce, rec.RecordSeq, s.selfServeNow())

if err != nil || !ok {

return err

}

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

index 811671ae9..aa1f2622a 100644

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

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

@@ -136,6 +136,11 @@ type SelfServeMove struct {

FinishGate string

NotifiedSeq uint64

+ // AnnounceSeq is a landed write's record_seq whose serial bump has not

+ // succeeded yet (final review F9): set with the request's confirmation,

+ // cleared by the bump that announces it, and bumped for on every tick

+ // whatever the current request's state. Zero is none owed.

+ AnnounceSeq uint64

EventSeq int64

Outcome, OutcomeReason, Warning string

@@ -270,6 +275,11 @@ func (s *Store) migrateSelfServe() error {

return err

}

+ // Final review F9: the landed write still owed its serial bump.

+ // Additive.

+ if _, err := s.db.Exec(`ALTER TABLE self_serve_moves ADD COLUMN announce_seq INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") {

+ return err

+ }

// Task 9: an event's failed deliveries, so one that can never be

// delivered is dead-lettered (DeliverSelfServeEvent). Additive.

if _, err := s.db.Exec(`ALTER TABLE self_serve_events ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") {

@@ -325,14 +335,14 @@ const selfServeMoveCols = `id, account, app, host, incarnation, from_region, to_

phase_step, stage_request, precheck_request, rounds, estimate_seconds, estimate_terms, reason,

health_attempts, health_consecutive_ok, health_first_ok_at, health_last_probe_at, health_last_error, finish_gate,

notified_seq, event_seq, outcome, outcome_reason, warning, abort_requested_at, source_reopened_at, abort_finished_at,

- effects_done, release_pending, open_record_pending, created_at, updated_at`

+ effects_done, release_pending, open_record_pending, created_at, updated_at, announce_seq`

type ssScanner interface{ Scan(dest ...any) error }

func scanSelfServeMove(r ssScanner) (*SelfServeMove, error) {

var (

m SelfServeMove

- inc, expSeq, expInc, notified int64

+ inc, expSeq, expInc, notified, announce int64

participants string

phaseStarted, step, window, expiryDeadline, stagingStart int64

expiryPending, issued, proposed, confirmed, consumed int64

@@ -348,7 +358,7 @@ func scanSelfServeMove(r ssScanner) (*SelfServeMove, error) {

&m.PhaseStep, &stageReq, &precheckReq, &m.Rounds, &m.EstimateSeconds, &m.EstimateTerms, &m.Reason,

&m.HealthAttempts, &m.HealthConsecutiveOK, &firstOK, &lastProbe, &m.HealthLastError, &m.FinishGate,

&notified, &m.EventSeq, &m.Outcome, &m.OutcomeReason, &m.Warning, &abortAt, &reopenedAt, &abortDone,

- &effects, &release, &openRec, &created, &updated); err != nil {

+ &effects, &release, &openRec, &created, &updated, &announce); err != nil {

return nil, err

}

var err error

@@ -364,6 +374,9 @@ func scanSelfServeMove(r ssScanner) (*SelfServeMove, error) {

if m.NotifiedSeq, err = ssStoredUint("notified_seq", notified); err != nil {

return nil, err

}

+ if m.AnnounceSeq, err = ssStoredUint("announce_seq", announce); err != nil {

+ return nil, err

+ }

if participants != "" {

if err := json.Unmarshal([]byte(participants), &m.Participants); err != nil {

return nil, fmt.Errorf("console: self-serve move %s participants: %w", m.ID, err)

@@ -694,16 +707,30 @@ WHERE id = ? AND phase = ? AND outcome = '' AND (request_consumed <> 0 OR reques

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

+ return s.ConfirmSelfServeRequestLanded(id, nonce, 0, now)

+// ConfirmSelfServeRequestLanded is ConfirmSelfServeRequest for a write

+// that landed at record_seq landed: the same statement records that

+// sequence as owed its serial bump (announce_seq, final review F9), so a

+// crash or a failed bump after the confirmation leaves it owed. Zero owes

+// nothing.

+func (s *Store) ConfirmSelfServeRequestLanded(id, nonce string, landed uint64, now time.Time) (bool, error) {

if ssNanos(now) <= 0 {

return false, fmt.Errorf("console: self-serve move %s: a confirmation needs a time", id)

}

+ seq, err := ssUint("announce_seq", landed)

+ if err != nil {

+ return false, err

+ }

return ssWon(s.db.Exec(`UPDATE self_serve_moves SET

request_confirmed = ?, request_expired = 0,

+ announce_seq = MAX(announce_seq, ?),

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

+ ssNanos(now), seq, ssNanos(now), id, nonce))

}

// ConsumeSelfServeRequest consumes a confirmed request whose wait is

@@ -942,13 +969,14 @@ func (s *Store) SetSelfServeWarning(id, warning string) error {

}

// SetSelfServeNotifiedSeq records the record_seq the serial was last bumped

-// for; a later tick bumps again while the record is ahead of it.

+// for, and clears the owed announcement (announce_seq) it covers: a bump

+// announces every landed write up to 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)

+ return s.setSelfServe("notified seq", id, `notified_seq = MAX(notified_seq, ?), announce_seq = CASE WHEN announce_seq <= ? THEN 0 ELSE announce_seq END`, v, v)

}

// SetSelfServeReopened records when the source reported reopened (abort).

diff --git a/internal/e2e/selfserve/catch_up_test.go b/internal/e2e/selfserve/catch_up_test.go

new file mode 100644

index 000000000..ed7792542

--- /dev/null

+++ b/internal/e2e/selfserve/catch_up_test.go

@@ -0,0 +1,72 @@

+//go:build e2e

+package selfserve

+import (

+ "testing"

+// TestLandedWriteIsAnnouncedBehindTheNextRequest (final review F9): a

+// landed write whose serial bump failed stays owed on the row (its landed

+// sequence) until a bump succeeds, whatever the current request's state.

+// The serial fails through a round's confirmation and the tick after; the

+// round's report lets the job consume it and issue the next request, whose

+// record PUT is held (unconfirmed); with the serial restored, the next tick

+// (and the one after a restart) announces the earlier landed write before

+// any other record transition.

+func TestLandedWriteIsAnnouncedBehindTheNextRequest(t *testing.T) {

+ for _, restart := range []bool{false, true} {

+ name := "same console"

+ if restart {

+ name = "after a restart"

+ }

+ t.Run(name, func(t *testing.T) {

+ w := newJobWorld(t)

+ row := w.admit()

+ w.watch = row.ID

+ w.phase(row.ID, "staging", 60)

+ w.fs.mu.Lock()

+ w.fs.serialFailing = true

+ w.fs.mu.Unlock()

+ // A request confirmed with the serial failing, and the tick after.

+ var landed uint64

+ w.until(20, "a request confirmed and unannounced", func() bool {

+ r := w.row(row.ID)

+ landed = w.record().RecordSeq

+ return r.Request.Confirmed && !r.Request.Consumed && r.NotifiedSeq < landed

+ })

+ w.down()

+ w.tick()

+ if r := w.row(row.ID); r.NotifiedSeq >= landed {

+ t.Fatalf("notified_seq %d with the serial failing, record_seq %d", r.NotifiedSeq, landed)

+ }

+ first := w.row(row.ID).Request.Nonce

+ // The round's report; the next request's record PUT is held.

+ w.up()

+ w.boxes()

+ w.fs.setPut(func(string, []byte) putFault { return putHeld })

+ w.consoleUntil(4, "the next request issued", func() bool {

+ r := w.row(row.ID)

+ return r.Request.Nonce != first && !r.Request.Confirmed && !r.Request.Consumed

+ })

+ if r := w.row(row.ID); r.NotifiedSeq >= landed {

+ t.Fatalf("notified_seq %d before the serial was restored", r.NotifiedSeq)

+ }

+ w.down()

+ w.fs.mu.Lock()

+ w.fs.serialFailing = false

+ w.fs.mu.Unlock()

+ if restart {

+ w.restart()

+ }

+ puts := w.fs.recordPuts()

+ w.tick()

+ if r := w.row(row.ID); r.NotifiedSeq < landed {

+ t.Fatalf("the landed write (record_seq %d) is still unannounced behind the next request: notified_seq %d, %s", landed, r.NotifiedSeq, describe(r))

+ }

+ if n := w.fs.recordPuts(); n != puts {

+ t.Fatalf("%d record transitions landed before the announcement", n-puts)

+ }

+ })

+ }

diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go

index 653f2ff56..91a9c798b 100644

--- a/internal/e2e/selfserve/job_world_test.go

+++ b/internal/e2e/selfserve/job_world_test.go

@@ -52,6 +52,8 @@ type faultStore struct {

// seen, when set, sees every Put of the console (any key) with its

// context, before the Put runs.

seen func(ctx context.Context, key string)

+ // serialFailing fails every serial bump while set.

+ serialFailing bool

}

type putFault int

@@ -75,6 +77,10 @@ func (f *faultStore) Put(ctx context.Context, key string, body []byte, ifMatch s

decide, seen := f.put, f.seen

if key == release.SerialKey {

f.bumps++

+ if f.serialFailing {

+ f.mu.Unlock()

+ return "", errLostReply

+ }

}

if f.puts == nil {

f.puts = map[string]int{}