amadan
e2e: the self-serve job's witnesses from requested to ready (Review Focus 1: TestEveryBoundaryRecovers)
The console's journal and pass (console.TestHooks) beside the bridge's two real boxes, over a store that can lose a write's reply or commit it after the client gave up, with console restarts on the same SQLite file: every boundary before the barrier under every crash point, the reconciliation table, the requests' identities, the admission's fence, the replaced participant at every gate, the facts, the book, ready's maintenance and the cut handoff, the deadline races and the lossless pre-offline abort.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
8 files changed,
+2564
−0
internal/e2e/selfserve/job_boundary_test.go+256 −0internal/e2e/selfserve/job_phases_test.go+501 −0internal/e2e/selfserve/job_races_test.go+318 −0internal/e2e/selfserve/job_reconcile_test.go+217 −0internal/e2e/selfserve/job_requests_test.go+325 −0internal/e2e/selfserve/job_resume_test.go+360 −0internal/e2e/selfserve/job_test.go+76 −0internal/e2e/selfserve/job_world_test.go+511 −0
diff --git a/internal/e2e/selfserve/job_boundary_test.go b/internal/e2e/selfserve/job_boundary_test.go| new file mode 100644 |
| index 000000000..348624efd |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_boundary_test.go |
| @@ -0,0 +1,256 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| +) |
| + |
| +// jobPhaseOrder is the order the job's phases move in; reverting and done |
| +// may follow any of them, and nothing moves backwards. |
| +var jobPhaseOrder = map[string]int{"requested": 0, "reserving": 1, "queued": 2, "reports": 3, "staging": 4, "prechecks": 5, |
| + "ready": 6, "restage": 7, "barrier": 8, "reverting": 100} |
| + |
| +// jobBoundary is one record write the job makes before the cut's barrier: |
| +// the request's kind and the phase that writes it, how the witness brings |
| +// the row there, and when the move is past it. |
| +type jobBoundary struct { |
| + name, kind, phase string |
| + setup func(w *jobWorld, id string) |
| + past func(w *jobWorld, id string) bool |
| +} |
| + |
| +func phaseAtLeast(phase string) func(w *jobWorld, id string) bool { |
| + return func(w *jobWorld, id string) bool { |
| + r := w.row(id) |
| + return jobPhaseOrder[r.Phase] >= jobPhaseOrder[phase] && r.Outcome == "" |
| + } |
| +} |
| + |
| +func finishedRow(w *jobWorld, id string) bool { |
| + r := w.row(id) |
| + return r.Outcome != "" && !r.ReleasePending && r.EffectsDone |
| +} |
| + |
| +var jobBoundaries = []jobBoundary{ |
| + {name: "queue", kind: "queue", phase: "reserving", past: phaseAtLeast("reports")}, |
| + {name: "staging", kind: "stage", phase: "reports", past: phaseAtLeast("prechecks")}, |
| + {name: "restage", kind: "stage", phase: "staging", past: phaseAtLeast("prechecks")}, |
| + {name: "precheck", kind: "precheck", phase: "prechecks", past: phaseAtLeast("ready")}, |
| + {name: "estimate", kind: "estimate", phase: "prechecks", past: phaseAtLeast("ready")}, |
| + {name: "maintain", kind: "maintain", phase: "ready", |
| + setup: func(w *jobWorld, id string) { w.phase(id, "ready", 60); w.advance(11 * time.Minute) }, |
| + past: func(w *jobWorld, id string) bool { |
| + r := w.row(id) |
| + return r.Phase == "ready" && r.Request.Kind == "maintain" && r.Request.Consumed && r.Request.Confirmed |
| + }}, |
| + {name: "abort-queued", kind: "abort", phase: "reverting", |
| + setup: func(w *jobWorld, id string) { w.phase(id, "reports", 60); w.cancel(id) }, past: finishedRow}, |
| + {name: "abort", kind: "abort", phase: "reverting", |
| + setup: func(w *jobWorld, id string) { w.phase(id, "prechecks", 60); w.cancel(id) }, past: finishedRow}, |
| + {name: "finish", kind: "finish", phase: "reverting", |
| + setup: func(w *jobWorld, id string) { w.phase(id, "prechecks", 60); w.cancel(id) }, past: finishedRow}, |
| + {name: "cut-restage", kind: "stage", phase: "restage", |
| + setup: func(w *jobWorld, id string) { |
| + w.phase(id, "ready", 60) |
| + now := time.Now().Add(w.skew) |
| + if ok, err := w.cst.MarkSelfServeCut(id, now, now.Add(5*time.Minute)); err != nil || !ok { |
| + w.t.Fatalf("cut = %v %v", ok, err) |
| + } |
| + }, past: phaseAtLeast("barrier")}, |
| +} |
| + |
| +// jobFault is how the boundary's write goes wrong. |
| +type jobFault struct { |
| + name string |
| + point string // the crash point's step, "" for none |
| + put putFault |
| +} |
| + |
| +var jobFaults = []jobFault{ |
| + {name: "crash before the request", point: "issue"}, |
| + {name: "crash with the request persisted", point: "write"}, |
| + {name: "crash after the write", point: "written"}, |
| + {name: "crash before the confirmation", point: "confirm"}, |
| + {name: "crash before the bump", point: "bump"}, |
| + {name: "crash before notified_seq", point: "notified"}, |
| + {name: "lost reply", put: putLost}, |
| + {name: "commit after the context was cancelled", point: "written", put: putHeld}, |
| +} |
| + |
| +// TestEveryBoundaryRecovers is Review Focus 1 for the job's writes before |
| +// the barrier (queue, staging, restage, precheck, estimate, the maintenance |
| +// round, the queued abort, the abort, its finish, the cut's restage): at |
| +// every boundary a crash before the request, with it persisted, after the |
| +// write, before the confirmation, before the bump and before notified_seq, |
| +// a lost reply, and a store that commits the write after the client's |
| +// context was cancelled; each crash recovered by a console that reopens the |
| +// same SQLite file, with and without a second crash during the recovery. |
| +// The row reconciles forward and never regresses, the request in flight |
| +// keeps its persisted identity (no second request of its kind is issued |
| +// while it is unconfirmed), the late commit is confirmed without a second |
| +// write, and the serial is bumped for the landed write. |
| +func TestEveryBoundaryRecovers(t *testing.T) { |
| + for _, b := range jobBoundaries { |
| + for _, f := range jobFaults { |
| + for _, second := range []bool{false, true} { |
| + if second && f.point == "" { |
| + continue |
| + } |
| + name := b.name + "/" + f.name |
| + if second { |
| + name += "/second crash" |
| + } |
| + t.Run(name, func(t *testing.T) { everyBoundary(t, b, f, second) }) |
| + } |
| + } |
| + } |
| + t.Run("ordinary progression bumps once per landed write", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + bumps, confirms := 0, 0 |
| + for p, n := range w.hits { |
| + switch { |
| + case strings.HasSuffix(p, ":bump"): |
| + bumps += n |
| + case strings.HasSuffix(p, ":confirm"): |
| + confirms += n |
| + } |
| + } |
| + if bumps == 0 || bumps != confirms || w.fs.bumpCount() != bumps { |
| + t.Fatalf("%d confirmations, %d job bumps, %d serial writes; want one bump per landed write", confirms, bumps, w.fs.bumpCount()) |
| + } |
| + }) |
| +} |
| + |
| +func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + if b.setup != nil { |
| + b.setup(w, row.ID) |
| + } |
| + inFlight := func(r *console.SelfServeMove) bool { |
| + return r.Phase == b.phase && r.Request.Kind == b.kind && !r.RequestIssued.IsZero() && !r.Request.Consumed && !r.Request.Confirmed |
| + } |
| + at := func(r *console.SelfServeMove) bool { |
| + if r.Phase != b.phase { |
| + return false |
| + } |
| + if b.kind == "estimate" && r.PhaseStep != "estimate" { |
| + return false |
| + } |
| + if b.kind == "finish" && r.PhaseStep != "finish" { |
| + return false |
| + } |
| + return true |
| + } |
| + var nonce string |
| + if f.put != putRun { |
| + fired := false |
| + w.fs.setPut(func(key string, body []byte) putFault { |
| + r := w.row(row.ID) |
| + if fired || !inFlight(r) { |
| + return putRun |
| + } |
| + fired, nonce = true, r.Request.Nonce |
| + return f.put |
| + }) |
| + } |
| + if f.point != "" { |
| + w.crashWhen = func(point string, r *console.SelfServeMove) bool { |
| + return point == b.kind+":"+f.point && at(r) && (f.put != putHeld || nonce != "") |
| + } |
| + } |
| + bumpsBefore := w.fs.bumpCount() |
| + w.until(80, "at the boundary's fault", func() bool { return (f.point != "" && w.crashWhen == nil) || (f.point == "" && nonce != "") }) |
| + if f.point == "" { |
| + w.fs.setPut(nil) |
| + } |
| + if c := w.crashed; c != nil && inFlight(c) { |
| + nonce = c.Request.Nonce |
| + } |
| + if f.point != "" { |
| + w.fs.setPut(nil) |
| + if f.put == putHeld { |
| + for _, err := range w.fs.release(w.ctx) { |
| + if err != nil { |
| + t.Fatalf("the late commit: %v", err) |
| + } |
| + } |
| + } |
| + w.restart() |
| + if second { |
| + w.crashWhen = func(point string, r *console.SelfServeMove) bool { |
| + return strings.HasPrefix(point, b.kind+":") && at(r) |
| + } |
| + } |
| + } |
| + snapsFrom := len(w.snaps) - 1 |
| + if f.put == putHeld && !second { |
| + // The late commit is the write: the recovery confirms it from the |
| + // record and never writes it again. |
| + writes, puts := w.hits[b.kind+":write"], w.fs.recordPuts() |
| + w.until(20, "the late commit confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Nonce != nonce || r.Request.Confirmed |
| + }) |
| + if r := w.row(row.ID); r.Request.Nonce == nonce && (w.hits[b.kind+":write"] != writes || w.fs.recordPuts() != puts) { |
| + t.Fatalf("the recovery wrote again (%d writes, %d puts; before %d, %d): %s", w.hits[b.kind+":write"], w.fs.recordPuts(), writes, puts, describe(r)) |
| + } |
| + } |
| + w.until(80, b.name+" past", func() bool { |
| + if w.crashed != nil && second && w.crashWhen == nil && f.point != "" { |
| + // The second crash: restart again, once. |
| + w.crashed, second = nil, false |
| + w.restart() |
| + } |
| + return b.past(w, row.ID) |
| + }) |
| + |
| + // Forward only. |
| + last := -1 |
| + for _, s := range w.snaps { |
| + o := jobPhaseOrder[s.Phase] |
| + if s.Outcome == "" && o < last { |
| + t.Fatalf("the row went back to %s after %d (%s)", s.Phase, last, describe(&s)) |
| + } |
| + if o > last { |
| + last = o |
| + } |
| + } |
| + if nonce != "" { |
| + // The request in flight keeps its identity until it is confirmed: |
| + // no other request of its kind and phase is issued meanwhile, and |
| + // it is the one confirmed. |
| + confirmed := false |
| + for _, s := range w.snaps[snapsFrom:] { |
| + if s.Request.Kind != b.kind || s.Request.Phase != b.phase || confirmed { |
| + continue |
| + } |
| + if s.Request.Nonce != nonce { |
| + t.Fatalf("request %s was replaced by %s before it was confirmed (%s)", nonce, s.Request.Nonce, describe(&s)) |
| + } |
| + if s.Request.Confirmed { |
| + confirmed = true |
| + } |
| + } |
| + final := w.row(row.ID) |
| + if !confirmed && !(final.Request.Nonce == nonce && final.Request.Confirmed) && !strings.HasPrefix(b.name, "abort") && b.name != "finish" { |
| + // A clearing request's confirmation and its consumption by the |
| + // finaliser can share one tick; every other must be seen. |
| + t.Fatalf("request %s was never confirmed (final %s)", nonce, describe(final)) |
| + } |
| + if final.NotifiedSeq == 0 { |
| + t.Fatalf("notified_seq is 0 after the landed write") |
| + } |
| + } |
| + if w.fs.bumpCount() <= bumpsBefore { |
| + t.Fatalf("no serial bump after the write landed") |
| + } |
| +} |
diff --git a/internal/e2e/selfserve/job_phases_test.go b/internal/e2e/selfserve/job_phases_test.go| new file mode 100644 |
| index 000000000..93bcae47b |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_phases_test.go |
| @@ -0,0 +1,501 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/status" |
| +) |
| + |
| +// toReports brings a new row to reports with the boxes still, then lets |
| +// them answer the queued move once and stops them again, so a witness can |
| +// change an answer before the console decides. |
| +func (w *jobWorld) toReports() *console.SelfServeMove { |
| + w.t.Helper() |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + w.up() |
| + w.boxes() |
| + w.down() |
| + return row |
| +} |
| + |
| +// doctor rewrites one of the move's box reports, as a box answering |
| +// otherwise would have. |
| +func (w *jobWorld) doctor(id, role, box string, edit func(r *moves.OfflineReport)) { |
| + w.t.Helper() |
| + r, ok := w.report(id, role, box) |
| + if !ok { |
| + w.t.Fatalf("no %s report from %s", role, box) |
| + } |
| + edit(&r) |
| + w.putReport(r) |
| +} |
| + |
| +// TestReportsDecideFromFacts: the queued move's reports decide from the |
| +// facts the boxes report, each refusal its own sentence, before anything |
| +// is copied (the record never reaches staging): a lane that fails, a stray |
| +// entry, files the archive refuses, a data path the target holds (through |
| +// the real target box), a fact the box left out, free disk the box could |
| +// not read, and an object already in the target bucket. |
| +func TestReportsDecideFromFacts(t *testing.T) { |
| + for _, c := range []struct { |
| + name, reason string |
| + edit func(w *jobWorld, id string) |
| + }{ |
| + {"a failed lane", "no staging lane is configured from ie to uk", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleLane, host.E2ESourceBox, func(r *moves.OfflineReport) { r.Lane.OK, r.Lane.Error = false, "denied" }) |
| + }}, |
| + {"a stray entry", "shop.example.com has a stray entry beside its files directory: /data/shop.example.com-extra", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleSource, host.E2ESourceBox, func(r *moves.OfflineReport) { |
| + r.Precheck.Strays = moves.StraysFact{Known: true, Entries: []moves.StrayEntry{{Path: "/data/shop.example.com-extra", Kind: moves.StrayDir, Entries: 2, Known: true}}} |
| + }) |
| + }}, |
| + {"an empty stray directory passes", "", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleSource, host.E2ESourceBox, func(r *moves.OfflineReport) { |
| + r.Precheck.Strays = moves.StraysFact{Known: true, Entries: []moves.StrayEntry{{Path: "/data/shop.example.com-empty", Kind: moves.StrayDir, Known: true}}} |
| + }) |
| + }}, |
| + {"files the archive refuses", "shop.example.com's files cannot be archived: too many entries", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleSource, host.E2ESourceBox, func(r *moves.OfflineReport) { |
| + r.Precheck.ArchiveEligible = moves.Fact{Known: true, Detail: "too many entries"} |
| + }) |
| + }}, |
| + {"a fact the box left out", "could not determine the route on uk-box", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleTarget, host.E2ETargetBox, func(r *moves.OfflineReport) { r.Ready.RouteAbsent = moves.Fact{} }) |
| + }}, |
| + {"free disk unread", "the target box has too little free disk for shop.example.com", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleTarget, host.E2ETargetBox, func(r *moves.OfflineReport) { r.Ready.DiskFree = moves.Fact{Detail: "unread"} }) |
| + }}, |
| + {"objects in the target bucket", "the target bucket already holds objects for shop.example.com", func(w *jobWorld, id string) { |
| + w.doctor(id, moves.OfflineRoleTarget, host.E2ETargetBox, func(r *moves.OfflineReport) { |
| + r.Ready.MovesPrefix.Objects = map[string]moves.PrefixObject{"db.ltx": {ETag: "x", Size: 1}} |
| + }) |
| + }}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.toReports() |
| + c.edit(w, row.ID) |
| + if c.reason == "" { |
| + w.consoleUntil(5, "staging", func() bool { return w.row(row.ID).Phase == "staging" }) |
| + return |
| + } |
| + w.up() |
| + r := w.finished(row.ID, 20) |
| + if r.Outcome != "aborted" || r.OutcomeReason != c.reason { |
| + t.Fatalf("finished %s; want %q", describe(r), c.reason) |
| + } |
| + if w.hits["stage:write"] != 0 { |
| + t.Fatal("the move was staged") |
| + } |
| + }) |
| + } |
| + t.Run("a data path the real target holds", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + tgt := w.e.Box(host.E2ETargetBox) |
| + env := filepath.Join(tgt.DataDir(), host.E2EHost+".env") |
| + if err := os.WriteFile(env, []byte("A=1\n"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + row := w.admit() |
| + r := w.finished(row.ID, 30) |
| + if want := "the target box already holds data for shop.example.com: " + env; r.OutcomeReason != want { |
| + t.Fatalf("finished %s; want %q", describe(r), want) |
| + } |
| + }) |
| +} |
| + |
| +// TestPrechecksCompareTheBook: the prechecks require the stager's book of |
| +// the last round to be exactly what the target's staging prefix holds, and |
| +// the target's instances prefix to be empty. |
| +func TestPrechecksCompareTheBook(t *testing.T) { |
| + prechecking := func(w *jobWorld) *console.SelfServeMove { |
| + row := w.admit() |
| + w.phase(row.ID, "prechecks", 60) |
| + w.down() |
| + w.consoleUntil(5, "the precheck confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "precheck" && r.Request.Confirmed |
| + }) |
| + w.up() |
| + w.boxes() |
| + w.down() |
| + return row |
| + } |
| + t.Run("a key the book does not have", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := prechecking(w) |
| + w.doctor(row.ID, moves.OfflineRoleStager, host.E2ESourceBox, func(r *moves.OfflineReport) { |
| + for k := range r.Staged { |
| + delete(r.Staged, k) |
| + break |
| + } |
| + }) |
| + w.up() |
| + r := w.finished(row.ID, 30) |
| + if r.OutcomeReason != "the staging prefix for shop.example.com does not match what this move staged" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| + t.Run("objects under the instances prefix", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := prechecking(w) |
| + w.doctor(row.ID, moves.OfflineRoleTarget, host.E2ETargetBox, func(r *moves.OfflineReport) { |
| + r.Ready.InstancesPrefix.Objects = map[string]moves.PrefixObject{"db.ltx": {ETag: "x", Size: 1}} |
| + }) |
| + w.up() |
| + r := w.finished(row.ID, 30) |
| + if r.OutcomeReason != "the target bucket already holds objects for shop.example.com" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| + t.Run("the book matches", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := prechecking(w) |
| + w.up() |
| + w.phase(row.ID, "ready", 20) |
| + }) |
| +} |
| + |
| +// TestReadyRestagesAndExpires: ready is idle between maintenance rounds |
| +// (no step deadline); a round is due 10 minutes after the last ended, with |
| +// its own allowance (max(5 min, 2 × the copy)) from its issue, and is |
| +// consumed back to idle; the row never enters restage or barrier without |
| +// the cut, and the window is never moved. Past the 24-hour window the move |
| +// aborts "the move was not confirmed in time". |
| +func TestReadyRestagesAndExpires(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + ready := w.phase(row.ID, "ready", 60) |
| + window := ready.WindowDeadline |
| + if !ready.StepDeadline.IsZero() || ready.NextMaintenanceAt.IsZero() { |
| + t.Fatalf("ready is not idle: step %v next %v", ready.StepDeadline, ready.NextMaintenanceAt) |
| + } |
| + for round := 1; round <= 2; round++ { |
| + w.advance(9 * time.Minute) |
| + w.boxes() |
| + w.tick() |
| + if r := w.row(row.ID); r.Request.Kind == "maintain" && !r.Request.Consumed && r.Request.Nonce != "" && round == 2 && r.RequestIssued.After(ready.NextMaintenanceAt) { |
| + t.Fatalf("a round was issued before it was due: %s", describe(r)) |
| + } |
| + w.advance(2 * time.Minute) |
| + w.tick() |
| + r := w.row(row.ID) |
| + if r.Request.Kind != "maintain" || r.Request.Consumed { |
| + t.Fatalf("round %d not issued: %s", round, describe(r)) |
| + } |
| + if got := r.Request.Deadline.Sub(r.RequestIssued); got != 5*time.Minute || !r.StepDeadline.Equal(r.Request.Deadline) { |
| + t.Fatalf("round %d's allowance %v (step %v, request %v)", round, got, r.StepDeadline, r.Request.Deadline) |
| + } |
| + w.until(10, "the round consumed", func() bool { return w.row(row.ID).Request.Consumed }) |
| + r = w.row(row.ID) |
| + if r.Phase != "ready" || !r.WindowDeadline.Equal(window) || !r.StepDeadline.IsZero() || !r.NextMaintenanceAt.After(ready.NextMaintenanceAt) { |
| + t.Fatalf("after round %d: %s window %v (was %v) step %v next %v", round, describe(r), r.WindowDeadline, window, r.StepDeadline, r.NextMaintenanceAt) |
| + } |
| + ready = r |
| + } |
| + for _, s := range w.snaps { |
| + if s.Phase == "restage" || s.Phase == "barrier" { |
| + t.Fatalf("the row entered %s without a cut", s.Phase) |
| + } |
| + } |
| + w.advance(window.Sub(time.Now().Add(w.skew)) + time.Minute) |
| + r := w.finished(row.ID, 30) |
| + if r.OutcomeReason != "the move was not confirmed in time" || w.record().Move != nil { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| +} |
| + |
| +// TestDeadlinesEvaluateBeforeEvidence: a wait's deadline is decided from |
| +// the row before any report is read, so evidence arriving after it is |
| +// never what moves the row: the reports' and a staging round's. |
| +func TestDeadlinesEvaluateBeforeEvidence(t *testing.T) { |
| + t.Run("reports", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + w.advance(6 * time.Minute) |
| + w.up() |
| + w.boxes() // the reports arrive, passing, after the deadline |
| + r := w.finished(row.ID, 20) |
| + if r.Outcome != "aborted" || w.hits["stage:write"] != 0 { |
| + t.Fatalf("finished %s, stage writes %d", describe(r), w.hits["stage:write"]) |
| + } |
| + }) |
| + t.Run("a staging round", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "staging", 60) |
| + w.down() |
| + w.consoleUntil(5, "a round confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "stage" && r.Request.Confirmed && !r.Request.Consumed |
| + }) |
| + w.advance(11 * time.Minute) |
| + w.up() |
| + w.boxes() |
| + r := w.finished(row.ID, 20) |
| + if r.Outcome != "aborted" || r.OutcomeReason != "could not determine the staged copy on ie-box" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + for _, s := range w.snaps { |
| + if s.Phase == "prechecks" { |
| + t.Fatal("the late round moved the row to the prechecks") |
| + } |
| + } |
| + }) |
| +} |
| + |
| +// TestQueueOrderAndDisjointProgress: three rows. Two share the boxes and go |
| +// in the order they were requested: the second waits "waiting for an |
| +// earlier move on this box" while the first has not reserved, then |
| +// "waiting for another move on this box" while it holds them, and |
| +// proceeds once the first is finished. The third, on disjoint boxes, |
| +// reserves while the second waits. |
| +func TestQueueOrderAndDisjointProgress(t *testing.T) { |
| + w := newJobWorld(t) |
| + for _, h := range []string{"b.shop.example.com", "c.shop.example.com"} { |
| + if err := provision.CreateRecord(w.ctx, w.raw, provision.Record{Schema: 1, Host: h, Slug: strings.Split(h, ".")[0], App: host.E2EApp, |
| + Account: w.acct, Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // The first row's queue cannot be written yet: it stays reserving. |
| + w.fs.setPut(func(key string, body []byte) putFault { |
| + if strings.Contains(key, "/"+host.E2EHost) && strings.Contains(string(body), provision.OfflineQueued) { |
| + return putFail |
| + } |
| + return putRun |
| + }) |
| + first := w.admit() |
| + second := w.admitFor("b.shop.example.com", w.participants()) |
| + disjoint := provision.Participants{ |
| + Source: []provision.Participant{{BoxID: "x-box-id", Label: "x-box", InstanceID: "i-x"}}, |
| + Target: provision.Participant{BoxID: "y-box-id", Label: "y-box", InstanceID: "i-y"}, |
| + } |
| + for _, p := range []provision.Participant{disjoint.Source[0], disjoint.Target} { |
| + // Reports too old to judge identity by (and, publishing no edges |
| + // file, retired boxes to the fleet's evidence): the job reserves |
| + // on the pinned ids and never calls them replaced. |
| + if err := status.WriteBox(w.ctx, w.raw, status.Box{Label: p.Label, GeneratedAt: time.Now().Add(-time.Hour).UTC(), |
| + Metrics: &status.BoxMetrics{BoxID: p.BoxID, InstanceID: p.InstanceID}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + third := w.admitFor("c.shop.example.com", disjoint) |
| + thirdHeld := false |
| + for i := 0; i < 3; i++ { |
| + w.tick() |
| + for _, e := range w.reservations() { |
| + if e.ID == third.ID && w.row(second.ID).Phase == "reserving" { |
| + thirdHeld = true |
| + } |
| + } |
| + } |
| + if r := w.row(second.ID); r.Phase != "reserving" || r.Reason != "waiting for an earlier move on this box" { |
| + t.Fatalf("the second row: %s reason %q", describe(r), r.Reason) |
| + } |
| + if r := w.row(first.ID); r.Phase != "reserving" { |
| + t.Fatalf("the first row: %s", describe(r)) |
| + } |
| + if !thirdHeld { |
| + t.Fatalf("the disjoint row did not reserve while the second waited: %s, reservations %v", describe(w.row(third.ID)), w.reservations()) |
| + } |
| + // The first queues and holds the boxes: the second waits on it. |
| + w.fs.setPut(nil) |
| + w.consoleUntil(10, "the first queued", func() bool { return w.row(first.ID).Phase == "reports" }) |
| + w.consoleUntil(5, "the second busy", func() bool { return w.row(second.ID).Reason == "waiting for another move on this box" }) |
| + if r := w.row(second.ID); r.Phase != "reserving" { |
| + t.Fatalf("the second row moved past the first: %s", describe(r)) |
| + } |
| + // The first is cancelled and finished: the second takes the boxes. |
| + w.cancel(first.ID) |
| + w.finished(first.ID, 20) |
| + w.consoleUntil(10, "the second queued", func() bool { |
| + r := w.row(second.ID) |
| + return r.Phase == "queued" || r.Phase == "reports" || r.Outcome != "" |
| + }) |
| + if r := w.row(second.ID); r.Outcome != "" { |
| + t.Fatalf("the second row: %s", describe(r)) |
| + } |
| +} |
| + |
| +// TestCutHandoffWithAnOutstandingMaintenanceRound: a cut requested while a |
| +// maintenance round is outstanding adopts it as the cut's first round |
| +// (restage, adopted: nothing new is written); a cut with none outstanding |
| +// enters restage at issue, which writes its own round; either way the row |
| +// reaches the barrier. A cancel after the cut is refused. |
| +func TestCutHandoffWithAnOutstandingMaintenanceRound(t *testing.T) { |
| + cut := func(w *jobWorld, id string) { |
| + now := time.Now().Add(w.skew) |
| + if ok, err := w.cst.MarkSelfServeCut(id, now, now.Add(5*time.Minute)); err != nil || !ok { |
| + w.t.Fatalf("cut = %v %v", ok, err) |
| + } |
| + } |
| + t.Run("an outstanding round", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + w.advance(11 * time.Minute) |
| + w.consoleUntil(5, "a round confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "maintain" && r.Request.Confirmed |
| + }) |
| + round := w.row(row.ID).Request.Nonce |
| + cut(w, row.ID) |
| + w.tick() |
| + if r := w.row(row.ID); r.Phase != "ready" || r.Request.Nonce != round { |
| + t.Fatalf("before the round's report: %s", describe(r)) |
| + } |
| + w.up() |
| + r := w.phase(row.ID, "barrier", 20) |
| + adopted := false |
| + for _, s := range w.snaps { |
| + if s.Phase == "restage" && s.PhaseStep == "adopted" { |
| + adopted = true |
| + } |
| + } |
| + if !adopted || len(w.nonces("stage", "restage")) != 0 || r.Request.Nonce != round { |
| + t.Fatalf("adopted %v, restage requests %v, %s", adopted, w.nonces("stage", "restage"), describe(r)) |
| + } |
| + if ok, _ := w.cst.RequestSelfServeAbort(row.ID, []string{"requested", "reserving", "queued", "reports", "staging", "prechecks", "ready"}, time.Now()); ok { |
| + t.Fatal("a cancel after the cut was accepted") |
| + } |
| + }) |
| + t.Run("none outstanding", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + cut(w, row.ID) |
| + w.phase(row.ID, "barrier", 20) |
| + issued := false |
| + for _, s := range w.snaps { |
| + if s.Phase == "restage" && s.PhaseStep == "issue" { |
| + issued = true |
| + } |
| + } |
| + if !issued || len(w.nonces("stage", "restage")) != 1 { |
| + t.Fatalf("restage issue %v, its requests %v", issued, w.nonces("stage", "restage")) |
| + } |
| + if ok, _ := w.cst.RequestSelfServeAbort(row.ID, []string{"ready", "restage", "barrier"}, time.Now()); ok { |
| + t.Fatal("a cancel after the cut was accepted") |
| + } |
| + }) |
| +} |
| + |
| +// TestCutHandoffSurvivesACrash: the substep the cut's advance chose is in |
| +// the advancing statement, so a crash after either branch (and another |
| +// during its recovery) resumes that continuation; a completed maintenance |
| +// round and the cut in both orders; and an uncommitted round at the cut |
| +// gets the cut's allowance, so recovery at its old deadline does not |
| +// abort. |
| +func TestCutHandoffSurvivesACrash(t *testing.T) { |
| + cut := func(w *jobWorld, id string) { |
| + now := time.Now().Add(w.skew) |
| + if ok, err := w.cst.MarkSelfServeCut(id, now, now.Add(5*time.Minute)); err != nil || !ok { |
| + w.t.Fatalf("cut = %v %v", ok, err) |
| + } |
| + } |
| + t.Run("issue, crashed twice", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + cut(w, row.ID) |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "stage:issue" && r.Phase == "restage" } |
| + w.until(5, "the crash", func() bool { return w.crashWhen == nil }) |
| + w.restart() |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "stage:written" && r.Phase == "restage" } |
| + w.until(5, "the second crash", func() bool { return w.crashWhen == nil }) |
| + nonce := w.crashed.Request.Nonce |
| + w.restart() |
| + r := w.phase(row.ID, "barrier", 20) |
| + if n := w.nonces("stage", "restage"); len(n) != 1 || n[0] != nonce || r.Request.Nonce != nonce { |
| + t.Fatalf("restage's requests %v, want only %s", n, nonce) |
| + } |
| + }) |
| + t.Run("adopted, crashed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + w.advance(11 * time.Minute) |
| + w.consoleUntil(5, "a round confirmed", func() bool { return w.row(row.ID).Request.Kind == "maintain" && w.row(row.ID).Request.Confirmed }) |
| + cut(w, row.ID) |
| + w.up() |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "advance:restage->barrier" } |
| + w.until(10, "the crash", func() bool { return w.crashWhen == nil }) |
| + w.restart() |
| + w.phase(row.ID, "barrier", 10) |
| + if n := w.nonces("stage", "restage"); len(n) != 0 { |
| + t.Fatalf("an adopted round was written again: %v", n) |
| + } |
| + }) |
| + t.Run("the round consumed, then the cut", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.advance(11 * time.Minute) |
| + w.until(10, "the round consumed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "maintain" && r.Request.Consumed |
| + }) |
| + cut(w, row.ID) |
| + w.phase(row.ID, "barrier", 20) |
| + if len(w.nonces("stage", "restage")) != 1 { |
| + t.Fatalf("restage's requests %v; want its own round", w.nonces("stage", "restage")) |
| + } |
| + }) |
| + t.Run("the cut, then the stale consume", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.advance(11 * time.Minute) |
| + w.at("maintain:consume", func() { cut(w, row.ID) }) |
| + w.until(10, "the round adopted", func() bool { return w.row(row.ID).Phase == "restage" || w.row(row.ID).Phase == "barrier" }) |
| + w.phase(row.ID, "barrier", 10) |
| + if len(w.nonces("stage", "restage")) != 0 { |
| + t.Fatalf("the round was not adopted: restage wrote %v", w.nonces("stage", "restage")) |
| + } |
| + }) |
| + t.Run("an uncommitted round at the cut", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.fs.setPut(func(string, []byte) putFault { return putFail }) |
| + w.advance(11 * time.Minute) |
| + w.consoleUntil(3, "the round issued", func() bool { return w.row(row.ID).Request.Kind == "maintain" }) |
| + old := w.row(row.ID).Request.Deadline |
| + w.advance(time.Minute) |
| + cut(w, row.ID) |
| + if r := w.row(row.ID); !r.Request.Deadline.After(old) { |
| + t.Fatalf("the cut left the round's deadline at %v (was %v)", r.Request.Deadline, old) |
| + } |
| + w.advance(old.Sub(time.Now().Add(w.skew)) + 30*time.Second) // past the old deadline, inside the cut's |
| + w.fs.setPut(nil) |
| + r := w.phase(row.ID, "barrier", 20) |
| + if r.Outcome != "" || r.Request.Expired { |
| + t.Fatalf("the round expired at its old deadline: %s", describe(r)) |
| + } |
| + }) |
| +} |
diff --git a/internal/e2e/selfserve/job_races_test.go b/internal/e2e/selfserve/job_races_test.go| new file mode 100644 |
| index 000000000..b49e3d360 |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_races_test.go |
| @@ -0,0 +1,318 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// staging writes are the record writes that carry a staged_at change. |
| +func stagingWrite(body []byte) bool { return strings.Contains(string(body), `"staged_at"`) } |
| + |
| +// toMaintaining brings a row to ready and advances the console to its |
| +// first maintenance round, with the round's writes failing: the round is |
| +// issued and outstanding, unconfirmed. It answers the row then. |
| +func (w *jobWorld) toMaintaining() *console.SelfServeMove { |
| + w.t.Helper() |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.failKinds["maintain"] = true |
| + w.advance(11 * time.Minute) |
| + w.consoleUntil(3, "the round issued", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "maintain" && !r.Request.Consumed && !r.Request.Confirmed |
| + }) |
| + return w.row(row.ID) |
| +} |
| + |
| +// tickAtDeadline ticks the console one second before the outstanding |
| +// request's deadline, so its write is made in time, and lets the clock |
| +// pass the deadline after that write and before its confirmation read. |
| +func (w *jobWorld) tickAtDeadline(r *console.SelfServeMove, during func()) { |
| + w.t.Helper() |
| + w.skew += r.Request.Deadline.Sub(time.Now().Add(w.skew)) - time.Second |
| + w.at(r.Request.Kind+":written", func() { |
| + console.TestHooks.Advance(2 * time.Second) |
| + if during != nil { |
| + during() |
| + } |
| + }) |
| + w.tick() |
| +} |
| + |
| +// voided reports whether the snapshots saw the request with nonce voided. |
| +func (w *jobWorld) voided(nonce string) bool { |
| + for _, s := range w.snaps { |
| + if s.Request.Nonce == nonce && s.Request.Voided { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func (w *jobWorld) markCut(id string) bool { |
| + now := time.Now().Add(w.skew) |
| + ok, err := w.cst.MarkSelfServeCut(id, now, now.Add(5*time.Minute)) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + return ok |
| +} |
| + |
| +// TestStaleRequestDeadlineLosesToTheCut: the expiry decision is one UPDATE |
| +// against the deadline and the cut state the worker read with the request. |
| +func TestStaleRequestDeadlineLosesToTheCut(t *testing.T) { |
| + t.Run("a cut during the confirmation read extends the deadline", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.tickAtDeadline(r, func() { |
| + if !w.markCut(r.ID) { |
| + t.Error("the cut was refused") |
| + } |
| + }) |
| + got := w.row(r.ID) |
| + if got.Request.Expired || got.Outcome != "" || got.Request.Nonce != r.Request.Nonce || w.hits["maintain:expire"] == 0 { |
| + t.Fatalf("after the stale expiry: %s (expiry attempts %d)", describe(got), w.hits["maintain:expire"]) |
| + } |
| + w.failKinds["maintain"] = false |
| + w.phase(r.ID, "barrier", 20) |
| + }) |
| + t.Run("the expiry first refuses the cut", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.crashWhen = func(p string, _ *console.SelfServeMove) bool { return p == "maintain:fence" } |
| + w.tickAtDeadline(r, nil) |
| + if got := w.row(r.ID); !got.Request.Expired { |
| + t.Fatalf("no expiry: %s", describe(got)) |
| + } |
| + if w.markCut(r.ID) { |
| + t.Fatal("the cut was accepted after the expiry") |
| + } |
| + }) |
| + t.Run("after a cut the extended deadline expires", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + if !w.markCut(r.ID) { |
| + t.Fatal("the cut was refused") |
| + } |
| + r = w.row(r.ID) |
| + w.tickAtDeadline(r, nil) |
| + w.failKinds["maintain"] = false |
| + got := w.finished(r.ID, 30) |
| + if got.Outcome != "aborted" || !w.voided(r.Request.Nonce) || w.hits["maintain:expire"] < 1 { |
| + t.Fatalf("after the extended deadline: %s", describe(got)) |
| + } |
| + }) |
| + for _, point := range []string{"maintain:fence", "maintain:void-expired"} { |
| + t.Run("a won expiry, a crash at "+point, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.crashWhen = func(p string, _ *console.SelfServeMove) bool { return p == point } |
| + w.tickAtDeadline(r, nil) |
| + if w.crashWhen != nil { |
| + t.Fatal("no crash") |
| + } |
| + w.restart() |
| + w.failKinds["maintain"] = false |
| + decided := w.hits["maintain:expire"] |
| + got := w.finished(r.ID, 30) |
| + if got.Outcome != "aborted" || !w.voided(r.Request.Nonce) { |
| + t.Fatalf("after the restart: %s", describe(got)) |
| + } |
| + if w.hits["maintain:expire"] != decided { |
| + t.Fatalf("the expiry was decided again after the restart (%d, was %d)", w.hits["maintain:expire"], decided) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// TestReplayStopsAtTheDeadlineInsideOneInvocation: a request whose write |
| +// failed before its deadline and whose confirmation read returns after it |
| +// is not written again: the expiry is decided and the request fenced and |
| +// classified; the same with the write having committed is confirmed. (The |
| +// rule is selfServeApply's for every kind with a deadline; the |
| +// maintenance round witnesses it here, the open in Task 8.) |
| +func TestReplayStopsAtTheDeadlineInsideOneInvocation(t *testing.T) { |
| + t.Run("not committed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + writes := w.hits["maintain:write"] |
| + w.tickAtDeadline(r, nil) |
| + if w.hits["maintain:write"] != writes+1 { |
| + t.Fatalf("written %d times in the tick, want once (before the deadline)", w.hits["maintain:write"]-writes) |
| + } |
| + got := w.row(r.ID) |
| + if !got.Request.Voided || got.AbortRequestedAt.IsZero() { |
| + t.Fatalf("after the deadline: %s", describe(got)) |
| + } |
| + }) |
| + t.Run("committed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.failKinds["maintain"] = false |
| + lost := false |
| + w.fs.setPut(func(_ string, body []byte) putFault { |
| + if !lost && stagingWrite(body) { |
| + lost = true |
| + return putLost |
| + } |
| + return putRun |
| + }) |
| + w.tickAtDeadline(r, nil) |
| + got := w.row(r.ID) |
| + if !got.Request.Confirmed || got.Request.Voided || got.Request.Expired { |
| + t.Fatalf("the committed round: %s", describe(got)) |
| + } |
| + }) |
| +} |
| + |
| +// TestExpiredRequestIsNeverReplayed: a request whose deadline passed |
| +// unconfirmed is classified, never replayed: persisted but never written, |
| +// the record is fenced, the request voided and the move aborts; written |
| +// before the crash, its receipt confirms it; written between the job's |
| +// observation and its fence, the fenced observation carries the receipt |
| +// and confirms it. (The open's own case, past the no-return boundary, is |
| +// Task 8's TestExpiredOpenIsNeverReplayed.) |
| +func TestExpiredRequestIsNeverReplayed(t *testing.T) { |
| + t.Run("never written", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.restart() |
| + w.failKinds["maintain"] = false |
| + writes := w.hits["maintain:write"] |
| + w.skew += r.Request.Deadline.Sub(time.Now().Add(w.skew)) + time.Second |
| + got := w.finished(r.ID, 30) |
| + if got.Outcome != "aborted" || !w.voided(r.Request.Nonce) || w.hits["maintain:write"] != writes { |
| + t.Fatalf("finished %s, %d writes after the deadline", describe(got), w.hits["maintain:write"]-writes) |
| + } |
| + if rec := w.record(); rec.LastMove == nil || rec.LastMove.ID != r.ID { |
| + t.Fatalf("the record after the abort: %+v", rec.LastMove) |
| + } |
| + }) |
| + t.Run("written between the observation and the fence", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + w.failKinds["maintain"] = false |
| + held := false |
| + w.fs.setPut(func(_ string, body []byte) putFault { |
| + if !held && stagingWrite(body) { |
| + held = true |
| + return putHeld |
| + } |
| + return putRun |
| + }) |
| + w.at("maintain:fence", func() { |
| + for _, err := range w.fs.release(w.ctx) { |
| + if err != nil { |
| + t.Errorf("the late round: %v", err) |
| + } |
| + } |
| + }) |
| + w.tickAtDeadline(r, nil) |
| + got := w.row(r.ID) |
| + if !got.Request.Confirmed || got.Request.Voided || got.Request.Expired || got.Request.Nonce != r.Request.Nonce { |
| + t.Fatalf("the round that won the fence: %s", describe(got)) |
| + } |
| + }) |
| +} |
| + |
| +// TestCancelBetweenExpiryAndVoid: the expiry wins, a cancel then requests |
| +// the abort, the fence finds no receipt: the void keeps the cancel's time |
| +// and reason, and the abort runs once. |
| +func TestCancelBetweenExpiryAndVoid(t *testing.T) { |
| + w := newJobWorld(t) |
| + r := w.toMaintaining() |
| + var cancelled time.Time |
| + w.at("maintain:fence", func() { |
| + w.cancel(r.ID) |
| + cancelled = w.row(r.ID).AbortRequestedAt |
| + }) |
| + w.tickAtDeadline(r, nil) |
| + w.failKinds["maintain"] = false |
| + got := w.finished(r.ID, 30) |
| + if got.Outcome != "aborted" || got.OutcomeReason != "the move was cancelled" || !got.AbortRequestedAt.Equal(cancelled) || !got.Request.Voided && got.Request.Nonce == r.Request.Nonce { |
| + t.Fatalf("finished %s, abort at %v (cancelled %v)", describe(got), got.AbortRequestedAt, cancelled) |
| + } |
| + if n := w.hits["abort:write"]; n != 1 { |
| + t.Fatalf("the abort was written %d times", n) |
| + } |
| +} |
| + |
| +// TestOperatorRoundSupersedesTheJobsWait: an operator restage moves |
| +// staged_at past the job's confirmed round without a receipt; the job |
| +// abandons its wait and issues its own fresh round, never adopting the |
| +// operator's; in staging and in ready, and across a crash between the |
| +// abandon and the issue. |
| +func TestOperatorRoundSupersedesTheJobsWait(t *testing.T) { |
| + supersede := func(t *testing.T, w *jobWorld, id, kind string, crash bool) { |
| + w.down() |
| + w.consoleUntil(5, "a round confirmed", func() bool { |
| + r := w.row(id) |
| + return r.Request.Kind == kind && r.Request.Confirmed && !r.Request.Consumed |
| + }) |
| + s1 := w.row(id).Request |
| + w.edit(func(rec *provision.Record) error { |
| + return provision.OfflineRestage(id, stamp(rec), provision.Receipt{})(rec) |
| + }) |
| + s2 := *w.record().Move.StagedAt |
| + w.up() |
| + w.boxes() // the stager answers S2 only |
| + if st, _ := w.report(id, moves.OfflineRoleStager, host.E2ESourceBox); !st.StageRequest.Equal(s2) { |
| + t.Fatalf("the stager answered %v, want the operator's %v", st.StageRequest, s2) |
| + } |
| + w.down() |
| + if crash { |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { |
| + return strings.HasSuffix(p, ":issue") && r.RequestAbandoned |
| + } |
| + } |
| + w.consoleUntil(4, "the wait abandoned", func() bool { return w.row(id).Request.Nonce != s1.Nonce || w.row(id).Request.Consumed }) |
| + if crash { |
| + w.consoleUntil(4, "the crash", func() bool { return w.crashWhen == nil }) |
| + w.restart() |
| + } |
| + w.consoleUntil(4, "a fresh round", func() bool { r := w.row(id); return r.Request.Nonce != s1.Nonce && !r.Request.Consumed }) |
| + s3 := w.row(id).Request |
| + if !s3.Proposed.After(s2) || s3.Kind != kind { |
| + t.Fatalf("the fresh round %+v, want a %s after %v", s3, kind, s2) |
| + } |
| + w.up() |
| + } |
| + t.Run("staging", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + supersede(t, w, row.ID, "stage", false) |
| + w.phase(row.ID, "ready", 60) |
| + }) |
| + t.Run("staging, 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) |
| + supersede(t, w, row.ID, "stage", true) |
| + w.phase(row.ID, "ready", 60) |
| + }) |
| + t.Run("ready", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "ready", 60) |
| + w.advance(11 * time.Minute) |
| + supersede(t, w, row.ID, "maintain", false) |
| + w.until(10, "the round consumed", func() bool { return w.row(row.ID).Request.Consumed }) |
| + if !w.markCut(row.ID) { |
| + t.Fatal("the cut was refused") |
| + } |
| + w.phase(row.ID, "barrier", 20) |
| + }) |
| +} |
diff --git a/internal/e2e/selfserve/job_reconcile_test.go b/internal/e2e/selfserve/job_reconcile_test.go| new file mode 100644 |
| index 000000000..cc5c4f937 |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_reconcile_test.go |
| @@ -0,0 +1,217 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// operatorTo takes the record of move id from staging to state with the |
| +// operator's transitions (empty receipts), as the operator doors would. |
| +func (w *jobWorld) operatorTo(id, state string) { |
| + w.t.Helper() |
| + steps := []struct { |
| + state string |
| + mut func(rec *provision.Record) error |
| + }{ |
| + {provision.OfflineStopping, func(rec *provision.Record) error { |
| + return provision.OfflineStartStaged(id, "op@example.com", stamp(rec), provision.Receipt{})(rec) |
| + }}, |
| + {provision.OfflineSealing, provision.OfflineSeal(id, host.E2ESourceBox, []string{host.E2ESourceBox}, provision.Receipt{})}, |
| + {provision.OfflineSealed, provision.OfflineSealedBy(id, host.E2ESourceBox, provision.MoveSource{})}, |
| + {provision.OfflineLanding, provision.OfflineFlip(id, provision.Receipt{})}, |
| + {provision.OfflineLanded, provision.OfflineLandedBy(id, host.E2ETargetRegion)}, |
| + {provision.OfflineOpen, func(rec *provision.Record) error { |
| + return provision.OfflineOpenTransition(id, stamp(rec), provision.Receipt{})(rec) |
| + }}, |
| + {provision.OfflineFinishing, provision.OfflineFinishingTransition(id, provision.Receipt{})}, |
| + } |
| + if state == provision.OfflineReverting { |
| + w.edit(provision.OfflineAbort(id, provision.Receipt{})) |
| + return |
| + } |
| + for _, st := range steps { |
| + w.edit(st.mut) |
| + if st.state == state { |
| + return |
| + } |
| + } |
| + w.t.Fatalf("no operator path to %s", state) |
| +} |
| + |
| +// TestReconcileForwardNeverRegresses is the reconciliation table, row by |
| +// row: a row behind its record is moved forward to the record's set's |
| +// first phase (at the substep the record dictates), a row already in the |
| +// set stays, and a row whose record says the move ended, is another's or |
| +// is gone is finished without the record being touched. |
| +func TestReconcileForwardNeverRegresses(t *testing.T) { |
| + // The record ahead of a ready row, by an operator's transitions. |
| + for _, c := range []struct{ state, phase, step string }{ |
| + {provision.OfflineStopping, "stopping", "seal"}, |
| + {provision.OfflineSealing, "sealing", ""}, |
| + {provision.OfflineSealed, "sealing", ""}, |
| + {provision.OfflineLanding, "landing", ""}, |
| + {provision.OfflineLanded, "landed", ""}, |
| + {provision.OfflineOpen, "opened", ""}, |
| + {provision.OfflineFinishing, "finishing", "gate2"}, |
| + {provision.OfflineReverting, "reverting", "finish"}, |
| + } { |
| + t.Run("ready row, record "+c.state, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + 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 != "" { |
| + t.Fatalf("reconciled to %s", describe(r)) |
| + } |
| + for i := 0; i < 3; i++ { |
| + w.tick() |
| + } |
| + if r := w.row(row.ID); r.Phase != c.phase { |
| + t.Fatalf("after more ticks the row is %s", describe(r)) |
| + } |
| + }) |
| + } |
| + |
| + // Rows already in the record's set stay. |
| + stays := func(t *testing.T, w *jobWorld, id, phase string) { |
| + t.Helper() |
| + for i := 0; i < 4; i++ { |
| + w.tick() |
| + if r := w.row(id); r.Phase != phase || r.Outcome != "" { |
| + t.Fatalf("tick %d: the row moved: %s", i, describe(r)) |
| + } |
| + } |
| + } |
| + t.Run("stopping before its start stays on offline-staging", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + w.setPhase(row.ID, "stopping", "start") |
| + stays(t, w, row.ID, "stopping") |
| + }) |
| + t.Run("stopping with its start pending stays on offline-staging", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + rec := w.record() |
| + w.setPhase(row.ID, "stopping", "start") |
| + w.sql(`UPDATE self_serve_moves SET request_issued = ?, request_kind = 'start', request_phase = 'stopping', request_nonce = 'rq_pendingstart', |
| + request_expect_inc = ?, request_expect_seq = ?, request_proposed = ?, request_confirmed = 0, request_consumed = 0, request_voided = 0, request_deadline = 0 |
| +WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now().UnixNano(), row.ID) |
| + stays(t, w, row.ID, "stopping") |
| + if r := w.row(row.ID); r.Request.Nonce != "rq_pendingstart" || r.Request.Voided || r.Request.Consumed { |
| + t.Fatalf("the pending start was resolved: %s", describe(r)) |
| + } |
| + }) |
| + t.Run("landing before its flip stays on offline-sealed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + w.operatorTo(row.ID, provision.OfflineSealed) |
| + w.consoleUntil(3, "sealing", func() bool { return w.row(row.ID).Phase == "sealing" }) |
| + w.setPhase(row.ID, "landing", "") |
| + stays(t, w, row.ID, "landing") |
| + }) |
| + for _, phase := range []string{"health", "finishing"} { |
| + t.Run(phase+" survives ticks at offline-open", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "ready", 60) |
| + w.down() |
| + w.operatorTo(row.ID, provision.OfflineOpen) |
| + w.consoleUntil(3, "opened", func() bool { return w.row(row.ID).Phase == "opened" }) |
| + step := "" |
| + if phase == "finishing" { |
| + step = "gate1" |
| + } |
| + w.setPhase(row.ID, phase, step) |
| + stays(t, w, row.ID, phase) |
| + }) |
| + } |
| + t.Run("reports and prechecks stay on their records", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + stays(t, w, row.ID, "reports") |
| + w.up() |
| + w.phase(row.ID, "prechecks", 60) |
| + w.down() |
| + stays(t, w, row.ID, "prechecks") |
| + }) |
| + |
| + // Rows behind their records move forward. |
| + t.Run("reports row, record offline-staging", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "prechecks", 60) |
| + w.down() |
| + w.setPhase(row.ID, "reports", "") |
| + w.consoleUntil(3, "staging", func() bool { return w.row(row.ID).Phase == "staging" }) |
| + if r := w.row(row.ID); r.PhaseStep != "round" || r.StagingStartedAt.IsZero() { |
| + t.Fatalf("entered staging %s, started %v", describe(r), r.StagingStartedAt) |
| + } |
| + }) |
| + t.Run("reserving row, record offline-queued", func(t *testing.T) { |
| + 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, "reserving", "") |
| + w.consoleUntil(3, "queued", func() bool { return w.row(row.ID).Phase == "queued" }) |
| + w.consoleUntil(3, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + }) |
| + |
| + // Rows whose record ended the move, carries another or is gone. |
| + t.Run("no move and last_move names the row", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + w.edit(func(rec *provision.Record) error { |
| + return provision.OfflineAbortAt(row.ID, stamp(rec), provision.Receipt{})(rec) |
| + }) |
| + r := w.finished(row.ID, 5) |
| + if r.Outcome != "aborted" || len(w.reservations()) != 0 { |
| + t.Fatalf("finished %s, reservations %v", describe(r), w.reservations()) |
| + } |
| + }) |
| + t.Run("another move is in flight", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + other := moveID() |
| + w.edit(provision.OfflineStage(other, "", host.E2ETargetRegion, "op@example.com", |
| + provision.StageTarget{Stager: host.E2ESourceBox, Bucket: "data-uk", Region: "eu-west-2"}, time.Now(), "")) |
| + before := w.record() |
| + r := w.finished(row.ID, 5) |
| + if r.Outcome != "aborted" || r.OutcomeReason != "another move is in flight" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + if after := w.record(); after.Move == nil || after.Move.ID != other || after.RecordSeq != before.RecordSeq { |
| + t.Fatalf("the record was touched: %+v (seq %d, was %d)", after.Move, after.RecordSeq, before.RecordSeq) |
| + } |
| + }) |
| + t.Run("the record is gone", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + if err := w.raw.Delete(w.ctx, provision.RecordKey(w.acct, host.E2EApp, host.E2EHost)); err != nil { |
| + t.Fatal(err) |
| + } |
| + r := w.finished(row.ID, 5) |
| + if r.Outcome != "aborted" || r.OutcomeReason != "the instance was recreated" || len(w.reservations()) != 0 { |
| + t.Fatalf("finished %s, reservations %v", describe(r), w.reservations()) |
| + } |
| + }) |
| +} |
diff --git a/internal/e2e/selfserve/job_requests_test.go b/internal/e2e/selfserve/job_requests_test.go| new file mode 100644 |
| index 000000000..a5008780a |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_requests_test.go |
| @@ -0,0 +1,325 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "encoding/json" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// nonces are the distinct request nonces the snapshots saw for kind in |
| +// phase, in order. |
| +func (w *jobWorld) nonces(kind, phase string) []string { |
| + var out []string |
| + for _, s := range w.snaps { |
| + if s.Request.Kind == kind && s.Request.Phase == phase && (len(out) == 0 || out[len(out)-1] != s.Request.Nonce) { |
| + out = append(out, s.Request.Nonce) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// TestRequestsAreNeverReissuedWhileUnconsumed: an unconfirmed request whose |
| +// writes keep failing is replayed under its own nonce, never replaced; a |
| +// confirmed one whose report never comes is waited on, never reissued; both |
| +// then complete under the nonce they were issued with. |
| +func TestRequestsAreNeverReissuedWhileUnconsumed(t *testing.T) { |
| + t.Run("unconfirmed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.fs.setPut(func(string, []byte) putFault { return putFail }) |
| + w.until(10, "a restage outstanding", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "stage" && r.Request.Phase == "staging" && !r.Request.Confirmed && !r.Request.Consumed |
| + }) |
| + for i := 0; i < 6; i++ { |
| + w.boxes() |
| + w.tick() |
| + } |
| + if n := w.nonces("stage", "staging"); len(n) != 1 { |
| + t.Fatalf("the failing restage was reissued: %v", n) |
| + } |
| + issued := w.nonces("stage", "staging")[0] |
| + w.fs.setPut(nil) |
| + w.until(10, "the restage confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Nonce != issued || r.Request.Confirmed |
| + }) |
| + if r := w.row(row.ID); r.Request.Nonce != issued || w.record().Move.SelfServe.LastRequest != issued { |
| + t.Fatalf("confirmed %s, the record carries %s; want %s", describe(r), w.record().Move.SelfServe.LastRequest, issued) |
| + } |
| + }) |
| + t.Run("confirmed", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.down() |
| + w.consoleUntil(10, "a round confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "stage" && r.Request.Confirmed && !r.Request.Consumed |
| + }) |
| + waiting := w.row(row.ID).Request.Nonce |
| + for i := 0; i < 6; i++ { |
| + w.tick() |
| + } |
| + if r := w.row(row.ID); r.Request.Nonce != waiting || r.Request.Consumed { |
| + t.Fatalf("the confirmed wait changed: %s (was %s)", describe(r), waiting) |
| + } |
| + w.up() |
| + w.phase(row.ID, "ready", 60) |
| + }) |
| +} |
| + |
| +// TestStaleEvidenceNeverSatisfiesANewRequest: a restage issued but not |
| +// written (a crash with the request persisted): the previous round's report |
| +// does not satisfy it; the replay after the crash lands the proposed stamp |
| +// it was issued with, and only a report carrying that stamp moves the row. |
| +func TestStaleEvidenceNeverSatisfiesANewRequest(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.crashWhen = func(point string, r *console.SelfServeMove) bool { |
| + return point == "stage:write" && r.Phase == "staging" |
| + } |
| + w.until(10, "the restage's crash", func() bool { return w.crashWhen == nil }) |
| + issued := w.crashed.Request |
| + prev, ok := w.report(row.ID, moves.OfflineRoleStager, host.E2ESourceBox) |
| + if !ok || prev.StageRequest.Equal(issued.Proposed) { |
| + t.Fatalf("the previous round's report %+v", prev) |
| + } |
| + w.restart() |
| + w.down() |
| + for i := 0; i < 5; i++ { |
| + w.tick() |
| + } |
| + r := w.row(row.ID) |
| + rec := w.record() |
| + switch { |
| + case r.Request.Nonce != issued.Nonce || !r.Request.Confirmed || r.Request.Consumed: |
| + t.Fatalf("after the replay: %s; want %s confirmed and waiting", describe(r), issued.Nonce) |
| + case rec.Move.StagedAt == nil || !rec.Move.StagedAt.Equal(issued.Proposed) || !r.StageRequest.Equal(issued.Proposed): |
| + t.Fatalf("the replay landed staged_at %v (row %v); want the persisted %v", rec.Move.StagedAt, r.StageRequest, issued.Proposed) |
| + case r.Phase != "staging": |
| + t.Fatalf("the previous round's report moved the row: %s", describe(r)) |
| + } |
| + // A stand-in for the stager answering the old stamp again changes |
| + // nothing either. |
| + prev.At = time.Now().UTC() |
| + w.putReport(prev) |
| + for i := 0; i < 3; i++ { |
| + w.tick() |
| + } |
| + if r := w.row(row.ID); r.Request.Nonce != issued.Nonce || r.Request.Consumed { |
| + t.Fatalf("a report of the old stamp satisfied the new request: %s", describe(r)) |
| + } |
| + w.up() |
| + w.until(20, "the round answered", func() bool { return w.row(row.ID).Request.Nonce != issued.Nonce }) |
| + st, _ := w.report(row.ID, moves.OfflineRoleStager, host.E2ESourceBox) |
| + if !st.StageRequest.Equal(issued.Proposed) { |
| + t.Fatalf("the row moved on a report of %v, want %v", st.StageRequest, issued.Proposed) |
| + } |
| +} |
| + |
| +// TestAbandonedAdmissionIsFencedAndConditional: a queue request that stays |
| +// unconfirmed past 30 seconds on a record with no move is given up only |
| +// through the fence. A queue write that commits after the job's negative |
| +// re-read but before the fence is in the fenced observation, so the job |
| +// continues as queued; one that never lands is fenced out, the row voided |
| +// and finished durably before the reservations go, and its late commit |
| +// cannot land. |
| +func TestAbandonedAdmissionIsFencedAndConditional(t *testing.T) { |
| + queueing := func(w *jobWorld, id string) { |
| + t.Helper() |
| + held := false |
| + queue := func(body []byte) bool { return strings.Contains(string(body), provision.OfflineQueued) } |
| + w.fs.setPut(func(key string, body []byte) putFault { |
| + switch { |
| + case !queue(body): |
| + return putRun |
| + case !held: |
| + held = true |
| + return putHeld |
| + } |
| + return putFail |
| + }) |
| + w.consoleUntil(10, "the queue outstanding", func() bool { |
| + r := w.row(id) |
| + return r.Request.Kind == "queue" && !r.Request.Confirmed && held |
| + }) |
| + w.fs.setPut(func(_ string, body []byte) putFault { |
| + if queue(body) { |
| + return putFail |
| + } |
| + return putRun |
| + }) |
| + w.advance(31 * time.Second) |
| + } |
| + t.Run("the queue commits before the fence", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + queueing(w, row.ID) |
| + nonce := w.row(row.ID).Request.Nonce |
| + w.at("queue:fence", func() { |
| + for _, err := range w.fs.release(w.ctx) { |
| + if err != nil { |
| + t.Errorf("the late queue: %v", err) |
| + } |
| + } |
| + }) |
| + w.tick() |
| + if w.on["queue:fence"] != nil { |
| + t.Fatal("the job never fenced the queue") |
| + } |
| + w.fs.setPut(nil) |
| + r := w.phase(row.ID, "reports", 10) |
| + if rec := w.record(); rec.Move == nil || rec.Move.ID != row.ID || rec.Move.SelfServe.LastRequest != nonce || r.Outcome != "" { |
| + t.Fatalf("after the fence: record %+v, row %s", rec.Move, describe(r)) |
| + } |
| + }) |
| + t.Run("the queue never lands", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + queueing(w, row.ID) |
| + if len(w.reservations()) == 0 { |
| + t.Fatal("no reservation before the fence") |
| + } |
| + w.crash = "finalise:release" |
| + w.tick() |
| + r := w.row(row.ID) |
| + if !r.Request.Voided || r.Outcome != "aborted" || r.OutcomeReason != "the move could not be started" || !r.ReleasePending || len(w.reservations()) == 0 { |
| + t.Fatalf("before the release: %s (release pending %v, reservations %v)", describe(r), r.ReleasePending, w.reservations()) |
| + } |
| + for _, err := range w.fs.release(w.ctx) { |
| + if err == nil { |
| + t.Fatal("the queue written before the fence landed after it") |
| + } |
| + } |
| + w.fs.setPut(nil) |
| + w.restart() |
| + r = w.finished(row.ID, 5) |
| + if rec := w.record(); rec.Move != nil || len(w.reservations()) != 0 || r.ReleasePending { |
| + t.Fatalf("after the release: move %+v, reservations %v", rec.Move, w.reservations()) |
| + } |
| + }) |
| +} |
| + |
| +// TestReplacedParticipantAtEveryGate: a pinned box whose current status |
| +// report names another instance aborts the move "box <label> was replaced", |
| +// at the reservation, at the reports and at the prechecks. |
| +func TestReplacedParticipantAtEveryGate(t *testing.T) { |
| + replace := func(w *jobWorld, label string) { |
| + w.e.Box(label).SetInstanceID("i-replacement") |
| + if err := w.e.Status(w.ctx); err != nil { |
| + w.t.Fatal(err) |
| + } |
| + } |
| + t.Run("reserving", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + replace(w, host.E2ETargetBox) |
| + w.down() |
| + r := w.finished(row.ID, 10) |
| + if r.OutcomeReason != "box uk-box was replaced" || len(w.reservations()) != 0 || w.record().Move != nil { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| + t.Run("reports", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + w.up() |
| + replace(w, host.E2ESourceBox) |
| + w.down() |
| + r := w.finished(row.ID, 10) |
| + if r.OutcomeReason != "box ie-box was replaced" || len(w.reservations()) != 0 || w.record().Move != nil { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| + t.Run("prechecks", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.phase(row.ID, "prechecks", 60) |
| + replace(w, host.E2ETargetBox) |
| + r := w.finished(row.ID, 30) |
| + if r.OutcomeReason != "box uk-box was replaced" || len(w.reservations()) != 0 || w.record().Move != nil { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| +} |
| + |
| +// TestReservingWaitsThenAborts: a box another operation holds keeps the row |
| +// reserving, "waiting for another move on this box", with nothing written |
| +// to the record; past the 24-hour window it aborts "the move was not |
| +// started in time" and the other operation's reservation is untouched. |
| +func TestReservingWaitsThenAborts(t *testing.T) { |
| + w := newJobWorld(t) |
| + src := w.e.Box(host.E2ESourceBox) |
| + held := moves.OfflineReservation{Kind: moves.OfflineKindCleanup, ID: "cleanup-1", Host: "other.example.com", Account: w.acct, App: host.E2EApp, |
| + ReservedAt: time.Now().Add(-time.Hour).UTC()} |
| + body, err := json.Marshal(moves.OfflineActiveBody(map[string]moves.OfflineReservation{src.BoxID(): held})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := w.raw.Put(w.ctx, moves.OfflineActiveKey, body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + row := w.admit() |
| + before := w.record() |
| + w.until(6, "waiting", func() bool { return w.row(row.ID).Reason == "waiting for another move on this box" }) |
| + if r := w.row(row.ID); r.Phase != "reserving" || w.record().RecordSeq != before.RecordSeq { |
| + t.Fatalf("while waiting: %s, record seq %d (was %d)", describe(r), w.record().RecordSeq, before.RecordSeq) |
| + } |
| + w.advance(24*time.Hour + time.Minute) |
| + r := w.finished(row.ID, 6) |
| + if r.OutcomeReason != "the move was not started in time" || r.Outcome != "aborted" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + if left := w.reservations(); len(left) != 1 || left[src.BoxID()].ID != "cleanup-1" { |
| + t.Fatalf("reservations after the abort: %v", left) |
| + } |
| +} |
| + |
| +// TestEligibilityRechecksBeforeQueue: the record is checked again, fresh, |
| +// once the boxes are reserved and before the queue write: a run opened |
| +// since the request, or a class the record no longer has, aborts with the |
| +// check's sentence and writes no move. |
| +func TestEligibilityRechecksBeforeQueue(t *testing.T) { |
| + for _, c := range []struct { |
| + name, reason string |
| + edit func(*provision.Record) error |
| + }{ |
| + {"an open run", "a run is open on this instance: run_1. wait for it to finish or cancel it, then move the instance", func(rec *provision.Record) error { |
| + rec.ActiveRun = &provision.RunState{ID: "run_1", State: provision.RunQueued, Deadline: time.Now().Add(time.Hour).UTC()} |
| + return nil |
| + }}, |
| + {"steered", host.E2EHost + " is steered or runs under a unit. offline moves cannot carry either yet", func(rec *provision.Record) error { |
| + rec.Steering = "canary" |
| + return nil |
| + }}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.at("reserve", func() { w.edit(c.edit) }) |
| + r := w.finished(row.ID, 10) |
| + if r.Outcome != "aborted" || r.OutcomeReason != c.reason || w.record().Move != nil || len(w.reservations()) != 0 { |
| + t.Fatalf("finished %s, move %+v", describe(r), w.record().Move) |
| + } |
| + if w.hits["queue:write"] != 0 { |
| + t.Fatalf("the queue was written %d times", w.hits["queue:write"]) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/e2e/selfserve/job_resume_test.go b/internal/e2e/selfserve/job_resume_test.go| new file mode 100644 |
| index 000000000..8596a5f05 |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_resume_test.go |
| @@ -0,0 +1,360 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "encoding/json" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// TestConfirmedRequestResumesAfterACrash: a confirmed request whose wait |
| +// is satisfied, crashed before its consumption, resumes the wait on the |
| +// next tick: no second request, one bump for its one landed write. |
| +func TestConfirmedRequestResumesAfterACrash(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "stage:consume" } |
| + w.until(10, "the crash before the consume", func() bool { return w.crashWhen == nil }) |
| + confirmed := w.crashed.Request.Nonce |
| + bumps, confirms := w.hits["stage:bump"], w.hits["stage:confirm"] |
| + w.restart() |
| + w.until(5, "the round consumed", func() bool { return w.row(row.ID).Request.Nonce != confirmed }) |
| + for _, s := range w.snaps { |
| + if s.Request.Kind == "stage" && s.Request.Phase == "staging" && s.Request.Nonce != confirmed && !s.Request.Consumed && s.RequestIssued.Before(w.crashed.RequestIssued) { |
| + t.Fatalf("a second request before the consume: %s", describe(&s)) |
| + } |
| + } |
| + // One bump for the confirmed write; the next round's own, if issued. |
| + if got := w.hits["stage:bump"] - bumps; got != w.hits["stage:confirm"]-confirms { |
| + t.Fatalf("%d bumps for %d confirmations since the crash", got, w.hits["stage:confirm"]-confirms) |
| + } |
| + w.phase(row.ID, "ready", 60) |
| +} |
| + |
| +// TestSamePhaseSequencesConsume: inside a phase that writes more than |
| +// once, each satisfied request is consumed immediately before the next is |
| +// issued: staging round after round, and the estimate after the precheck. |
| +// (stopping's start then seal and finishing's two gates are Task 8's.) |
| +func TestSamePhaseSequencesConsume(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + rounds, estimates := 0, 0 |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { |
| + switch { |
| + case p == "stage:issue" && r.Phase == "staging" && r.Request.Phase == "staging": |
| + rounds++ |
| + if !r.Request.Consumed || r.PhaseStep != "round" { |
| + t.Errorf("a round issued over %s", describe(r)) |
| + } |
| + case p == "stage:issue" && r.Phase == "staging": |
| + // The first round in staging follows the reports' request, |
| + // consumed by the round's own consume. |
| + rounds++ |
| + if !r.Request.Consumed { |
| + t.Errorf("a round issued over %s", describe(r)) |
| + } |
| + case p == "estimate:issue": |
| + estimates++ |
| + if r.Request.Kind != "precheck" || !r.Request.Consumed || r.PhaseStep != "estimate" { |
| + t.Errorf("the estimate issued over %s", describe(r)) |
| + } |
| + } |
| + return false |
| + } |
| + w.phase(row.ID, "ready", 60) |
| + w.crashWhen = nil |
| + if rounds < 1 || estimates < 1 { |
| + t.Fatalf("%d staging rounds issued in staging, %d estimates", rounds, estimates) |
| + } |
| + if st, _ := w.report(row.ID, "stager", host.E2ESourceBox); st.Rounds < 2 { |
| + t.Fatalf("the stager ran %d rounds", st.Rounds) |
| + } |
| +} |
| + |
| +// TestForeignRequestStopsTheCaller: a phase that finds another phase's |
| +// request outstanding never interprets it and writes nothing of its own; |
| +// reconciliation resolves it next (fenced, then voided or confirmed) and |
| +// the phase then issues its own. |
| +func TestForeignRequestStopsTheCaller(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.down() |
| + w.consoleUntil(10, "reports", func() bool { return w.row(row.ID).Phase == "reports" }) |
| + w.up() |
| + w.boxes() |
| + w.down() |
| + // The reports' stage request is issued and its write fails, then the |
| + // row is put in staging, whose own requests are staging's. |
| + w.failKinds["stage"] = true |
| + w.consoleUntil(3, "the stage request outstanding", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "stage" && r.Request.Phase == "reports" && !r.Request.Confirmed |
| + }) |
| + foreign := w.row(row.ID).Request.Nonce |
| + w.setPhase(row.ID, "staging", "round") |
| + writes := w.hits["stage:write"] |
| + w.tick() |
| + if r := w.row(row.ID); w.hits["stage:write"] != writes || r.Request.Nonce != foreign && !r.Request.Voided { |
| + t.Fatalf("the staging phase interpreted the reports' request: %s, %d writes", describe(r), w.hits["stage:write"]-writes) |
| + } |
| + w.failKinds["stage"] = false |
| + w.consoleUntil(5, "the foreign request resolved", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Nonce != foreign || r.Request.Voided || r.Request.Confirmed |
| + }) |
| + if !w.voided(foreign) && w.hits["stage:fence"] == 0 { |
| + t.Fatalf("the foreign request was resolved without the fence") |
| + } |
| +} |
| + |
| +// TestNoOpMutationsStillConfirm: a write whose transition changes nothing |
| +// else still carries its receipt and is confirmed: an estimate equal to |
| +// the record's, and an abort on a record an operator already reverted. |
| +func TestNoOpMutationsStillConfirm(t *testing.T) { |
| + t.Run("an equal estimate", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "estimate:issue" } |
| + w.until(60, "the estimate's crash", func() bool { return w.crashWhen == nil }) |
| + seconds := int(w.row(row.ID).EstimateSeconds) |
| + w.edit(provision.OfflineSetEstimate(row.ID, seconds, provision.Receipt{})) |
| + w.restart() |
| + w.phase(row.ID, "ready", 10) |
| + rec := w.record() |
| + if rec.Move.SelfServe.EstimateSeconds != seconds || rec.Move.SelfServe.LastRequest != w.row(row.ID).Request.Nonce { |
| + t.Fatalf("the estimate write: %+v, row %s", rec.Move.SelfServe, describe(w.row(row.ID))) |
| + } |
| + }) |
| + t.Run("a repeated abort", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "prechecks", 60) |
| + w.cancel(row.ID) |
| + w.at("abort:issue", func() { w.edit(provision.OfflineAbort(row.ID, provision.Receipt{})) }) |
| + r := w.finished(row.ID, 40) |
| + if r.Outcome != "aborted" || w.hits["abort:confirm"] != 1 { |
| + t.Fatalf("finished %s, abort confirmations %d", describe(r), w.hits["abort:confirm"]) |
| + } |
| + }) |
| +} |
| + |
| +// TestBoundWritesNeverLandAfterAVoid: a request's write is bound to the |
| +// record it read. Another writer advancing the record makes the write fail |
| +// and the request voided, and its late commit never lands; a record |
| +// deleted and recreated at the same sequence refuses it on the |
| +// incarnation. |
| +func TestBoundWritesNeverLandAfterAVoid(t *testing.T) { |
| + t.Run("another writer", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + held := false |
| + w.fs.setPut(func(_ string, body []byte) putFault { |
| + if !held && stagingWrite(body) { |
| + held = true |
| + return putHeld |
| + } |
| + return putRun |
| + }) |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "stage:written" && r.Phase == "staging" } |
| + w.until(10, "the held restage", func() bool { return w.crashWhen == nil }) |
| + w.down() |
| + nonce := w.crashed.Request.Nonce |
| + w.fs.setPut(nil) |
| + w.edit(func(rec *provision.Record) error { |
| + return provision.OfflinePrecheck(row.ID, stamp(rec), provision.Receipt{})(rec) |
| + }) |
| + for _, err := range w.fs.release(w.ctx) { |
| + if err == nil { |
| + t.Fatal("the held write landed on the advanced record") |
| + } |
| + } |
| + w.restart() |
| + voids := w.hits["stage:void"] |
| + w.consoleUntil(5, "the void", func() bool { return w.hits["stage:void"] > voids }) |
| + if r := w.row(row.ID); r.Request.Nonce == nonce && !r.Request.Voided { |
| + t.Fatalf("after the void: %s", describe(r)) |
| + } |
| + w.up() |
| + w.phase(row.ID, "ready", 60) |
| + if rec := w.record(); rec.Move.SelfServe.LastRequest == nonce { |
| + t.Fatal("the voided request's receipt is on the record") |
| + } |
| + }) |
| + t.Run("another writer between the issue and the write", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + var nonce string |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { |
| + if p == "stage:write" && r.Phase == "staging" && nonce == "" { |
| + nonce = r.Request.Nonce |
| + w.edit(func(rec *provision.Record) error { |
| + return provision.OfflinePrecheck(row.ID, stamp(rec), provision.Receipt{})(rec) |
| + }) |
| + } |
| + return false |
| + } |
| + w.until(10, "the restage written", func() bool { return nonce != "" }) |
| + w.crashWhen = nil |
| + if rec := w.record(); rec.Move.SelfServe.LastRequest == nonce { |
| + t.Fatal("the write bound to the earlier record landed on the advanced one") |
| + } |
| + if w.hits["stage:void"] == 0 { |
| + t.Fatal("the request was not voided") |
| + } |
| + w.phase(row.ID, "ready", 60) |
| + }) |
| + t.Run("recreated at the same sequence", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == "stage:write" && r.Phase == "staging" } |
| + w.until(10, "the restage persisted", func() bool { return w.crashWhen == nil }) |
| + w.down() |
| + nonce := w.crashed.Request.Nonce |
| + key := provision.RecordKey(w.acct, host.E2EApp, host.E2EHost) |
| + raw, _, err := w.raw.Get(w.ctx, key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var body map[string]any |
| + if err := json.Unmarshal(raw, &body); err != nil { |
| + t.Fatal(err) |
| + } |
| + body["incarnation"] = 2 |
| + delete(body, "move") |
| + fresh, _ := json.Marshal(body) |
| + if err := w.raw.Delete(w.ctx, key); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := w.raw.Put(w.ctx, key, fresh, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := w.record() |
| + w.restart() |
| + r := w.finished(row.ID, 10) |
| + after := w.record() |
| + if r.OutcomeReason != "the instance was recreated" || after.Move != nil || after.RecordSeq != before.RecordSeq || !w.voided(nonce) { |
| + t.Fatalf("finished %s; the recreated record %+v (seq %d, was %d)", describe(r), after.Move, after.RecordSeq, before.RecordSeq) |
| + } |
| + }) |
| +} |
| + |
| +// TestSubstepsResumeAfterEveryCrash: a phase resumes the substep recorded, |
| +// never its first write: a crash after the precheck was consumed and |
| +// before the estimate's issue re-enters at estimate; with the estimate |
| +// outstanding it resumes it; abort, then finish, likewise. |
| +func TestSubstepsResumeAfterEveryCrash(t *testing.T) { |
| + for _, c := range []struct{ point, phase, step string }{ |
| + {"estimate:issue", "prechecks", "estimate"}, |
| + {"estimate:write", "prechecks", "estimate"}, |
| + {"finish:issue", "reverting", "finish"}, |
| + {"finish:write", "reverting", "finish"}, |
| + } { |
| + t.Run(c.point, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + if c.phase == "reverting" { |
| + w.phase(row.ID, "prechecks", 60) |
| + w.cancel(row.ID) |
| + } |
| + w.crashWhen = func(p string, r *console.SelfServeMove) bool { return p == c.point } |
| + w.until(80, "the crash", func() bool { return w.crashWhen == nil }) |
| + first := w.hits[strings.Split(c.point, ":")[0]+":issue"] |
| + w.restart() |
| + if r := w.row(row.ID); r.Phase != c.phase || r.PhaseStep != c.step { |
| + t.Fatalf("after the crash: %s", describe(r)) |
| + } |
| + prior := map[string]int{"precheck": w.hits["precheck:issue"], "abort": w.hits["abort:issue"]} |
| + if c.phase == "reverting" { |
| + w.finished(row.ID, 40) |
| + } else { |
| + w.phase(row.ID, "ready", 20) |
| + } |
| + if w.hits["precheck:issue"] != prior["precheck"] || w.hits["abort:issue"] != prior["abort"] { |
| + t.Fatalf("the phase's first write was issued again (precheck %d, abort %d)", w.hits["precheck:issue"]-prior["precheck"], w.hits["abort:issue"]-prior["abort"]) |
| + } |
| + _ = first |
| + }) |
| + } |
| +} |
| + |
| +// TestAbandonedRequestsNeverStrandAbort: a confirmed wait that will never |
| +// be satisfied is abandoned when the abort comes, so the abort's own |
| +// request issues: a stage round whose copy never reports, then a cancel; |
| +// a precheck whose box died, then the deadline. |
| +func TestAbandonedRequestsNeverStrandAbort(t *testing.T) { |
| + t.Run("a round that never reports, then a cancel", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "staging", 60) |
| + w.down() |
| + w.consoleUntil(5, "a round confirmed", func() bool { |
| + r := w.row(row.ID) |
| + return r.Request.Kind == "stage" && r.Request.Confirmed && !r.Request.Consumed |
| + }) |
| + w.cancel(row.ID) |
| + w.consoleUntil(6, "the abort issued", func() bool { r := w.row(row.ID); return r.Request.Kind == "abort" }) |
| + w.up() |
| + w.finished(row.ID, 40) |
| + }) |
| + t.Run("a precheck whose box died, then the deadline", func(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "prechecks", 60) |
| + w.down() |
| + w.consoleUntil(5, "the precheck confirmed", func() bool { r := w.row(row.ID); return r.Request.Kind == "precheck" && r.Request.Confirmed }) |
| + w.advance(6 * time.Minute) |
| + w.consoleUntil(8, "the abort issued", func() bool { r := w.row(row.ID); return r.Request.Kind == "abort" }) |
| + w.up() |
| + r := w.finished(row.ID, 40) |
| + if r.OutcomeReason != "could not determine the source's facts on ie-box" { |
| + t.Fatalf("finished %s", describe(r)) |
| + } |
| + }) |
| +} |
| + |
| +// TestLateEvidenceStillClearsTheMove: a wait whose request carries no |
| +// deadline (the abort's finish) is never expired by it: the reverting |
| +// facts arriving after the reverting phase's 10 minutes still finish the |
| +// move and release the reservations. |
| +func TestLateEvidenceStillClearsTheMove(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.phase(row.ID, "prechecks", 60) |
| + w.down() |
| + w.cancel(row.ID) |
| + w.consoleUntil(10, "reverting at finish", func() bool { r := w.row(row.ID); return r.Phase == "reverting" && r.PhaseStep == "finish" }) |
| + w.advance(11 * time.Minute) |
| + for i := 0; i < 4; i++ { |
| + w.tick() |
| + } |
| + if r := w.row(row.ID); r.Outcome != "" { |
| + t.Fatalf("the reverting deadline finished the move: %s", describe(r)) |
| + } |
| + w.up() |
| + r := w.finished(row.ID, 30) |
| + if r.Outcome != "aborted" || len(w.reservations()) != 0 || w.record().Move != nil { |
| + t.Fatalf("finished %s, reservations %v", describe(r), w.reservations()) |
| + } |
| +} |
diff --git a/internal/e2e/selfserve/job_test.go b/internal/e2e/selfserve/job_test.go| new file mode 100644 |
| index 000000000..2f397e649 |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_test.go |
| @@ -0,0 +1,76 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The self-serve job from requested to ready, and its reconciliation |
| +// (self-serve moves plan 2 Task 7), witnessed end to end: the console's |
| +// journal and pass over the bridge's two real boxes. |
| + |
| +// TestStagingReadyRule: a requested row reserves, queues, takes the boxes' |
| +// reports, stages round after round until the ready rule holds (at least |
| +// two rounds, the last under 16 MiB), prechecks and is ready, the record |
| +// still staging with the job's receipt. |
| +func TestStagingReadyRule(t *testing.T) { |
| + w := newJobWorld(t) |
| + row := w.admit() |
| + r := w.phase(row.ID, "ready", 60) |
| + rec := w.record() |
| + if rec.Move == nil || rec.Move.State != provision.OfflineStaging || rec.Move.SelfServe == nil || rec.Move.SelfServe.LastRequest != r.Request.Nonce { |
| + t.Fatalf("record at ready: %+v; row %s", rec.Move, describe(r)) |
| + } |
| + st, ok := w.report(row.ID, "stager", "ie-box") |
| + if !ok || st.Rounds < 2 { |
| + t.Fatalf("stager report %+v", st) |
| + } |
| +} |
| + |
| +// cancel marks the row's abort as the cancel door does. |
| +func (w *jobWorld) cancel(id string) { |
| + w.t.Helper() |
| + ok, err := w.cst.RequestSelfServeAbort(id, []string{"requested", "reserving", "queued", "reports", "staging", "prechecks", "ready"}, time.Now().Add(w.skew).UTC()) |
| + if err != nil || !ok { |
| + w.t.Fatalf("cancel %s = %v, %v (%s)", id, ok, err, describe(w.row(id))) |
| + } |
| +} |
| + |
| +// TestAbortBeforeOfflineIsLossless: a cancel in each phase before the cut |
| +// returns the move with nothing lost: no record move, last_move aborted |
| +// with no offline time, the source never stopped (reopened at the cancel), |
| +// the reservations released, the row finished "the move was cancelled". |
| +func TestAbortBeforeOfflineIsLossless(t *testing.T) { |
| + for _, at := range []string{"requested", "reserving", "reports", "staging", "ready"} { |
| + t.Run(at, func(t *testing.T) { |
| + w := newJobWorld(t) |
| + region := w.record().Region |
| + row := w.admit() |
| + if at != "requested" { |
| + w.phase(row.ID, at, 60) |
| + } |
| + w.cancel(row.ID) |
| + r := w.finished(row.ID, 60) |
| + rec := w.record() |
| + switch { |
| + case r.Outcome != "aborted" || r.OutcomeReason != "the move was cancelled": |
| + t.Fatalf("outcome %q %q", r.Outcome, r.OutcomeReason) |
| + case rec.Move != nil || rec.Region != region: |
| + t.Fatalf("record after the abort: move %+v region %q", rec.Move, rec.Region) |
| + case !r.SourceReopenedAt.Equal(r.AbortRequestedAt) || r.AbortFinishedAt.IsZero(): |
| + t.Fatalf("abort facts: requested %v reopened %v finished %v", r.AbortRequestedAt, r.SourceReopenedAt, r.AbortFinishedAt) |
| + case len(w.reservations()) != 0: |
| + t.Fatalf("reservations left: %v", w.reservations()) |
| + } |
| + if at != "requested" && at != "reserving" { |
| + if rec.LastMove == nil || rec.LastMove.ID != row.ID || rec.LastMove.Outcome != "aborted" || rec.LastMove.OfflineS != 0 || rec.LastMove.Request == "" { |
| + t.Fatalf("last_move %+v", rec.LastMove) |
| + } |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go| new file mode 100644 |
| index 000000000..3c43920ba |
| --- /dev/null |
| +++ b/internal/e2e/selfserve/job_world_test.go |
| @@ -0,0 +1,511 @@ |
| +//go:build e2e |
| + |
| +package selfserve |
| + |
| +import ( |
| + "context" |
| + "database/sql" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/console" |
| + "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The self-serve job's world (self-serve moves plan 2 Task 7): the bridge's |
| +// two real boxes over one deployment store, and beside them a real console |
| +// (its SQLite journal, its self-serve pass driven through |
| +// console.TestHooks.Tick) whose view of the bucket goes through faultStore, |
| +// so a witness can lose a write's reply, hold a write and commit it after |
| +// the console gave up on it, or count the serial bumps. A console restart |
| +// closes the store and opens the same database file again, under the same |
| +// console name (a console restarted on its own host takes its lease back). |
| + |
| +// faultStore is the console's bucket: the deployment store, with the |
| +// faults a witness arms. |
| +type faultStore struct { |
| + release.Store |
| + mu sync.Mutex |
| + // put, when set, decides each record Put of the console: run (the |
| + // write reaches the store), lost (its reply is lost after it ran) or |
| + // held (it does not run now; release runs it later against the ETag it |
| + // was sent with, as a store committing after the client's context was |
| + // cancelled would). |
| + put func(key string, body []byte) putFault |
| + held []heldPut |
| + // failing, when set, fails every record Put while it answers true. |
| + failing func() bool |
| + bumps int |
| + puts map[string]int |
| +} |
| + |
| +type putFault int |
| + |
| +const ( |
| + putRun putFault = iota |
| + putLost |
| + putHeld |
| + putFail |
| +) |
| + |
| +type heldPut struct { |
| + key, ifMatch string |
| + body []byte |
| +} |
| + |
| +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 |
| + if key == release.SerialKey { |
| + f.bumps++ |
| + } |
| + if f.puts == nil { |
| + f.puts = map[string]int{} |
| + } |
| + f.mu.Unlock() |
| + fault := putRun |
| + if decide != nil && provision.IsRecordKey(key) { |
| + fault = decide(key, body) |
| + } |
| + if f.failing != nil && f.failing() && provision.IsRecordKey(key) { |
| + fault = putFail |
| + } |
| + switch fault { |
| + case putHeld: |
| + f.mu.Lock() |
| + f.held = append(f.held, heldPut{key: key, ifMatch: ifMatch, body: append([]byte(nil), body...)}) |
| + f.mu.Unlock() |
| + return "", context.DeadlineExceeded |
| + case putFail: |
| + return "", errLostReply |
| + } |
| + tag, err := f.Store.Put(ctx, key, body, ifMatch) |
| + if err == nil && provision.IsRecordKey(key) { |
| + f.mu.Lock() |
| + f.puts[key]++ |
| + f.mu.Unlock() |
| + } |
| + if fault == putLost && err == nil { |
| + return "", errLostReply |
| + } |
| + return tag, err |
| +} |
| + |
| +// DeleteIfMatch forwards the deployment store's conditional delete. |
| +func (f *faultStore) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| + return f.Store.(release.ConditionalDeleter).DeleteIfMatch(ctx, key, etag) |
| +} |
| + |
| +// release runs every held Put now, as the store committing it late; each |
| +// answers what the store answered (a CAS conflict when the record moved). |
| +func (f *faultStore) release(ctx context.Context) []error { |
| + f.mu.Lock() |
| + held := f.held |
| + f.held = nil |
| + f.mu.Unlock() |
| + var errs []error |
| + for _, h := range held { |
| + _, err := f.Store.Put(ctx, h.key, h.body, h.ifMatch) |
| + if err == nil { |
| + f.mu.Lock() |
| + f.puts[h.key]++ |
| + f.mu.Unlock() |
| + } |
| + errs = append(errs, err) |
| + } |
| + return errs |
| +} |
| + |
| +func (f *faultStore) setPut(fn func(key string, body []byte) putFault) { |
| + f.mu.Lock() |
| + f.put = fn |
| + f.mu.Unlock() |
| +} |
| + |
| +func (f *faultStore) bumpCount() int { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return f.bumps |
| +} |
| + |
| +func (f *faultStore) recordPuts() int { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + n := 0 |
| + for _, v := range f.puts { |
| + n += v |
| + } |
| + return n |
| +} |
| + |
| +// jobWorld is the boxes, the console and its journal. |
| +type jobWorld struct { |
| + t *testing.T |
| + ctx context.Context |
| + e *host.E2E |
| + raw release.Store |
| + fs *faultStore |
| + dir string |
| + cst *console.Store |
| + srv *console.Server |
| + now time.Time // the console's clock: the wall clock plus skew |
| + skew time.Duration |
| + acct string |
| + crash string // the crash point armed, "" for none |
| + hits map[string]int |
| + // watch is the row the witness follows: crashWhen, when set, decides |
| + // at every crash point from the point and the row as it is then |
| + // (once: it is cleared when it fires), crashed is the row at the |
| + // crash, and snaps the row after every tick. |
| + watch string |
| + crashWhen func(point string, r *console.SelfServeMove) bool |
| + crashed *console.SelfServeMove |
| + snaps []console.SelfServeMove |
| + // on runs a witness's action at a crash point (once). |
| + on map[string]func() |
| + // failKinds are the request kinds whose record writes fail (the |
| + // write's own Put, between its write and written points; a fence or |
| + // any other write still runs). |
| + failKinds map[string]bool |
| + failing bool |
| +} |
| + |
| +const jobConsole = "console-e2e-job" |
| + |
| +// newJobWorld builds the bridge with a lifetime record (incarnation 1) and |
| +// a console beside it in self-serve mode. |
| +func newJobWorld(t *testing.T) *jobWorld { |
| + t.Helper() |
| + raw := release.NewDirStore(t.TempDir()) |
| + e := host.NewE2E(t, raw, host.E2ELifetime()) |
| + w := &jobWorld{t: t, ctx: context.Background(), e: e, raw: raw, fs: &faultStore{Store: raw}, dir: t.TempDir(), acct: e.Account(), hits: map[string]int{}, |
| + failKinds: map[string]bool{}} |
| + w.fs.failing = func() bool { return w.failing } |
| + w.open() |
| + t.Cleanup(func() { |
| + console.TestHooks.Fault(nil) |
| + if w.cst != nil { |
| + w.cst.Close() |
| + } |
| + }) |
| + console.TestHooks.Fault(w.fault) |
| + // The boxes publish their edges files and status blocks. |
| + w.boxes() |
| + return w |
| +} |
| + |
| +// open opens the console's database and builds the console on it. |
| +func (w *jobWorld) open() { |
| + w.t.Helper() |
| + cst, err := console.Open(filepath.Join(w.dir, "console.db")) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + srv, err := console.NewServer(console.Config{Origin: "http://127.0.0.1", Keymail: "http://keymail.invalid", |
| + Registry: filepath.Join(w.dir, "registry.db"), Releases: w.fs, ReleasePub: w.e.ReleasePub(), Key: w.e.ReleaseKey(), AppScheme: "http", |
| + HomeRegion: host.E2EHomeRegion, SecretsPub: w.e.SecretsPub(), SelfServeMoves: "offline"}, cst) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + console.TestHooks.NameConsole(srv, jobConsole) |
| + w.cst, w.srv = cst, srv |
| +} |
| + |
| +// restart is a console crash and restart: the same database file opened |
| +// again, a new console process under the same name. |
| +func (w *jobWorld) restart() { |
| + w.t.Helper() |
| + if err := w.cst.Close(); err != nil { |
| + w.t.Fatal(err) |
| + } |
| + w.open() |
| +} |
| + |
| +// fault is the console's crash points: the armed one crashes the tick. |
| +func (w *jobWorld) fault(point string) { |
| + w.hits[point]++ |
| + if kind, step, ok := strings.Cut(point, ":"); ok && w.failKinds[kind] { |
| + switch step { |
| + case "write": |
| + w.failing = true |
| + case "written": |
| + w.failing = false |
| + } |
| + } |
| + if fn := w.on[point]; fn != nil { |
| + delete(w.on, point) |
| + fn() |
| + } |
| + if w.crash != "" && point == w.crash { |
| + w.crash = "" |
| + console.TestHooks.Crash(point) |
| + } |
| + if w.crashWhen != nil && w.watch != "" { |
| + r, err := w.cst.SelfServeMove(w.watch) |
| + if err == nil && r != nil && w.crashWhen(point, r) { |
| + w.crashWhen, w.crashed = nil, r |
| + console.TestHooks.Crash(point) |
| + } |
| + } |
| +} |
| + |
| +// at runs fn once, when the job reaches point. |
| +func (w *jobWorld) at(point string, fn func()) { |
| + if w.on == nil { |
| + w.on = map[string]func(){} |
| + } |
| + w.on[point] = fn |
| +} |
| + |
| +// sql runs a statement on the console's database file, for a witness that |
| +// puts a row where the job itself has no way to (another task's phase). |
| +func (w *jobWorld) sql(stmt string, args ...any) { |
| + w.t.Helper() |
| + db, err := sql.Open("sqlite", filepath.Join(w.dir, "console.db")) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec(stmt, args...); err != nil { |
| + w.t.Fatal(err) |
| + } |
| +} |
| + |
| +// boxes runs one agent tick and one status tick on every box that is up. |
| +func (w *jobWorld) boxes() { |
| + w.t.Helper() |
| + if err := w.e.Pass(w.ctx); err != nil { |
| + w.t.Logf("boxes: %v", err) |
| + } |
| + if err := w.e.Status(w.ctx); err != nil { |
| + w.t.Logf("status: %v", err) |
| + } |
| +} |
| + |
| +// tick runs one console pass at the console's clock and answers the crash |
| +// point it stopped at, if any. |
| +func (w *jobWorld) tick() string { |
| + w.t.Helper() |
| + w.now = time.Now().Add(w.skew) |
| + crashed := console.TestHooks.Tick(w.ctx, w.srv, w.now) |
| + if w.watch != "" { |
| + if r, err := w.cst.SelfServeMove(w.watch); err == nil && r != nil { |
| + w.snaps = append(w.snaps, *r) |
| + } |
| + } |
| + return crashed |
| +} |
| + |
| +// advance moves the console's clock (never the boxes'). |
| +func (w *jobWorld) advance(d time.Duration) { w.skew += d } |
| + |
| +// participants pins both boxes as the self-serve door would. |
| +func (w *jobWorld) participants() provision.Participants { |
| + src, tgt := w.e.Box(host.E2ESourceBox), w.e.Box(host.E2ETargetBox) |
| + return provision.Participants{ |
| + Source: []provision.Participant{{BoxID: src.BoxID(), Label: src.Label(), InstanceID: src.InstanceID()}}, |
| + Target: provision.Participant{BoxID: tgt.BoxID(), Label: tgt.Label(), InstanceID: tgt.InstanceID()}, |
| + } |
| +} |
| + |
| +var idSeq int |
| + |
| +// moveID is a fresh well-formed move id. |
| +func moveID() string { |
| + idSeq++ |
| + b := []byte("mvjobaaaaaaaaaaaaaaaaaaaaa") |
| + for n, i := idSeq, len(b)-1; n > 0; n, i = n/26, i-1 { |
| + b[i] = byte('a' + n%26) |
| + } |
| + return "mv_" + string(b) |
| +} |
| + |
| +// admit writes a requested row for the instance, as the request door does |
| +// once its checks pass. |
| +func (w *jobWorld) admit() *console.SelfServeMove { |
| + w.t.Helper() |
| + return w.admitFor(host.E2EHost, w.participants()) |
| +} |
| + |
| +func (w *jobWorld) admitFor(hostname string, p provision.Participants) *console.SelfServeMove { |
| + w.t.Helper() |
| + rec := w.recordOf(hostname) |
| + now := time.Now().Add(w.skew).UTC() |
| + row := console.SelfServeMove{ID: moveID(), Account: w.acct, App: host.E2EApp, Host: hostname, Incarnation: rec.Incarnation, |
| + From: "", To: host.E2ETargetRegion, Participants: p, DisplayName: "Shop's account", RequestedBy: "token:tok shop-admin", |
| + Phase: "requested", EstimateSeconds: 120, WindowDeadline: now.Add(24 * time.Hour), CreatedAt: now, UpdatedAt: now, PhaseStartedAt: now} |
| + ok, open, err := w.cst.AdmitSelfServeMove(row) |
| + if err != nil || !ok { |
| + w.t.Fatalf("admit = %v %+v %v", ok, open, err) |
| + } |
| + return w.row(row.ID) |
| +} |
| + |
| +// row is the journal's row. |
| +func (w *jobWorld) row(id string) *console.SelfServeMove { |
| + w.t.Helper() |
| + r, err := w.cst.SelfServeMove(id) |
| + if err != nil || r == nil { |
| + w.t.Fatalf("row %s: %v %v", id, r, err) |
| + } |
| + return r |
| +} |
| + |
| +// record is the instance's record, read from the deployment store. |
| +func (w *jobWorld) record() *provision.Record { |
| + w.t.Helper() |
| + return w.recordOf(host.E2EHost) |
| +} |
| + |
| +func (w *jobWorld) recordOf(hostname string) *provision.Record { |
| + w.t.Helper() |
| + rec, _, err := provision.GetRecordETag(w.ctx, w.raw, w.acct, host.E2EApp, hostname) |
| + if err != nil { |
| + w.t.Fatalf("record %s: %v", hostname, err) |
| + } |
| + return rec |
| +} |
| + |
| +// edit writes the record as another writer (an operator, a box) would. |
| +func (w *jobWorld) edit(mut func(*provision.Record) error) { |
| + w.t.Helper() |
| + if _, err := provision.SetRecordMove(w.ctx, w.raw, w.acct, host.E2EApp, host.E2EHost, mut); err != nil { |
| + w.t.Fatal(err) |
| + } |
| +} |
| + |
| +// until runs rounds (the boxes, then one console tick) until done holds, |
| +// failing after max rounds. A crash stops nothing here: the next round |
| +// ticks again. |
| +func (w *jobWorld) until(max int, what string, done func() bool) { |
| + w.t.Helper() |
| + for i := 0; i < max; i++ { |
| + if done() { |
| + return |
| + } |
| + w.boxes() |
| + w.tick() |
| + } |
| + if !done() { |
| + w.t.Fatalf("after %d rounds, still not %s", max, what) |
| + } |
| +} |
| + |
| +// phase waits for row id to reach phase. |
| +func (w *jobWorld) phase(id, phase string, max int) *console.SelfServeMove { |
| + w.t.Helper() |
| + w.until(max, "in "+phase, func() bool { |
| + r := w.row(id) |
| + return r.Phase == phase || r.Outcome != "" |
| + }) |
| + r := w.row(id) |
| + if r.Phase != phase { |
| + w.t.Fatalf("row %s is %s (%s %q), want %s", id, r.Phase, r.Outcome, r.OutcomeReason, phase) |
| + } |
| + return r |
| +} |
| + |
| +// finished waits for row id's outcome. |
| +func (w *jobWorld) finished(id string, max int) *console.SelfServeMove { |
| + w.t.Helper() |
| + w.until(max, "finished", func() bool { |
| + r := w.row(id) |
| + return r.Outcome != "" && !r.ReleasePending && r.EffectsDone |
| + }) |
| + return w.row(id) |
| +} |
| + |
| +// reservations is active.json's table, by box key. |
| +func (w *jobWorld) reservations() map[string]moves.OfflineReservation { |
| + w.t.Helper() |
| + raw, _, err := w.raw.Get(w.ctx, moves.OfflineActiveKey) |
| + if err != nil { |
| + return nil |
| + } |
| + table, err := moves.ParseOfflineActive(raw) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + return table |
| +} |
| + |
| +// report reads one of the move's box reports. |
| +func (w *jobWorld) report(id, role, box string) (moves.OfflineReport, bool) { |
| + w.t.Helper() |
| + raw, _, err := w.raw.Get(w.ctx, moves.OfflineReportKey(host.E2EHost, id, role, box)) |
| + if err != nil { |
| + return moves.OfflineReport{}, false |
| + } |
| + var r moves.OfflineReport |
| + if err := json.Unmarshal(raw, &r); err != nil { |
| + w.t.Fatal(err) |
| + } |
| + return r, true |
| +} |
| + |
| +// putReport writes a box report as a box would, for the witnesses that |
| +// stand in for a box's answer. |
| +func (w *jobWorld) putReport(r moves.OfflineReport) { |
| + w.t.Helper() |
| + body, err := json.Marshal(r) |
| + if err != nil { |
| + w.t.Fatal(err) |
| + } |
| + key := moves.OfflineReportKey(r.Host, r.Op, r.Role, r.Box) |
| + _, etag, _ := w.raw.Get(w.ctx, key) |
| + if _, err := w.raw.Put(w.ctx, key, body, etag); err != nil { |
| + w.t.Fatal(err) |
| + } |
| +} |
| + |
| +// describe is a row's request state, for a failure message. |
| +func describe(r *console.SelfServeMove) string { |
| + return fmt.Sprintf("phase %s/%s outcome %q %q request %s %s confirmed %v consumed %v voided %v expired %v", |
| + r.Phase, r.PhaseStep, r.Outcome, r.OutcomeReason, r.Request.Kind, r.Request.Nonce, r.Request.Confirmed, r.Request.Consumed, r.Request.Voided, r.Request.Expired) |
| +} |
| + |
| +// consoleUntil ticks the console alone (the boxes stay still) until done. |
| +func (w *jobWorld) consoleUntil(max int, what string, done func() bool) { |
| + w.t.Helper() |
| + for i := 0; i < max; i++ { |
| + if done() { |
| + return |
| + } |
| + w.tick() |
| + } |
| + if !done() { |
| + w.t.Fatalf("after %d console ticks, still not %s", max, what) |
| + } |
| +} |
| + |
| +// down takes both boxes down (their passes and status ticks are skipped). |
| +func (w *jobWorld) down() { |
| + w.e.Box(host.E2ESourceBox).Down() |
| + w.e.Box(host.E2ETargetBox).Down() |
| +} |
| + |
| +func (w *jobWorld) up() { |
| + w.e.Box(host.E2ESourceBox).Up() |
| + w.e.Box(host.E2ETargetBox).Up() |
| +} |
| + |
| +// setPhase puts the row in phase/step as another task's step would, with |
| +// no deadline. |
| +func (w *jobWorld) setPhase(id, phase, step string) { |
| + w.t.Helper() |
| + w.sql(`UPDATE self_serve_moves SET phase = ?, phase_step = ?, step_deadline = 0 WHERE id = ?`, phase, step, id) |
| +} |
| + |
| +// stamp is a transition's stamp proposed from the record, as a door's. |
| +func stamp(rec *provision.Record) time.Time { return provision.ProposeStamp(rec, time.Now()) } |