amadan

carlos / platform Public

console: an operator's start of a self-serve staged move takes it: the job yields (aborted, another move is in flight) and leaves the reservations to the operator's clearing door

Fix round 1, Important 4: reconciliation detects a record started without
the job's start receipt (selfServeTakenOver) and finishes the row with
YieldSelfServeMove (release_pending = 0); selfServeDoorCleared releases a
yielded row's reservations when the operator clears the record.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 50eeec86f019f8fc44891a7d1659d39844f572a5 parent aa692a3
5 files changed, +122 −4
  • internal/console/apiinstancemove_offline.go +6 −1
  • internal/console/selfserve_job.go +35 −0
  • internal/console/selfserve_store.go +14 −2
  • internal/e2e/selfserve/cut_test.go +35 −0
  • internal/e2e/selfserve/job_reconcile_test.go +32 −1
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 160d84085..be6a7db0b 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -1135,7 +1135,12 @@ func (s *Server) selfServeDoorCleared(id string, rec *provision.Record) bool {
if row == nil {
return false
}
- if row.Outcome != "" || rec.LastMove == nil || rec.LastMove.ID != id {
+ if row.Outcome != "" {
+ // Finished already: the finaliser releases a pending release; a
+ // move the job yielded to an operator is released by this door.
+ return row.ReleasePending
+ }
+ if rec.LastMove == nil || rec.LastMove.ID != id {
return true
}
if out, conf := selfServeOutstanding(row); out && !conf && rec.RecordSeq > row.Request.ExpectSeq {
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index f0a0d803d..7f2323a90 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -377,6 +377,20 @@ func (s *Server) selfServeReconcile(ctx context.Context, t *leaseToken, row *Sel
}
return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, msgSelfServeAnotherMove, "")
}
+ if selfServeTakenOver(row, mv) {
+ // An operator's start took the move offline: the move is the
+ // operator's from here, and the job yields it (its reservations
+ // stay with the move; the operator's clearing door releases them).
+ if unresolved() {
+ return true, s.selfServeResolveFenced(ctx, t, row)
+ }
+ TestHooks.point("finalise:yield")
+ if err := s.store.YieldSelfServeMove(row.ID, msgSelfServeAnotherMove, now); err != nil {
+ return true, err
+ }
+ log.Printf("console: self-serve move %s (%s): an operator started the move; the job yields it", row.ID, row.Host)
+ return true, s.selfServeFinaliseRest(ctx, t, row.ID)
+ }
set, ok := selfServeSets[mv.State]
if !ok || containsPhase(set.phases, row.Phase) {
return false, nil
@@ -423,6 +437,27 @@ func (s *Server) selfServeReconcile(ctx context.Context, t *leaseToken, row *Sel
return true, nil
}
+// selfServeTakenOver reports whether an operator took the move offline: the
+// record is past its start (stopping through finishing) and the row never
+// started it (the row is before stopping, or at the start substep without
+// the start the record's receipt names).
+func selfServeTakenOver(row *SelfServeMove, mv *provision.MoveState) bool {
+ switch mv.State {
+ case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding,
+ provision.OfflineLanded, provision.OfflineOpen, provision.OfflineFinishing:
+ default:
+ return false
+ }
+ if selfServeRank[row.Phase] > selfServeRank[ssStopping] || row.Phase == ssReverting {
+ return false
+ }
+ if row.Phase == ssStopping && row.PhaseStep != ssStepStart {
+ return false
+ }
+ ours := row.Request.Kind == "start" && mv.SelfServe != nil && mv.SelfServe.LastRequest == row.Request.Nonce
+ return !ours
+}
+
// selfServeOutcomeReason is a finished move's reason: none for done; for an
// abort, the reason the job recorded when it decided it, else a
// cancellation's.
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index e65abde9e..f308161d4 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -951,6 +951,18 @@ func (s *Store) SetSelfServeAbortFinished(id string, at time.Time) error {
// resolved first). Matching nothing is not an error: the row is finished
// already, or a request is outstanding (the caller reads the row again).
func (s *Store) FinishSelfServeMove(id, outcome, reason, warning string, now time.Time) error {
+ return s.finishSelfServeMove(id, outcome, reason, warning, true, now)
+}
+
+// YieldSelfServeMove is FinishSelfServeMove for a move an operator took
+// over (an operator's start of a self-serve staged move): outcome aborted
+// with reason, and release_pending = 0, since the reservations are the
+// operator's move's now and its own clearing door releases them.
+func (s *Store) YieldSelfServeMove(id, reason string, now time.Time) error {
+ return s.finishSelfServeMove(id, provision.OfflineOutcomeAborted, reason, "", false, now)
+}
+
+func (s *Store) finishSelfServeMove(id, outcome, reason, warning string, release bool, now time.Time) error {
if outcome == "" {
return fmt.Errorf("console: self-serve move %s: an outcome is required", id)
}
@@ -961,11 +973,11 @@ func (s *Store) FinishSelfServeMove(id, outcome, reason, warning string, now tim
defer tx.Rollback()
res, err := tx.Exec(`UPDATE self_serve_moves SET
outcome = ?, outcome_reason = ?, warning = CASE WHEN ? <> '' THEN ? ELSE warning END,
- phase = 'done', phase_started_at = ?, step_deadline = 0, release_pending = 1, effects_done = 0, open_record_pending = 0,
+ phase = 'done', phase_started_at = ?, step_deadline = 0, release_pending = ?, effects_done = 0, open_record_pending = 0,
request_consumed = CASE WHEN request_confirmed <> 0 THEN 1 ELSE request_consumed END,
updated_at = ?
WHERE id = ? AND outcome = '' AND NOT (request_issued <> 0 AND request_consumed = 0 AND request_confirmed = 0)`,
- outcome, reason, warning, warning, ssNanos(now), ssNanos(now), id)
+ outcome, reason, warning, warning, ssNanos(now), ssBool(release), ssNanos(now), id)
won, err := ssWon(res, err)
if err != nil {
return err
diff --git a/internal/e2e/selfserve/cut_test.go b/internal/e2e/selfserve/cut_test.go
index 56be85c0f..b795db25b 100644
--- a/internal/e2e/selfserve/cut_test.go
+++ b/internal/e2e/selfserve/cut_test.go
@@ -616,6 +616,41 @@ func TestOperatorDoorsFinaliseASelfServeMove(t *testing.T) {
t.Fatalf("finished %s, reservations %v", describe(r), w.reservations())
}
})
+ t.Run("an operator's start takes the move", func(t *testing.T) {
+ // The fix round's pin (Controller ruling 5): an operator starts a
+ // self-serve staged move; the job yields it ("another move is in
+ // flight") without releasing the reservations, and writes nothing
+ // more; the operator's own clearing door releases them.
+ w := newJobWorld(t)
+ row := w.admit()
+ w.phase(row.ID, "ready", 60)
+ if code, out := w.door("POST", "", `{"region":"uk"}`); code != 202 {
+ t.Fatalf("start = %d %s", code, out)
+ }
+ w.consoleUntil(3, "yielded", func() bool { return w.row(row.ID).Outcome != "" })
+ r := w.row(row.ID)
+ if r.Outcome != "aborted" || r.OutcomeReason != "another move is in flight" || r.ReleasePending || !w.reservedBy(row.ID) {
+ t.Fatalf("yielded %s, release %v, reservations %v", describe(r), r.ReleasePending, w.reservations())
+ }
+ seq := w.record().RecordSeq
+ for i := 0; i < 3; i++ {
+ w.tick()
+ }
+ if got := w.record().RecordSeq; got != seq {
+ t.Fatalf("the job wrote the operator's record after yielding (seq %d, was %d)", got, seq)
+ }
+ // The operator aborts and finishes: the door releases.
+ if code, out := w.door("POST", "/abort", ""); code != 202 {
+ t.Fatalf("abort = %d %s", code, out)
+ }
+ for i := 0; i < 40 && w.record().Move != nil; i++ {
+ w.boxes()
+ w.door("POST", "/finish", "")
+ }
+ if rec := w.record(); rec.Move != nil || len(w.reservations()) != 0 {
+ t.Fatalf("after the operator's finish: %+v, reservations %v", rec.Move, w.reservations())
+ }
+ })
t.Run("the operator's abort of a queued move", func(t *testing.T) {
w := newJobWorld(t)
row := w.admit()
diff --git a/internal/e2e/selfserve/job_reconcile_test.go b/internal/e2e/selfserve/job_reconcile_test.go
index 4006c2f13..8970ebc23 100644
--- a/internal/e2e/selfserve/job_reconcile_test.go
+++ b/internal/e2e/selfserve/job_reconcile_test.go
@@ -61,11 +61,16 @@ func TestReconcileForwardNeverRegresses(t *testing.T) {
{provision.OfflineFinishing, "finishing", "gate2"},
{provision.OfflineReverting, "reverting", "finish"},
} {
- t.Run("ready row, record "+c.state, func(t *testing.T) {
+ t.Run("started row, record "+c.state, func(t *testing.T) {
+ // The job started the move (Task 8: a row that never started it
+ // yields to the operator who did, below).
w := newJobWorld(t)
row := w.admit()
w.phase(row.ID, "ready", 60)
w.down()
+ if c.state != provision.OfflineReverting {
+ w.setPhase(row.ID, "stopping", "seal")
+ }
w.operatorTo(row.ID, c.state)
w.consoleUntil(3, "reconciled to "+c.phase, func() bool { return w.row(row.ID).Phase == c.phase })
if r := w.row(row.ID); r.PhaseStep != c.step || r.Outcome != "" {
@@ -83,6 +88,29 @@ func TestReconcileForwardNeverRegresses(t *testing.T) {
})
}
+ // Task 8: an operator's start of a ready row's move takes it: the job
+ // yields (aborted "another move is in flight", the reservations left to
+ // the operator's move), whatever state the operator took it to.
+ for _, state := range []string{provision.OfflineStopping, provision.OfflineSealed, provision.OfflineLanded, provision.OfflineOpen, provision.OfflineFinishing} {
+ t.Run("ready row, an operator's move at "+state, func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.phase(row.ID, "ready", 60)
+ w.down()
+ before := w.record().RecordSeq
+ w.operatorTo(row.ID, state)
+ after := w.record().RecordSeq
+ w.consoleUntil(3, "yielded", func() bool { return w.row(row.ID).Outcome != "" })
+ r := w.row(row.ID)
+ if r.Outcome != "aborted" || r.OutcomeReason != "another move is in flight" || r.ReleasePending || !w.reservedBy(row.ID) {
+ t.Fatalf("yielded %s release %v reservations %v", describe(r), r.ReleasePending, w.reservations())
+ }
+ if rec := w.record(); rec.RecordSeq != after || before == after {
+ t.Fatalf("the job wrote the operator's record: seq %d, the operator left %d", rec.RecordSeq, after)
+ }
+ })
+ }
+
// Rows already in the record's set stay.
stays := func(t *testing.T, w *jobWorld, id, phase string) {
t.Helper()
@@ -124,6 +152,7 @@ WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now()
row := w.admit()
w.phase(row.ID, "ready", 60)
w.down()
+ w.setPhase(row.ID, "stopping", "seal")
w.operatorTo(row.ID, provision.OfflineSealed)
w.consoleUntil(3, "sealing", func() bool { return w.row(row.ID).Phase == "sealing" })
w.setPhase(row.ID, "landing", "")
@@ -135,6 +164,7 @@ WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now()
row := w.admit()
w.phase(row.ID, "ready", 60)
w.down()
+ w.setPhase(row.ID, "stopping", "seal")
w.operatorTo(row.ID, provision.OfflineOpen)
w.consoleUntil(3, "opened", func() bool { return w.row(row.ID).Phase == "opened" })
step := ""
@@ -275,6 +305,7 @@ WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now()
row := w.admit()
w.phase(row.ID, "ready", 60)
w.down()
+ w.setPhase(row.ID, "stopping", "seal")
w.operatorTo(row.ID, provision.OfflineLanded)
w.consoleUntil(3, "landed", func() bool { return w.row(row.ID).Phase == "landed" })
w.setPhase(row.ID, "open-uncertain", "")