amadan

carlos / platform Public

console: reconciliation never moves a row to a lower rank, and abandons a confirmed wait the record went past

An explicit phase rank guards the forward advance: a row ahead of every
phase its record admits is logged and left. A confirmed request whose
wait the record has passed is abandoned on the advance unless the target
phase consumes it (the queue into queued) or waits on it (the reports'
stage request into staging), so a confirmed queue an operator went past
no longer blocks staging until the round's deadline. Both witnessed in
TestReconcileForwardNeverRegresses (RED before).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude bf78a2690bfc472b747ad73320c1f2e2971615a8 parent 2ced088
2 files changed, +64 −5
  • internal/console/selfserve_job.go +27 −5
  • internal/e2e/selfserve/job_reconcile_test.go +37 −0
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 58fc90009..1ff64be53 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -147,6 +147,17 @@ var selfServeSets = map[string]struct {
provision.OfflineReverting: {[]string{ssReverting}, ssReverting, ssStepFinish},
}
+// selfServeRank orders the phases a row moves through: reconciliation
+// never moves a row to a lower rank. reverting ranks with landed, the last
+// phase it can be entered from, so a reverting row may still be moved
+// forward to the post-open phases by a record an operator opened, and
+// never back into the pre-copy.
+var selfServeRank = map[string]int{
+ ssRequested: 0, ssReserving: 1, ssQueued: 2, ssReports: 3, ssStaging: 4, ssPrechecks: 5, ssReady: 6,
+ ssRestage: 7, ssBarrier: 8, ssStopping: 9, ssSealing: 10, ssLanding: 11, ssLanded: 12, ssReverting: 12,
+ ssOpenUncertain: 13, ssOpened: 14, ssHealth: 15, ssFinishing: 16, ssDone: 17,
+}
+
func containsPhase(phases []string, p string) bool {
for _, x := range phases {
if x == p {
@@ -364,15 +375,26 @@ func (s *Server) selfServeReconcile(ctx context.Context, t *leaseToken, row *Sel
// The record carries this move and the row is behind it: the queue
// landed (reserving, requested), or a write the row has not caught up
// with did (a crash between a confirmation and its advance, an
- // operator's transition). Forward to the set's first phase, consuming
- // a confirmed request whose wait the record has passed.
+ // operator's transition). Forward to the set's first phase, never to a
+ // lower rank: a row ahead of every phase its record admits is logged
+ // and left.
+ if selfServeRank[set.first] < selfServeRank[row.Phase] {
+ log.Printf("console: self-serve move %s (%s): the record is %s, behind the row's %s; the row is left where it is",
+ row.ID, row.Host, mv.State, row.Phase)
+ return false, nil
+ }
+ // A confirmed request is consumed by the advance when the set's first
+ // phase is where its wait ends (the queue's, into queued), and kept
+ // when that phase waits on it (the reports' stage request, into
+ // staging); any other confirmed wait the record has passed can never
+ // be satisfied there and is abandoned, never retained across phases.
nonce := ""
if selfServeWaiting(row, "queue") && set.first == ssQueued {
nonce = row.Request.Nonce
}
- if out, conf := selfServeOutstanding(row); out && conf && nonce == "" && set.first != ssStaging {
- // A confirmed wait the record has passed (an operator went past
- // it): it is left, never retained across phases.
+ kept := selfServeWaiting(row, "stage") && set.first == ssStaging
+ if out, conf := selfServeOutstanding(row); out && conf && nonce == "" && !kept {
+ TestHooks.point(row.Request.Kind + ":abandon")
if _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline); err != nil {
return true, err
}
diff --git a/internal/e2e/selfserve/job_reconcile_test.go b/internal/e2e/selfserve/job_reconcile_test.go
index de5d2e604..1eee16e66 100644
--- a/internal/e2e/selfserve/job_reconcile_test.go
+++ b/internal/e2e/selfserve/job_reconcile_test.go
@@ -6,6 +6,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/console"
"github.com/carlosframework/platform/internal/host"
"github.com/carlosframework/platform/internal/provision"
)
@@ -224,6 +225,42 @@ WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now()
t.Fatalf("the other move was touched: %+v", after.Move)
}
})
+ t.Run("a confirmed queue an operator went past is abandoned", func(t *testing.T) {
+ // The queue is confirmed and the job stops before consuming it;
+ // an operator starts the pre-copy. The row is moved forward to
+ // staging and the queue's wait, which can never be satisfied
+ // there, is abandoned, so staging issues its own round.
+ w := newJobWorld(t)
+ row := w.admit()
+ w.watch = row.ID
+ w.crashWhen = func(p string, _ *console.SelfServeMove) bool { return p == "advance:reserving->queued" }
+ w.consoleUntil(10, "the crash before the queue's advance", func() bool { return w.crashWhen == nil })
+ queue := w.crashed.Request.Nonce
+ w.edit(func(rec *provision.Record) error {
+ return provision.OfflineQueuedToStaging(row.ID, stamp(rec), provision.Receipt{})(rec)
+ })
+ w.restart()
+ w.consoleUntil(3, "staging", func() bool { return w.row(row.ID).Phase == "staging" })
+ if r := w.row(row.ID); r.Request.Nonce == queue && !r.RequestAbandoned {
+ t.Fatalf("the queue's wait was kept: %s", describe(r))
+ }
+ w.phase(row.ID, "prechecks", 40)
+ })
+ t.Run("a target of a lower rank is refused", func(t *testing.T) {
+ // A row ahead of every phase its record admits (barrier on an
+ // offline-queued record) is never moved back to the set's first.
+ w := newJobWorld(t)
+ row := w.admit()
+ w.down()
+ w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" })
+ w.setPhase(row.ID, "barrier", "")
+ for i := 0; i < 3; i++ {
+ w.tick()
+ }
+ if r := w.row(row.ID); r.Phase != "barrier" {
+ t.Fatalf("the row went back: %s", describe(r))
+ }
+ })
t.Run("the record is gone", func(t *testing.T) {
w := newJobWorld(t)
row := w.admit()