amadan

carlos / platform Public

self-serve job: queue and start check the very record they bind to (final review F1)

selfServeApply runs the request's Check on the judged read that gives the
request its incarnation and record_seq, so a writer that changes the record
after the checks (an instance PATCH steering a shared-pool instance) meets
a sequence conflict and the next decision refuses the steered record. The
queue and the start pass their eligibility checks as that Check.

Witness: TestQueueAndStartBindTheValidatedRecord (e2e).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude eda53b5940c12f6120777f6c11927d226606a4b3 parent f7cfd32
3 files changed, +151 −27
  • internal/console/selfserve_apply.go +40 −2
  • internal/console/selfserve_job.go +42 −25
  • internal/e2e/selfserve/validated_bind_test.go +69 −0
diff --git a/internal/console/selfserve_apply.go b/internal/console/selfserve_apply.go
index d3a105ee6..734e9cbe3 100644
--- a/internal/console/selfserve_apply.go
+++ b/internal/console/selfserve_apply.go
@@ -33,6 +33,38 @@ import (
type request struct {
Kind, Phase string
Deadline time.Time
+ // Check, when set, is the request's eligibility, run on the very
+ // judged read the request takes its incarnation and record_seq from
+ // (final review F1): the checks and the bound write refer to one
+ // record version, so a writer that changes the record after the checks
+ // meets a sequence conflict, never a bind to a version nobody checked.
+ // ok false ends the step with err (the check's own disposition: a
+ // refusal waited out, an abort) and nothing is issued. It runs before
+ // every attempt, a resume of a request already outstanding included;
+ // 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)
+}
+
+// requestChecked is selfServeApply's answer when the request's Check
+// stopped it: err is the check's disposition, the step's answer.
+type requestChecked struct{ err error }
+
+func (r *requestChecked) Error() string {
+ if r.err == nil {
+ return "console: self-serve move: the request's checks stopped it"
+ }
+ return r.err.Error()
+}
+func (r *requestChecked) Unwrap() error { return r.err }
+
+// checkedOut reports whether err is a Check's stop, and the step's answer.
+func checkedOut(err error) (error, bool) {
+ var c *requestChecked
+ if errors.As(err, &c) {
+ return c.err, true
+ }
+ return nil, false
}
// The ends of a step that selfServeApply can answer.
@@ -133,8 +165,8 @@ func selfServeMutationRefused(err error) bool {
// 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,
+// 1. the judged record gives the expected incarnation and record_seq, and
+// the request's Check runs on that same read; 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
@@ -159,6 +191,12 @@ func (s *Server) selfServeApply(ctx context.Context, t *leaseToken, row *SelfSer
if err != nil {
return nil, err
}
+ if req.Check != nil {
+ if ok, err := req.Check(rec); !ok {
+ return nil, &requestChecked{err: err}
+ }
+ }
+ TestHooks.point(req.Kind + ":validated")
nonce, err := newRequestNonce()
if err != nil {
return nil, err
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 3f68e6431..c75c4019f 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -896,35 +896,40 @@ func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *Sel
return nil
}
- // Reserved: the fresh judged record, checked again before the queue.
- rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
- if err != nil {
- return err
- }
- if out, conf := selfServeOutstanding(row); !(out && !conf && row.Request.Kind == "queue") {
- // A queue already issued is replayed as it was decided; a new one
- // is decided on this record.
- if ref := selfServeBusy(rec); ref != nil {
- return s.selfServeRefused(row, ref, now)
+ // Reserved: the fresh judged record, checked again before the queue,
+ // on the very read the queue is bound to (F1).
+ out, conf := selfServeOutstanding(row)
+ check := func(rec *provision.Record) (bool, error) {
+ if !(out && !conf && row.Request.Kind == "queue") {
+ // A queue already issued is replayed as it was decided; a new
+ // one is decided on this record.
+ if ref := selfServeBusy(rec); ref != nil {
+ return false, s.selfServeRefused(row, ref, now)
+ }
+ if ref := s.offlineRecordChecks(ctx, row.Account, row.App, row.Host, rec); ref != nil {
+ return false, s.selfServeRefused(row, ref, now)
+ }
}
- if ref := s.offlineRecordChecks(ctx, row.Account, row.App, row.Host, rec); ref != nil {
- return s.selfServeRefused(row, ref, now)
+ var ref *offlineRefusal
+ sel, ref = s.offlineRegionChecks(ctx, row.Account, row.App, row.Host, rec, s.homeRegionOut(row.To), true, true, "")
+ if ref != nil {
+ return false, s.selfServeRefused(row, ref, now)
}
- }
- sel, ref = s.offlineRegionChecks(ctx, row.Account, row.App, row.Host, rec, s.homeRegionOut(row.To), true, true, "")
- if ref != nil {
- return s.selfServeRefused(row, ref, now)
- }
- if label := selfServeSelectionMoved(row, sel); label != "" {
- return s.selfServeAbort(row, ssReplaced(label), now)
+ if label := selfServeSelectionMoved(row, sel); label != "" {
+ return false, s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ return true, nil
}
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}, nil,
+ _, err = s.selfServeApply(ctx, t, row, request{Kind: "queue", Phase: ssReserving, Check: check}, 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)
})
if err != nil {
+ if err, ok := checkedOut(err); ok {
+ return err
+ }
var refusal *requestRefusal
if errors.As(err, &refusal) {
// The record refuses the queue: the request stays outstanding
@@ -1711,23 +1716,35 @@ func (s *Server) selfServeStopping(ctx context.Context, t *leaseToken, row *Self
if row.PhaseStep != ssStepStart || (out && row.Request.Kind != "start") {
return nil
}
- if !(out && !conf) {
+ // The start validation in full, on the very read the start is
+ // bound to (F1); a start already issued is replayed as decided.
+ check := func(rec *provision.Record) (bool, error) {
+ if out && !conf {
+ return true, nil
+ }
+ if selfServeOurs(rec, row) == nil || rec.Move.State != provision.OfflineStaging {
+ return false, nil // the record moved on: reconciliation reads it next tick
+ }
if ref := s.offlineRecordChecks(ctx, row.Account, row.App, row.Host, rec); ref != nil {
- return s.selfServeRefused(row, ref, now)
+ return false, s.selfServeRefused(row, ref, now)
}
sel, ref := s.offlineRegionChecks(ctx, row.Account, row.App, row.Host, rec, s.homeRegionOut(row.To), false, true, "")
if ref != nil {
- return s.selfServeRefused(row, ref, now)
+ return false, s.selfServeRefused(row, ref, now)
}
if label := selfServeSelectionMoved(row, sel); label != "" {
- return s.selfServeAbort(row, ssReplaced(label), now)
+ return false, s.selfServeAbort(row, ssReplaced(label), now)
}
+ return true, nil
}
- _, err := s.selfServeApply(ctx, t, row, request{Kind: "start", Phase: ssStopping}, nil,
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "start", Phase: ssStopping, Check: check}, nil,
func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
return provision.OfflineStartStaged(row.ID, row.RequestedBy, at, rc)
})
if err != nil {
+ if err, ok := checkedOut(err); ok {
+ return err
+ }
return s.selfServePreOfflineRefusal(row, err, now)
}
if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
diff --git a/internal/e2e/selfserve/validated_bind_test.go b/internal/e2e/selfserve/validated_bind_test.go
new file mode 100644
index 000000000..1864fdbc9
--- /dev/null
+++ b/internal/e2e/selfserve/validated_bind_test.go
@@ -0,0 +1,69 @@
+//go:build e2e
+
+package selfserve
+
+import (
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/console"
+ "github.com/carlosframework/platform/internal/host"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// TestQueueAndStartBindTheValidatedRecord (final review F1): the queue and
+// the start are each bound to the very record their checks judged. An
+// ordinary instance PATCH that steers the shared-pool instance between the
+// checks and the bound write (a record offlineRecordChecks refuses) never
+// lets the transition land on the steered record: the bound write meets a
+// sequence conflict, and the next decision refuses the steered record.
+func TestQueueAndStartBindTheValidatedRecord(t *testing.T) {
+ steer := func(t *testing.T, w *jobWorld) {
+ // The PATCH's write on a console with lifetimes on: sequenced.
+ before := w.record().RecordSeq
+ if _, err := provision.SetRecordMove(provision.WithRecordSequence(w.ctx), w.raw, w.acct, host.E2EApp, host.E2EHost,
+ func(rec *provision.Record) error { rec.Steering = "latency"; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ if after := w.record().RecordSeq; after == before {
+ t.Fatalf("the PATCH did not advance record_seq (%d)", after)
+ }
+ }
+ t.Run("queue", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ w.at("queue:validated", func() { steer(t, w) })
+ w.until(20, "the row ended", func() bool { return w.row(row.ID).Outcome != "" || w.record().Move != nil })
+ if rec := w.record(); rec.Move != nil {
+ t.Fatalf("the queue landed on the steered record: %+v", rec.Move)
+ }
+ if r := w.row(row.ID); r.Phase == "queued" || r.Outcome == "" {
+ t.Fatalf("the steered record was not refused: %s", describe(r))
+ }
+ })
+ t.Run("start", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ cutSetup("")(w, row.ID)
+ started := false
+ w.fs.setPut(func(key string, body []byte) putFault {
+ if strings.Contains(string(body), `"`+provision.OfflineStopping+`"`) {
+ started = true
+ }
+ return putRun
+ })
+ w.at("start:validated", func() { steer(t, w) })
+ w.run(60, "the row ended", row.ID, func(r *console.SelfServeMove) bool { return r.Outcome != "" || started })
+ if started {
+ t.Fatal("the start was written on the steered record")
+ }
+ if r := w.row(row.ID); r.Outcome == "" || w.hits["start:validated"] == 0 {
+ t.Fatalf("the steered record was not refused: %s", describe(r))
+ }
+ if rec := w.record(); rec.LastMove != nil && rec.LastMove.OfflineS != 0 {
+ t.Fatalf("the instance went offline: %+v", rec.LastMove)
+ }
+ })
+}