amadan

carlos / platform Public

self-serve job: a superseded round's abandonment sets the replacement's fresh deadline (final review F8)

selfServeSuperseded abandons the wait with the replacement round's fresh
step deadline in the same statement (selfServeReplacementDeadline: the
issuing phase's deadline, the maintenance round's copy allowance in ready),
never the old round's remainder, so a crash between the abandon and the
issue cannot exhaust the replacement before it starts.

Witness: TestOperatorRoundSupersedesTheJobsWait, three new cases (staging,
prechecks, restage) superseded ten seconds before the old deadline with a
restart between the abandon and the issue.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e971dae9e2597d589271aff2259afc711b752a5d parent 5591ccb
2 files changed, +95 −9
  • internal/console/selfserve_job.go +26 −9
  • internal/e2e/selfserve/job_races_test.go +69 −0
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 75299e38c..a07a5dbc1 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -1125,8 +1125,11 @@ func selfServeStagerReport(row *SelfServeMove, mv *provision.MoveState, ev offli
// later than the confirmed request's proposed stamp, so an operator round
// moved it without a receipt. The job abandons its wait (one UPDATE) and
// issues a fresh bound round on the next step; it never adopts the
-// operator's round.
-func (s *Server) selfServeSuperseded(row *SelfServeMove, mv *provision.MoveState) (bool, error) {
+// operator's round. The abandonment sets the replacement round's fresh
+// step deadline in the same statement (final review F8), so a crash
+// between the abandon and the issue never leaves the replacement on the
+// old round's remainder.
+func (s *Server) selfServeSuperseded(row *SelfServeMove, mv *provision.MoveState, now time.Time) (bool, error) {
var stamp *time.Time
switch row.Request.Kind {
case "stage", "maintain":
@@ -1140,10 +1143,24 @@ func (s *Server) selfServeSuperseded(row *SelfServeMove, mv *provision.MoveState
return false, nil
}
TestHooks.point(row.Request.Kind + ":abandon")
- _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline)
+ _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, selfServeReplacementDeadline(row, now))
return true, err
}
+// selfServeReplacementDeadline is the fresh allowance of the round that
+// replaces a superseded one, by the phase the round was issued in: the
+// phase's own deadline (a staging round's, the prechecks' wait, the cut's
+// copy allowance), and in ready the maintenance round's copy allowance.
+func selfServeReplacementDeadline(row *SelfServeMove, now time.Time) time.Time {
+ if row.Request.Phase == ssReady {
+ return now.Add(selfServeCopyAllowance(row))
+ }
+ if d := selfServePhaseDeadline(row, row.Request.Phase, now); !d.IsZero() {
+ return d
+ }
+ return row.StepDeadline
+}
+
// selfServeStaging is the pre-copy, round after round: a round's report
// (the stager's, for the confirmed request's stamp) either meets the ready
// rule (or the bound: 30 minutes or 10 rounds) and the row goes to the
@@ -1156,7 +1173,7 @@ func (s *Server) selfServeStaging(ctx context.Context, t *leaseToken, row *SelfS
return nil
}
if selfServeWaiting(row, "stage", "maintain") {
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
@@ -1249,7 +1266,7 @@ func (s *Server) selfServePrechecks(ctx context.Context, t *leaseToken, row *Sel
}
if row.PhaseStep == ssStepPrecheck {
if selfServeWaiting(row, "precheck") {
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
fleet, ok := s.readOfflineFleet(ctx)
@@ -1359,7 +1376,7 @@ func (s *Server) selfServeReady(ctx context.Context, t *leaseToken, row *SelfSer
// An operator round superseded the outstanding one: it is
// abandoned, and the next step finds none and restage writes the
// cut's own round.
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
@@ -1384,7 +1401,7 @@ func (s *Server) selfServeReady(ctx context.Context, t *leaseToken, row *SelfSer
_, err := s.selfServeApply(ctx, t, row, request{Kind: "maintain", Phase: ssReady}, nil, selfServeRestageMut(row.ID))
return s.selfServePreOfflineRefusal(row, err, now)
}
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
@@ -1435,7 +1452,7 @@ func (s *Server) selfServeRestage(ctx context.Context, t *leaseToken, row *SelfS
return err
}
if selfServeWaiting(row, "stage") && row.Request.Phase == ssRestage {
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
@@ -1660,7 +1677,7 @@ func (s *Server) selfServeBarrier(ctx context.Context, t *leaseToken, row *SelfS
})
return s.selfServePreOfflineRefusal(row, err, now)
}
- if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ if sup, err := s.selfServeSuperseded(row, mv, now); sup || err != nil {
return err
}
fleet, ok := s.readOfflineFleet(ctx)
diff --git a/internal/e2e/selfserve/job_races_test.go b/internal/e2e/selfserve/job_races_test.go
index e3ce7c221..70f12fb7a 100644
--- a/internal/e2e/selfserve/job_races_test.go
+++ b/internal/e2e/selfserve/job_races_test.go
@@ -335,6 +335,75 @@ func TestOperatorRoundSupersedesTheJobsWait(t *testing.T) {
t.Fatalf("restage's own rounds: %v", n)
}
})
+ // Final review F8: the abandonment sets the replacement round's fresh
+ // step deadline in the same statement. Superseded ten seconds before
+ // the old round's deadline, and the console crashing between the
+ // abandon and the issue, the restarted console past the old deadline
+ // issues the replacement round on its fresh allowance and does not
+ // expire.
+ nearDeadline := func(t *testing.T, w *jobWorld, id, kind, phase string, operator func(rec *provision.Record) error) {
+ w.down()
+ w.consoleUntil(5, "a round confirmed", func() bool {
+ r := w.row(id)
+ return r.Phase == phase && r.Request.Kind == kind && r.Request.Confirmed && !r.Request.Consumed
+ })
+ s1 := w.row(id)
+ old := s1.StepDeadline
+ if old.IsZero() {
+ t.Fatalf("no step deadline on the confirmed round: %s", describe(s1))
+ }
+ w.skew += old.Sub(time.Now().Add(w.skew)) - 10*time.Second
+ w.edit(operator)
+ w.crashWhen = func(p string, r *console.SelfServeMove) bool {
+ return strings.HasSuffix(p, ":issue") && r.RequestAbandoned
+ }
+ w.consoleUntil(4, "the crash between the abandon and the issue", func() bool { return w.crashWhen == nil })
+ if r := w.crashed; !r.StepDeadline.After(old.Add(time.Minute)) {
+ t.Fatalf("the abandonment kept the old round's deadline %v (now %v), want a fresh allowance", r.StepDeadline, old)
+ }
+ w.restart()
+ w.advance(30 * time.Second) // past the old round's deadline
+ w.consoleUntil(4, "a fresh round", func() bool {
+ r := w.row(id)
+ return r.Outcome != "" || (r.Request.Nonce != s1.Request.Nonce && r.Request.Kind == kind && !r.Request.Consumed)
+ })
+ if r := w.row(id); r.Outcome != "" || r.Phase != phase {
+ t.Fatalf("the replacement round expired on the old deadline: %s", describe(r))
+ }
+ w.up()
+ }
+ restage := func(rec *provision.Record) error {
+ return provision.OfflineRestage(rec.Move.ID, stamp(rec), provision.Receipt{})(rec)
+ }
+ t.Run("staging, near the old deadline, crashed between the abandon and the issue", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ w.phase(row.ID, "staging", 60)
+ nearDeadline(t, w, row.ID, "stage", "staging", restage)
+ w.phase(row.ID, "ready", 60)
+ })
+ t.Run("prechecks, near the old deadline, crashed between the abandon and the issue", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ w.phase(row.ID, "prechecks", 60)
+ nearDeadline(t, w, row.ID, "precheck", "prechecks", func(rec *provision.Record) error {
+ return provision.OfflinePrecheck(rec.Move.ID, stamp(rec), provision.Receipt{})(rec)
+ })
+ w.phase(row.ID, "ready", 60)
+ })
+ t.Run("restage, near the old deadline, crashed between the abandon and the issue", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ w.phase(row.ID, "ready", 60)
+ if !w.markCut(row.ID) {
+ t.Fatal("the cut was refused")
+ }
+ nearDeadline(t, w, row.ID, "stage", "restage", restage)
+ w.phase(row.ID, "barrier", 20)
+ })
t.Run("ready", func(t *testing.T) {
w := newJobWorld(t)
row := w.admit()