carlos / platform Public

self-serve job: a queue request persists its staging coordinates and replays them (final review F10)

The queue request is issued with the staging bucket and region it writes
(request_stage_bucket, request_stage_region, additive columns); a replay
sends the identical body. When the participants advertise other
coordinates before the queue's fence, selfServeApply answers
errRequestStageMoved without writing and the job fences the request (its
receipt confirms it; absent, it is voided and a new request carries the new
coordinates). A queue persisted before the columns existed is fenced the
same way.

Witness: TestQueueReplayKeepsItsStagingCoordinates (e2e), capturing every
queue PUT body across failed writes and a restart.

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

claude 1 day ago 76bb215861d69530579da393158456c51fde1ccd parent 0f46b21

4 files changed, +144 −7

internal/console/selfserve_apply.go +17 −0
internal/console/selfserve_job.go +7 −1
internal/console/selfserve_store.go +21 −6
internal/e2e/selfserve/queue_replay_test.go +99 −0

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

index 391a1ffc4..87efbf6bf 100644

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

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

@@ -44,6 +44,12 @@ type request struct {

// the caller decides what a resume checks (a request already decided

// is replayed as it was decided).

Check func(rec *provision.Record) (ok bool, err error)

+ // Stage, when set, points at the staging coordinates the request's

+ // mutation writes (the queue's), read after Check: a new request

+ // persists them, and a request already outstanding is replayed only

+ // while they are the ones it was issued with; otherwise the answer is

+ // errRequestStageMoved and nothing is written (final review F10).

+ Stage *provision.StageTarget

}

// requestChecked is selfServeApply's answer when the request's Check

@@ -89,6 +95,12 @@ var (

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

+ // errRequestStageMoved: the outstanding request was issued with other

+ // staging coordinates than the ones the participants advertise now.

+ // Its replay would send another destination under the same identity,

+ // so the caller fences it (and a new request is issued once it is

+ // voided).

+ errRequestStageMoved = errors.New("console: self-serve move: the staging coordinates changed under the outstanding request")

// errRecordUnreadable wraps a judged read that could not tell what the

// record says (unreadable or corrupt): nothing is decided from it, and

// an open whose write may have landed goes to open-uncertain (Task 8).

@@ -202,6 +214,9 @@ func (s *Server) selfServeApply(ctx context.Context, t *leaseToken, row *SelfSer

return nil, err

}

pr := persistedRequest{Kind: req.Kind, Phase: req.Phase, Nonce: nonce, ExpectInc: rec.Incarnation, ExpectSeq: rec.RecordSeq}

+ if req.Stage != nil {

+ pr.StageBucket, pr.StageRegion = req.Stage.Bucket, req.Stage.Region

+ }

if proposed != nil {

pr.Proposed = proposed.UTC()

} else {

@@ -228,6 +243,8 @@ func (s *Server) selfServeApply(ctx context.Context, t *leaseToken, row *SelfSer

return nil, errRequestForeign

case existing.Confirmed && !existing.Consumed:

return rec, nil

+ case req.Stage != nil && (existing.StageBucket != req.Stage.Bucket || existing.StageRegion != req.Stage.Region):

+ return nil, errRequestStageMoved

}

return s.selfServeRun(ctx, t, row, *existing, mut, false)

}

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

index e8cf4db7b..48b90322b 100644

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

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

@@ -923,7 +923,7 @@ func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *Sel

}

ss := provision.SelfServeMove{ID: row.ID, Incarnation: row.Incarnation, DisplayName: row.DisplayName,

RequestedBy: row.RequestedBy, EstimateSeconds: int(row.EstimateSeconds)}

- _, err = s.selfServeApply(ctx, t, row, request{Kind: "queue", Phase: ssReserving, Check: check}, nil,

+ _, err = s.selfServeApply(ctx, t, row, request{Kind: "queue", Phase: ssReserving, Check: check, Stage: &sel.Stage}, nil,

func(at time.Time, rc provision.Receipt) func(*provision.Record) error {

return provision.OfflineQueue(row.ID, row.From, row.To, row.RequestedBy, sel.Stage, row.Participants, ss, at, rc)

})

@@ -931,6 +931,12 @@ func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *Sel

if err, ok := checkedOut(err); ok {

return err

}

+ if errors.Is(err, errRequestStageMoved) {

+ // Fenced so the old body can never land; voided, the next

+ // tick issues a new queue with the new coordinates (F10).

+ log.Printf("console: self-serve move %s: %v (fencing the queue)", row.ID, err)

+ return s.selfServeResolveFenced(ctx, t, row)

+ }

var refusal *requestRefusal

if errors.As(err, &refusal) {

// The record refuses the queue: the request stays outstanding

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

index aa1f2622a..ef779333b 100644

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

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

@@ -92,6 +92,11 @@ type persistedRequest struct {

Proposed, Deadline, CutRequestedAt time.Time

Confirmed, Consumed, Voided bool

Expired bool

+ // StageBucket and StageRegion are the staging coordinates a queue

+ // request was issued with (final review F10): its replay sends them

+ // again, never ones recomputed from a later fleet. Empty for every

+ // other kind, and for a queue issued before the columns existed.

+ StageBucket, StageRegion string

}

// SelfServeMove is one self_serve_moves row.

@@ -275,6 +280,13 @@ func (s *Store) migrateSelfServe() error {

return err

}

+ // Final review F10: the staging coordinates a queue request was issued

+ // with. Additive.

+ for _, col := range []string{"request_stage_bucket", "request_stage_region"} {

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

+ 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") {

@@ -335,7 +347,8 @@ 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, announce_seq`

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

+ request_stage_bucket, request_stage_region`

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

@@ -358,7 +371,8 @@ 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, &announce); err != nil {

+ &effects, &release, &openRec, &created, &updated, &announce,

+ &m.Request.StageBucket, &m.Request.StageRegion); err != nil {

return nil, err

}

var err error

@@ -612,9 +626,10 @@ func (s *Store) selfServeRequest(id string) (*persistedRequest, error) {

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

+ request_proposed, request_deadline, cut_requested_at, request_confirmed, request_consumed, request_voided, request_expired,

+ request_stage_bucket, request_stage_region

FROM self_serve_moves WHERE id = ?`, id).Scan(&issued, &pr.Kind, &pr.Phase, &pr.Nonce, &expInc, &expSeq,

- &proposed, &deadline, &cut, &confirmed, &consumed, &voided, &expired)

+ &proposed, &deadline, &cut, &confirmed, &consumed, &voided, &expired, &pr.StageBucket, &pr.StageRegion)

if errors.Is(err, sql.ErrNoRows) {

return nil, nil

}

@@ -678,11 +693,11 @@ func (s *Store) issueSelfServeRequest(id, kind, phase string, pr persistedReques

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([]any{ssNanos(now), kind, phase, pr.Nonce, inc, seq, ssNanos(pr.Proposed), pr.StageBucket, pr.StageRegion}, 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 = ?, `+set+`,

+ request_proposed = ?, request_stage_bucket = ?, request_stage_region = ?, `+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)`, args...)

won, err := ssWon(res, err)

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

new file mode 100644

index 000000000..a36d1dce0

--- /dev/null

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

@@ -0,0 +1,99 @@

+//go:build e2e

+package selfserve

+import (

+ "encoding/json"

+ "regexp"

+ "strings"

+ "testing"

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

+// TestQueueReplayKeepsItsStagingCoordinates (final review F10): the queue

+// request persists the staging coordinates it was issued with, and every

+// replay sends the identical body. When the target advertises another

+// data bucket before the queue's fence, the outstanding request is fenced

+// and voided and a new request (a new nonce) carries the new coordinates:

+// no request identity is ever sent with two destinations. Across the

+// queue's failed writes, with and without a console restart.

+func TestQueueReplayKeepsItsStagingCoordinates(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)

+ nonceOf := regexp.MustCompile(`"last_request":"(rq_[0-9a-f]+)"`)

+ bucketOf := regexp.MustCompile(`"stage_bucket":"([^"]*)"`)

+ buckets := map[string]map[string]bool{} // nonce -> buckets sent

+ failing := true

+ w.fs.setPut(func(key string, body []byte) putFault {

+ b := string(body)

+ if !strings.Contains(b, `"offline-queued"`) {

+ return putRun

+ }

+ if n, bk := nonceOf.FindStringSubmatch(b), bucketOf.FindStringSubmatch(b); n != nil && bk != nil {

+ if buckets[n[1]] == nil {

+ buckets[n[1]] = map[string]bool{}

+ }

+ buckets[n[1]][bk[1]] = true

+ }

+ if failing {

+ return putFail

+ }

+ return putRun

+ })

+ row := w.admit()

+ w.watch = row.ID

+ w.until(20, "a queue PUT sent", func() bool { return len(buckets) > 0 })

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

+ if w.row(row.ID).Request.Kind != "queue" || first == "" {

+ t.Fatalf("no queue request outstanding: %s", describe(w.row(row.ID)))

+ }

+ // The target advertises another data bucket (its edges file);

+ // the boxes stay still from here so the file stands.

+ key := "control/edges/" + host.E2ETargetBox + ".json"

+ raw, etag, err := w.raw.Get(w.ctx, key)

+ if err != nil {

+ t.Fatal(err)

+ }

+ var f map[string]any

+ if err := json.Unmarshal(raw, &f); err != nil {

+ t.Fatal(err)

+ }

+ f["data_bucket"] = "data-uk-moved"

+ body, _ := json.Marshal(f)

+ if _, err := w.raw.Put(w.ctx, key, body, etag); err != nil {

+ t.Fatal(err)

+ }

+ if restart {

+ w.restart()

+ }

+ w.tick() // a replay, still failing

+ failing = false

+ w.consoleUntil(10, "queued", func() bool {

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

+ return r.Phase != "requested" && r.Phase != "reserving" || r.Outcome != ""

+ })

+ for n, bs := range buckets {

+ if len(bs) != 1 {

+ t.Fatalf("request %s was sent with %d destinations: %v", n, len(bs), bs)

+ }

+ rec := w.record()

+ if rec.Move == nil || rec.Move.StageBucket != "data-uk-moved" {

+ t.Fatalf("the queue landed with %+v, want the new bucket", rec.Move)

+ }

+ if rec.Move.SelfServe.LastRequest == first {

+ t.Fatal("the new coordinates landed under the first request's identity")

+ }

+ if r := w.row(row.ID); r.Outcome != "" {

+ t.Fatalf("the move ended: %s", describe(r))

+ }

+ })

+ }