amadan

carlos / platform Public

console, e2e: the estimate's defaults then medians, the two-minute floor, before and after staging; first-observation marks; no rates from aborted moves; sequenced events delivered once and dead-lettered after five failures; the operator status's self_serve and log; the cut's timings and events end to end

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 587a797115821ec7dac0e6b990e6719cefdf042c parent 109a0b6
3 files changed, +669 −0
  • internal/console/selfserve_estimate_test.go +341 −0
  • internal/console/selfserve_events_test.go +256 −0
  • internal/e2e/selfserve/cut_test.go +72 −0
diff --git a/internal/console/selfserve_estimate_test.go b/internal/console/selfserve_estimate_test.go
new file mode 100644
index 000000000..28b01e671
--- /dev/null
+++ b/internal/console/selfserve_estimate_test.go
@@ -0,0 +1,341 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The estimate and the timings (self-serve moves plan 2 Task 9; spec
+// §6.3): the defaults until a region pair has five finished moves' timings
+// for a term, then the median of the last twenty; the marks as the
+// console first observed them, insert-only; the five move_timings rows a
+// finished move derives, and none from an aborted one.
+
+// ssTiming inserts one move_timings row for the lane ie -> uk.
+func ssTiming(t *testing.T, st *Store, id, phase string, bytes int64, seconds float64, at time.Time) {
+ t.Helper()
+ if _, err := st.InsertMoveTiming(MoveTiming{ID: id, Phase: phase, From: "ie", To: "uk", Bytes: bytes, Seconds: seconds, At: at}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// ssRate is how long n bytes take at rate bytes a second.
+func ssRate(n int64, rate float64) time.Duration {
+ return time.Duration(float64(n) / rate * float64(time.Second))
+}
+
+// ssSeconds is d to the nearest second, for comparing terms.
+func ssSeconds(d time.Duration) int64 { return int64(d.Round(time.Second) / time.Second) }
+
+func TestEstimateUsesDefaultsThenMedians(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ const remaining, files, replica = 100 * mib, 10 * mib, 100 * mib
+ defaults := estimateTerms{
+ Stop: 20 * time.Second,
+ Copy: 10*time.Second + ssRate(remaining, 25*mib),
+ Land: 30*time.Second + ssRate(files+replica, 40*mib),
+ Open: 30 * time.Second,
+ Promote: 10*time.Second + ssRate(replica, 70*mib),
+ }
+ check := func(t *testing.T, from, to string, want estimateTerms) {
+ t.Helper()
+ got, seconds := s.estimate(from, to, remaining, files, replica)
+ for _, c := range []struct {
+ name string
+ got, want time.Duration
+ }{{"stop", got.Stop, want.Stop}, {"copy", got.Copy, want.Copy}, {"land", got.Land, want.Land}, {"open", got.Open, want.Open}, {"promote", got.Promote, want.Promote}} {
+ if ssSeconds(c.got) != ssSeconds(c.want) {
+ t.Errorf("%s: %v, want %v", c.name, c.got, c.want)
+ }
+ }
+ total := want.Stop + want.Copy + want.Land + want.Open + want.Promote
+ if wantSec := int((total+time.Minute-1)/time.Minute) * 60; seconds != wantSec && wantSec >= 120 {
+ t.Errorf("seconds %d, want %d (%v in whole minutes)", seconds, wantSec, total)
+ }
+ }
+
+ t.Run("no timings: the defaults", func(t *testing.T) { check(t, "ie", "uk", defaults) })
+
+ // Four finished moves: still fewer than five per term.
+ at := ssT0
+ addMove := func(n int, stop, open float64, copyRate, landRate, promoteRate float64) {
+ at = at.Add(time.Minute)
+ id := fmt.Sprintf("mv_estimate%02d", n)
+ ssTiming(t, st, id, "stop", 0, stop, at)
+ ssTiming(t, st, id, "open", 0, open, at)
+ ssTiming(t, st, id, "copy", 100*mib, 100*mib/copyRate, at)
+ ssTiming(t, st, id, "land", 110*mib, 110*mib/landRate, at)
+ ssTiming(t, st, id, "promote", 100*mib, 100*mib/promoteRate, at)
+ }
+ for i := 0; i < 4; i++ {
+ addMove(i, 60, 90, 10*mib, 5*mib, 2*mib)
+ }
+ t.Run("four moves: still the defaults", func(t *testing.T) { check(t, "ie", "uk", defaults) })
+
+ // A fifth, an outlier: the median is the four's.
+ addMove(4, 600, 900, 1*mib, 1*mib, 1*mib)
+ medians := estimateTerms{
+ Stop: 60 * time.Second,
+ Copy: 10*time.Second + 10*time.Second,
+ Land: 30*time.Second + 22*time.Second,
+ Open: 90 * time.Second,
+ Promote: 10*time.Second + 50*time.Second,
+ }
+ t.Run("five moves: the medians", func(t *testing.T) { check(t, "ie", "uk", medians) })
+ t.Run("another region pair keeps the defaults", func(t *testing.T) {
+ check(t, "ie", "us", defaults)
+ check(t, "uk", "ie", defaults)
+ })
+
+ // A rate row without bytes is no rate: twenty newer copy rows carrying
+ // no bytes change nothing.
+ for i := 0; i < 20; i++ {
+ at = at.Add(time.Minute)
+ ssTiming(t, st, fmt.Sprintf("mv_nobytes%02d", i), "copy", 0, 1, at)
+ }
+ t.Run("rows without bytes are not rates", func(t *testing.T) { check(t, "ie", "uk", medians) })
+
+ // The last twenty only: twenty newer moves move every median.
+ for i := 0; i < 20; i++ {
+ addMove(10+i, 30, 45, 20*mib, 10*mib, 4*mib)
+ }
+ t.Run("the median of the last twenty", func(t *testing.T) {
+ check(t, "ie", "uk", estimateTerms{
+ Stop: 30 * time.Second,
+ Copy: 10*time.Second + 5*time.Second,
+ Land: 30*time.Second + 11*time.Second,
+ Open: 45 * time.Second,
+ Promote: 10*time.Second + 25*time.Second,
+ })
+ })
+}
+
+func TestEstimateMinimumIsTwoMinutes(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ // The defaults alone with nothing to carry are 100 seconds.
+ if _, seconds := s.estimate("ie", "uk", 0, 0, 0); seconds != 120 {
+ t.Fatalf("an empty instance's estimate is %d s, want 120", seconds)
+ }
+ // Five fast moves: every term a second, still two minutes.
+ for i := 0; i < 5; i++ {
+ id := fmt.Sprintf("mv_fast%02d", i)
+ for _, p := range []string{"stop", "open"} {
+ ssTiming(t, st, id, p, 0, 1, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ for _, p := range []string{"copy", "land", "promote"} {
+ ssTiming(t, st, id, p, 1<<40, 1, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ }
+ if terms, seconds := s.estimate("ie", "uk", mib, mib, mib); seconds != 120 || terms.Stop != time.Second || terms.Open != time.Second {
+ t.Fatalf("a fast lane's estimate is %d s (%+v), want 120", seconds, terms)
+ }
+ // Whole minutes, rounded up: 121 seconds is three minutes.
+ st2 := ssStore(t)
+ s2 := &Server{store: st2}
+ for i := 0; i < 5; i++ {
+ ssTiming(t, st2, fmt.Sprintf("mv_slow%02d", i), "stop", 0, 41, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ // 41 + 10 + 30 + 30 + 10 = 121 s.
+ if _, seconds := s2.estimate("ie", "uk", 0, 0, 0); seconds != 180 {
+ t.Fatalf("a 121-second estimate is %d s, want 180", seconds)
+ }
+}
+
+func TestEstimateBeforeAndAfterStaging(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ const replica, files = 3 << 30, 4 * mib
+ for i := 0; i < 5; i++ {
+ ssTiming(t, st, fmt.Sprintf("mv_lane%02d", i), "copy", 100*mib, 2, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ // Before a move is requested the final copy is the whole replica (the
+ // door's estimate).
+ before, beforeSec := s.estimate("ie", "uk", replica, files, replica)
+
+ // Once staged, the last round's copied and deferred bytes.
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ row := ssLoad(t, st, ssID)
+ mv := &provision.MoveState{Stager: "uk-1-label"}
+ ev := offlineEvidence{reports: []offlineReport{
+ {OfflineReport: moves.OfflineReport{Role: moves.OfflineRoleStager, Box: "uk-1-label", Phase: moves.OfflinePhaseStaged,
+ ReplicaBytes: replica, RoundCopiedBytes: 6 * mib, RoundDeferredBytes: 4 * mib}},
+ {OfflineReport: moves.OfflineReport{Role: moves.OfflineRoleSource, Box: "ie-1-label", Phase: moves.OfflinePhasePrecheck,
+ Precheck: &moves.SourcePrecheck{FilesBytes: files}}},
+ }}
+ if err := s.selfServeRefineEstimate(row, mv, ev); err != nil {
+ t.Fatal(err)
+ }
+ row = ssLoad(t, st, ssID)
+ after := estimateTermsOf(row.EstimateTerms)
+ want, wantSec := s.estimate("ie", "uk", 10*mib, files, replica)
+ var raw estimateTerms
+ if err := json.Unmarshal([]byte(row.EstimateTerms), &raw); err != nil {
+ t.Fatal(err)
+ }
+ switch {
+ case after != want || int(row.EstimateSeconds) != wantSec:
+ t.Fatalf("the refined estimate %+v (%d s), want %+v (%d s)", after, row.EstimateSeconds, want, wantSec)
+ case after.Copy >= before.Copy || int(row.EstimateSeconds) >= beforeSec:
+ t.Fatalf("staging did not shorten the copy: before %v (%d s), after %v (%d s)", before.Copy, beforeSec, after.Copy, row.EstimateSeconds)
+ case after.Land != before.Land || after.Promote != before.Promote || after.Stop != before.Stop || after.Open != before.Open:
+ t.Fatalf("staging changed a term it does not measure: before %+v, after %+v", before, after)
+ case after.Copy != 10*time.Second+ssRate(10*mib, 50*mib):
+ t.Fatalf("the refined copy %v is not the lane's median rate (50 MiB/s)", after.Copy)
+ }
+}
+
+// ssMarks writes every mark for id, a second apart from at, with the copy,
+// landing and replica bytes.
+func ssMarks(t *testing.T, st *Store, id string, at time.Time, skip ...string) {
+ t.Helper()
+ skipped := map[string]bool{}
+ for _, m := range skip {
+ skipped[m] = true
+ }
+ for i, m := range []struct {
+ mark string
+ bytes int64
+ }{
+ {"offline_at", 0}, {"stopped_at", 0}, {"seal_at", 0}, {"sealed_at", 40 * mib}, {"flip_at", 0},
+ {"landed_at", 120 * mib}, {"landed_replica", 100 * mib}, {"open_at", 0}, {"opened_at", 0}, {"first_ok_at", 0},
+ } {
+ if skipped[m.mark] {
+ continue
+ }
+ // Each mark is i*i seconds after at, so every phase has its own
+ // length.
+ if err := st.MarkSelfServeTiming(id, m.mark, at.Add(time.Duration(i*i)*time.Second), m.bytes); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+func TestTimingMarksAreFirstObservationTimes(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ row := ssLoad(t, st, ssID)
+ first, later := ssT0.Add(time.Minute), ssT0.Add(2*time.Minute)
+
+ s.selfServeMark(row, "offline_at", first, 0)
+ s.selfServeMark(row, "offline_at", later, 0)
+ if err := st.MarkSelfServeTiming(ssID, "sealed_at", first, 40*mib); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.MarkSelfServeTiming(ssID, "sealed_at", later, 99*mib); err != nil {
+ t.Fatal(err)
+ }
+ marks, err := st.SelfServeTimings(ssID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m := marks["offline_at"]; !m.At.Equal(first) || m.Bytes != 0 {
+ t.Fatalf("offline_at = %+v, want the first observation %v", m, first)
+ }
+ if m := marks["sealed_at"]; !m.At.Equal(first) || m.Bytes != 40*mib {
+ t.Fatalf("sealed_at = %+v, want the first observation's time and bytes", m)
+ }
+ // One timing event per mark, in the mark's own transaction.
+ timing := 0
+ for _, e := range ssEvents(t, st, ssID) {
+ if e.Phase == "requested:timing" {
+ timing++
+ }
+ }
+ if timing != 2 {
+ t.Fatalf("%d timing events, want one per mark: %+v", timing, ssEvents(t, st, ssID))
+ }
+ if err := st.MarkSelfServeTiming("mv_absent", "offline_at", first, 0); err == nil {
+ t.Fatal("a mark for an absent row was written")
+ }
+
+ // The landed mark goes with the advance that first sees the landing,
+ // and only if the advance wins.
+ ssPhase(t, st, ssID, ssLanding, ssT0.Add(time.Hour))
+ landed := ssT0.Add(3 * time.Minute)
+ if ok, err := st.AdvanceSelfServeToStep(ssID, ssSealing, ssLanded, "", "", time.Time{}, time.Time{}, "", landed,
+ ssMarkTx(ssID, "landed_at", landed, 120*mib)); err != nil || ok {
+ t.Fatalf("a lost advance = %v, %v", ok, err)
+ }
+ if marks, _ := st.SelfServeTimings(ssID); marks["landed_at"].At != (time.Time{}) {
+ t.Fatalf("a lost advance wrote the landed mark: %+v", marks["landed_at"])
+ }
+ if ok, err := st.AdvanceSelfServeToStep(ssID, ssLanding, ssLanded, "", "", time.Time{}, time.Time{}, "", landed,
+ ssMarkTx(ssID, "landed_at", landed, 120*mib)); err != nil || !ok {
+ t.Fatalf("the advance = %v, %v", ok, err)
+ }
+ if marks, _ := st.SelfServeTimings(ssID); !marks["landed_at"].At.Equal(landed) || marks["landed_at"].Bytes != 120*mib {
+ t.Fatalf("landed_at = %+v", marks["landed_at"])
+ }
+}
+
+func TestAbortedMovesContributeNoRates(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ ctx := context.Background()
+ finish := func(id, host, outcome string, skip ...string) *SelfServeMove {
+ t.Helper()
+ ssAdmit(t, st, ssMove(id, host, ssT0, "ie-1", "uk-1"))
+ ssMarks(t, st, id, ssT0, skip...)
+ if err := st.FinishSelfServeMove(id, outcome, "", "", ssT0.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ row := ssLoad(t, st, id)
+ for i := 0; i < 2; i++ { // the effects step again after a crash
+ if err := s.selfServeTimingsOnFinish(ctx, row); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return row
+ }
+ finish("mv_abortedabortedabortedabort", "a."+ssHost, provision.OfflineOutcomeAborted)
+ if got, err := st.MoveTimings("ie", "uk", "stop", false, 20); err != nil || len(got) != 0 {
+ t.Fatalf("an aborted move left timings: %+v, %v", got, err)
+ }
+
+ finish("mv_donedonedonedonedonedonedo", "b."+ssHost, provision.OfflineOutcomeDone)
+ // offline 0, stopped 1, seal 4, sealed 9, flip 16, landed 25, replica
+ // 36, open 49, opened 64, first ok 81 seconds.
+ want := map[string]MoveTiming{
+ "stop": {Seconds: 1},
+ "copy": {Seconds: 5, Bytes: 40 * mib},
+ "land": {Seconds: 9, Bytes: 120 * mib},
+ "open": {Seconds: 15},
+ "promote": {Seconds: 32, Bytes: 100 * mib},
+ }
+ for phase, w := range want {
+ got, err := st.MoveTimings("ie", "uk", phase, false, 20)
+ if err != nil || len(got) != 1 {
+ t.Fatalf("%s: %+v, %v; want one row", phase, got, err)
+ }
+ if g := got[0]; g.ID != "mv_donedonedonedonedonedonedo" || g.Seconds != w.Seconds || g.Bytes != w.Bytes || g.From != "ie" || g.To != "uk" {
+ t.Errorf("%s: %+v, want %+v", phase, g, w)
+ }
+ }
+
+ // A done move without a mark contributes no row for its phase.
+ finish("mv_nofirstoknofirstoknofirsto", "c."+ssHost, provision.OfflineOutcomeDone, "first_ok_at", "seal_at")
+ for phase, n := range map[string]int{"promote": 1, "copy": 1, "stop": 2, "land": 2, "open": 2} {
+ if got, _ := st.MoveTimings("ie", "uk", phase, false, 20); len(got) != n {
+ t.Errorf("%s: %d rows, want %d", phase, len(got), n)
+ }
+ }
+
+ // Rates come only from finished moves: five aborted moves with slow
+ // copies change no term.
+ before, _ := s.estimate("ie", "uk", 100*mib, 0, 0)
+ for i := 0; i < 5; i++ {
+ finish(fmt.Sprintf("mv_slowabort%02d", i), fmt.Sprintf("s%d.%s", i, ssHost), provision.OfflineOutcomeAborted)
+ }
+ if after, _ := s.estimate("ie", "uk", 100*mib, 0, 0); after != before {
+ t.Fatalf("aborted moves changed the estimate: %+v, then %+v", before, after)
+ }
+}
diff --git a/internal/console/selfserve_events_test.go b/internal/console/selfserve_events_test.go
new file mode 100644
index 000000000..bfea686de
--- /dev/null
+++ b/internal/console/selfserve_events_test.go
@@ -0,0 +1,256 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The events (self-serve moves plan 2 Task 9; spec §8): every phase
+// change, refusal, fact decision, lane probe, health probe and timing is
+// an event, sequenced per row in the transaction of the change; the pass
+// delivers each once as an audit row, and dead-letters one it cannot
+// deliver after five attempts; the operator's status shows the move and,
+// with ?log=1, its events.
+
+// ssDeliver is the pass's delivery of e.
+func ssDeliver(st *Store, e SelfServeEvent, crash bool) error {
+ return st.DeliverSelfServeEvent(context.Background(), e, func(tx *sql.Tx) error {
+ if err := auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{Action: "instance-move", TargetType: "instance", Target: e.Host, Message: e.Phase, Detail: e.Reason}); err != nil {
+ return err
+ }
+ if crash {
+ return errors.New("the console stopped after the audit insert")
+ }
+ return nil
+ })
+}
+
+func TestEventsAreSequencedAndDeliveredOnce(t *testing.T) {
+ st := ssStore(t)
+ now := ssT0.Add(time.Second)
+ ssAdmit(t, st, ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1"))
+ ssPhase(t, st, ssID, ssPrechecks, ssT0.Add(time.Hour))
+
+ // Two prechecks, each decided: two distinct facts events.
+ for i, nonce := range []string{"p1", "p2"} {
+ ssIssue(t, st, ssID, "precheck", ssPrechecks, nonce, now)
+ if ok, err := st.ConfirmSelfServeRequest(ssID, nonce, now); err != nil || !ok {
+ t.Fatalf("confirm %s = %v, %v", nonce, ok, err)
+ }
+ if ok, err := st.ConsumeSelfServeRequest(ssID, nonce, ssStepPrecheck, now.Add(time.Minute),
+ ssEventTx(ssID, ssPrechecks+":facts", now.Add(time.Duration(i)*time.Second), "", "passed")); err != nil || !ok {
+ t.Fatalf("consume %s = %v, %v", nonce, ok, err)
+ }
+ }
+ // A consume that loses its predicate inserts no event.
+ if ok, err := st.ConsumeSelfServeRequest(ssID, "p2", ssStepEstimate, now, ssEventTx(ssID, ssPrechecks+":facts", now, "", "passed")); err != nil || ok {
+ t.Fatalf("a second consume = %v, %v", ok, err)
+ }
+ // A refusal, a probe and a waiting reason are events too.
+ if _, err := st.AbortSelfServe(ssID, "the source box has no route for shop.example.com", now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := st.AbortSelfServe(ssID, "a second abort", now); err != nil {
+ t.Fatal(err)
+ }
+ if err := st.RecordSelfServeHealth(ssID, now, false, "the probe was answered 503"); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < 2; i++ {
+ if err := st.SetSelfServeReason(ssID, msgSelfServeWaitingEarlier); err != nil {
+ t.Fatal(err)
+ }
+ }
+ evs := ssEvents(t, st, ssID)
+ var phases []string
+ for i, e := range evs {
+ if e.Seq != int64(i+1) {
+ t.Errorf("event %d has seq %d", i, e.Seq)
+ }
+ phases = append(phases, e.Phase)
+ }
+ if want := "requested prechecks:facts prechecks:facts prechecks:refusal health:probe prechecks:waiting"; strings.Join(phases, " ") != want {
+ t.Fatalf("events %q, want %q", strings.Join(phases, " "), want)
+ }
+ if e := evs[3]; e.Reason != "the source box has no route for shop.example.com" {
+ t.Fatalf("the refusal's event %+v", e)
+ }
+ if e := evs[4]; e.Reason != "" || e.Detail != "failed: the probe was answered 503" {
+ t.Fatalf("the probe's event %+v", e)
+ }
+
+ // A delivery failure leaves the event undelivered and counts it.
+ pending, err := st.UnemittedSelfServeEvents(100)
+ if err != nil || len(pending) != len(evs) {
+ t.Fatalf("undelivered %d, %v; want %d", len(pending), err, len(evs))
+ }
+ first := pending[0]
+ if err := ssDeliver(st, first, true); err == nil {
+ t.Fatal("a crashed delivery succeeded")
+ }
+ if n := auditCount(t, st); n != 0 {
+ t.Fatalf("a crash after the audit insert left %d audit rows", n)
+ }
+ if left, _ := st.UnemittedSelfServeEvents(100); len(left) != len(evs) {
+ t.Fatal("a crashed delivery marked its event emitted")
+ }
+ var attempts int
+ if err := st.db.QueryRow(`SELECT attempts FROM self_serve_events WHERE id = ? AND seq = ?`, first.ID, first.Seq).Scan(&attempts); err != nil || attempts != 1 {
+ t.Fatalf("attempts = %d, %v; want 1", attempts, err)
+ }
+ // Delivered again, then a third time from a stale snapshot: one audit
+ // row.
+ for i := 0; i < 2; i++ {
+ if err := ssDeliver(st, first, false); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, e := range pending[1:] {
+ if err := ssDeliver(st, e, false); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if n := auditCount(t, st); n != len(evs) {
+ t.Fatalf("%d audit rows for %d events", n, len(evs))
+ }
+ if left, _ := st.UnemittedSelfServeEvents(100); len(left) != 0 {
+ t.Fatalf("undelivered after delivery: %+v", left)
+ }
+}
+
+func TestUndeliveredEventsStartThePass(t *testing.T) {
+ ctx := context.Background()
+ // stage leaves a finished row with nothing owed but its events.
+ stage := func(t *testing.T, fx *consoleFixture, acct string) {
+ t.Helper()
+ m := ssMove(ssID, ssHost, ssT0, "ie-1", "uk-1")
+ m.Account = acct
+ ssAdmit(t, fx.st, m)
+ if err := fx.st.FinishSelfServeMove(ssID, "done", "", "", ssT0); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.db.Exec(`UPDATE self_serve_moves SET effects_done = 1, release_pending = 0 WHERE id = ?`, ssID); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ t.Run("delivered, the pass stops", func(t *testing.T) {
+ fx, _ := ssPassFixture(t)
+ stage(t, fx, "bab")
+ fx.srv.selfServeScanAtStart(ctx)
+ if !fx.srv.selfServeMode() {
+ t.Fatal("undelivered events did not start the pass")
+ }
+ fx.srv.selfServeIteration(ctx)
+ if n := auditCount(t, fx.st); n != 2 {
+ t.Fatalf("%d audit rows, want the admission's and the outcome's", n)
+ }
+ fx.srv.selfServeIteration(ctx)
+ if fx.srv.selfServeMode() {
+ t.Fatal("recovery mode stayed on with every event delivered")
+ }
+ })
+
+ t.Run("undeliverable, dead-lettered after five attempts", func(t *testing.T) {
+ fx, _ := ssPassFixture(t)
+ stage(t, fx, "gone") // no such account: the audit row cannot be written
+ fx.srv.selfServeScanAtStart(ctx)
+ var logs strings.Builder
+ restore := captureLogTo(&logs)
+ for i := 0; i < selfServeDeliverAttempts-1; i++ {
+ fx.srv.selfServeIteration(ctx)
+ }
+ if left, _ := fx.st.UnemittedSelfServeEvents(10); len(left) != 2 || !fx.srv.selfServeMode() {
+ restore()
+ t.Fatalf("after %d failed attempts: %d undelivered, mode %v; want both still owed", selfServeDeliverAttempts-1, len(left), fx.srv.selfServeMode())
+ }
+ fx.srv.selfServeIteration(ctx)
+ fx.srv.selfServeIteration(ctx)
+ fx.srv.selfServeIteration(ctx)
+ restore()
+ if left, _ := fx.st.UnemittedSelfServeEvents(10); len(left) != 0 {
+ t.Fatalf("after %d attempts %d undelivered", selfServeDeliverAttempts, len(left))
+ }
+ for _, e := range ssEvents(t, fx.st, ssID) {
+ if !e.Emitted || !strings.HasPrefix(e.Detail, "undeliverable:") {
+ t.Fatalf("event %+v, want emitted with an undeliverable detail", e)
+ }
+ }
+ if n := strings.Count(logs.String(), "undeliverable"); n != 2 {
+ t.Fatalf("logged %d times, want once per event: %q", n, logs.String())
+ }
+ if n := auditCount(t, fx.st); n != 0 {
+ t.Fatalf("%d audit rows", n)
+ }
+ if fx.srv.selfServeMode() {
+ t.Fatal("recovery mode stayed on for dead-lettered events")
+ }
+ })
+}
+
+func TestOfflineStatusShowsSelfServeAndLog(t *testing.T) {
+ get := func(t *testing.T, w *ssWorld, query string) (offlineView, string) {
+ t.Helper()
+ code, out := w.req(http.MethodGet, query, "")
+ if code != http.StatusOK {
+ t.Fatalf("GET status%s = %d %s", query, code, out)
+ }
+ return w.view(code, out), out
+ }
+
+ t.Run("a self-serve move", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ row := w.prepared()
+ v, out := get(t, w, "")
+ if v.SelfServe == nil || v.SelfServe.ID != row.ID || v.SelfServe.State != "preparing" || v.SelfServe.Kind != provision.MoveKindOffline {
+ t.Fatalf("self_serve = %+v in %s", v.SelfServe, out)
+ }
+ if v.Events != nil || strings.Contains(out, `"events"`) {
+ t.Fatalf("events without ?log=1: %s", out)
+ }
+ if err := w.fx.st.MarkSelfServeTiming(row.ID, "offline_at", time.Now(), 0); err != nil {
+ t.Fatal(err)
+ }
+ // Delivered or not, the log shows every event, oldest first.
+ evs, err := w.fx.st.UnemittedSelfServeEvents(10)
+ if err != nil || len(evs) != 2 {
+ t.Fatalf("events %+v, %v", evs, err)
+ }
+ if err := ssDeliver(w.fx.st, evs[0], false); err != nil {
+ t.Fatal(err)
+ }
+ v, out = get(t, w, "?log=1")
+ if len(v.Events) != 2 || v.Events[0].Phase != ssRequested || v.Events[1].Phase != "requested:timing" || v.Events[1].Reason != "offline_at" || v.Events[0].At > v.Events[1].At {
+ t.Fatalf("events %+v in %s", v.Events, out)
+ }
+ w.queued(row)
+ w.edit(provision.OfflineQueuedToStaging(row.ID, time.Now(), provision.Receipt{}))
+ if v, out := get(t, w, "?log=1"); v.SelfServe == nil || v.SelfServe.ID != row.ID || v.State != provision.OfflineStaging || len(v.Events) != 2 {
+ t.Fatalf("staging: self_serve %+v events %+v in %s", v.SelfServe, v.Events, out)
+ }
+ })
+
+ t.Run("an operator move shows neither", func(t *testing.T) {
+ w := newSelfServeWorld(t, false)
+ w.ok("/stage", ssBody, provision.OfflineStaging)
+ _, out := get(t, w, "?log=1")
+ var keys map[string]json.RawMessage
+ if err := json.Unmarshal([]byte(out), &keys); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := keys["self_serve"]; ok {
+ t.Fatalf("an operator move shows self_serve: %s", out)
+ }
+ if _, ok := keys["events"]; ok {
+ t.Fatalf("an operator move shows events: %s", out)
+ }
+ })
+}
diff --git a/internal/e2e/selfserve/cut_test.go b/internal/e2e/selfserve/cut_test.go
index 5e00d65c7..489fd6605 100644
--- a/internal/e2e/selfserve/cut_test.go
+++ b/internal/e2e/selfserve/cut_test.go
@@ -142,6 +142,78 @@ func TestCutRunsEveryPhaseToDone(t *testing.T) {
}
}
+// TestCutRecordsTimingsAndDeliversEvents: a move run from the cut to done
+// leaves every timing mark at the console's first observation, in the
+// cut's order, with the copy's and the landing's bytes; the finaliser
+// derives the five move_timings rows on the region pair; and every event
+// (the phases, the fact decisions, the lane, the probes, the marks) is
+// delivered once, none dead-lettered.
+func TestCutRecordsTimingsAndDeliversEvents(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.toCut()
+ r := w.done(row.ID)
+ if r.Outcome != "done" {
+ t.Fatalf("finished %s", describe(r))
+ }
+ marks, err := w.cst.SelfServeTimings(row.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ order := []string{"offline_at", "stopped_at", "seal_at", "sealed_at", "flip_at", "landed_at", "open_at", "opened_at"}
+ for i, m := range order {
+ if marks[m].At.IsZero() {
+ t.Fatalf("no %s mark: %+v", m, marks)
+ }
+ if i > 0 && marks[m].At.Before(marks[order[i-1]].At) {
+ t.Fatalf("%s %v is before %s %v", m, marks[m].At, order[i-1], marks[order[i-1]].At)
+ }
+ }
+ switch {
+ case marks["first_ok_at"].At.Before(marks["open_at"].At) || marks["first_ok_at"].At.IsZero():
+ t.Fatalf("first_ok_at %v, open_at %v", marks["first_ok_at"].At, marks["open_at"].At)
+ case marks["landed_replica"].Bytes <= 0 || marks["landed_at"].Bytes <= marks["landed_replica"].Bytes:
+ t.Fatalf("landed_at %d bytes, landed_replica %d: want the files and the replica, and the replica", marks["landed_at"].Bytes, marks["landed_replica"].Bytes)
+ }
+ for _, phase := range []string{"stop", "copy", "land", "open", "promote"} {
+ got, err := w.cst.MoveTimings("", host.E2ETargetRegion, phase, false, 20)
+ if err != nil || len(got) != 1 || got[0].ID != row.ID {
+ t.Fatalf("%s timings %+v, %v; want the move's one row", phase, got, err)
+ }
+ }
+ if got, _ := w.cst.MoveTimings("", host.E2ETargetRegion, "land", true, 20); len(got) != 1 || got[0].Bytes != marks["landed_at"].Bytes {
+ t.Fatalf("land's rate row %+v, want the landed bytes %d", got, marks["landed_at"].Bytes)
+ }
+ // The terminal event is the last tick's: one more delivers it.
+ for i := 0; i < 3; i++ {
+ if left, err := w.cst.UnemittedSelfServeEvents(10); err == nil && len(left) == 0 {
+ break
+ }
+ w.tick()
+ }
+ evs, err := w.cst.SelfServeEvents(row.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ seen := map[string]int{}
+ for i, e := range evs {
+ switch {
+ case e.Seq != int64(i+1):
+ t.Fatalf("event %d has seq %d", i, e.Seq)
+ case !e.Emitted || strings.HasPrefix(e.Detail, "undeliverable:"):
+ t.Fatalf("event %+v not delivered", e)
+ }
+ seen[e.Phase]++
+ }
+ for _, p := range []string{"barrier:facts", "health:probe", "stopping:timing", "landed:timing", "done"} {
+ if seen[p] == 0 {
+ t.Errorf("no %s event in %v", p, seen)
+ }
+ }
+ if seen["done"] != 1 {
+ t.Errorf("%d terminal events", seen["done"])
+ }
+}
+
// openRecord is the move's open record.
func (w *jobWorld) openRecord(id string) moves.OpenRecord {
w.t.Helper()