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 "encoding/json"
6 "errors"
7 "os"
8 "path/filepath"
9 "sort"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/activator"
15 "github.com/carlosframework/platform/internal/appconfig"
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/hostguard"
18 "github.com/carlosframework/platform/internal/hostwrite"
19 "github.com/carlosframework/platform/internal/lease"
20 "github.com/carlosframework/platform/internal/moves"
21 "github.com/carlosframework/platform/internal/park"
22 "github.com/carlosframework/platform/internal/provision"
23 "github.com/carlosframework/platform/internal/registry"
24)
25
26// The source's finalization harness: the landing harness driven through
27// restored, then committed the way Task 7 will (the record's probing and
28// probed steps, MoveCommit, and hostclaim.Commit with the source's location
29// park scoped by sourceLocationParkScope), and the SOURCE box (ie-box) given
30// a real registry holding its row "moving" under the move with two timers,
31// the config ack the box last wrote, and the database's marker and
32// quarantine beside the database the cut harness left at the flat path.
33// The source's park engine runs over the fake versioned store and the same
34// fake lease table as the landing's (fakeParkLeases).
35
36const finalAckBody = `{"v":1,"host":"a.go.example.com"}`
37
38type finalHarness struct {
39 *landingHarness
40 reg *registry.Registry
41 finalAt string // the source's crash point; "" crashes nowhere
42 finalDone bool
43 finalHits []string
44 crashAll map[string]bool // crash once at every point not in it (the replay test)
45 deletes int // DeleteHeld calls that deleted
46}
47
48func newFinalHarness(t *testing.T, opts ...p1Option) *finalHarness {
49 t.Helper()
50 return newFinalHarnessFrom(t, newLandingHarness(t, opts...))
51}
52
53// newFinalHarnessFrom is newFinalHarness over a landing harness the caller
54// prepared before the flip.
55func newFinalHarnessFrom(t *testing.T, lh *landingHarness) *finalHarness {
56 t.Helper()
57 lh.flip(t)
58 return newFinalHarnessFlipped(t, lh)
59}
60
61// newFinalHarnessFlipped is newFinalHarnessFrom over a landing harness the
62// caller already flipped (and may have changed after the flip).
63func newFinalHarnessFlipped(t *testing.T, lh *landingHarness) *finalHarness {
64 t.Helper()
65 h := &finalHarness{landingHarness: lh}
66 h.land(t)
67 h.installed(t)
68 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
69 if err != nil {
70 t.Fatal(err)
71 }
72 t.Cleanup(func() { reg.Close() })
73 h.reg = reg
74 src := h.phase1Harness.route
75 src.Status, src.HoldOp = "moving", p1ID
76 if err := reg.Upsert(src); err != nil {
77 t.Fatal(err)
78 }
79 for _, tm := range []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000},
80 {Host: p1Host, Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000}} {
81 if err := reg.UpsertTimer(tm); err != nil {
82 t.Fatal(err)
83 }
84 }
85 dd := h.cfg.DataDir
86 for name, body := range map[string]string{p1Host + ".db.gen": "cur", p1Host + ".db.corrupt": "quarantined",
87 p1Host + ".config.env": "SECRET=1\n", p1Host + "-sidecar.env": "SIDE=1\n"} {
88 if err := os.WriteFile(filepath.Join(dd, name), []byte(body), 0o600); err != nil {
89 t.Fatal(err)
90 }
91 }
92 if _, err := h.s.Put(context.Background(), appconfig.ConfigAckKey(h.srcBox, p1Host), []byte(finalAckBody), ""); err != nil {
93 t.Fatal(err)
94 }
95 h.cfg.Route = func(host string) (*registry.Route, error) { return reg.Get(host) }
96 h.cfg.OwnsRoute = func(host string) bool {
97 rt, err := reg.Get(host)
98 return err == nil && rt != nil && rt.Provisioned
99 }
100 h.cfg.DeleteHeld = func(host, op string) (bool, error) {
101 ok, err := reg.DeleteIfHoldWithTimers(host, op)
102 if ok {
103 h.deletes++
104 }
105 return ok, err
106 }
107 h.cfg.Guard = hostguard.Guard{DataDir: dd, LockDir: t.TempDir()}
108 h.cfg.Local = park.Local{DataDir: dd, Box: h.srcBox}
109 h.cfg.ParkLeases = h.leases
110 h.cfg.ParkView = sourceTestViewFor(park.PlacementRegion(h.src), park.PlacementRegion(h.dst))
111 h.cfg.ParkAuthorize = sourceTestAuthorizeFor(h.srcBox, park.PlacementRegion(h.src))
112 h.cfg.ParkLocalAuthorize = sourceTestLocalAuthorizeFor(h.srcBox, park.PlacementRegion(h.src))
113 h.cfg.parkFault = h.lcfg.parkFault
114 h.cfg.finalFault = func(point string) error {
115 h.finalHits = append(h.finalHits, point)
116 if h.crashAll != nil && !h.crashAll[point] {
117 h.crashAll[point] = true
118 return errCrash
119 }
120 if point == h.finalAt && !h.finalDone {
121 h.finalDone = true
122 return errCrash
123 }
124 return nil
125 }
126 return h
127}
128
129// advanceToProbed walks the record and the claim from restored through
130// probing to probed, as Task 6 will.
131func (h *finalHarness) advanceToProbed(t *testing.T) {
132 t.Helper()
133 ctx := context.Background()
134 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
135 for _, step := range [][2]string{{provision.LandRestored, provision.LandProbing}, {provision.LandProbing, provision.LandProbed}} {
136 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveLandingPhase(g, step[0], step[1], nil)); err != nil {
137 t.Fatal(err)
138 }
139 obs, err := hostclaim.Read(ctx, h.s, p1Host)
140 if err != nil {
141 t.Fatal(err)
142 }
143 if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.SetMovePhase(p1ID, hostclaim.KindLanding, step[0], step[1])); err != nil {
144 t.Fatal(err)
145 }
146 }
147}
148
149// commitRecord is the record's commit alone (MoveCommit).
150func (h *finalHarness) commitRecord(t *testing.T) {
151 t.Helper()
152 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
153 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveCommit(g, h.clock.Now())); err != nil {
154 t.Fatal(err)
155 }
156}
157
158// commitClaim is the claim's commit (hostclaim.Commit), with the source's
159// location park scoped from the flip (H7) and the target's staging lane.
160func (h *finalHarness) commitClaim(t *testing.T) {
161 t.Helper()
162 ctx := context.Background()
163 rec := h.read(t, p1Host)
164 obs, err := hostclaim.Read(ctx, h.s, p1Host)
165 if err != nil {
166 t.Fatal(err)
167 }
168 scope, err := sourceLocationParkScope(rec, *obs.Claim.Holder)
169 if err != nil {
170 t.Fatal(err)
171 }
172 staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
173 if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.Commit(p1ID, park.PlacementRegion(h.src), h.srcBox, park.PlacementRegion(h.dst), scope, staging, obligationID)); err != nil {
174 t.Fatal(err)
175 }
176}
177
178func (h *finalHarness) commit(t *testing.T) {
179 t.Helper()
180 h.advanceToProbed(t)
181 h.commitRecord(t)
182 h.commitClaim(t)
183}
184
185// sourcePass runs one pass as the source box.
186func (h *finalHarness) sourcePass(t *testing.T) { t.Helper(); h.pass(t) }
187
188// crashSource runs one source pass that must reach the crash point.
189func (h *finalHarness) crashSource(t *testing.T) {
190 t.Helper()
191 h.sourcePass(t)
192 if !h.finalDone {
193 t.Fatalf("the pass never reached %s (points reached: %v)", h.finalAt, h.finalHits)
194 }
195}
196
197func (h *finalHarness) sourceJournal(t *testing.T) *moves.Journal {
198 t.Helper()
199 j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
200 if err != nil {
201 t.Fatal(err)
202 }
203 return j
204}
205
206// live lists which of the source's artifacts are at their live paths.
207func (h *finalHarness) live(t *testing.T) []string {
208 t.Helper()
209 var out []string
210 for _, p := range append(testSourcePaths(), filepath.Join(".moves", p1Host, p1ID)) {
211 if _, err := os.Lstat(filepath.Join(h.cfg.DataDir, p)); err == nil {
212 out = append(out, p)
213 } else if !errors.Is(err, os.ErrNotExist) {
214 t.Fatal(err)
215 }
216 }
217 return out
218}
219
220// parked lists every regular file under .parked/<inc>/<host>/, as
221// "<retirement>/<rest>", sorted.
222func (h *finalHarness) parked(t *testing.T) []string {
223 t.Helper()
224 root := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host)
225 var out []string
226 filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
227 if err == nil && !d.IsDir() {
228 rel, _ := filepath.Rel(root, p)
229 out = append(out, filepath.ToSlash(rel))
230 }
231 return nil
232 })
233 sort.Strings(out)
234 return out
235}
236
237// untouched asserts that nothing of the source's was finalized.
238func (h *finalHarness) untouched(t *testing.T, why string) {
239 t.Helper()
240 if got, want := h.live(t), []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}; strings.Join(got, " ") != strings.Join(want, " ") {
241 t.Fatalf("%s: live artifacts %v, want %v", why, got, want)
242 }
243 if got := h.parked(t); len(got) != 0 {
244 t.Fatalf("%s: parked %v", why, got)
245 }
246 if rt, err := h.reg.Get(p1Host); err != nil || rt == nil {
247 t.Fatalf("%s: the row: %+v %v", why, rt, err)
248 }
249 if tm, _ := h.reg.TimersFor(p1Host); len(tm) != 2 {
250 t.Fatalf("%s: timers %+v", why, tm)
251 }
252 if m := h.read(t, p1Host).Move; m == nil || m.SourceParkedAt != nil {
253 t.Fatalf("%s: record %+v", why, m)
254 }
255 if _, _, err := h.s.Get(context.Background(), appconfig.ConfigAckKey(h.srcBox, p1Host)); err != nil {
256 t.Fatalf("%s: the config ack: %v", why, err)
257 }
258 for _, step := range []string{finalStepRemote, finalStepInstallation, finalStepLocal + "-plan", finalStepRowDeleted} {
259 if h.sourceJournal(t).Done(step) {
260 t.Fatalf("%s: the journal has %s", why, step)
261 }
262 }
263 if calls := h.vs.callsOn("parked/"); len(calls) != 0 {
264 t.Fatalf("%s: the source's location was parked: %v", why, calls)
265 }
266}
267
268// finalized asserts the end state of a source finalization of an
269// unregistered source: every artifact in one retirement under the engine's
270// layout with the op's journal and a receipt naming the op, nothing live,
271// the row and its timers gone, the config ack moved into the report
272// prefix, source_parked_at stamped, and the source's location parked.
273func (h *finalHarness) finalized(t *testing.T) string {
274 t.Helper()
275 ctx := context.Background()
276 if got := h.live(t); len(got) != 0 {
277 t.Fatalf("still live: %v", got)
278 }
279 parked := h.parked(t)
280 if len(parked) == 0 {
281 t.Fatal("nothing parked")
282 }
283 ret, _, _ := strings.Cut(parked[0], "/")
284 var rest []string
285 for _, p := range parked {
286 r, tail, _ := strings.Cut(p, "/")
287 if r != ret {
288 t.Fatalf("more than one retirement: %v", parked)
289 }
290 rest = append(rest, tail)
291 }
292 jpath := "artifacts/.moves/" + p1Host + "/" + p1ID + "/journal.json"
293 want := []string{"artifacts/" + p1Host + "/upload.txt", "artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.corrupt", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env",
294 "artifacts/" + p1Host + ".config.env", "artifacts/" + p1Host + "-sidecar.env", localParkReceiptName}
295 var got []string
296 sawJournal := false
297 for _, r := range rest {
298 switch {
299 case r == jpath:
300 sawJournal = true
301 case strings.HasPrefix(r, "artifacts/.moves/"):
302 // the op journal directory's other contents (the archive scratch)
303 default:
304 got = append(got, r)
305 }
306 }
307 sort.Strings(want)
308 if strings.Join(got, " ") != strings.Join(want, " ") || !sawJournal {
309 t.Fatalf("parked %v (journal %v), want %v and the journal", got, sawJournal, want)
310 }
311 art := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, "artifacts")
312 for name, body := range map[string]string{p1Host + ".db": "the stopped database", p1Host + "/upload.txt": "hello", p1Host + ".env": "APP_MODE=prod\n", p1Host + ".db.gen": "cur"} {
313 if b, err := os.ReadFile(filepath.Join(art, name)); err != nil || string(b) != body {
314 t.Fatalf("parked %s: %q %v", name, b, err)
315 }
316 }
317 var receipt localParkReceipt
318 if b, err := os.ReadFile(filepath.Join(filepath.Dir(art), localParkReceiptName)); err != nil || json.Unmarshal(b, &receipt) != nil || receipt.Op != p1ID || receipt.Retirement != ret || receipt.Incarnation != h.inc {
319 t.Fatalf("receipt %+v %v", receipt, err)
320 }
321 var pj moves.Journal
322 if b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, jpath)); err != nil || json.Unmarshal(b, &pj) != nil {
323 t.Fatalf("the parked journal: %v", err)
324 }
325 if pj.Steps[finalStepRowDeleted] != "done" || !pj.Done(finalStepParked) || pj.Steps[finalStepLocal] != ret || pj.Steps[finalStepInstallation] != "unregistered" {
326 t.Fatalf("the parked journal's steps: %v", pj.Steps)
327 }
328 if rt, err := h.reg.Get(p1Host); err != nil || rt != nil {
329 t.Fatalf("the row: %+v %v", rt, err)
330 }
331 if tm, err := h.reg.TimersFor(p1Host); err != nil || len(tm) != 0 {
332 t.Fatalf("timers: %+v %v", tm, err)
333 }
334 if _, _, err := h.s.Get(ctx, appconfig.ConfigAckKey("ie-box", p1Host)); !errors.Is(err, os.ErrNotExist) {
335 t.Fatalf("the config ack is still there: %v", err)
336 }
337 if b, _, err := h.s.Get(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID)+finalConfigAckName); err != nil || string(b) != finalAckBody {
338 t.Fatalf("the config ack's copy: %q %v", b, err)
339 }
340 m := h.read(t, p1Host).Move
341 if m == nil || m.State != provision.MoveCommitted || m.SourceParkedAt == nil {
342 t.Fatalf("record: %+v", m)
343 }
344 // The source's location is archived and gone from its live keys; its
345 // park-replica is still owed until the horizon (the regional runner's).
346 for k := range h.f.buckets["data-ie"] {
347 if strings.HasPrefix(k, cutRoot+"/") {
348 t.Fatalf("still at the source's location: %s", k)
349 }
350 }
351 archived := 0
352 for k := range h.f.buckets["data-ie"] {
353 if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") && strings.Contains(k, "/instances/") {
354 archived++
355 }
356 }
357 if archived == 0 {
358 t.Fatal("nothing archived from the source's location")
359 }
360 if held := h.leases.holding(); len(held) != 0 {
361 t.Fatalf("leases still held: %v", held)
362 }
363 return ret
364}
365
366// Review Focus 5: the source finalizes only a committed record whose row
367// this move still holds (or whose journal says the row is deleted).
368func TestSourceParkWaitsForCommit(t *testing.T) {
369 t.Parallel()
370 t.Run("a landing record", func(t *testing.T) {
371 h := newFinalHarness(t)
372 h.advanceToProbed(t)
373 h.sourcePass(t)
374 h.untouched(t, "landing/probed")
375 })
376 t.Run("a committed record whose claim is not committed", func(t *testing.T) {
377 h := newFinalHarness(t)
378 h.advanceToProbed(t)
379 h.commitRecord(t)
380 h.sourcePass(t)
381 h.untouched(t, "claim at probed")
382 })
383 t.Run("a committed record whose row another op holds", func(t *testing.T) {
384 h := newFinalHarness(t)
385 h.commit(t)
386 if ok, err := h.reg.ReleaseHold(p1Host, p1ID); err != nil || !ok {
387 t.Fatal(ok, err)
388 }
389 if err := h.reg.SetHold(p1Host, "restoring", p1ID2); err != nil {
390 t.Fatal(err)
391 }
392 h.sourcePass(t)
393 h.untouched(t, "row held by another op")
394 if rt, _ := h.reg.Get(p1Host); rt.HoldOp != p1ID2 {
395 t.Fatalf("the other op's row: %+v", rt)
396 }
397 })
398 t.Run("a committed record with no row and no journalled delete", func(t *testing.T) {
399 h := newFinalHarness(t)
400 h.commit(t)
401 if err := h.reg.Delete(p1Host); err != nil {
402 t.Fatal(err)
403 }
404 h.sourcePass(t)
405 if got := h.live(t); len(got) != 8 || len(h.parked(t)) != 0 || h.read(t, p1Host).Move.SourceParkedAt != nil {
406 t.Fatalf("finalized without the row: live %v parked %v", got, h.parked(t))
407 }
408 })
409 t.Run("then the commit", func(t *testing.T) {
410 h := newFinalHarness(t)
411 h.advanceToProbed(t)
412 h.sourcePass(t)
413 h.commitRecord(t)
414 h.commitClaim(t)
415 h.sourcePass(t)
416 h.finalized(t)
417 })
418}
419
420// The end to end finalization of an unregistered source (ruling H5(b)): the
421// database, its marker and quarantine, the files directory and the env are
422// gone from the live paths and in one retirement under the engine's layout;
423// the row and its timers rows are gone; source_parked_at is set; the op
424// journal is parked last; a further pass changes nothing.
425func TestSourceParkRetiresInstallationAndRow(t *testing.T) {
426 t.Parallel()
427 h := newFinalHarness(t)
428 h.commit(t)
429 if c := h.claim(t); len(c.Obligations) != 2 {
430 t.Fatalf("the commit queued %+v (want the source's location and the target's lane, no park-local)", c.Obligations)
431 }
432 h.sourcePass(t)
433 h.finalized(t)
434 if h.deletes != 1 {
435 t.Fatalf("the row was deleted %d times", h.deletes)
436 }
437 before := h.parked(t)
438 h.freshWorker()
439 h.sourcePass(t)
440 if after := h.parked(t); strings.Join(after, " ") != strings.Join(before, " ") || h.deletes != 1 {
441 t.Fatalf("a further pass changed the park: %v", after)
442 }
443 if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
444 t.Fatal("a finalized source still routes the record here")
445 }
446}
447
448// Past the horizon the engine acknowledges the source's location park in
449// the same run, and the finalization goes on.
450func TestSourceParkPastTheHorizonAcknowledges(t *testing.T) {
451 t.Parallel()
452 h := newFinalHarness(t)
453 h.commit(t)
454 h.clock.Advance(park.SignatureAge + cutRequestLimit + time.Minute)
455 h.sourcePass(t)
456 h.finalized(t)
457 for _, o := range h.claim(t).Obligations {
458 if o.Kind == "park-replica" && o.Region == "ie" {
459 t.Fatalf("the source's location park is still owed: %+v", o)
460 }
461 }
462}
463
464// The remote half comes first: while the archive is incomplete (a delete
465// the store accepts but does not land) nothing local moves.
466func TestSourceParkWaitsForTheRemoteHalf(t *testing.T) {
467 t.Parallel()
468 h := newFinalHarness(t)
469 h.commit(t)
470 stuck := true
471 h.vs.skipDelete = func(key string) bool { return stuck && strings.HasPrefix(key, cutRoot+"/") }
472 h.sourcePass(t)
473 if got := h.live(t); len(got) != 8 {
474 t.Fatalf("the local half ran before the remote: live %v", got)
475 }
476 if j := h.sourceJournal(t); j.Done(finalStepRemote) || j.Done(finalStepLocal+"-plan") {
477 t.Fatalf("journal: %v", j.Steps)
478 }
479 if rt, _ := h.reg.Get(p1Host); rt == nil {
480 t.Fatal("the row went before the remote half")
481 }
482 stuck = false
483 h.sourcePass(t)
484 h.finalized(t)
485}
486
487// Round-2 finding 12: a crash between the row's delete and the journal's
488// "done" (and before MoveSourceParked) is replayed: the record's region is
489// the target's and the row is gone, so only the op journal routes the
490// record back here, and the replay treats the pending delete as done and
491// stamps source_parked_at.
492func TestSourceParkSurvivesACrashAfterTheRowDelete(t *testing.T) {
493 t.Parallel()
494 for _, point := range []string{"after-row-pending", "after-row-delete"} {
495 t.Run(point, func(t *testing.T) {
496 h := newFinalHarness(t)
497 h.commit(t)
498 h.finalAt = point
499 h.crashSource(t)
500 if j := h.sourceJournal(t); j.Steps[finalStepRowDeleted] != "pending" {
501 t.Fatalf("journal: %v", j.Steps)
502 }
503 rt, _ := h.reg.Get(p1Host)
504 if (point == "after-row-delete") != (rt == nil) {
505 t.Fatalf("the row after %s: %+v", point, rt)
506 }
507 if h.read(t, p1Host).Move.SourceParkedAt != nil {
508 t.Fatal("stamped before the crash point")
509 }
510 if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleSource {
511 t.Fatal("the journal does not route the record back to the source")
512 }
513 h.freshWorker()
514 h.sourcePass(t)
515 h.finalized(t)
516 if h.deletes != 1 {
517 t.Fatalf("deleted %d times", h.deletes)
518 }
519 })
520 }
521}
522
523// Every crash point, each once, with a fresh worker after each: the replay
524// converges on the same end state, with each artifact renamed once, the row
525// deleted once and one retirement.
526func TestSourceParkReplayIsIdempotent(t *testing.T) {
527 t.Parallel()
528 h := newFinalHarness(t)
529 h.commit(t)
530 h.crashAll = map[string]bool{}
531 h.parkAt = "delete"
532 for i := 0; i < 40 && len(h.live(t)) > 0; i++ {
533 h.freshWorker()
534 h.sourcePass(t)
535 }
536 h.finalized(t)
537 if h.deletes != 1 {
538 t.Fatalf("deleted %d times", h.deletes)
539 }
540 want := []string{"after-" + finalStepRemote, "after-" + finalStepInstallation, "after-" + finalStepLocal + "-plan",
541 "after-" + finalStepLocal,
542 "after-row-pending", "after-row-delete", "after-" + finalStepRowDeleted, "after-config-ack-copy", "after-" + finalStepConfigAck,
543 "after-source-parked-pending", "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
544 for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env"} {
545 want = append(want, "after-rename-unsynced:"+p, "after-rename:"+p)
546 }
547 for _, p := range want {
548 if !h.crashAll[p] {
549 t.Errorf("never crashed at %s (crashed at %v)", p, h.crashAll)
550 }
551 }
552 if !h.parkDone {
553 t.Error("the park engine never crashed at its delete")
554 }
555}
556
557// A registered source (the commit queued its park-local because the
558// inventory lists paths of the journalled stamp S) retires its installation
559// through park.Local under the guard, which completes the park-local; what
560// the inventory does not list is parked by the moves-owned rename.
561func TestSourceParkRetiresARegisteredInstallation(t *testing.T) {
562 t.Parallel()
563 h := newFinalHarness(t)
564 dd := h.cfg.DataDir
565 for _, d := range []string{"hostguard", ".parked"} {
566 if err := os.MkdirAll(filepath.Join(dd, d), 0o770); err != nil {
567 t.Fatal(err)
568 }
569 }
570 rec := `{"v":1,"host":"` + p1Host + `","incarnation":"` + h.inc + `","active_stamp":1,"retired_stamps":[]}`
571 if err := os.WriteFile(filepath.Join(dd, "hostguard", p1Host+".json"), []byte(rec), 0o660); err != nil {
572 t.Fatal(err)
573 }
574 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
575 if c.Op.Journal[hostclaim.JournalSourceStamp] != "1" {
576 t.Fatalf("fixture: source stamp %q", c.Op.Journal[hostclaim.JournalSourceStamp])
577 }
578 if c.Inventory == nil {
579 c.Inventory = map[string]hostclaim.InventoryEntry{}
580 }
581 c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}
582 })
583 h.commit(t)
584 var chore hostclaim.Obligation
585 for _, o := range h.claim(t).Obligations {
586 if o.Kind == "park-local" {
587 chore = o
588 }
589 }
590 if chore.Box != "ie-box" {
591 t.Fatalf("the commit queued no park-local for the source: %+v", h.claim(t).Obligations)
592 }
593 h.sourcePass(t)
594 c := h.claim(t)
595 for _, o := range c.Obligations {
596 if o.Kind == "park-local" {
597 t.Fatalf("the park-local is still owed: %+v", o)
598 }
599 }
600 if len(c.Inventory["ie-box"].Paths) != 0 {
601 t.Fatalf("inventory: %+v", c.Inventory["ie-box"])
602 }
603 if got := h.live(t); len(got) != 0 {
604 t.Fatalf("still live: %v", got)
605 }
606 parked := strings.Join(h.parked(t), " ")
607 for _, p := range []string{chore.Retirement + "/artifacts/" + p1Host + ".db", chore.Retirement + "/artifacts/" + p1Host + "/upload.txt", chore.Retirement + "/.journal.json"} {
608 if !strings.Contains(parked, p) {
609 t.Fatalf("park.Local's retirement lacks %s: %s", p, parked)
610 }
611 }
612 j := h.parkedJournal(t)
613 if j.Steps[finalStepInstallation] != "retired" {
614 t.Fatalf("installation step %q", j.Steps[finalStepInstallation])
615 }
616 ret := j.Steps[finalStepLocal]
617 if ret == chore.Retirement || !strings.Contains(parked, ret+"/artifacts/"+p1Host+".db.corrupt") {
618 t.Fatalf("the unregistered quarantine is not in the moves-owned retirement %s: %s", ret, parked)
619 }
620 if captured, err := hostguard.Capture(dd, p1Host); err != nil || captured.Stamp != 0 {
621 t.Fatalf("the installation record is still active: %+v %v", captured, err)
622 }
623 if rt, _ := h.reg.Get(p1Host); rt != nil || h.read(t, p1Host).Move.SourceParkedAt == nil {
624 t.Fatal("the row or the stamp")
625 }
626}
627
628// parkedJournal reads the op journal from wherever the finalization parked it.
629func (h *finalHarness) parkedJournal(t *testing.T) moves.Journal {
630 t.Helper()
631 var j moves.Journal
632 for _, p := range h.parked(t) {
633 if strings.HasSuffix(p, "/.moves/"+p1Host+"/"+p1ID+"/journal.json") {
634 b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, p))
635 if err != nil || json.Unmarshal(b, &j) != nil {
636 t.Fatalf("parked journal: %v", err)
637 }
638 return j
639 }
640 }
641 t.Fatalf("no parked journal in %v", h.parked(t))
642 return j
643}
644
645// The rename moves only what its plan recorded: after the plan, an
646// artifact replaced at its live path (another identity) and one that
647// appears where the plan found nothing are both left where they are,
648// untouched; everything recorded is parked.
649func TestSourceParkLeavesWhatItDidNotRecord(t *testing.T) {
650 t.Parallel()
651 h := newFinalHarness(t)
652 h.commit(t)
653 dd := h.cfg.DataDir
654 if err := os.Remove(filepath.Join(dd, p1Host+".db.corrupt")); err != nil {
655 t.Fatal(err)
656 }
657 h.finalAt = "after-" + finalStepLocal + "-plan"
658 h.crashSource(t)
659 envPath := filepath.Join(dd, p1Host+".env")
660 if err := os.Rename(envPath, envPath+".old"); err != nil {
661 t.Fatal(err)
662 }
663 if err := os.WriteFile(envPath, []byte("SOMEONE=else\n"), 0o600); err != nil {
664 t.Fatal(err)
665 }
666 if err := os.WriteFile(filepath.Join(dd, p1Host+".db.corrupt"), []byte("new"), 0o600); err != nil {
667 t.Fatal(err)
668 }
669 h.freshWorker()
670 h.sourcePass(t)
671 for name, body := range map[string]string{p1Host + ".env": "SOMEONE=else\n", p1Host + ".db.corrupt": "new"} {
672 if b, err := os.ReadFile(filepath.Join(dd, name)); err != nil || string(b) != body {
673 t.Fatalf("%s was touched: %q %v", name, b, err)
674 }
675 }
676 parked := strings.Join(h.parked(t), " ")
677 if strings.Contains(parked, p1Host+".env") || strings.Contains(parked, p1Host+".db.corrupt") || !strings.Contains(parked, "artifacts/"+p1Host+".db ") {
678 t.Fatalf("parked %s", parked)
679 }
680 if h.read(t, p1Host).Move.SourceParkedAt == nil {
681 t.Fatal("the finalization did not complete")
682 }
683}
684
685// H7: the source's location park's horizon starts at the flip, so a final
686// checkpoint request signed after the stop evidence was collected, and
687// before the flip, is inside it; a park scoped from the stop evidence is
688// refused before the engine runs.
689func TestSourceLocationParkScopeCoversTheFinalCheckpoint(t *testing.T) {
690 t.Parallel()
691 checked := time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)
692 flipped := checked.Add(40 * time.Second)
693 signed := checked.Add(30 * time.Second) // the final checkpoint, after CheckedAt
694 rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, FlippedAt: &flipped}}
695 scope, err := sourceLocationParkScope(rec, hostclaim.Holder{Location: cutRoot + "/", Incarnation: "inc"})
696 if err != nil || scope.Location != cutRoot || !scope.EvidenceAt.Equal(flipped) || scope.RequestLimit != cutRequestLimit {
697 t.Fatalf("scope %+v %v", scope, err)
698 }
699 horizon := scope.EvidenceAt.Add(park.SignatureAge + scope.RequestLimit)
700 if latest := signed.Add(park.SignatureAge + cutRequestLimit); latest.After(horizon) {
701 t.Fatalf("a request signed at %s can land until %s, after the horizon %s", signed, latest, horizon)
702 }
703 if early := checked.Add(park.SignatureAge + cutRequestLimit); !signed.Add(park.SignatureAge + cutRequestLimit).After(early) {
704 t.Fatal("fixture: the stop evidence's horizon would have covered it too")
705 }
706 if _, err := sourceLocationParkScope(provision.Record{Move: &provision.MoveState{}}, hostclaim.Holder{}); err == nil {
707 t.Fatal("a scope without the flip")
708 }
709
710 h := newFinalHarness(t)
711 h.advanceToProbed(t)
712 h.commitRecord(t)
713 rec = h.read(t, p1Host)
714 obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
715 if err != nil {
716 t.Fatal(err)
717 }
718 early := hostclaim.ParkScope{Location: cutRoot, Incarnation: h.inc, EvidenceAt: rec.Move.FlippedAt.Add(-time.Minute), RequestLimit: cutRequestLimit}
719 staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
720 if _, err := hostclaim.Update(context.Background(), h.s, obs, hostclaim.Commit(p1ID, "ie", "ie-box", "uk", early, staging, obligationID)); err != nil {
721 t.Fatal(err)
722 }
723 h.sourcePass(t)
724 h.untouched(t, "a horizon before the flip")
725}
726
727// The routing rule for the states after the flip (H5(a)): this box's
728// region names the role, and the row or the op's journal says it has work.
729func TestMovesParticipantRecord(t *testing.T) {
730 dd := t.TempDir()
731 owns := false
732 cfg := moveCfg{Region: "ie", DataDir: dd, OwnsRoute: func(string) bool { return owns }}
733 rec := func(state, from, to string) provision.Record {
734 return provision.Record{Host: p1Host, Region: to, Move: &provision.MoveState{ID: p1ID, State: state, From: from, To: to}}
735 }
736 if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleNone {
737 t.Fatalf("no row and no journal: %q", got)
738 }
739 owns = true
740 for _, c := range []struct {
741 rec provision.Record
742 want string
743 }{
744 {rec(provision.MoveCommitted, "ie", "uk"), roleSource},
745 {rec(provision.MoveRollingBack, "ie", "uk"), roleSource},
746 {rec(provision.MoveReverting, "ie", "uk"), roleSource},
747 {rec(provision.MoveCommitted, "uk", "ie"), roleTarget},
748 {rec(provision.MoveCommitted, "fr", "uk"), roleNone},
749 {rec(provision.MoveLanding, "ie", "uk"), roleNone},
750 {rec(provision.MovePreparing, "ie", "uk"), roleNone},
751 {rec(provision.MoveCommitted, "ie", "ie"), roleNone},
752 {provision.Record{Host: p1Host, Move: &provision.MoveState{State: provision.MoveCommitted, From: "ie", To: "uk"}}, roleNone},
753 {provision.Record{Host: p1Host}, roleNone},
754 } {
755 if got := movesParticipantRecord(cfg, c.rec); got != c.want {
756 t.Errorf("%+v: %q, want %q", c.rec.Move, got, c.want)
757 }
758 }
759 owns = false
760 j := &moves.Journal{Host: p1Host, Op: p1ID}
761 if err := j.Commit(dd); err != nil {
762 t.Fatal(err)
763 }
764 if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleSource {
765 t.Fatalf("the op's journal: %q", got)
766 }
767 if got := movesParticipantRecord(cfg, provision.Record{Host: p1Host, Move: &provision.MoveState{ID: "../x", State: provision.MoveCommitted, From: "ie", To: "uk"}}); got != roleNone {
768 t.Fatalf("an unusable op id: %q", got)
769 }
770}
771
772// Production leaves the finalization unarmed (no park engine leases until
773// Plan 5, ruling G17) but wires the registry's delete; an unarmed pass
774// changes nothing.
775func TestSourceParkWaitsUnarmedInProduction(t *testing.T) {
776 t.Parallel()
777 h := newFinalHarness(t)
778 h.commit(t)
779 for name, unarm := range map[string]func(*moveCfg){
780 "no park leases": func(c *moveCfg) { c.ParkLeases = nil },
781 "no park view": func(c *moveCfg) { c.ParkView = nil },
782 "no park authorizer": func(c *moveCfg) { c.ParkAuthorize = nil },
783 "no claim local authorizer": func(c *moveCfg) { c.ParkLocalAuthorize = nil },
784 } {
785 saved := h.cfg
786 unarm(&h.cfg)
787 if sourceFinalArmed(h.cfg) {
788 t.Fatalf("%s: armed", name)
789 }
790 h.sourcePass(t)
791 h.untouched(t, name)
792 h.cfg = saved
793 }
794 var prod moveCfg
795 reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
796 if err != nil {
797 t.Fatal(err)
798 }
799 defer reg.Close()
800 hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
801 armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
802 if prod.DeleteHeld == nil || prod.Guard.DataDir == "" || prod.Local.DataDir == "" || prod.Versioned == nil {
803 t.Fatalf("production wiring: %+v", prod)
804 }
805 prod.Route = func(string) (*registry.Route, error) { return nil, nil }
806 if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || sourceFinalArmed(prod) {
807 t.Fatal("production armed the finalization")
808 }
809}
810
811// The rename helper on its own: absent candidates, a refused path, and a
812// destination that holds something else.
813func TestParkLocalArtifactsRefusals(t *testing.T) {
814 dd := t.TempDir()
815 j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
816 mark := func(step, value string) error { j.Mark(step, value); return nil }
817 p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", NewID: func() string { return "r1" }}
818 for _, bad := range []string{"../x", "/abs", ".parked/x", "hostguard", "a/../b", ""} {
819 p.Paths = []string{bad}
820 if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
821 t.Errorf("%q was accepted", bad)
822 }
823 }
824 if len(j.Steps) != 0 {
825 t.Fatalf("a refused park journalled %v", j.Steps)
826 }
827 if err := os.WriteFile(filepath.Join(dd, "a"), []byte("a"), 0o600); err != nil {
828 t.Fatal(err)
829 }
830 p.Paths = []string{"a", "absent"}
831 p.Fault = func(point string) error {
832 if point == "after-rename:a" {
833 return errCrash
834 }
835 return nil
836 }
837 if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
838 t.Fatalf("crash: %v", err)
839 }
840 // Something else at the destination stops the park; nothing more moves.
841 dst := filepath.Join(parkedArtifactsDir(dd, "inc", p1Host, "r1"), "a")
842 if err := os.Rename(dst, filepath.Join(dd, "a")); err != nil {
843 t.Fatal(err)
844 }
845 if err := os.WriteFile(dst, []byte("other"), 0o600); err != nil {
846 t.Fatal(err)
847 }
848 p.Fault = nil
849 if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
850 t.Fatal("a foreign destination was accepted")
851 }
852 if b, _ := os.ReadFile(filepath.Join(dd, "a")); string(b) != "a" {
853 t.Fatalf("the live artifact moved: %q", b)
854 }
855 os.Remove(dst)
856 if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
857 t.Fatalf("resume: %q %v", r, err)
858 }
859 if _, err := os.Lstat(filepath.Join(dd, "a")); !errors.Is(err, os.ErrNotExist) {
860 t.Fatalf("a is still live: %v", err)
861 }
862 // Another plan under the same step is refused.
863 p.Paths = []string{"b"}
864 j.Unmark("s")
865 if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
866 t.Fatal("a plan for other paths was accepted")
867 }
868}
869
870// testSourcePaths is what the tests expect the source to keep at its flat
871// live paths and park: the database with its marker, journal, quarantine
872// and litestream shadow, the files directory, <host>.env, and the two
873// rendered env files (fix round 1, ruling 3).
874func testSourcePaths() []string {
875 db := p1Host + ".db"
876 return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", p1Host, p1Host + ".env",
877 p1Host + ".config.env", p1Host + "-sidecar.env"}
878}
879
880// sourceTestView and sourceTestAuthorize are the park view and authorizer
881// the tests inject for the source's location park, in place of Plan 5's
882// owners (fix round 1, ruling 4: production never substitutes stand-ins).
883// The target's region is authoritative after the commit, so every version
884// under the location is selected; a chore is authorized only when it is
885// this move's, for ie-box or ie, under the move's committed landing with
886// ie-box its source.
887func sourceTestView(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
888 return sourceTestViewFor("ie", "uk")(ctx, obs)
889}
890
891// sourceTestViewFor is sourceTestView for a source seen from region with
892// the target's region authoritative (the harness's sides: intoHome's is ca
893// with @home authoritative).
894func sourceTestViewFor(region, authoritative string) func(context.Context, hostclaim.Observed) (park.View, error) {
895 return func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
896 if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
897 return park.View{}, park.ErrUnsafe
898 }
899 location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
900 return park.View{ClaimETag: obs.ETag, Region: region, AuthoritativeRegion: authoritative,
901 GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
902 }
903}
904
905// sourceTestLocalAuthorize is the claim's local authorizer (M9) for the
906// source's park-local: sourceTestAuthorize over the claim park.Local read.
907func sourceTestLocalAuthorize(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
908 return sourceTestLocalAuthorizeFor("ie-box", "ie")(ctx, c, o)
909}
910
911// sourceTestLocalAuthorizeFor is sourceTestLocalAuthorize for the source
912// box and its engine region.
913func sourceTestLocalAuthorizeFor(box, region string) func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error {
914 authorize := sourceTestAuthorizeFor(box, region)
915 return func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
916 return authorize(ctx, hostclaim.Observed{Host: p1Host, Claim: c}, o)
917 }
918}
919
920func sourceTestAuthorize(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
921 return sourceTestAuthorizeFor("ie-box", "ie")(ctx, obs, o)
922}
923
924// sourceTestAuthorizeFor is sourceTestAuthorize for the source box and its
925// engine region.
926func sourceTestAuthorizeFor(box, region string) func(context.Context, hostclaim.Observed, hostclaim.Obligation) error {
927 return func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
928 c := obs.Claim
929 if obs.Absent || c == nil || c.Op.ID != p1ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted ||
930 len(c.Op.Participants) != 2 || c.Op.Participants[0] != box || o.Op != p1ID {
931 return park.ErrUnsafe
932 }
933 if (o.Kind == "park-local" && o.Box == box) || (o.Kind == "park-replica" && o.Region == region) {
934 return nil
935 }
936 return park.ErrUnsafe
937 }
938}
939
940// Fix round 1, ruling 1: a crash after source_parked_at is stamped and
941// before the journal is parked, then the close (which clears the move, so
942// no record routes here any more), then a fresh source pass: the local
943// sweep parks the journal, because the close proves the stamp landed.
944func TestSourceParkJournalSurvivesTheClose(t *testing.T) {
945 t.Parallel()
946 h := newFinalHarness(t)
947 h.commit(t)
948 h.finalAt = "after-source-parked-record"
949 h.crashSource(t)
950 if j := h.sourceJournal(t); j.Done(finalStepParked) && j.Steps[finalStepParked] == "done" {
951 t.Fatalf("fixture: the stamp is journalled done: %v", j.Steps)
952 }
953 ctx := context.Background()
954 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
955 for _, mut := range []func(*provision.Record) error{provision.MoveOpened(g, h.clock.Now()), provision.MoveClose(g, h.clock.Now())} {
956 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, mut); err != nil {
957 t.Fatal(err)
958 }
959 }
960 if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID {
961 t.Fatalf("fixture: not closed: %+v", rec)
962 }
963 if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
964 t.Fatal("fixture: a closed record still routes here")
965 }
966 h.freshWorker()
967 h.sourcePass(t)
968 if got := h.live(t); len(got) != 0 {
969 t.Fatalf("still live after the close: %v", got)
970 }
971 j := h.parkedJournal(t)
972 if j.Steps[finalStepParked] != "done" {
973 t.Fatalf("the parked journal's stamp: %v", j.Steps)
974 }
975}
976
977// The sweep never touches a journal that has not reached the stamp, nor one
978// whose stamp is pending while the record still carries the op unstamped: a
979// journal of another op (a landing's, say) under a record with no move, and
980// this op's journal stopped right after it journalled the stamp pending,
981// before the stamp's CAS.
982func TestSourceParkSweepLeavesUnstampedJournals(t *testing.T) {
983 t.Parallel()
984 h := newFinalHarness(t)
985 h.commit(t)
986 // Another op's journal that would park cleanly if the sweep took it: a
987 // complete local plan and retirement, everything but the stamp step.
988 plan, err := json.Marshal(localParkPlan{Retirement: "r0", Incarnation: h.inc})
989 if err != nil {
990 t.Fatal(err)
991 }
992 otherSteps := map[string]string{finalStepLocal + "-plan": string(plan), finalStepLocal: "r0", finalStepRowDeleted: "done"}
993 other, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2)
994 if err != nil {
995 t.Fatal(err)
996 }
997 for k, v := range otherSteps {
998 other.Mark(k, v)
999 }
1000 if err := other.Commit(h.cfg.DataDir); err != nil {
1001 t.Fatal(err)
1002 }
1003 h.finalAt = "after-source-parked-pending"
1004 h.crashSource(t)
1005 if j := h.sourceJournal(t); j.Steps[finalStepParked] != "pending" || h.read(t, p1Host).Move.SourceParkedAt != nil {
1006 t.Fatalf("fixture: %v", j.Steps)
1007 }
1008 h.cfg.ParkLeases = nil // unarmed: only the sweep runs
1009 h.freshWorker()
1010 h.sourcePass(t)
1011 for _, op := range []string{p1ID, p1ID2} {
1012 if !moveJournalPresent(h.cfg.DataDir, p1Host, op) {
1013 t.Fatalf("the sweep took %s's journal before its stamp", op)
1014 }
1015 }
1016 if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2); err != nil || len(j.Steps) != len(otherSteps) {
1017 t.Fatalf("the sweep wrote into another op's journal: %v %v", j.Steps, err)
1018 }
1019}
1020
1021// Fix round 1, ruling 2: a replay that finds an artifact already at its
1022// destination (a stop after the rename, before either sync) syncs the
1023// source's parent and every destination parent before it goes on; and the
1024// rename offers an interruption point between the rename and the syncs.
1025func TestParkLocalArtifactsSyncsOnReplay(t *testing.T) {
1026 dd := t.TempDir()
1027 if err := os.MkdirAll(filepath.Join(dd, "d", "e"), 0o755); err != nil {
1028 t.Fatal(err)
1029 }
1030 if err := os.WriteFile(filepath.Join(dd, "d", "e", "a"), []byte("a"), 0o600); err != nil {
1031 t.Fatal(err)
1032 }
1033 j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
1034 mark := func(step, value string) error { j.Mark(step, value); return nil }
1035 var points []string
1036 p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", Paths: []string{"d/e/a"}, NewID: func() string { return "r1" },
1037 Fault: func(point string) error {
1038 points = append(points, point)
1039 if point == "after-rename-unsynced:d/e/a" {
1040 return errCrash
1041 }
1042 return nil
1043 }}
1044 if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
1045 t.Fatalf("no interruption between the rename and its syncs: %v (points %v)", err, points)
1046 }
1047 var synced []string
1048 saved := retireSync
1049 retireSync = func(path string) error { synced = append(synced, path); return saved(path) }
1050 t.Cleanup(func() { retireSync = saved })
1051 p.Fault = nil
1052 if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
1053 t.Fatalf("replay: %q %v", r, err)
1054 }
1055 art := parkedArtifactsDir(dd, "inc", p1Host, "r1")
1056 for _, want := range []string{filepath.Join(dd, "d", "e"), filepath.Join(art, "d", "e"), filepath.Join(art, "d"), art, filepath.Dir(art), filepath.Join(dd, ".parked")} {
1057 found := false
1058 for _, s := range synced {
1059 found = found || s == want
1060 }
1061 if !found {
1062 t.Errorf("the replay did not sync %s (synced %v)", want, synced)
1063 }
1064 }
1065}
1066