amadan

carlos / platform Public

self-serve job: the open record's PUT runs under the lease (final review F11)

selfServeLanded checks the lease afresh and writes the open record under
selfServeWriteCtx (the lease's deadline); offlineOpenRecord takes a guard
it calls immediately before its create-only PUT, so a console that lost the
lease starts no PUT. The operator door passes no guard.

Witness: TestOpenRecordWriteNeedsTheLease (e2e), with faultStore.seen
inspecting the PUT's context deadline.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 34148e26dcc73d3e5bd156ad007bca1f601c049a parent 3e5d6a8
4 files changed, +111 −4
  • internal/console/apiinstancemove_offline.go +11 −2
  • internal/console/selfserve_job.go +17 −1
  • internal/e2e/selfserve/job_world_test.go +7 −1
  • internal/e2e/selfserve/open_record_lease_test.go +76 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 58e93d930..44c5241ea 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -1558,7 +1558,7 @@ func (s *Server) offlineVerbDoor(verb string) http.HandlerFunc {
if verb == offlineVerbOpen {
// The open record goes first (create-only, or this move's own
// adopted), and the open lands at its stamp.
- open, err := s.offlineOpenRecord(ctx, d.rec, mv, ev, provision.ProposeStamp(d.rec, time.Now()))
+ open, err := s.offlineOpenRecord(ctx, d.rec, mv, ev, provision.ProposeStamp(d.rec, time.Now()), nil)
if errors.Is(err, errOpenRecordInTheWay) {
jsonError(w, http.StatusConflict, msgOpenRecordInTheWay)
return
@@ -1588,7 +1588,11 @@ var errOpenRecordInTheWay = errors.New("console: offline move: another move's op
// adopted (a retry after a crash between the record and the open): its
// OpenedAt is the stamp the open carries. Anything else at the key is
// errOpenRecordInTheWay. A read or write that fails is the error.
-func (s *Server) offlineOpenRecord(ctx context.Context, rec *provision.Record, mv *provision.MoveState, ev offlineEvidence, proposed time.Time) (moves.OpenRecord, error) {
+//
+// guard, when set, is checked immediately before the PUT (the self-serve
+// job's fresh lease check, final review F11); its error is the answer and
+// nothing is written. The operator door passes nil.
+func (s *Server) offlineOpenRecord(ctx context.Context, rec *provision.Record, mv *provision.MoveState, ev offlineEvidence, proposed time.Time, guard func() error) (moves.OpenRecord, error) {
key := moves.OpenRecordKey(rec.Host, mv.ID)
if key == "" {
return moves.OpenRecord{}, errOpenRecordInTheWay
@@ -1614,6 +1618,11 @@ func (s *Server) offlineOpenRecord(ctx context.Context, rec *provision.Record, m
if err != nil {
return moves.OpenRecord{}, err
}
+ if guard != nil {
+ if err := guard(); err != nil {
+ return moves.OpenRecord{}, err
+ }
+ }
if _, err := s.cfg.Releases.Put(ctx, key, body, ""); err != nil {
if !errors.Is(err, release.ErrCASConflict) {
return moves.OpenRecord{}, err
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index c75c4019f..75299e38c 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -1952,7 +1952,23 @@ func (s *Server) selfServeLanded(ctx context.Context, t *leaseToken, row *SelfSe
}
}
TestHooks.point("open:record")
- open, err := s.offlineOpenRecord(ctx, rec, mv, ev, provision.ProposeStamp(rec, now))
+ // The open record's PUT under the record transitions' lease-bound
+ // context, the lease checked afresh immediately before it (F11).
+ guard := func() error {
+ if !t.valid(s.selfServeNow()) {
+ return errSelfServeLeaseLost
+ }
+ return nil
+ }
+ if err := guard(); err != nil {
+ return err
+ }
+ wctx, cancel := s.selfServeWriteCtx(ctx, t)
+ open, err := s.offlineOpenRecord(wctx, rec, mv, ev, provision.ProposeStamp(rec, now), guard)
+ cancel()
+ if errors.Is(err, errSelfServeLeaseLost) {
+ return err
+ }
if errors.Is(err, errOpenRecordInTheWay) {
return s.selfServeAbort(row, msgOpenRecordInTheWay, now)
}
diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go
index c11f290da..0b5ebaa6e 100644
--- a/internal/e2e/selfserve/job_world_test.go
+++ b/internal/e2e/selfserve/job_world_test.go
@@ -48,6 +48,9 @@ type faultStore struct {
unreadable bool
bumps int
puts map[string]int
+ // 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)
}
type putFault int
@@ -68,7 +71,7 @@ var errLostReply = errors.New("faultStore: the reply was lost")
func (f *faultStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
f.mu.Lock()
- decide := f.put
+ decide, seen := f.put, f.seen
if key == release.SerialKey {
f.bumps++
}
@@ -76,6 +79,9 @@ func (f *faultStore) Put(ctx context.Context, key string, body []byte, ifMatch s
f.puts = map[string]int{}
}
f.mu.Unlock()
+ if seen != nil {
+ seen(ctx, key)
+ }
fault := putRun
if decide != nil && provision.IsRecordKey(key) {
fault = decide(key, body)
diff --git a/internal/e2e/selfserve/open_record_lease_test.go b/internal/e2e/selfserve/open_record_lease_test.go
new file mode 100644
index 000000000..49437687f
--- /dev/null
+++ b/internal/e2e/selfserve/open_record_lease_test.go
@@ -0,0 +1,76 @@
+//go:build e2e
+
+package selfserve
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/console"
+ "github.com/carlosframework/platform/internal/host"
+ "github.com/carlosframework/platform/internal/moves"
+)
+
+// TestOpenRecordWriteNeedsTheLease (final review F11): the open record's
+// create-only PUT runs under the lease-bound context the record
+// transitions use, its deadline the lease's, and a fresh lease check
+// immediately before it. A console that loses the lease while it is about
+// to write the open record (renewal fails, the lease runs out, another
+// console takes it) starts no PUT afterwards; the console holding the
+// lease writes it with the lease's deadline on its context.
+func TestOpenRecordWriteNeedsTheLease(t *testing.T) {
+ w := newJobWorld(t)
+ r := w.cutTo("landed")
+ key := moves.OpenRecordKey(host.E2EHost, r.ID)
+ var (
+ lost bool
+ afterPut int
+ deadline time.Time
+ )
+ w.fs.seen = func(ctx context.Context, k string) {
+ if k != key {
+ return
+ }
+ if lost {
+ afterPut++
+ }
+ deadline, _ = ctx.Deadline()
+ }
+ w.at("open:record", func() {
+ // Another console takes the lease once this one's runs out; this
+ // console's clock passes its lease.
+ body, _ := json.Marshal(map[string]any{"console": "console-other", "until": time.Now().Add(w.skew + 10*time.Minute).UTC()})
+ _, etag, _ := w.raw.Get(w.ctx, "control/offline-moves/console.json")
+ if _, err := w.raw.Put(w.ctx, "control/offline-moves/console.json", body, etag); err != nil {
+ t.Fatal(err)
+ }
+ console.TestHooks.Advance(2 * time.Minute)
+ lost = true
+ })
+ w.consoleUntil(3, "at the open record", func() bool { return lost })
+ if afterPut != 0 {
+ t.Fatalf("%d open record PUTs started after the lease was lost", afterPut)
+ }
+ if _, _, err := w.raw.Get(w.ctx, key); err == nil {
+ t.Fatal("the open record was written after the lease was lost")
+ }
+
+ // The lease back (the other console's has run out): the open record
+ // is written under the lease's deadline.
+ w.restart()
+ lost = false
+ body, _ := json.Marshal(map[string]any{"console": "console-other", "until": time.Now().Add(w.skew - time.Second).UTC()})
+ _, etag, _ := w.raw.Get(w.ctx, "control/offline-moves/console.json")
+ if _, err := w.raw.Put(w.ctx, "control/offline-moves/console.json", body, etag); err != nil {
+ t.Fatal(err)
+ }
+ w.run(20, "the open record written", r.ID, func(*console.SelfServeMove) bool {
+ _, _, err := w.raw.Get(w.ctx, key)
+ return err == nil
+ })
+ if deadline.IsZero() || deadline.After(time.Now().Add(w.skew+70*time.Second)) {
+ t.Fatalf("the open record's PUT ran with deadline %v, want the lease's (60 s)", deadline)
+ }
+}