amadan

carlos / platform Public

console: the start and stage doors reserve boxes and write bound to their observation

The operator start and stage doors build an offlineSelection from
offlineSelect's fleet, reserve it (the staged start and the restage retry
the stage's own reservation), and write the record with SetRecordMoveAt
bound to the incarnation and record_seq of the judged record the gate read
before reserving, with the stamp proposed from it. A changed record is the
new 409 "the instance changed while the move was being started. try
again"; a new move's door releases its reservation once a read shows the
record without the move, a staged move's door keeps the stage's. Finish and
the finish of an abort release through releaseBoxes. offlineTakeLock and
offlineReleaseLock are gone.

TestOfflineStaleLockIsReplaced: a stale lock naming this door's own host
is reclaimed by fencing that host's record after the door observed it, so
the first start now answers the 409 and the retry starts.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f6e695c63ea1f8472789f82a04f9b0fd94138514 parent b7fe0e6
2 files changed, +117 −105
  • internal/console/apiinstancemove_offline.go +108 −105
  • internal/console/apiinstancemove_offline_test.go +9 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 3b2c34b09..9a775ae7d 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -613,8 +613,10 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re
if !ok {
return
}
+ sel := offlineOperatorSelection(fleet, d.rec.Region, to, provision.StageTarget{})
// A staged instance starts the move it is staged for: staging to
- // stopping, the instance offline from now. The lock is the stage's.
+ // stopping, the instance offline from now. The reservation is the
+ // stage's, retried as its own.
if mv, staged := offlineStaged(d.rec); staged {
if to != mv.To {
writeRefusal(w, s.offlineStagedFor(d.host, mv.To))
@@ -628,21 +630,19 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re
writeRefusal(w, ref)
return
}
- s.offlineAdvance(w, r, d, offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ s.offlineAdvanceBound(w, r, d, sel, mv.ID, func(at time.Time) func(*provision.Record) error {
return provision.OfflineStartStaged(mv.ID, d.addr, at, provision.Receipt{})
- }))
+ })
return
}
- s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
- return offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
- return provision.OfflineStart(id, d.rec.Region, to, d.addr, at, "")
- })
+ s.offlineBegin(w, r, d, sel, func(id string, at time.Time) func(*provision.Record) error {
+ return provision.OfflineStart(id, d.rec.Region, to, d.addr, at, "")
})
}
// handleAPIInstanceOfflineStage is the pre-copy door. On a record at rest it
// creates the move in offline-staging, checked exactly as start is and
-// under the same one-move lock, and the source region's stager runs one
+// under the same reservation, and the source region's stager runs one
// round on the serial bump. On a record already staging for that region it
// moves staged_at forward (OfflineRestage), so the stager runs one more
// round; each round copies only what is new (the receipts).
@@ -664,9 +664,10 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re
writeRefusal(w, ref)
return
}
- s.offlineAdvance(w, r, d, offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
+ sel := offlineOperatorSelection(fleet, d.rec.Region, to, provision.StageTarget{Stager: mv.Stager, Bucket: mv.StageBucket, Region: mv.StageRegion})
+ s.offlineAdvanceBound(w, r, d, sel, mv.ID, func(at time.Time) func(*provision.Record) error {
return provision.OfflineRestage(mv.ID, at, provision.Receipt{})
- }))
+ })
return
}
for _, b := range append(fleet.boxesAt(d.rec.Region), fleet.boxesAt(to)...) {
@@ -696,17 +697,23 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re
}
}
st := provision.StageTarget{Stager: moves.OfflineStager(names), Bucket: tgt.DataBucket, Region: tgt.DataBucketRegion}
- s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
- return offlineStamped(time.Now(), func(at time.Time) func(*provision.Record) error {
- return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, at, "")
- })
+ s.offlineBegin(w, r, d, offlineOperatorSelection(fleet, d.rec.Region, to, st), func(id string, at time.Time) func(*provision.Record) error {
+ return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, at, "")
})
}
// offlineBegin mints the move's id and writes the transition begin makes
-// with it (OfflineStart or OfflineStage) under the one-move lock, then
-// bumps the serial and answers 202 with the status view.
-func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineDoor, begin func(id string) func(*provision.Record) error) {
+// with it (OfflineStart or OfflineStage), then bumps the serial and answers
+// 202 with the status view. Predecessor-bound admission (self-serve moves
+// plan 2 Task 3): the door decided on d.rec, the judged record its gate read
+// before any reservation; the stamp is proposed from it
+// (provision.ProposeStamp), the reservation names its incarnation and
+// record_seq, and the write is bound to them (provision.SetRecordMoveAt),
+// one attempt. A record that changed since, including the fence of a reclaim
+// that took this door's stale reservation and a host recreated at a new
+// incarnation, refuses the write: the door answers 409 and releases its
+// reservation, and never reads again to retry.
+func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineDoor, sel offlineSelection, begin func(id string, at time.Time) func(*provision.Record) error) {
ctx := r.Context()
host := d.host
id, err := moves.NewMoveID(rand.Reader)
@@ -715,33 +722,47 @@ func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineD
jsonError(w, http.StatusBadGateway, "could not mint a move id")
return
}
- start := begin(id)
+ start := begin(id, provision.ProposeStamp(d.rec, time.Now()))
probe := *d.rec
if err := start(&probe); err != nil {
jsonError(w, http.StatusConflict, offlineConflict(err))
return
}
- // One offline move at a time: the lock is taken before the record is
- // written, and released if the write fails.
- if ref := s.offlineTakeLock(ctx, d, id); ref != nil {
+ // The boxes are reserved before the record is written. A failed
+ // reservation writes nothing.
+ if ref := s.reserveBoxes(ctx, sel, s.offlineReservationOf(d, id)); ref != nil {
writeRefusal(w, ref)
return
}
- // The write is bounded well under offlineLockGrace, so the lock cannot be
- // judged stale while it is still in flight.
+ // The write is bounded well under offlineLockGrace, so the reservation
+ // cannot be judged stale while it is still in flight.
wctx, cancel := context.WithTimeout(ctx, offlineStartWrite)
defer cancel()
- got, err := s.offlineApply(wctx, d, start)
+ got, err := s.offlineApplyBound(wctx, d, start)
if err != nil {
// A failed write does not prove nothing was written (a commit whose
- // reply was lost). The lock is never released here: if the record
- // carries this move the start succeeded, and if it does not the
- // lock goes stale and frees itself.
+ // reply was lost): if the record carries this move the start
+ // succeeded. This read confirms; it never retries the write.
rctx, rcancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer rcancel()
- if cur, _, rerr := s.readRecord(rctx, d.acct.Sqid, d.app, host); rerr == nil && cur.Move.Offline() && cur.Move.ID == id {
+ cur, _, rerr := s.readRecord(rctx, d.acct.Sqid, d.app, host)
+ switch {
+ case rerr == nil && cur.Move.Offline() && cur.Move.ID == id:
got, err = cur, nil
+ case errors.Is(err, release.ErrCASConflict):
+ // Not the record the door decided on. The reservation is
+ // released only once the record is read without this move;
+ // otherwise it goes stale and is reclaimed on evidence.
+ if rerr == nil {
+ if rel := s.releaseBoxes(context.WithoutCancel(ctx), moves.OfflineKindMove, id); rel != nil {
+ log.Printf("console: offline move %s: releasing the reservation of %s: %v", host, id, rel)
+ }
+ }
+ jsonError(w, http.StatusConflict, offlineChangedMsg)
+ return
}
+ // Any other failure keeps the reservation: if the record carries
+ // the move it is the move's, and if it does not it goes stale.
}
if err != nil {
s.offlineWriteFailed(w, d, err)
@@ -751,6 +772,56 @@ func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineD
writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, got))
}
+// offlineChangedMsg is the start and stage doors' refusal of a bound write
+// whose record changed after the door observed it (self-serve moves plan 2
+// Task 3).
+const offlineChangedMsg = "the instance changed while the move was being started. try again"
+
+// offlineReservationOf is the door's reservation for move id: d's host,
+// account and app, bound to the record the door observed.
+func (s *Server) offlineReservationOf(d offlineDoor, id string) moves.OfflineReservation {
+ return moves.OfflineReservation{Kind: moves.OfflineKindMove, ID: id, Host: d.host, Account: d.acct.Sqid, App: d.app,
+ ReservedAt: time.Now().UTC(), ExpectSeq: d.rec.RecordSeq, ExpectIncarnation: d.rec.Incarnation}
+}
+
+// offlineApplyBound is one transition bound to the record the door
+// observed: its incarnation and record_seq, one attempt.
+func (s *Server) offlineApplyBound(ctx context.Context, d offlineDoor, mut func(*provision.Record) error) (*provision.Record, error) {
+ return provision.SetRecordMoveAt(ctx, s.cfg.Releases, d.acct.Sqid, d.app, d.host,
+ provision.Expect{Incarnation: d.rec.Incarnation, Seq: d.rec.RecordSeq}, mut)
+}
+
+// offlineAdvanceBound is a staged move's start or restage: the move's own
+// reservation retried (it is the stage's, so an exact retry writes nothing),
+// then the transition bound to d.rec with its stamp proposed from it, the
+// serial bumped and 202 with the status view. A record that changed since
+// the door observed it is the 409 and the reservation is kept: it belongs
+// to the move the record still carries.
+func (s *Server) offlineAdvanceBound(w http.ResponseWriter, r *http.Request, d offlineDoor, sel offlineSelection, id string, mut func(at time.Time) func(*provision.Record) error) {
+ ctx := r.Context()
+ m := mut(provision.ProposeStamp(d.rec, time.Now()))
+ probe := *d.rec
+ if err := m(&probe); err != nil {
+ s.offlineWriteFailed(w, d, err)
+ return
+ }
+ if ref := s.reserveBoxes(ctx, sel, s.offlineReservationOf(d, id)); ref != nil {
+ writeRefusal(w, ref)
+ return
+ }
+ rec, err := s.offlineApplyBound(ctx, d, m)
+ if errors.Is(err, release.ErrCASConflict) {
+ jsonError(w, http.StatusConflict, offlineChangedMsg)
+ return
+ }
+ if err != nil {
+ s.offlineWriteFailed(w, d, err)
+ return
+ }
+ s.bumpSerial(ctx)
+ writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, rec))
+}
+
// offlineStartWrite bounds the start door's record write.
const offlineStartWrite = 30 * time.Second
@@ -772,7 +843,9 @@ func (s *Server) offlineWriteFailed(w http.ResponseWriter, d offlineDoor, err er
// offlineWrite applies one transition and bumps the serial, or answers the
// refusal itself. When the move it leaves is cleared (finish, or the finish
-// that ends an abort) it releases the lock.
+// that ends an abort) it releases the move's reservation; a failure is
+// logged and left, and the next reserver reclaims it on evidence (the record
+// no longer carries the move).
func (s *Server) offlineWrite(w http.ResponseWriter, r *http.Request, d offlineDoor, mut func(*provision.Record) error) (*provision.Record, bool) {
ctx := r.Context()
rec, err := s.offlineApply(ctx, d, mut)
@@ -782,7 +855,9 @@ func (s *Server) offlineWrite(w http.ResponseWriter, r *http.Request, d offlineD
}
s.bumpSerial(ctx)
if rec.Move == nil && d.rec.Move != nil {
- s.offlineReleaseLock(ctx, d.host, d.rec.Move.ID)
+ if err := s.releaseBoxes(context.WithoutCancel(ctx), moves.OfflineKindMove, d.rec.Move.ID); err != nil {
+ log.Printf("console: offline move %s: releasing the reservation of %s: %v", d.host, d.rec.Move.ID, err)
+ }
}
return rec, true
}
@@ -795,83 +870,11 @@ func (s *Server) offlineAdvance(w http.ResponseWriter, r *http.Request, d offlin
}
}
-// offlineLockGrace is how long a lock is believed whatever its record says:
-// the start door takes the lock, then writes the record.
+// offlineLockGrace is how long a reservation is believed whatever its
+// record says: a reserver reserves, then writes the record. Past it a
+// reservation is reclaimable, on evidence only (offlineReclaimable).
const offlineLockGrace = 2 * time.Minute
-// offlineTakeLock takes control/offline-moves/active.json for this host and
-// move, create-only. A lock already there is live while the record it names
-// still carries that move, or is younger than offlineLockGrace (MV-O30 naming
-// its host); otherwise it is stale (a crashed start, a failed release) and is
-// replaced under its etag. A
-// lock that cannot be read is not guessed at.
-func (s *Server) offlineTakeLock(ctx context.Context, d offlineDoor, id string) *offlineRefusal {
- unreadable := &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the instance"}
- body, err := json.Marshal(moves.OfflineActive{Host: d.host, Op: id, Account: d.acct.Sqid, App: d.app, At: time.Now().UTC()})
- if err != nil {
- return unreadable
- }
- _, err = s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, "")
- if err == nil {
- return nil
- }
- if !errors.Is(err, release.ErrCASConflict) {
- log.Printf("console: offline move %s: taking the lock: %v", d.host, err)
- return unreadable
- }
- raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey)
- var held moves.OfflineActive
- if err != nil || json.Unmarshal(raw, &held) != nil || held.Host == "" {
- log.Printf("console: offline move %s: reading the lock: %v", d.host, err)
- return unreadable
- }
- busy := &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s is moving offline. move one instance at a time", held.Host)}
- // A lock is taken before its record is written, so a record without the
- // move is only a stale lock once the holder has had time to write it.
- if time.Since(held.At) < offlineLockGrace {
- return busy
- }
- other, _, err := s.readRecord(ctx, held.Account, held.App, held.Host)
- switch {
- case errors.Is(err, os.ErrNotExist):
- case err != nil:
- log.Printf("console: offline move %s: reading the lock holder %s: %v", d.host, held.Host, err)
- return unreadable
- case other.Move.Offline() && other.Move.ID == held.Op:
- return busy
- }
- if _, err := s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, etag); err != nil {
- if errors.Is(err, release.ErrCASConflict) {
- return busy
- }
- log.Printf("console: offline move %s: replacing a stale lock: %v", d.host, err)
- return unreadable
- }
- return nil
-}
-
-// offlineReleaseLock deletes the lock under the etag it read, when it is
-// this move's. A failure is logged and left: the next start finds the lock
-// stale (its record no longer carries the move) and replaces it.
-func (s *Server) offlineReleaseLock(ctx context.Context, host, id string) {
- ctx = context.WithoutCancel(ctx)
- raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey)
- if errors.Is(err, os.ErrNotExist) {
- return
- }
- var held moves.OfflineActive
- if err != nil || json.Unmarshal(raw, &held) != nil {
- log.Printf("console: offline move %s: reading the lock to release it: %v", host, err)
- return
- }
- if held.Host != host || held.Op != id {
- return
- }
- if err := release.DeleteIfMatch(ctx, s.cfg.Releases, moves.OfflineActiveKey, etag); err != nil {
- log.Printf("console: offline move %s: releasing the lock: %v", host, err)
- }
-}
-
// ---- the step each state waits for ----
// The verbs.
diff --git a/internal/console/apiinstancemove_offline_test.go b/internal/console/apiinstancemove_offline_test.go
index ba7162c75..dae55a041 100644
--- a/internal/console/apiinstancemove_offline_test.go
+++ b/internal/console/apiinstancemove_offline_test.go
@@ -1399,8 +1399,17 @@ func TestOfflineStaleLockIsReplaced(t *testing.T) {
put(x, moves.OfflineActive{Host: "beta.shop.example.com", Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now().Add(-time.Hour)})
x.start()
// A lock whose record carries another move of that host is not this one's.
+ // Reclaiming it fences this host's own record after the door observed
+ // it, so the door's bound write refuses (self-serve moves plan 2 Task 3)
+ // and releases; the stale lock is gone, and the retry starts.
y := newOfflineWorld(t, oneEach...)
put(y, moves.OfflineActive{Host: offlineHost, Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now().Add(-time.Hour)})
+ if code, out := y.req(http.MethodPost, "", `{"region":"uk"}`); code != http.StatusConflict || errorText(out) != "the instance changed while the move was being started. try again" {
+ t.Fatalf("start over its own host's stale lock = %d %s", code, out)
+ }
+ if _, held := y.lock(); held {
+ t.Fatal("the stale lock outlived its reclaim")
+ }
y.start()
// A young lock is believed even before its record carries the move: the
// start door takes the lock, then writes the record.