amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ed25519"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10 "strconv"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/activator"
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/hostwrite"
18 "github.com/carlosframework/platform/internal/lease"
19 "github.com/carlosframework/platform/internal/moves"
20 "github.com/carlosframework/platform/internal/park"
21 "github.com/carlosframework/platform/internal/provision"
22 "github.com/carlosframework/platform/internal/registry"
23 "github.com/carlosframework/platform/internal/release"
24)
25
26// The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6). The
27// fixture is the cut harness driven to a revert, either from stopping (the
28// stop evidence never held: the claim still moving under the source, no
29// baselines on the record) or from stopped after execution was given up
30// (the claim reverting, the baselines recorded). Its row follows the holds
31// the cut and the revert write; the hold's status change, the open and the
32// probe (Task 9, BLOCKED-ON-CLAIM C1) are faked; the target region's box
33// (uk-box) runs the lane park through the real park engine over the same
34// buckets with a fake lease table. Everything else is the real revert: the
35// cut's evidence and giveUpExecution, revertClaim, hostclaim's and
36// provision's own transitions.
37
38type revertHarness struct {
39 *cutHarness
40 tdir string
41 tleases *fakeParkLeases
42 tw *movesWorker
43 probe func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
44 probes []recoveryProbeRequest
45 restatus []string // status:op of each status change asked for
46 opens []string // the op of each open asked for
47 events []string
48 rvAt string
49 rvDone bool
50 rvHits []string
51}
52
53func newRevertHarness(t *testing.T, opts ...p1Option) *revertHarness {
54 t.Helper()
55 return newRevertHarnessOn(t, newCutHarness(t, opts...))
56}
57
58// newRevertHarnessOn is newRevertHarness over a cut harness the caller
59// built (a landing harness's, whose source replica is a real one).
60func newRevertHarnessOn(t *testing.T, ch *cutHarness) *revertHarness {
61 t.Helper()
62 h := &revertHarness{cutHarness: ch, tdir: t.TempDir(), tleases: &fakeParkLeases{}}
63 // The row follows the cut's hold, as holdMoving keeps it: held moving
64 // under the op, a row held by another op refused.
65 h.cfg.HoldMoving = func(host, op string) error {
66 h.holds = append(h.holds, host+"="+op)
67 if h.route.HoldOp != "" && h.route.HoldOp != op {
68 return fmt.Errorf("%s is held %s by %s", host, h.route.Status, h.route.HoldOp)
69 }
70 if h.route.HoldOp == "" {
71 h.route.Status, h.route.HoldOp = "moving", op
72 }
73 return nil
74 }
75 return h
76}
77
78// fromStopping drives the cut to a revert at stopping: the evidence never
79// held, the deadline passed, the record reverting with no baselines and the
80// claim still moving under the source.
81func (h *revertHarness) fromStopping(t *testing.T) {
82 t.Helper()
83 h.ready(t)
84 h.ev.stopped = false
85 h.cut(t)
86 h.pass(t)
87 h.clock.Advance(2 * time.Minute)
88 h.pass(t)
89 m := h.move(t)
90 if m.State != provision.MoveReverting || m.CutPhase != provision.CutStopping || m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
91 t.Fatalf("the fixture did not revert at stopping: %+v", m)
92 }
93 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != h.srcBox {
94 t.Fatalf("the fixture's claim: %+v", c.Op)
95 }
96 if h.route.Status != "moving" || h.route.HoldOp != p1ID {
97 t.Fatalf("the fixture's row: %+v", h.route)
98 }
99}
100
101// fromStopped drives the cut to a revert at stopped after execution was
102// given up: the baselines recorded and the claim reverting at evidence.
103func (h *revertHarness) fromStopped(t *testing.T) {
104 t.Helper()
105 h.ready(t)
106 h.crashAt = "after-exec-none"
107 h.cut(t)
108 h.crashPass(t)
109 h.clock.Advance(2 * time.Minute)
110 h.pass(t)
111 m := h.move(t)
112 if m.State != provision.MoveReverting || m.CutPhase != provision.CutStopped || m.Source == nil || m.Source.Files == nil {
113 t.Fatalf("the fixture did not revert at stopped: %+v", m)
114 }
115 if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence {
116 t.Fatalf("the fixture's claim: %+v", c.Op)
117 }
118}
119
120// arm wires the revert's hooks on the source box.
121func (h *revertHarness) arm(t *testing.T) {
122 t.Helper()
123 h.cfg.RestatusHold = func(host, status, op string, validate func() error) (bool, error) {
124 h.restatus = append(h.restatus, status+":"+op)
125 if validate != nil {
126 if err := validate(); err != nil {
127 return false, err
128 }
129 }
130 if host != p1Host || h.route.HoldOp != op {
131 return false, nil
132 }
133 h.route.Status = status
134 return true, nil
135 }
136 h.cfg.ReleaseHeld = func(host, op string) (bool, error) {
137 h.opens = append(h.opens, op)
138 if op == "" || h.route.HoldOp != op {
139 return false, nil
140 }
141 h.route.Status, h.route.HoldOp = "ready", ""
142 return true, nil
143 }
144 h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
145 h.events = append(h.events, ev.Event)
146 return nil
147 }
148 h.cfg.recoveryProbe = func(ctx context.Context, _ release.Store, _ provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
149 h.probes = append(h.probes, req)
150 if h.probe != nil {
151 return h.probe(req)
152 }
153 if err := req.Record(ctx, revertRendered()); err != nil {
154 return recoveryProbeWaiting, err
155 }
156 return recoveryProbeAnswered, nil
157 }
158 h.cfg.rvFault = func(point string) error {
159 h.rvHits = append(h.rvHits, point)
160 if point == h.rvAt && !h.rvDone {
161 h.rvDone = true
162 return errCrash
163 }
164 return nil
165 }
166}
167
168// revertRendered are the inputs the fake probe renders: not the pinned ones.
169func revertRendered() provision.MoveInputs {
170 return provision.MoveInputs{BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9", Release: "v7"}
171}
172
173// laneCfg is the target region's box, its lane park armed.
174func (h *revertHarness) laneCfg() moveCfg {
175 c := h.target()
176 c.DataDir = h.tdir
177 armParkFakes(&c, h.s, h.tleases, park.PlacementRegion(c.Region), roleTarget)
178 return c
179}
180
181// tPass runs one pass as the target region's box.
182func (h *revertHarness) tPass(t *testing.T) {
183 t.Helper()
184 if h.tw == nil {
185 h.tw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
186 }
187 h.tw.cfg = h.laneCfg()
188 if err := movesPass(context.Background(), h.s, h.tw); err != nil {
189 t.Fatal(err)
190 }
191}
192
193func (h *revertHarness) verifications() int {
194 n := 0
195 for _, p := range h.rvHits {
196 if p == "verify" {
197 n++
198 }
199 }
200 return n
201}
202
203func (h *revertHarness) revPhase(t *testing.T) string {
204 t.Helper()
205 return h.claim(t).Op.Journal[hostclaim.JournalRevPhase]
206}
207
208// laneChore is the lane's park the revert queued in the target region.
209func (h *revertHarness) laneChore(t *testing.T) *hostclaim.Obligation {
210 t.Helper()
211 for _, o := range h.claim(t).Obligations {
212 if o.Op == p1ID && o.Kind == "park-replica" && o.Region == park.PlacementRegion(h.dst) && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) {
213 return &o
214 }
215 }
216 return nil
217}
218
219// laneKeys are the objects still in the move's lane in the target bucket.
220func (h *revertHarness) laneKeys() []string {
221 var out []string
222 for k := range h.f.buckets[h.dstBucket] {
223 if strings.HasPrefix(k, moves.ScopeLocation(p1Host, p1ID)+"/") {
224 out = append(out, k)
225 }
226 }
227 return out
228}
229
230// pastHorizon moves the clock past the lane park's acknowledgement horizon.
231func (h *revertHarness) pastHorizon(t *testing.T, o *hostclaim.Obligation) {
232 t.Helper()
233 until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit)
234 if d := until.Sub(h.clock.now); d > 0 {
235 h.clock.Advance(d + time.Second)
236 }
237}
238
239// reopened asserts the source reopened: the claim reverting/opened under
240// the source, the record still reverting with ReopenedAt, the row open.
241func (h *revertHarness) reopened(t *testing.T) {
242 t.Helper()
243 m := h.move(t)
244 if m == nil || m.State != provision.MoveReverting || m.ReopenedAt == nil {
245 t.Fatalf("the source did not reopen: %+v", m)
246 }
247 c := h.claim(t)
248 if c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != h.srcBox || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseOpened {
249 t.Fatalf("the claim after the reopen: %+v", c.Op)
250 }
251 if h.route.HoldOp != "" || h.route.Status != "ready" {
252 t.Fatalf("the row is not open: %+v", h.route)
253 }
254}
255
256// drive runs the source's and the target's passes, moving the clock past the
257// lane park's horizon once the source has opened, until the record closes.
258func (h *revertHarness) drive(t *testing.T) {
259 t.Helper()
260 for i := 0; i < 12; i++ {
261 h.pass(t)
262 h.tPass(t)
263 if h.read(t, p1Host).Move == nil {
264 return
265 }
266 if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
267 if o := h.laneChore(t); o != nil {
268 h.pastHorizon(t, o)
269 }
270 }
271 }
272 t.Fatalf("the revert never closed: %+v, claim %+v, hits %v", h.move(t), h.claim(t).Op, h.rvHits)
273}
274
275// closed asserts the end: the record at rest in ie with LastMove rolled_back,
276// the claim serving under the source with this move recorded as released,
277// the row open, the source's journal parked.
278func (h *revertHarness) closed(t *testing.T) {
279 t.Helper()
280 rec := h.read(t, p1Host)
281 if rec.Move != nil || rec.Region != h.src || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" {
282 t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
283 }
284 c := h.claim(t)
285 if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != h.srcBox || hostclaim.RequireReleasedMove(p1ID)(c) != nil {
286 t.Fatalf("the claim after the close: %+v", c.Op)
287 }
288 if h.route.HoldOp != "" || h.route.Status != "ready" {
289 t.Fatalf("the row after the close: %+v", h.route)
290 }
291 if moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
292 t.Fatal("the source's journal of the move is still live")
293 }
294 parked, _ := filepath.Glob(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, "*", "artifacts", ".moves", p1Host, p1ID))
295 if len(parked) != 1 {
296 t.Fatalf("the source's journal is not parked once: %v", parked)
297 }
298 for _, s := range h.restatus {
299 if s != "restoring:"+p1ID {
300 t.Fatalf("a status change other than restoring under the move id: %v", h.restatus)
301 }
302 }
303}
304
305// TestRevertFromStoppingComputesEvidenceFirst (§7.4 evidence, J7, J8): a
306// revert that began at stopping first computes the stop evidence, and starts
307// nothing until it holds: the row stays held moving under the move id and is
308// fenced again, the claim stays moving under the source, no report is
309// written and nothing is probed. Once the tree is proven stopped, the cut's
310// giveUpExecution publishes the complete stopped-source.json and gives up
311// execution citing it, revertClaim writes the claim half (reverting, rev_phase
312// evidence, the lane's park queued), the row goes restoring under the move id,
313// and rev_phase steps evidence to probing (no baselines, so no verify), op.phase
314// following (J1).
315func TestRevertFromStoppingComputesEvidenceFirst(t *testing.T) {
316 t.Parallel()
317 h := newRevertHarness(t)
318 h.fromStopping(t)
319 h.arm(t)
320 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
321 fences, calls := h.fences, h.ev.calls
322 h.pass(t)
323 if h.ev.calls != calls+1 || h.fences != fences+1 {
324 t.Fatalf("the evidence was not asked again behind a fresh fence: evidence %d->%d fences %d->%d", calls, h.ev.calls, fences, h.fences)
325 }
326 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != "" {
327 t.Fatalf("the claim moved before the evidence held: %+v", c.Op)
328 }
329 if _, _, err := h.s.Get(context.Background(), h.stoppedKey()); !errors.Is(err, os.ErrNotExist) {
330 t.Fatalf("a stopped report before the evidence held: %v", err)
331 }
332 if len(h.probes) != 0 || len(h.restatus) != 0 || h.route.Status != "moving" || h.route.HoldOp != p1ID {
333 t.Fatalf("something started before the evidence held: probes %d restatus %v row %+v", len(h.probes), h.restatus, h.route)
334 }
335 if m := h.move(t); m.State != provision.MoveReverting || m.Source != nil {
336 t.Fatalf("the record: %+v", m)
337 }
338
339 // The tree is proven stopped; a crash right after the report leaves the
340 // claim still moving under the source: nothing has started.
341 h.ev.stopped = true
342 h.crashAt, h.crashed = "after-report", false
343 h.crashPass(t)
344 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
345 t.Fatalf("the claim after the report alone: %+v", c.Op)
346 }
347 if len(h.probes) != 0 || len(h.restatus) != 0 {
348 t.Fatalf("something started before execution was given up: probes %d restatus %v", len(h.probes), h.restatus)
349 }
350 h.freshWorker()
351 h.pass(t)
352 rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
353 if err != nil || !rep.Complete() || rep.OpID != p1ID || rep.Role != "source" {
354 t.Fatalf("the stopped report: %v %+v", err, rep)
355 }
356 c := h.claim(t)
357 if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() {
358 t.Fatalf("the claim half: %+v", c.Op)
359 }
360 if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
361 t.Fatalf("rev_phase %q op.phase %q, want probing for both (no baselines)", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
362 }
363 if o := h.laneChore(t); o == nil || o.Park.Incarnation != h.inc || o.Park.RequestLimit != cutRequestLimit || o.Owner != "region:uk" {
364 t.Fatalf("the lane's park: %+v", o)
365 }
366 if h.verifications() != 0 {
367 t.Fatalf("a revert from stopping verified %d times: no baseline was recorded", h.verifications())
368 }
369 if m := h.move(t); m.Source != nil {
370 t.Fatalf("the revert recorded baselines: %+v", m.Source)
371 }
372 if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
373 t.Fatalf("the row: %+v, want restoring under the move id", h.route)
374 }
375 if len(h.probes) != 1 {
376 t.Fatalf("%d probes, want 1", len(h.probes))
377 }
378 req := h.probes[0]
379 if req.RollbackID != "" || !req.Clean || req.Generation != "" || req.SourceGeneration != "cur" || req.Coverage != "" || req.Attempts != recoveryProbeAttempts {
380 t.Fatalf("the probe request: %+v", req)
381 }
382
383 // The probe answers: probed, reopened, the row opened, the claim opened.
384 h.probe = nil
385 h.pass(t)
386 h.reopened(t)
387 m := h.move(t)
388 if m.RecoveryInputs == nil || m.RecoveryInputs.BundleETags["default"] != "e9" || m.Inputs == nil || m.Inputs.BundleETags["default"] != "e1" {
389 t.Fatalf("recovery inputs %+v, move.inputs %+v", m.RecoveryInputs, m.Inputs)
390 }
391 if len(h.opens) != 1 || h.opens[0] != p1ID {
392 t.Fatalf("the open: %v", h.opens)
393 }
394}
395
396// TestRevertVerifyThenProbeNeverVerifiesAgain (§7.4 verify and probing;
397// witness wRollbackJournal for the revert, INV-2): a revert from stopped
398// verifies its files and timers against the stopped baselines, then journals
399// probing before the first attempt. The probe's process writes (the files and
400// the timers change) and the agent restarts: the next pass runs the probe
401// only, never the verification, and the revert is not stuck.
402func TestRevertVerifyThenProbeNeverVerifiesAgain(t *testing.T) {
403 t.Parallel()
404 h := newRevertHarness(t)
405 h.fromStopped(t)
406 h.arm(t)
407 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
408 h.pass(t)
409 if h.verifications() != 1 {
410 t.Fatalf("verified %d times, want once", h.verifications())
411 }
412 c := h.claim(t)
413 if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
414 t.Fatalf("rev_phase %q op.phase %q, want probing", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
415 }
416 if len(h.probes) != 1 || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
417 t.Fatalf("probes %d row %+v", len(h.probes), h.route)
418 }
419 // The probe's process writes, and the agent restarts.
420 if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("written by the probe's process"), 0o644); err != nil {
421 t.Fatal(err)
422 }
423 h.timers = append(h.timers, registry.Timer{Host: p1Host, Name: "added", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
424 h.freshWorker()
425 h.pass(t)
426 if h.verifications() != 1 || len(h.probes) != 2 {
427 t.Fatalf("after the restart: verified %d times, %d probes; want 1 and 2", h.verifications(), len(h.probes))
428 }
429 if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
430 t.Fatalf("the revert stopped on its own probe's writes: %+v", m)
431 }
432 h.probe = nil
433 h.freshWorker()
434 h.pass(t)
435 h.reopened(t)
436 if h.verifications() != 1 || len(h.probes) != 3 {
437 t.Fatalf("verified %d times, %d probes; want 1 and 3", h.verifications(), len(h.probes))
438 }
439}
440
441// TestRevertVerifyMismatchIsStuck (INV-4): a revert from stopped whose files
442// or timers no longer match the stopped baselines stops for an operator with
443// the baseline-mismatch sentence before anything is probed; rev_phase stays
444// verify, the row stays held restoring.
445func TestRevertVerifyMismatchIsStuck(t *testing.T) {
446 t.Parallel()
447 for _, what := range []string{"files", "timers"} {
448 t.Run(what, func(t *testing.T) {
449 h := newRevertHarness(t)
450 h.fromStopped(t)
451 h.arm(t)
452 if what == "files" {
453 if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
454 t.Fatal(err)
455 }
456 } else {
457 h.timers = nil
458 }
459 h.pass(t)
460 m := h.move(t)
461 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Reason != provision.RevertCutDeadline {
462 t.Fatalf("not stuck with the mismatch: %+v", m)
463 }
464 if h.revPhase(t) != hostclaim.RevPhaseVerify || len(h.probes) != 0 {
465 t.Fatalf("rev_phase %q, %d probes", h.revPhase(t), len(h.probes))
466 }
467 if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
468 t.Fatalf("the row: %+v", h.route)
469 }
470 h.pass(t)
471 if h.verifications() != 1 || h.move(t).State != provision.MoveStuck {
472 t.Fatalf("a stuck revert ran again: %d verifications", h.verifications())
473 }
474 })
475 }
476}
477
478// retry is the operator's move resolve --retry: provision.MoveRetry through
479// the record writer, as the console door writes it (Task 7).
480func (h *revertHarness) retry(t *testing.T) {
481 t.Helper()
482 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
483 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetry(g)); err != nil {
484 t.Fatal(err)
485 }
486 if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
487 t.Fatalf("after the retry: %+v", m)
488 }
489}
490
491// TestRetryResumesARevertsRevPhase (Task 7's box side for a revert, J20): a
492// stuck revert resumes its rev_phase. Stuck at verify, each retry verifies
493// again (never skipped), unrepaired it is stuck again, repaired it goes on.
494// Stuck at probing after fifteen minutes of failed probes, a retry probes
495// again with a fresh fifteen minutes and never verifies again.
496func TestRetryResumesARevertsRevPhase(t *testing.T) {
497 t.Parallel()
498 t.Run("verify", func(t *testing.T) {
499 h := newRevertHarness(t)
500 h.fromStopped(t)
501 h.arm(t)
502 upload := filepath.Join(h.cfg.DataDir, p1Host, "upload.txt")
503 if err := os.WriteFile(upload, []byte("changed while stopped"), 0o644); err != nil {
504 t.Fatal(err)
505 }
506 h.pass(t)
507 if m := h.move(t); m.State != provision.MoveStuck {
508 t.Fatalf("not stuck: %+v", m)
509 }
510 h.retry(t)
511 h.pass(t)
512 if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || h.verifications() != 2 || len(h.probes) != 0 {
513 t.Fatalf("an unrepaired retry: %+v, %d verifications, %d probes", m, h.verifications(), len(h.probes))
514 }
515 if err := os.WriteFile(upload, []byte("hello"), 0o644); err != nil {
516 t.Fatal(err)
517 }
518 h.retry(t)
519 h.pass(t)
520 h.reopened(t)
521 if h.verifications() != 3 || len(h.probes) != 1 {
522 t.Fatalf("%d verifications, %d probes; want 3 and 1", h.verifications(), len(h.probes))
523 }
524 if got := h.move(t).Reason; got != provision.RevertCutDeadline {
525 t.Fatalf("the retry changed the revert's cause: %q", got)
526 }
527 })
528 t.Run("probing", func(t *testing.T) {
529 h := newRevertHarness(t)
530 h.fromStopped(t)
531 h.arm(t)
532 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
533 h.pass(t)
534 h.clock.Advance(10 * time.Minute)
535 h.pass(t)
536 if m := h.move(t); m.State != provision.MoveReverting {
537 t.Fatalf("stuck before fifteen minutes: %+v", m)
538 }
539 h.clock.Advance(6 * time.Minute)
540 h.pass(t)
541 m := h.move(t)
542 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
543 t.Fatalf("not stuck after fifteen minutes of failed probes: %+v", m)
544 }
545 if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
546 t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route)
547 }
548 probes := len(h.probes)
549 h.pass(t)
550 if len(h.probes) != probes {
551 t.Fatal("a stuck revert probed")
552 }
553 // The retry gives the probe fifteen minutes again.
554 h.retry(t)
555 h.pass(t)
556 if m := h.move(t); m.State != provision.MoveReverting || len(h.probes) != probes+1 {
557 t.Fatalf("after the retry: %+v, %d probes", m, len(h.probes))
558 }
559 h.clock.Advance(14 * time.Minute)
560 h.pass(t)
561 if m := h.move(t); m.State != provision.MoveReverting {
562 t.Fatalf("the retry's budget did not start again: %+v", m)
563 }
564 h.clock.Advance(2 * time.Minute)
565 h.pass(t)
566 if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
567 t.Fatalf("not stuck again: %+v", m)
568 }
569 h.retry(t)
570 h.probe = nil
571 h.pass(t)
572 h.reopened(t)
573 if h.verifications() != 1 {
574 t.Fatalf("verified %d times across the retries, want once", h.verifications())
575 }
576 })
577}
578
579// TestRevertOpensAndReleases (§7.4 opened, §7.11): after the reopen the
580// source waits for the lane park the revert queued in the target region. The
581// target region's box archives the lane under parked/<inc>/<host>/<ret>/moves/
582// through the park engine; the obligation is acknowledged only after its
583// horizon; and only then the claim goes to serving under the source and the
584// record closes with LastMove rolled_back, the offline time counted from the
585// cut to the reopen, and the source's journal parked.
586func TestRevertOpensAndReleases(t *testing.T) {
587 t.Parallel()
588 h := newRevertHarness(t)
589 h.fromStopped(t)
590 h.arm(t)
591 h.pass(t)
592 h.reopened(t)
593 o := h.laneChore(t)
594 if o == nil {
595 t.Fatal("no lane park queued")
596 }
597 lane := h.laneKeys()
598 if len(lane) == 0 {
599 t.Fatal("the fixture has no lane objects")
600 }
601 h.pass(t)
602 h.reopened(t)
603 if h.laneChore(t) == nil || len(h.events) == 0 || h.events[len(h.events)-1] == "rolled_back" {
604 t.Fatalf("the source closed before the lane was parked: %v", h.events)
605 }
606 // The target region's box archives the lane; the horizon has not passed.
607 h.tPass(t)
608 if left := h.laneKeys(); len(left) != 0 {
609 t.Fatalf("the lane still holds %v", left)
610 }
611 prefix := "parked/" + o.Park.Incarnation + "/" + p1Host + "/" + o.Retirement + "/moves/"
612 archived := 0
613 for k := range h.f.buckets["data-uk"] {
614 if strings.HasPrefix(k, prefix) {
615 archived++
616 }
617 }
618 if archived < len(lane) {
619 t.Fatalf("%d archived under %s, the lane had %d", archived, prefix, len(lane))
620 }
621 if h.laneChore(t) == nil {
622 t.Fatal("the lane park was acknowledged before its horizon")
623 }
624 h.pass(t)
625 h.reopened(t)
626 // Past the horizon the target acknowledges, and the source closes.
627 h.pastHorizon(t, o)
628 h.tPass(t)
629 if h.laneChore(t) != nil {
630 t.Fatal("the lane park was not acknowledged after its horizon")
631 }
632 if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened {
633 t.Fatalf("the claim before the source's pass: %+v", c.Op)
634 }
635 reopenedAt, cutAt := *h.move(t).ReopenedAt, *h.move(t).CutAt
636 h.pass(t)
637 h.closed(t)
638 if got, want := h.read(t, p1Host).LastMove.OfflineS, reopenedAt.Sub(cutAt).Seconds(); got != want || got <= 0 {
639 t.Fatalf("offline %v, want %v", got, want)
640 }
641 if n := countString(h.events, "rolled_back"); n != 1 {
642 t.Fatalf("the rolled_back event %d times: %v", n, h.events)
643 }
644 // Another pass on either box changes nothing.
645 before := h.claim(t)
646 h.pass(t)
647 h.tPass(t)
648 if after := h.claim(t); after.Gen != before.Gen {
649 t.Fatalf("a pass after the close wrote the claim (%d to %d)", before.Gen, after.Gen)
650 }
651}
652
653func countString(xs []string, s string) int {
654 n := 0
655 for _, x := range xs {
656 if x == s {
657 n++
658 }
659 }
660 return n
661}
662
663// Home part 2, Task 4 (found by TestMoveIntoHomePreFlipRevert): a real
664// recovery probe wake admits a writer, so the claim the revert's step read
665// before the probe is stale when the probe answers. The probed step reads
666// the claim again; with the step's own read it lost the CAS every pass, and
667// each pass's probe admitted one more fresh generation until the attempts
668// were spent. Here, on the default named move, the probe runs through the
669// production adapter and the claim runtime (Admit, Prepare, Restore,
670// Reserve, Materialize) with the production recovery owners, and admits
671// once; the source then reopens in the same pass.
672func TestRevertProbedAfterARealAdmission(t *testing.T) {
673 h := newRevertHarnessOn(t, newLandingHarness(t, realLaunchEnv).cutHarness)
674 h.fromStopped(t)
675 h.arm(t)
676 rec := revertRecovery(t, h)
677 h.pass(t)
678 h.reopened(t)
679 if rec.wakes != 1 || rec.materialized.Claim == nil {
680 t.Fatalf("the revert probe's starts: %d, materialized %v", rec.wakes, rec.materialized.Claim != nil)
681 }
682 h.pass(t)
683 h.reopened(t)
684 if rec.wakes != 1 {
685 t.Fatalf("a pass after the reopen probed again: %d starts", rec.wakes)
686 }
687}
688
689// The census condition on that re-read (census-ruling-home-part2): the
690// fresh claim is taken only when it differs from the step's own read by the
691// probe wake's own writes (its admission of a new generation for this box's
692// writer and its restore's re-tag of this box's database). Here the real
693// probe answers as above, and one foreign write lands after it and before
694// the probed CAS: a change to the op, to the holder or its session, or to a
695// claim field the probe never writes. The step refuses it as a conflict,
696// the claim stays at probing with the foreign change in place, and the row
697// stays held: the step is left to its next pass, as the stale CAS left it.
698func TestRevertProbedRefusesAForeignWrite(t *testing.T) {
699 for _, e := range []struct {
700 name string
701 edit func(*hostclaim.Claim)
702 }{
703 {"op target", func(c *hostclaim.Claim) { c.Op.Intent["target"] = "other-box" }},
704 {"op journal", func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalSrcGen] = "other" }},
705 {"holder session", func(c *hostclaim.Claim) {
706 c.Holder.Session++
707 c.Holder.Admissions[len(c.Holder.Admissions)-1].Session++
708 }},
709 {"holder writer box", func(c *hostclaim.Claim) { c.Holder.WriterBox = "uk-box" }},
710 {"quiet key", func(c *hostclaim.Claim) {
711 c.QuietAfter = map[string]time.Time{"uk-box": p1Created}
712 }},
713 {"chore", func(c *hostclaim.Claim) {
714 o := c.Obligations[0]
715 o.ID = "0123456789abcdef0123456789abcdef"
716 c.Obligations = append(c.Obligations, o)
717 }},
718 {"another box's inventory", func(c *hostclaim.Claim) {
719 c.Inventory["uk-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1}}
720 }},
721 } {
722 t.Run(e.name, func(t *testing.T) {
723 h := newRevertHarnessOn(t, newLandingHarness(t, realLaunchEnv).cutHarness)
724 h.fromStopped(t)
725 h.arm(t)
726 rec := revertRecovery(t, h)
727 probe, wrote := h.cfg.recoveryProbe, false
728 var foreign *hostclaim.Claim
729 var before, own hostclaim.Observed
730 read := func() hostclaim.Observed {
731 o, err := hostclaim.Read(context.Background(), h.s, p1Host)
732 if err != nil {
733 t.Fatal(err)
734 }
735 return o
736 }
737 h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
738 before = read()
739 out, err := probe(ctx, s, r, req)
740 if out == recoveryProbeAnswered && !wrote {
741 wrote = true
742 own = read()
743 rewriteP1Claim(t, h.s, e.edit)
744 foreign = h.claim(t)
745 }
746 return out, err
747 }
748 err := h.w().revertSource(context.Background(), h.s, h.read(t, p1Host), nil)
749 if !wrote || rec.wakes != 1 {
750 t.Fatalf("the probe did not answer once: wrote %v, %d starts", wrote, rec.wakes)
751 }
752 if !errors.Is(err, hostclaim.ErrConflict) {
753 t.Fatalf("the revert's probed took a foreign write: %v", err)
754 }
755 // The same change in place of the probe's own last write (the
756 // gen alone cannot tell them apart) is refused too, and the
757 // probe's own writes alone pass.
758 box := h.w().cfg.Box
759 if err := revertProbeWrites(box, h.cfg.DataDir, before, own); err != nil || own.Claim.Gen != before.Claim.Gen+2 {
760 t.Fatalf("the probe's own writes (gen %d to %d): %v", before.Claim.Gen, own.Claim.Gen, err)
761 }
762 swapped, err := revertClaimCopy(own.Claim)
763 if err != nil {
764 t.Fatal(err)
765 }
766 e.edit(swapped)
767 if err := revertProbeWrites(box, h.cfg.DataDir, before, hostclaim.Observed{Host: own.Host, ETag: own.ETag + "-foreign", Claim: swapped}); !errors.Is(err, hostclaim.ErrConflict) {
768 t.Fatalf("a foreign change at the probe's own gen was taken: %v", err)
769 }
770 c := h.claim(t)
771 if c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Gen != foreign.Gen {
772 t.Fatalf("the claim after the refusal: gen %d (foreign %d), %+v", c.Gen, foreign.Gen, c.Op)
773 }
774 if m := h.move(t); m.ReopenedAt != nil || h.route.HoldOp != p1ID {
775 t.Fatalf("the refusal reopened: %+v, row %+v", m, h.route)
776 }
777 })
778 }
779}
780
781// TestRevertProbedRefusesAnUnreplayableAdmission pins today's stall, so it
782// is explicit and not silent (amadan #300). revertProbeWrites replays the
783// probe's admission with hostclaim.AdmitWriter, which cannot vouch for a
784// loss-bound admission (AdmitWriterWithLoss) or a fence-bound one
785// (AdmitWriterWithFencedTargetLoss): their verified flags are hostclaim's
786// own and are not exposed, since exposing them would be a forging surface.
787// So the re-read refuses such an admission with hostclaim.ErrConflict and
788// the step stays at probing, as it did before the re-read existed. The later
789// fix is a hostclaim-owned function that re-verifies those admissions (#300).
790//
791// The recovery fixtures produce both kinds only on a return (a restoring
792// claim); revertProbeWrites reads no op kind, so the real admissions made
793// through the claim's writer are checked against it directly, beside a plain
794// admission over the same fixture that does replay. The revert step itself
795// is then driven with a probe whose one claim write is shaped as a
796// loss-bound admission.
797func TestRevertProbedRefusesAnUnreplayableAdmission(t *testing.T) {
798 admit := func(t *testing.T, f *rcFixture, pub ed25519.PublicKey) (hostclaim.Observed, hostclaim.Observed) {
799 t.Helper()
800 before := f.read()
801 in := f.selectAdmission(before)
802 after, err := f.lossWriter(pub).Admit(t.Context(), before, in.Admission, in.Baseline)
803 if err != nil {
804 t.Fatal(err)
805 }
806 if after.Claim.Gen != before.Claim.Gen+1 {
807 t.Fatalf("the admission wrote gen %d over %d", after.Claim.Gen, before.Claim.Gen)
808 }
809 return before, after
810 }
811 t.Run("a plain admission replays", func(t *testing.T) {
812 f := newRCFixture(t, rcClean)
813 before, after := admit(t, f, nil)
814 if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); err != nil {
815 t.Fatalf("the plain admission was refused: %v", err)
816 }
817 })
818 t.Run("loss-bound", func(t *testing.T) {
819 f := newRCFixture(t, rcClean)
820 pub := f.declare(false, []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20), rcCutSource()}, func(c *hostclaim.Claim) {
821 c.Holder.Session = 25
822 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
823 })
824 before, after := admit(t, f, pub)
825 if a := after.Claim.Holder.Admissions[len(after.Claim.Holder.Admissions)-1]; a.LossID != rcLossID {
826 t.Fatalf("not a loss-bound admission: %+v", a)
827 }
828 if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
829 t.Fatalf("a loss-bound admission was replayed: %v", err)
830 }
831 })
832 t.Run("fence-bound", func(t *testing.T) {
833 f := newRCFixture(t, rcClean)
834 record := f.fence()
835 before, after := admit(t, f, nil)
836 if after.Claim.Op.Journal["fence_admitted."+record.ID] == "" {
837 t.Fatalf("not a fence-bound admission: %+v", after.Claim.Op.Journal)
838 }
839 if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
840 t.Fatalf("a fence-bound admission was replayed: %v", err)
841 }
842 })
843 t.Run("the revert step stays at probing", func(t *testing.T) {
844 h := newRevertHarness(t)
845 h.fromStopped(t)
846 h.arm(t)
847 var admitted *hostclaim.Claim
848 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) {
849 if admitted == nil {
850 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
851 hd := c.Holder
852 session := c.Gen + 1
853 hd.Admissions = append(hd.Admissions, hostclaim.AdmissionRecord{Session: session, Box: h.srcBox, Generation: "lost-gen", Predecessor: hd.Generation, Mode: "cold", Region: "ie", PredecessorRegion: "ie", LossID: rcLossID})
854 hd.Session, hd.Stamp, hd.Generation, hd.WriterBox = session, session, "lost-gen", h.srcBox
855 c.Op.Journal["loss_admitted."+rcLossID] = strconv.FormatUint(session, 10)
856 })
857 admitted = h.claim(t)
858 }
859 return recoveryProbeAnswered, nil
860 }
861 err := h.w().revertSource(context.Background(), h.s, h.read(t, p1Host), nil)
862 if admitted == nil {
863 t.Fatal("the probe never ran")
864 }
865 if !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
866 t.Fatalf("the revert's probed took a loss-bound admission: %v", err)
867 }
868 c := h.claim(t)
869 if c.Gen != admitted.Gen || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
870 t.Fatalf("the claim after the refusal: gen %d (admitted %d), %+v", c.Gen, admitted.Gen, c.Op)
871 }
872 if m := h.move(t); m.ReopenedAt != nil || h.route.HoldOp != p1ID {
873 t.Fatalf("the refusal reopened: %+v, row %+v", m, h.route)
874 }
875 })
876}
877
878// TestRevertReplays: a crash after any of the revert's side effects, then a
879// restarted agent, still reaches the close, with the evidence asked once it
880// is journalled, the hold only ever restoring under the move id, and the
881// verification never run again once probing is journalled.
882func TestRevertReplays(t *testing.T) {
883 t.Parallel()
884 type entry struct {
885 name string
886 from func(*revertHarness, *testing.T)
887 cut []string // the cut's fault points (giveUpExecution)
888 points []string // the revert's
889 }
890 for _, e := range []entry{
891 {"from stopping", (*revertHarness).fromStopping, []string{"after-checkpoint", "after-report", "after-exec-none"},
892 []string{"after-revert-evidence", "after-revert-claim", "after-restatus", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
893 {"from stopped", (*revertHarness).fromStopped, nil,
894 []string{"after-restatus", "verify", "after-rev-verify", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
895 } {
896 for _, point := range append(append([]string{}, e.cut...), e.points...) {
897 t.Run(e.name+"/"+point, func(t *testing.T) {
898 h := newRevertHarness(t)
899 e.from(h, t)
900 if e.name == "from stopping" {
901 h.ev.stopped = true
902 }
903 h.arm(t)
904 isCut := false
905 for _, p := range e.cut {
906 isCut = isCut || p == point
907 }
908 // The fixture's own crash (fromStopped) has used the cut's
909 // seam: each case watches only its own.
910 crashed := func() bool { return h.rvDone }
911 if isCut {
912 h.crashAt, h.crashed = point, false
913 crashed = func() bool { return h.crashed }
914 } else {
915 h.rvAt = point
916 }
917 for i := 0; i < 8 && !crashed(); i++ {
918 h.pass(t)
919 h.tPass(t)
920 if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
921 if o := h.laneChore(t); o != nil {
922 h.pastHorizon(t, o)
923 }
924 }
925 }
926 if !crashed() {
927 t.Fatalf("never reached %s (revert %v, cut %v)", point, h.rvHits, h.hits)
928 }
929 h.freshWorker()
930 h.drive(t)
931 h.closed(t)
932 switch {
933 case e.name == "from stopping" && h.verifications() != 0:
934 t.Fatalf("a revert from stopping verified %d times", h.verifications())
935 case e.name == "from stopped" && point == "verify" && h.verifications() != 2:
936 t.Fatalf("verified %d times after a crash before the verification, want 2 (the crashed one counted)", h.verifications())
937 case e.name == "from stopped" && point != "verify" && h.verifications() != 1:
938 t.Fatalf("verified %d times, want once", h.verifications())
939 }
940 if len(h.probes) == 0 {
941 t.Fatal("never probed")
942 }
943 })
944 }
945 }
946}
947
948// TestRevertJournalSweptAfterTheClose (the local journal sweep, Tasks 4 and
949// 5): a crash after the close and before the source's journal is parked
950// leaves it to the sweep, which parks it once no record carries the move; a
951// journal whose revert is still in flight is left.
952func TestRevertJournalSweptAfterTheClose(t *testing.T) {
953 t.Parallel()
954 h := newRevertHarness(t)
955 h.fromStopped(t)
956 h.arm(t)
957 h.pass(t)
958 h.reopened(t)
959 // In flight: the sweep leaves the marked journal.
960 h.w().sweepFinalJournals(context.Background(), h.s, []provision.Record{h.read(t, p1Host)})
961 if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
962 t.Fatal("the sweep parked a journal whose revert is in flight")
963 }
964 h.rvAt = "after-close"
965 o := h.laneChore(t)
966 h.tPass(t)
967 h.pastHorizon(t, o)
968 h.tPass(t)
969 h.pass(t)
970 if !h.rvDone {
971 t.Fatalf("never reached the close: %v", h.rvHits)
972 }
973 if h.read(t, p1Host).Move != nil {
974 t.Fatal("the close did not land")
975 }
976 if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
977 t.Fatal("the crash after the close parked the journal")
978 }
979 h.freshWorker()
980 h.pass(t)
981 h.closed(t)
982}
983
984// TestRevertUnarmedInProduction (J18): production wires no recovery probe
985// until Task 9 and no park engine leases until Plan 5, so the revert's
986// recovery and the target's lane park are unarmed there. The claim half
987// still follows the record (Plan 2's revertClaim); nothing else moves.
988func TestRevertUnarmedInProduction(t *testing.T) {
989 t.Parallel()
990 if revertSourceArmed(moveCfg{Box: "ie-box", DataDir: "/d", RestatusHold: func(string, string, string, func() error) (bool, error) { return false, nil },
991 ReleaseHeld: func(string, string) (bool, error) { return false, nil }, Timers: func(string) ([]registry.Timer, error) { return nil, nil },
992 FilesDir: func(string) (string, string) { return "", "" }}) {
993 t.Fatal("the revert is armed without a recovery probe")
994 }
995 if revertLaneArmed(moveCfg{Box: "uk-box", Region: "uk", OwnBucket: "data-uk", Versioned: func(context.Context, string, string) (lease.VersionedStore, error) { return nil, nil }}) {
996 t.Fatal("the lane park is armed without the park engine's leases")
997 }
998 t.Run("from stopping", func(t *testing.T) {
999 h := newRevertHarness(t)
1000 h.fromStopping(t)
1001 h.ev.stopped = true
1002 h.pass(t)
1003 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
1004 t.Fatalf("an unarmed revert gave up execution: %+v", c.Op)
1005 }
1006 })
1007 t.Run("from stopped", func(t *testing.T) {
1008 h := newRevertHarness(t)
1009 h.fromStopped(t)
1010 lane := len(h.laneKeys())
1011 h.pass(t)
1012 c := h.claim(t)
1013 if c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence || h.route.Status != "moving" {
1014 t.Fatalf("an unarmed revert went on: %+v row %+v", c.Op, h.route)
1015 }
1016 tc := h.target()
1017 tc.DataDir = h.tdir
1018 h.tw = &movesWorker{cfg: tc, mirrors: map[string]lease.MirrorStore{}}
1019 if err := movesPass(context.Background(), h.s, h.tw); err != nil {
1020 t.Fatal(err)
1021 }
1022 if len(h.laneKeys()) != lane || len(h.tleases.acquired) != 0 {
1023 t.Fatal("an unarmed target parked the lane")
1024 }
1025 })
1026 t.Run("the claim half", func(t *testing.T) {
1027 // The record reverts and the claim half is lost: the unarmed pass
1028 // still writes it, as Plan 2's reverting arm did.
1029 h := newRevertHarness(t)
1030 h.ready(t)
1031 h.crashAt = "after-exec-none"
1032 h.cut(t)
1033 h.crashPass(t)
1034 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
1035 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
1036 t.Fatal(err)
1037 }
1038 h.pass(t)
1039 if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
1040 t.Fatalf("the claim half: %+v", c.Op)
1041 }
1042 })
1043}
1044
1045// A reverting record's target region box has neither the row nor a journal
1046// of the move (the revert never flipped), so the participant rule answers
1047// it as the target on the state alone: its one piece of work is the lane park
1048// the claim owes its region. Any other state still needs the row or the
1049// journal.
1050func TestRevertingTargetIsAParticipant(t *testing.T) {
1051 cfg := moveCfg{Region: "uk", DataDir: t.TempDir(), OwnsRoute: func(string) bool { return false }}
1052 rec := func(state string) provision.Record {
1053 return provision.Record{Host: p1Host, Region: "ie", Move: &provision.MoveState{ID: p1ID, State: state, From: "ie", To: "uk"}}
1054 }
1055 if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleTarget {
1056 t.Fatalf("a reverting record's target: %q", got)
1057 }
1058 for _, state := range []string{provision.MoveCommitted, provision.MoveRollingBack, provision.MoveStuck, provision.MoveCutting} {
1059 if got := movesParticipantRecord(cfg, rec(state)); got != roleNone {
1060 t.Fatalf("%s with no row and no journal: %q", state, got)
1061 }
1062 }
1063 cfg.Region = "fr"
1064 if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleNone {
1065 t.Fatalf("a third region: %q", got)
1066 }
1067}
1068
1069// restatusHold (production's RestatusHold): an op's own hold changes status
1070// under the host's hold lock, after the caller's validation, touching the
1071// stage-requested marker only when the change landed. A replay answers true
1072// and touches nothing; a row held by another op, an unheld row and a missing
1073// row answer false and are left; a validation error moves nothing.
1074func TestRestatusHoldIsConditional(t *testing.T) {
1075 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
1076 if err != nil {
1077 t.Fatal(err)
1078 }
1079 defer reg.Close()
1080 for _, rt := range []registry.Route{
1081 {Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
1082 {Host: "b.go.example.com", Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
1083 } {
1084 if err := reg.Upsert(rt); err != nil {
1085 t.Fatal(err)
1086 }
1087 }
1088 var written []string
1089 restatus := restatusHold(reg, func(host string) { written = append(written, host) })
1090 if ok, err := restatus(p1Host, "restoring", p1ID, nil); ok || err != nil {
1091 t.Fatalf("an unheld row: %v %v", ok, err)
1092 }
1093 if err := holdMoving(reg, nil)(p1Host, p1ID); err != nil {
1094 t.Fatal(err)
1095 }
1096 refused := errors.New("the record moved on")
1097 if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return refused }); ok || !errors.Is(err, refused) {
1098 t.Fatalf("a refused validation: %v %v", ok, err)
1099 }
1100 if rt, _ := reg.Get(p1Host); rt.Status != "moving" || len(written) != 0 {
1101 t.Fatalf("a refused validation moved the row: %+v, marker %d", rt, len(written))
1102 }
1103 for i := 0; i < 2; i++ {
1104 if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return nil }); !ok || err != nil {
1105 t.Fatalf("restatus %d: %v %v", i, ok, err)
1106 }
1107 }
1108 if rt, _ := reg.Get(p1Host); rt.Status != "restoring" || rt.HoldOp != p1ID || len(written) != 1 {
1109 t.Fatalf("row %+v; marker touched %d times", rt, len(written))
1110 }
1111 if ok, err := restatus(p1Host, "restoring", p1ID2, nil); ok || err != nil {
1112 t.Fatalf("another op's hold: %v %v", ok, err)
1113 }
1114 if err := holdMoving(reg, nil)("b.go.example.com", p1ID2); err != nil {
1115 t.Fatal(err)
1116 }
1117 if ok, err := restatus("b.go.example.com", "restoring", p1ID, nil); ok || err != nil {
1118 t.Fatalf("a row held by another op: %v %v", ok, err)
1119 }
1120 if rt, _ := reg.Get("b.go.example.com"); rt.Status != "moving" || rt.HoldOp != p1ID2 {
1121 t.Fatalf("another op's row moved: %+v", rt)
1122 }
1123 if ok, err := restatus("absent.example.com", "restoring", p1ID, nil); ok || err != nil {
1124 t.Fatalf("a missing row: %v %v", ok, err)
1125 }
1126 // Production wires it, and leaves the revert and the lane park unarmed.
1127 var prod moveCfg
1128 hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
1129 armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
1130 if prod.RestatusHold == nil || prod.ReleaseHeld == nil {
1131 t.Fatal("production does not wire the revert's row hooks")
1132 }
1133 prod.Box, prod.Region, prod.OwnBucket = "ie-box", "ie", "data-ie"
1134 if revertSourceArmed(prod) || revertLaneArmed(prod) {
1135 t.Fatal("production armed the revert")
1136 }
1137}
1138
1139// A revert from stopping whose files directory cannot be archived (a special
1140// file) is not refused again: nothing compares a files digest on this path
1141// (§7.4 evidence), so giveUpExecution publishes the report without one,
1142// records no baselines, and the revert goes on.
1143func TestRevertFromStoppingWithUnarchivableFiles(t *testing.T) {
1144 t.Parallel()
1145 h := newRevertHarness(t)
1146 h.fromStopping(t)
1147 if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
1148 t.Fatal(err)
1149 }
1150 h.arm(t)
1151 h.ev.stopped = true
1152 h.pass(t)
1153 h.reopened(t)
1154 rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
1155 if err != nil || !rep.Complete() || rep.FilesPlain != "" {
1156 t.Fatalf("the stopped report: %v %+v", err, rep)
1157 }
1158 if m := h.move(t); m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
1159 t.Fatalf("the record: %+v", m)
1160 }
1161 if h.verifications() != 0 || len(h.probes) != 1 {
1162 t.Fatalf("%d verifications, %d probes", h.verifications(), len(h.probes))
1163 }
1164}
1165
1166// TestRevertProbeAttemptsExhaustedIsStuck (§7.7; Task 6 fix round 1): a
1167// revert's probe has the recovery probe's nine attempts as well as the
1168// fifteen minutes. A probe that uses them all stops the revert for an
1169// operator with the recovery probe's sentence, long before fifteen minutes;
1170// rev_phase stays probing and the row held. A retry probes again with nine
1171// attempts and never verifies again.
1172func TestRevertProbeAttemptsExhaustedIsStuck(t *testing.T) {
1173 t.Parallel()
1174 h := newRevertHarness(t)
1175 h.fromStopped(t)
1176 h.arm(t)
1177 h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil }
1178 h.pass(t)
1179 m := h.move(t)
1180 if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Reason != provision.RevertCutDeadline {
1181 t.Fatalf("not stuck on the exhausted probe: %+v", m)
1182 }
1183 if len(h.probes) != 1 || h.probes[0].Attempts != recoveryProbeAttempts {
1184 t.Fatalf("probes %+v, want one with %d attempts", h.probes, recoveryProbeAttempts)
1185 }
1186 if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
1187 t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route)
1188 }
1189 h.retry(t)
1190 h.probe = nil
1191 h.pass(t)
1192 h.reopened(t)
1193 if len(h.probes) != 2 || h.probes[1].Attempts != recoveryProbeAttempts || h.verifications() != 1 {
1194 t.Fatalf("after the retry: probes %+v, %d verifications", h.probes, h.verifications())
1195 }
1196}
1197
1198// failStuckStore fails the record write that would mark the move stuck, as a
1199// store that refuses or loses the request does.
1200type failStuckStore struct {
1201 release.Store
1202 fail bool
1203}
1204
1205func (f *failStuckStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
1206 if f.fail && key == provision.RecordKey("bab", "app", p1Host) && strings.Contains(string(body), `"state":"stuck"`) {
1207 return "", errors.New("injected: the record write failed")
1208 }
1209 return f.Store.Put(ctx, key, body, ifMatch)
1210}
1211
1212// TestRevertProbeExhaustionSurvivesACrash (Task 6 fix round 1): once the
1213// revert's probe is exhausted (its attempts, or its fifteen minutes) the
1214// exhaustion is journalled before the stuck mark. A crash between the two, or
1215// a stuck mark the store refused, leaves the record reverting; the next pass
1216// marks it stuck without probing again and never grants a fresh window. Only
1217// a retry (the record back from stuck through MoveRetry) starts one.
1218func TestRevertProbeExhaustionSurvivesACrash(t *testing.T) {
1219 t.Parallel()
1220 for _, reason := range []string{provision.StuckRecoveryProbeAttempts, provision.StuckRevertProbeBudget} {
1221 for _, how := range []string{"crash", "failed record write"} {
1222 t.Run(reason+"/"+how, func(t *testing.T) {
1223 h := newRevertHarness(t)
1224 h.fromStopped(t)
1225 h.arm(t)
1226 exhausted := reason == provision.StuckRecoveryProbeAttempts
1227 // The probe records what it renders before each attempt, so
1228 // the record moves in the very pass the window ends.
1229 h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
1230 if err := req.Record(context.Background(), revertRendered()); err != nil {
1231 return recoveryProbeWaiting, err
1232 }
1233 if exhausted {
1234 return recoveryProbeExhausted, nil
1235 }
1236 return recoveryProbeWaiting, nil
1237 }
1238 if !exhausted {
1239 h.pass(t) // the window opens
1240 h.clock.Advance(16 * time.Minute)
1241 }
1242 probes := len(h.probes)
1243 fs := &failStuckStore{Store: h.s}
1244 if how == "crash" {
1245 h.rvAt = "after-probe-exhausted"
1246 } else {
1247 fs.fail = true
1248 }
1249 if err := movesPass(context.Background(), fs, h.w()); err != nil {
1250 t.Fatal(err)
1251 }
1252 if how == "crash" && !h.rvDone {
1253 t.Fatalf("never reached after-probe-exhausted: %v", h.rvHits)
1254 }
1255 if m := h.move(t); m.State != provision.MoveReverting {
1256 t.Fatalf("the fixture's stuck mark landed: %+v", m)
1257 }
1258 if len(h.probes) != probes+1 {
1259 t.Fatalf("%d probes in the exhausting pass, want 1", len(h.probes)-probes)
1260 }
1261 // The next pass, on a restarted agent with the clock where it
1262 // was: stuck, with no probe and no fresh window.
1263 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeAnswered, nil }
1264 h.freshWorker()
1265 h.pass(t)
1266 m := h.move(t)
1267 if m.State != provision.MoveStuck || m.StuckReason != reason {
1268 t.Fatalf("the next pass did not mark it stuck: %+v", m)
1269 }
1270 if len(h.probes) != probes+1 {
1271 t.Fatalf("the next pass probed again (%d probes)", len(h.probes)-probes)
1272 }
1273 // A retry starts a new window, and its probe answers.
1274 h.retry(t)
1275 h.pass(t)
1276 h.reopened(t)
1277 if h.verifications() != 1 {
1278 t.Fatalf("verified %d times, want once", h.verifications())
1279 }
1280 })
1281 }
1282 }
1283 // A retry's new window is a whole fifteen minutes, even with the earlier
1284 // exhaustion still journalled when the stuck mark landed.
1285 t.Run("the retry's window", func(t *testing.T) {
1286 h := newRevertHarness(t)
1287 h.fromStopped(t)
1288 h.arm(t)
1289 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
1290 h.pass(t)
1291 h.clock.Advance(16 * time.Minute)
1292 h.pass(t)
1293 if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
1294 t.Fatalf("not stuck: %+v", m)
1295 }
1296 h.retry(t)
1297 h.freshWorker()
1298 h.pass(t)
1299 h.clock.Advance(14 * time.Minute)
1300 h.pass(t)
1301 if m := h.move(t); m.State != provision.MoveReverting {
1302 t.Fatalf("the retry's window was not fresh: %+v", m)
1303 }
1304 h.clock.Advance(2 * time.Minute)
1305 h.pass(t)
1306 if m := h.move(t); m.State != provision.MoveStuck {
1307 t.Fatalf("the retry's window never ended: %+v", m)
1308 }
1309 })
1310}
1311