| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os/exec" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/hostbox" |
| 14 | "github.com/carlosframework/platform/internal/hostclaim" |
| 15 | "github.com/carlosframework/platform/internal/hostguard" |
| 16 | "github.com/carlosframework/platform/internal/hostwrite" |
| 17 | "github.com/carlosframework/platform/internal/moves" |
| 18 | "github.com/carlosframework/platform/internal/provision" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | const ( |
| 24 | paHost = "shop.bab.oncarlos.com" |
| 25 | paOp = "mv_probeattempts" |
| 26 | ) |
| 27 | |
| 28 | // paClaim is the landing claim after the first probe's admission: uk-box |
| 29 | // executes at probing and is the writer of session 20 in generation paOp, |
| 30 | // admitted cold from the source's session 15. |
| 31 | func paClaim(t *testing.T) *hostclaim.Claim { |
| 32 | t.Helper() |
| 33 | at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) |
| 34 | inc, err := hostclaim.Incarnation("bab", "jam", paHost, at) |
| 35 | if err != nil { |
| 36 | t.Fatal(err) |
| 37 | } |
| 38 | c := &hostclaim.Claim{V: 2, Host: paHost, Gen: 20, |
| 39 | Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + paHost + ".json", Account: "bab", App: "jam", CreatedAt: at, Incarnation: inc, Location: "instances/" + paHost, Generation: paOp, Stamp: 18, Session: 20, WriterBox: "uk-box", |
| 40 | Admissions: []hostclaim.AdmissionRecord{ |
| 41 | {Session: 15, Box: "ie-box", Generation: "src", Predecessor: "", Region: "ie", Mode: "cold"}, |
| 42 | {Session: 20, Box: "uk-box", Generation: paOp, Predecessor: "src", Region: "uk", PredecessorRegion: "ie", Mode: "cold", Coverage: "sealed"}}}, |
| 43 | Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: paOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}} |
| 44 | if err := c.Validate(); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | return c |
| 48 | } |
| 49 | |
| 50 | type paFixture struct { |
| 51 | t *testing.T |
| 52 | claims release.Store |
| 53 | rec *provision.Record |
| 54 | ops *movesProbeAttempts |
| 55 | start OperationStart |
| 56 | ids int |
| 57 | admit int |
| 58 | dataDir string |
| 59 | } |
| 60 | |
| 61 | func newPAFixture(t *testing.T) *paFixture { |
| 62 | t.Helper() |
| 63 | f := &paFixture{t: t, claims: release.NewDirStore(t.TempDir()), dataDir: t.TempDir(), admit: 2} |
| 64 | c := paClaim(t) |
| 65 | f.putClaim(c) |
| 66 | o := f.read() |
| 67 | f.rec = &provision.Record{Host: paHost, Account: "bab", App: "jam", Region: "uk", |
| 68 | Move: &provision.MoveState{ID: paOp, Incarnation: c.Holder.Incarnation, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: mpPins(), Landing: &provision.MoveLandingJournal{Attempts: 1}}} |
| 69 | f.ops = f.attempts() |
| 70 | in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Executable: "/opt/carlos/live/" + paHost, ExecutableSHA256: strings.Repeat("a", 64), Args: []string{"--socket", "/s"}, Env: []string{"CARLOS_MOVE_PROBE=1"}, ReadyPath: "/ready"} |
| 71 | h := o.Claim.Holder |
| 72 | f.start = OperationStart{Admitted: o, Policy: activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}, Inputs: in, |
| 73 | Owner: hostguard.FenceTarget{Host: paHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Op: paOp, Phase: hostclaim.PhaseLandingProbing, Box: "uk-box", Region: "uk", Enrollment: "enr-uk", BootID: "boot-1", ControllerID: "ctl-1", ClaimETag: o.ETag}} |
| 74 | return f |
| 75 | } |
| 76 | |
| 77 | // attempts is a fresh controller's provider over the same durable state. |
| 78 | func (f *paFixture) attempts() *movesProbeAttempts { |
| 79 | return &movesProbeAttempts{Claims: f.claims, DataDir: f.dataDir, Settle: 20 * time.Millisecond, Ready: time.Second, |
| 80 | Records: func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil }, |
| 81 | Admitted: func(context.Context, *provision.Record) (int, error) { return f.admit, nil }, |
| 82 | NewID: func() string { f.ids++; return "id-" + strings.Repeat("x", f.ids) }, |
| 83 | Now: func() time.Time { return time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC) }} |
| 84 | } |
| 85 | |
| 86 | func (f *paFixture) putClaim(c *hostclaim.Claim) { |
| 87 | f.t.Helper() |
| 88 | body, err := json.Marshal(c) |
| 89 | if err != nil { |
| 90 | f.t.Fatal(err) |
| 91 | } |
| 92 | key, err := hostclaim.Key(paHost) |
| 93 | if err != nil { |
| 94 | f.t.Fatal(err) |
| 95 | } |
| 96 | etag := "" |
| 97 | if _, current, err := f.claims.Get(f.t.Context(), key); err == nil { |
| 98 | etag = current |
| 99 | } |
| 100 | if _, err := f.claims.Put(f.t.Context(), key, body, etag); err != nil { |
| 101 | f.t.Fatal(err) |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | func (f *paFixture) read() hostclaim.Observed { |
| 106 | f.t.Helper() |
| 107 | o, err := hostclaim.Read(f.t.Context(), f.claims, paHost) |
| 108 | if err != nil { |
| 109 | f.t.Fatal(err) |
| 110 | } |
| 111 | return o |
| 112 | } |
| 113 | |
| 114 | // OpenProbe journals the first start's time once per record attempt and |
| 115 | // returns it to every later start, a restarted controller included; the |
| 116 | // next record attempt opens its own window. |
| 117 | func TestMovesProbeAttemptsOpenProbeIsJournalledOnce(t *testing.T) { |
| 118 | f := newPAFixture(t) |
| 119 | ctx := t.Context() |
| 120 | t1 := time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC) |
| 121 | got, err := f.ops.OpenProbe(ctx, f.start, t1) |
| 122 | if err != nil || !got.Equal(t1) { |
| 123 | t.Fatalf("open: %v %v", got, err) |
| 124 | } |
| 125 | if got, err := f.ops.OpenProbe(ctx, f.start, t1.Add(5*time.Second)); err != nil || !got.Equal(t1) { |
| 126 | t.Fatalf("a second start: %v %v", got, err) |
| 127 | } |
| 128 | if got, err := f.attempts().OpenProbe(ctx, f.start, t1.Add(9*time.Second)); err != nil || !got.Equal(t1) { |
| 129 | t.Fatalf("after a controller restart: %v %v", got, err) |
| 130 | } |
| 131 | // The next record attempt is a new cold admission: the same session |
| 132 | // never opens its window. |
| 133 | f.rec.Move.Landing.Attempts = 2 |
| 134 | t2 := t1.Add(time.Minute) |
| 135 | if _, err := f.ops.OpenProbe(ctx, f.start, t2); !errors.Is(err, hostbox.ErrChanged) { |
| 136 | t.Fatalf("the next record attempt on the same session: %v", err) |
| 137 | } |
| 138 | next := f.readmit(21) |
| 139 | if got, err := f.ops.OpenProbe(ctx, next, t2); err != nil || !got.Equal(t2) { |
| 140 | t.Fatalf("the next record attempt's admission: %v %v", got, err) |
| 141 | } |
| 142 | j, err := moves.ReadJournal(f.dataDir, paHost, paOp) |
| 143 | if err != nil || !j.Done("probe_opened_1") || !j.Done("probe_opened_2") || j.Steps["probe_session_20"] != "1" || j.Steps["probe_session_21"] != "2" { |
| 144 | t.Fatalf("journal: %+v %v", j, err) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | // readmit stores the claim after a new cold admission at session and |
| 149 | // returns that admission's start. |
| 150 | func (f *paFixture) readmit(session uint64) OperationStart { |
| 151 | f.t.Helper() |
| 152 | c := paClaim(f.t) |
| 153 | c.Holder.Session, c.Gen = session, session |
| 154 | c.Holder.Generation = "fresh" |
| 155 | c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: session, Box: "uk-box", Generation: "fresh", Predecessor: paOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"}) |
| 156 | f.putClaim(c) |
| 157 | s := f.start |
| 158 | s.Admitted = f.read() |
| 159 | s.Owner.Session, s.Owner.ClaimETag = session, s.Admitted.ETag |
| 160 | return s |
| 161 | } |
| 162 | |
| 163 | // A session is bound to the record attempt it first journalled (at |
| 164 | // Reserve(0) or OpenProbe): once the record's index moves past it, every |
| 165 | // method refuses and CurrentAttempt answers nothing, even across a |
| 166 | // controller restart. |
| 167 | func TestMovesProbeAttemptsBindTheSessionToOneRecordAttempt(t *testing.T) { |
| 168 | f := newPAFixture(t) |
| 169 | ctx := t.Context() |
| 170 | if err := f.ops.Reserve(ctx, f.start, 0); err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | if f.ops.CurrentAttempt(f.start).ID == "" { |
| 174 | t.Fatal("no reservation") |
| 175 | } |
| 176 | f.rec.Move.Landing.Attempts = 2 |
| 177 | if a := f.ops.CurrentAttempt(f.start); a.ID != "" { |
| 178 | t.Fatalf("a reservation of record attempt 1 answered for attempt 2: %+v", a) |
| 179 | } |
| 180 | if err := f.ops.Failed(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) { |
| 181 | t.Fatalf("Failed: %v", err) |
| 182 | } |
| 183 | if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) { |
| 184 | t.Fatalf("Reserve(0): %v", err) |
| 185 | } |
| 186 | if _, err := f.attempts().OpenProbe(ctx, f.start, time.Now()); !errors.Is(err, hostbox.ErrChanged) { |
| 187 | t.Fatalf("OpenProbe after a restart: %v", err) |
| 188 | } |
| 189 | if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); !errors.Is(err, hostbox.ErrChanged) { |
| 190 | t.Fatalf("Healthy: %v", err) |
| 191 | } |
| 192 | next := f.readmit(21) |
| 193 | if err := f.ops.Reserve(ctx, next, 0); err != nil { |
| 194 | t.Fatalf("the new admission: %v", err) |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | // Reserve, CurrentAttempt and Failed are bound to the record's attempt |
| 199 | // index and the admitted session: the reservation carries the owner's |
| 200 | // tuple and the launch digest, is stable between reservations, and a next |
| 201 | // start is reserved only after the last one was settled. |
| 202 | func TestMovesProbeAttemptsReserveBindsTheRecordsAttempt(t *testing.T) { |
| 203 | f := newPAFixture(t) |
| 204 | ctx := t.Context() |
| 205 | if a := f.ops.CurrentAttempt(f.start); a.ID != "" { |
| 206 | t.Fatalf("an attempt before any reservation: %+v", a) |
| 207 | } |
| 208 | if err := f.ops.Reserve(ctx, f.start, 0); err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | first := f.ops.CurrentAttempt(f.start) |
| 212 | digest, _ := ClaimLaunchDigest(f.start.Inputs) |
| 213 | o := f.start.Owner |
| 214 | if first.ID == "" || first.InvocationID == "" || first.ID == first.InvocationID || first.Session != o.Session || first.Box != o.Box || first.Enrollment != o.Enrollment || first.BootID != o.BootID || first.ControllerID != o.ControllerID || first.LaunchInputs != digest { |
| 215 | t.Fatalf("reservation: %+v", first) |
| 216 | } |
| 217 | if again := f.ops.CurrentAttempt(f.start); again != first { |
| 218 | t.Fatalf("CurrentAttempt moved: %+v", again) |
| 219 | } |
| 220 | if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != first { |
| 221 | t.Fatalf("Reserve(0) again is a validation: %v", err) |
| 222 | } |
| 223 | if err := f.ops.Reserve(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) { |
| 224 | t.Fatalf("start 1 before start 0 settled: %v", err) |
| 225 | } |
| 226 | if err := f.ops.Failed(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) { |
| 227 | t.Fatalf("settling a start that was not reserved: %v", err) |
| 228 | } |
| 229 | if err := f.ops.Failed(ctx, f.start, 0); err != nil { |
| 230 | t.Fatal(err) |
| 231 | } |
| 232 | if err := f.ops.Reserve(ctx, f.start, 1); err != nil { |
| 233 | t.Fatal(err) |
| 234 | } |
| 235 | second := f.ops.CurrentAttempt(f.start) |
| 236 | if second.ID == first.ID || second.InvocationID == first.InvocationID || second.Session != o.Session || second.LaunchInputs != digest { |
| 237 | t.Fatalf("start 1: %+v", second) |
| 238 | } |
| 239 | changed := f.start |
| 240 | changed.Inputs.Env = []string{"CARLOS_MOVE_PROBE=1", "OTHER=1"} |
| 241 | if err := f.ops.Failed(ctx, f.start, 1); err != nil { |
| 242 | t.Fatal(err) |
| 243 | } |
| 244 | if err := f.ops.Reserve(ctx, changed, 2); !errors.Is(err, hostbox.ErrChanged) { |
| 245 | t.Fatalf("other launch inputs: %v", err) |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // Every method refuses once the record or the claim no longer names the |
| 250 | // admitted probe: the record's attempt index outside what T11 admitted, a |
| 251 | // rollback reason, the record past probing, another session. |
| 252 | func TestMovesProbeAttemptsRefuseWhenTheProbeMovedOn(t *testing.T) { |
| 253 | for name, mutate := range map[string]func(*paFixture){ |
| 254 | "no attempt counted": func(f *paFixture) { f.rec.Move.Landing = nil }, |
| 255 | "beyond the admitted": func(f *paFixture) { f.rec.Move.Landing.Attempts = 3 }, |
| 256 | "none admitted": func(f *paFixture) { f.admit = 0 }, |
| 257 | "a rollback is marked": func(f *paFixture) { f.rec.Move.Reason = provision.RollbackProbeAttempts }, |
| 258 | "the record is probed": func(f *paFixture) { f.rec.Move.LandingPhase = provision.LandProbed }, |
| 259 | "another move": func(f *paFixture) { f.rec.Move.ID = "mv_other" }, |
| 260 | "another session": func(f *paFixture) { |
| 261 | c := paClaim(f.t) |
| 262 | c.Holder.Session, c.Gen = 21, 21 |
| 263 | c.Holder.Admissions[1].Session = 21 |
| 264 | f.putClaim(c) |
| 265 | }, |
| 266 | "the claim is probed": func(f *paFixture) { |
| 267 | c := paClaim(f.t) |
| 268 | c.Op.Phase = hostclaim.PhaseLandingProbed |
| 269 | f.putClaim(c) |
| 270 | }, |
| 271 | "a committed wake": func(f *paFixture) { f.start.Policy.Probe = false }, |
| 272 | } { |
| 273 | t.Run(name, func(t *testing.T) { |
| 274 | f := newPAFixture(t) |
| 275 | mutate(f) |
| 276 | ctx := t.Context() |
| 277 | if _, err := f.ops.OpenProbe(ctx, f.start, time.Now()); err == nil { |
| 278 | t.Fatal("OpenProbe") |
| 279 | } |
| 280 | if err := f.ops.Reserve(ctx, f.start, 0); err == nil { |
| 281 | t.Fatal("Reserve") |
| 282 | } |
| 283 | if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); err == nil { |
| 284 | t.Fatal("Healthy") |
| 285 | } |
| 286 | }) |
| 287 | } |
| 288 | } |
| 289 | |
| 290 | // Healthy is the 20 s settle with the process alive, then the second |
| 291 | // readiness request (moves §7.7); the start is journalled healthy only |
| 292 | // after both. |
| 293 | func TestMovesProbeAttemptsHealthySettlesAndAsksAgain(t *testing.T) { |
| 294 | f := newPAFixture(t) |
| 295 | ctx := t.Context() |
| 296 | if err := f.ops.Reserve(ctx, f.start, 0); err != nil { |
| 297 | t.Fatal(err) |
| 298 | } |
| 299 | cmd := exec.Command("sleep", "30") |
| 300 | if err := cmd.Start(); err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | t.Cleanup(func() { cmd.Process.Kill(); cmd.Wait() }) |
| 304 | began := time.Now() |
| 305 | asked := 0 |
| 306 | if err := f.ops.Healthy(ctx, f.start, 0, cmd, func(context.Context) error { |
| 307 | asked++ |
| 308 | if time.Since(began) < f.ops.Settle { |
| 309 | t.Error("asked before the settle") |
| 310 | } |
| 311 | return nil |
| 312 | }); err != nil || asked != 1 { |
| 313 | t.Fatalf("healthy: %v asked=%d", err, asked) |
| 314 | } |
| 315 | j, err := moves.ReadJournal(f.dataDir, paHost, paOp) |
| 316 | if err != nil || !strings.Contains(j.Steps["probe_start_1"], `"state":"healthy"`) { |
| 317 | t.Fatalf("journal: %+v %v", j, err) |
| 318 | } |
| 319 | |
| 320 | // A second request that fails, or a process that died in the settle, |
| 321 | // is not healthy. |
| 322 | g := newPAFixture(t) |
| 323 | if err := g.ops.Reserve(ctx, g.start, 0); err != nil { |
| 324 | t.Fatal(err) |
| 325 | } |
| 326 | refused := errors.New("503") |
| 327 | if err := g.ops.Healthy(ctx, g.start, 0, cmd, func(context.Context) error { return refused }); !errors.Is(err, refused) { |
| 328 | t.Fatalf("a failing second request: %v", err) |
| 329 | } |
| 330 | dead := exec.Command("/bin/true") |
| 331 | if err := dead.Run(); err != nil { |
| 332 | t.Fatal(err) |
| 333 | } |
| 334 | if err := g.ops.Healthy(ctx, g.start, 0, dead, func(context.Context) error { return nil }); err == nil { |
| 335 | t.Fatal("a dead process was healthy") |
| 336 | } |
| 337 | if j, _ := moves.ReadJournal(g.dataDir, paHost, paOp); strings.Contains(j.Steps["probe_start_1"], "healthy") { |
| 338 | t.Fatalf("journalled healthy: %v", j.Steps) |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | // Task 7: the committed wake's starts are journalled per admitted session, |
| 343 | // with no record attempt (the probe's are over); Healthy accepts at once. |
| 344 | func TestMovesProbeAttemptsCommittedWake(t *testing.T) { |
| 345 | f := newPAFixture(t) |
| 346 | ctx := t.Context() |
| 347 | c := paClaim(t) |
| 348 | c.Op.Phase = hostclaim.PhaseLandingCommitted |
| 349 | f.putClaim(c) |
| 350 | f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, "" |
| 351 | s := f.start |
| 352 | s.Admitted, s.Policy.Probe = f.read(), false |
| 353 | s.Owner.ClaimETag = s.Admitted.ETag |
| 354 | if err := f.ops.Reserve(ctx, s, 0); err != nil { |
| 355 | t.Fatal(err) |
| 356 | } |
| 357 | first := f.ops.CurrentAttempt(s) |
| 358 | if first.ID == "" || first.Session != s.Owner.Session { |
| 359 | t.Fatalf("reservation: %+v", first) |
| 360 | } |
| 361 | if err := f.ops.Failed(ctx, s, 0); err != nil { |
| 362 | t.Fatal(err) |
| 363 | } |
| 364 | if err := f.ops.Reserve(ctx, s, 1); err != nil || f.ops.CurrentAttempt(s).ID == first.ID { |
| 365 | t.Fatalf("start 1: %v", err) |
| 366 | } |
| 367 | asked := 0 |
| 368 | if err := f.ops.Healthy(ctx, s, 1, nil, func(context.Context) error { asked++; return nil }); err != nil || asked != 0 { |
| 369 | t.Fatalf("healthy: %v asked=%d", err, asked) |
| 370 | } |
| 371 | f.rec.Move.State = provision.MoveLanding |
| 372 | if err := f.ops.Reserve(ctx, s, 0); err == nil { |
| 373 | t.Fatal("a committed wake's start over a landing record") |
| 374 | } |
| 375 | } |
| 376 | |
| 377 | func recoveryPAFixture(t *testing.T, kind string) *paFixture { |
| 378 | t.Helper() |
| 379 | f := newPAFixture(t) |
| 380 | c := f.read().Claim |
| 381 | c.Op.Kind, c.Op.Phase = kind, hostclaim.PhaseProbing |
| 382 | c.Op.Exec, c.Holder.WriterBox = "ie-box", "ie-box" |
| 383 | c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing} |
| 384 | f.rec.Move.State = provision.MoveReverting |
| 385 | if kind == hostclaim.KindRestoring { |
| 386 | c.Op.Intent["rollback"] = plan10J2Rollback |
| 387 | f.rec.Move.State = provision.MoveRollingBack |
| 388 | f.rec.Move.Rollback = &provision.MoveRollback{ID: plan10J2Rollback, SourcePhase: provision.RbSourceProbing} |
| 389 | } |
| 390 | f.putClaim(c) |
| 391 | f.start.Admitted = f.read() |
| 392 | f.start.Owner.Box = "ie-box" |
| 393 | f.start.Owner.Phase = hostclaim.PhaseProbing |
| 394 | return f |
| 395 | } |
| 396 | |
| 397 | func TestRecoveryProbeAttemptsDurableBudget(t *testing.T) { |
| 398 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 399 | t.Run(kind, func(t *testing.T) { |
| 400 | f := recoveryPAFixture(t, kind) |
| 401 | for n := 0; n < 9; n++ { |
| 402 | // Three starts per wake, then a new admitted session and provider. |
| 403 | local := n % 3 |
| 404 | if local == 0 && n > 0 { |
| 405 | c := f.read().Claim |
| 406 | c.Gen++ |
| 407 | c.Holder.Session = c.Gen |
| 408 | f.putClaim(c) |
| 409 | f.start.Admitted = f.read() |
| 410 | f.start.Owner.Session = c.Gen |
| 411 | f.ops = f.attempts() |
| 412 | } |
| 413 | if _, err := f.ops.OpenProbe(t.Context(), f.start, time.Now()); err != nil { |
| 414 | t.Fatal(err) |
| 415 | } |
| 416 | if err := f.ops.Reserve(t.Context(), f.start, local); err != nil { |
| 417 | t.Fatalf("start %d: %v", n+1, err) |
| 418 | } |
| 419 | if f.ops.CurrentAttempt(f.start).ID == "" { |
| 420 | t.Fatal("no launch reservation") |
| 421 | } |
| 422 | if local == 0 { |
| 423 | if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil { |
| 424 | t.Fatal("idempotent reservation", err) |
| 425 | } |
| 426 | } |
| 427 | err := f.ops.Failed(t.Context(), f.start, local) |
| 428 | if n == 8 { |
| 429 | if !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 430 | t.Fatalf("ninth failure: %v", err) |
| 431 | } |
| 432 | } else if err != nil { |
| 433 | t.Fatal(err) |
| 434 | } |
| 435 | } |
| 436 | f.ops = f.attempts() |
| 437 | if err := f.ops.Reserve(t.Context(), f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 438 | t.Fatalf("restart lost count: %v", err) |
| 439 | } |
| 440 | }) |
| 441 | } |
| 442 | } |
| 443 | |
| 444 | func TestRecoveryProbeAttemptsOperatorRetryBudget(t *testing.T) { |
| 445 | t.Parallel() |
| 446 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 447 | t.Run(kind, func(t *testing.T) { |
| 448 | ctx := t.Context() |
| 449 | f := recoveryPAFixture(t, kind) |
| 450 | f.rec.Schema, f.rec.Slug, f.rec.Channel = 1, "shop", "stable" |
| 451 | f.rec.CreatedAt = f.start.Admitted.Claim.Holder.CreatedAt |
| 452 | if err := provision.CreateRecord(ctx, f.claims, *f.rec); err != nil { |
| 453 | t.Fatal(err) |
| 454 | } |
| 455 | restart := func() { |
| 456 | f.ops = f.attempts() |
| 457 | f.ops.Records = func(ctx context.Context, account, app, host string) (*provision.Record, error) { |
| 458 | return provision.GetRecord(ctx, f.claims, account, app, host) |
| 459 | } |
| 460 | } |
| 461 | restart() |
| 462 | g := provision.MoveGuard{ID: f.rec.Move.ID, Incarnation: f.rec.Move.Incarnation} |
| 463 | transition := func(mut func(*provision.Record) error) error { |
| 464 | _, err := provision.SetRecordMove(ctx, f.claims, f.rec.Account, f.rec.App, paHost, mut) |
| 465 | return err |
| 466 | } |
| 467 | newSession := func() { |
| 468 | c := f.read().Claim |
| 469 | c.Gen++ |
| 470 | c.Holder.Session = c.Gen |
| 471 | f.putClaim(c) |
| 472 | f.start.Admitted = f.read() |
| 473 | f.start.Owner.Session = c.Gen |
| 474 | restart() |
| 475 | } |
| 476 | var previous hostclaim.WakeRetryAttempt |
| 477 | for budget := 0; budget < 3; budget++ { |
| 478 | if budget > 0 { |
| 479 | if err := transition(provision.MoveMarkStuck(g, provision.StuckRecoveryProbeAttempts)); err != nil { |
| 480 | t.Fatal(err) |
| 481 | } |
| 482 | if err := transition(provision.MoveRetryRecovery(g)); err != nil { |
| 483 | t.Fatal(err) |
| 484 | } |
| 485 | if got := f.ops.CurrentAttempt(f.start); got.ID != "" { |
| 486 | t.Fatal("old reservation survived retry", got) |
| 487 | } |
| 488 | if err := f.ops.Failed(ctx, f.start, 2); !errors.Is(err, hostbox.ErrChanged) { |
| 489 | t.Fatalf("old failure settled after retry: %v", err) |
| 490 | } |
| 491 | } |
| 492 | rec, err := provision.GetRecord(ctx, f.claims, f.rec.Account, f.rec.App, paHost) |
| 493 | if err != nil || rec.Move.RecoveryRetryEpoch != uint64(budget) { |
| 494 | t.Fatalf("retry epoch not persisted: record=%+v err=%v", rec, err) |
| 495 | } |
| 496 | for n := 0; n < 9; n++ { |
| 497 | local := n % 3 |
| 498 | if local == 0 && n > 0 { |
| 499 | newSession() |
| 500 | } |
| 501 | if err := f.ops.Reserve(ctx, f.start, local); err != nil { |
| 502 | t.Fatalf("budget %d start %d: %v", budget, n+1, err) |
| 503 | } |
| 504 | attempt := f.ops.CurrentAttempt(f.start) |
| 505 | if attempt.ID == "" || attempt.ID == previous.ID { |
| 506 | t.Fatalf("start has no fresh reservation: %+v", attempt) |
| 507 | } |
| 508 | previous = attempt |
| 509 | if local == 0 { |
| 510 | restart() |
| 511 | if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != attempt { |
| 512 | t.Fatalf("reservation replay after restart: %v", err) |
| 513 | } |
| 514 | } |
| 515 | if budget > 0 { |
| 516 | if err := transition(provision.MoveRetryRecovery(g)); !errors.Is(err, provision.ErrMoveConflict) { |
| 517 | t.Fatalf("retry replay: %v", err) |
| 518 | } |
| 519 | } |
| 520 | err := f.ops.Failed(ctx, f.start, local) |
| 521 | if n == 8 { |
| 522 | if !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 523 | t.Fatalf("budget %d ninth failure: %v", budget, err) |
| 524 | } |
| 525 | } else if err != nil { |
| 526 | t.Fatal(err) |
| 527 | } |
| 528 | } |
| 529 | previousProvider := f.ops |
| 530 | restart() |
| 531 | if err := f.ops.Reserve(ctx, f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 532 | t.Fatalf("budget %d restart lost exhaustion: %v", budget, err) |
| 533 | } |
| 534 | f.ops = previousProvider |
| 535 | } |
| 536 | newSession() |
| 537 | if err := transition(provision.MoveRetryRecovery(g)); !errors.Is(err, provision.ErrMoveConflict) { |
| 538 | t.Fatalf("exhausted retry replay: %v", err) |
| 539 | } |
| 540 | if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 541 | t.Fatalf("retry replay or fresh session granted extra starts: %v", err) |
| 542 | } |
| 543 | }) |
| 544 | } |
| 545 | } |
| 546 | |
| 547 | func TestRecoveryProbeAttemptsLegacyBudget(t *testing.T) { |
| 548 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 549 | t.Run(kind, func(t *testing.T) { |
| 550 | f := recoveryPAFixture(t, kind) |
| 551 | rollback := paOp |
| 552 | if kind == hostclaim.KindRestoring { |
| 553 | rollback = plan10J2Rollback |
| 554 | } |
| 555 | raw, err := json.Marshal([]string{paOp, rollback, f.rec.Move.Incarnation}) |
| 556 | if err != nil { |
| 557 | t.Fatal(err) |
| 558 | } |
| 559 | j, err := moves.ReadJournal(f.dataDir, paHost, paOp) |
| 560 | if err != nil { |
| 561 | t.Fatal(err) |
| 562 | } |
| 563 | j.Mark("recovery_attempts_"+string(raw), "9") |
| 564 | if err := j.Commit(f.dataDir); err != nil { |
| 565 | t.Fatal(err) |
| 566 | } |
| 567 | if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) { |
| 568 | t.Fatalf("legacy exhaustion lost: %v", err) |
| 569 | } |
| 570 | }) |
| 571 | } |
| 572 | } |
| 573 | |
| 574 | func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) { |
| 575 | f := recoveryPAFixture(t, hostclaim.KindRestoring) |
| 576 | w := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }} |
| 577 | if w.recoveryAttemptsReady() { |
| 578 | t.Fatal("landing-only provider admitted recovery") |
| 579 | } |
| 580 | w.Operations = f.ops |
| 581 | if !w.recoveryAttemptsReady() { |
| 582 | t.Fatal("recovery provider not ready") |
| 583 | } |
| 584 | w.Operations = claimOperationRouter{hostclaim.KindRestoring: {Attempts: f.ops, ProbeReady: w.ProbeReady}} |
| 585 | if w.recoveryAttemptsReady() { |
| 586 | t.Fatal("missing revert provider admitted") |
| 587 | } |
| 588 | if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil { |
| 589 | t.Fatal(err) |
| 590 | } |
| 591 | oldKey := recoveryAttemptKey(f.start, 0) |
| 592 | c := f.read().Claim |
| 593 | c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz" |
| 594 | f.putClaim(c) |
| 595 | f.start.Admitted = f.read() |
| 596 | f.rec.Move.Rollback.ID = c.Op.Intent["rollback"] |
| 597 | if recoveryAttemptKey(f.start, 0) == oldKey { |
| 598 | t.Fatal("rollback not bound") |
| 599 | } |
| 600 | if got := f.ops.CurrentAttempt(f.start); got.ID != "" { |
| 601 | t.Fatal("reservation escaped its rollback") |
| 602 | } |
| 603 | if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil { |
| 604 | t.Fatal(err) |
| 605 | } |
| 606 | j, err := moves.ReadJournal(f.dataDir, paHost, paOp) |
| 607 | if err != nil { |
| 608 | t.Fatal(err) |
| 609 | } |
| 610 | if j.Steps[oldKey] != "1" || j.Steps[recoveryAttemptKey(f.start, 0)] != "1" { |
| 611 | t.Fatal("rollback budgets combined") |
| 612 | } |
| 613 | for _, mutate := range []func(*OperationStart){ |
| 614 | func(s *OperationStart) { s.Admitted.Claim.Holder.Incarnation = "other" }, |
| 615 | func(s *OperationStart) { s.Admitted.Claim.Op.ID = "other" }, |
| 616 | } { |
| 617 | copied := f.start |
| 618 | raw, _ := json.Marshal(f.start.Admitted.Claim) |
| 619 | var claim hostclaim.Claim |
| 620 | _ = json.Unmarshal(raw, &claim) |
| 621 | copied.Admitted.Claim = &claim |
| 622 | mutate(&copied) |
| 623 | if recoveryAttemptKey(copied, 0) == recoveryAttemptKey(f.start, 0) { |
| 624 | t.Fatal("budget key omitted identity") |
| 625 | } |
| 626 | } |
| 627 | } |
| 628 | |
| 629 | func TestRecoveryProbeAttemptsRefuseChangedRollback(t *testing.T) { |
| 630 | f := recoveryPAFixture(t, hostclaim.KindRestoring) |
| 631 | c := f.read().Claim |
| 632 | c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz" |
| 633 | f.putClaim(c) |
| 634 | if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, hostbox.ErrChanged) { |
| 635 | t.Fatalf("changed rollback admitted: %v", err) |
| 636 | } |
| 637 | } |
| 638 | |
amadan