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 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/activator"
15 "github.com/carlosframework/platform/internal/hostwrite"
16 "github.com/carlosframework/platform/internal/lease"
17 "github.com/carlosframework/platform/internal/moves"
18 "github.com/carlosframework/platform/internal/provision"
19 "github.com/carlosframework/platform/internal/registry"
20 "github.com/carlosframework/platform/internal/release"
21 "github.com/carlosframework/platform/internal/seeds"
22)
23
24// fakeMoveMirror keeps S3's ETag rules (lease's fakeMirror explains them):
25// a single-part object's ETag is its body, a multipart one's is
26// body+"-27", and a copy lands single-part with the body as its ETag. An
27// op log records ordering.
28//
29// A mirror round runs its copies and deletes side by side
30// (lease.DefaultMirrorWorkers), so buckets and ops are guarded by bmu
31// whenever a round can be running; a test reads them directly between
32// rounds. beforeCopy runs without the lock, so it may call the fake. Lock
33// order: mu, then bmu.
34type fakeMoveMirror struct {
35 bmu sync.Mutex
36 buckets map[string]map[string]fakeMoveObject
37 ops []string
38 beforeCopy func(srcKey string)
39
40 // The Phase 1 staging writer's state (movesphase1_harness_test.go):
41 // the guarded call log, a failure hook, and object versions.
42 mu sync.Mutex
43 at []stagingCall
44 failAt func(op, key string) error
45 failGet func(bucket, key string) error // a plain read that fails (the ledger, movescut_test.go)
46 versions int
47 vers map[string]string
48 // keepHistory makes the fake keep every version a later write
49 // superseded, readable and copyable by its id, as a versioned bucket
50 // does (off by default: the fake's older tests read the current
51 // version only).
52 keepHistory bool
53 history map[string]fakeMoveObject
54}
55
56type fakeMoveObject struct {
57 lease.MirrorObject
58 body string
59}
60
61func newFakeMoveMirror() *fakeMoveMirror {
62 return &fakeMoveMirror{buckets: map[string]map[string]fakeMoveObject{}}
63}
64func (f *fakeMoveMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) }
65func (f *fakeMoveMirror) putSized(bucket, key, etag string, size int64) {
66 f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: etag, Size: size}, etag})
67}
68func (f *fakeMoveMirror) putMultipart(bucket, key, body string, size int64) {
69 f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: body + "-27", Size: size}, body})
70}
71func (f *fakeMoveMirror) store(bucket, key string, o fakeMoveObject) {
72 f.bmu.Lock()
73 defer f.bmu.Unlock()
74 f.storeLocked(bucket, key, o)
75}
76
77func (f *fakeMoveMirror) storeLocked(bucket, key string, o fakeMoveObject) {
78 if f.buckets[bucket] == nil {
79 f.buckets[bucket] = map[string]fakeMoveObject{}
80 }
81 f.buckets[bucket][key] = o
82}
83func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) {
84 f.bmu.Lock()
85 defer f.bmu.Unlock()
86 out := map[string]lease.MirrorObject{}
87 for k, v := range f.buckets[bucket] {
88 if strings.HasPrefix(k, prefix) {
89 out[k] = v.MirrorObject
90 }
91 }
92 return out, nil
93}
94func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
95 f.bmu.Lock()
96 before := f.beforeCopy
97 f.bmu.Unlock()
98 if before != nil {
99 before(srcKey)
100 }
101 f.bmu.Lock()
102 defer f.bmu.Unlock()
103 o, ok := f.buckets[srcBucket][srcKey]
104 if !ok || (srcETag != "" && o.ETag != srcETag) {
105 return "", lease.ErrCopySourceChanged
106 }
107 f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey)
108 f.storeLocked(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: o.body, Size: o.Size}, o.body})
109 return o.body, nil
110}
111func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error {
112 f.bmu.Lock()
113 defer f.bmu.Unlock()
114 f.ops = append(f.ops, "delete "+bucket+"/"+key)
115 delete(f.buckets[bucket], key)
116 return nil
117}
118func (f *fakeMoveMirror) Get(_ context.Context, bucket, key string) ([]byte, error) {
119 if f.failGet != nil {
120 if err := f.failGet(bucket, key); err != nil {
121 return nil, err
122 }
123 }
124 f.bmu.Lock()
125 defer f.bmu.Unlock()
126 o, ok := f.buckets[bucket][key]
127 if !ok {
128 return nil, lease.ErrMirrorNotFound
129 }
130 return []byte(o.body), nil
131}
132func (f *fakeMoveMirror) Put(_ context.Context, bucket, key string, body []byte) error {
133 f.bmu.Lock()
134 defer f.bmu.Unlock()
135 f.ops = append(f.ops, "put "+bucket+"/"+key)
136 f.storeLocked(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "receipts", Size: int64(len(body))}, string(body)})
137 return nil
138}
139
140// bodies is a prefix's content by relative key: what a byte-exact
141// comparison compares (ETags differ across a multipart copy by design).
142func (f *fakeMoveMirror) bodies(bucket, prefix string) map[string]string {
143 f.bmu.Lock()
144 defer f.bmu.Unlock()
145 out := map[string]string{}
146 for k, v := range f.buckets[bucket] {
147 if strings.HasPrefix(k, prefix) {
148 out[strings.TrimPrefix(k, prefix)] = v.body
149 }
150 }
151 return out
152}
153
154type moveHarness struct {
155 s release.Store
156 f *fakeMoveMirror
157 cfg moveCfg
158 statuses []string
159 quiet bool
160 owns bool
161}
162
163func newMoveHarness(t *testing.T) *moveHarness {
164 t.Helper()
165 h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true, owns: true}
166 h.cfg = moveCfg{
167 Region: "ie",
168 OwnBucket: "data-ie",
169 OwnBucketRegion: "eu-west-1",
170 OwnPrefix: "instances",
171 DataDir: t.TempDir(),
172 Mirror: func(_ context.Context, dstRegion string) (lease.MirrorStore, error) {
173 return h.f, nil
174 },
175 Quiesced: func(string) bool { return h.quiet },
176 SetStatus: func(host, status string) error { h.statuses = append(h.statuses, host+"="+status); return nil },
177 OwnsRoute: func(host string) bool { return h.owns },
178 Checkpoint: func(_ context.Context, dbPath string) error {
179 h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath))
180 return nil
181 },
182 Now: time.Now,
183 ReadyDelta: defaultMoveReadyDelta, // what armMovesPass sets with the env unset
184 }
185 // The target's edges file: uk publishes a data bucket.
186 if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json",
187 []byte(`{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
188 t.Fatal(err)
189 }
190 return h
191}
192
193// worker wraps the harness cfg the way armMovesPass does in production —
194// a fresh worker per call, so memoized mirrors/nudges never leak between
195// the table's cases.
196func (h *moveHarness) worker() *movesWorker {
197 return &movesWorker{cfg: h.cfg, mirrors: map[string]lease.MirrorStore{}}
198}
199
200func (h *moveHarness) record(t *testing.T, host string, mut ...func(*provision.Record) error) provision.Record {
201 t.Helper()
202 rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab",
203 Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
204 if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
205 t.Fatal(err)
206 }
207 for _, m := range mut {
208 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil {
209 t.Fatal(err)
210 }
211 }
212 got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
213 if err != nil {
214 t.Fatal(err)
215 }
216 return *got
217}
218
219func (h *moveHarness) read(t *testing.T, host string) provision.Record {
220 t.Helper()
221 got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
222 if err != nil {
223 t.Fatal(err)
224 }
225 return *got
226}
227
228func TestMovesPassPrepareAndPromote(t *testing.T) {
229 h := newMoveHarness(t)
230 host := "a.go.example.com"
231 h.record(t, host, provision.MoveRequest("uk", time.Now()))
232 h.f.putSized("data-ie", "instances/"+host+"/generations/g/snapshot", "e1", 64<<20)
233 h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e2")
234
235 // Round 1: copies happen; the SAME pass does not promote, because the
236 // round copied more than the readiness bar (the 64 MiB snapshot): the
237 // cut's own round could have that much again to do.
238 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
239 t.Fatal(err)
240 }
241 if got := h.read(t, host); got.Move.State != provision.MovePreparing {
242 t.Fatalf("after the copying round: %s, want still preparing", got.Move.State)
243 }
244 staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
245 if len(staged) != 2 {
246 t.Fatalf("staging = %v", staged)
247 }
248 // Round 2: nothing to do → ready (zero bytes is under any bar).
249 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
250 t.Fatal(err)
251 }
252 got := h.read(t, host)
253 if got.Move.State != provision.MoveReady || got.Move.ReadyAt == nil {
254 t.Fatalf("after the quiet round: %+v, want ready", got.Move)
255 }
256 // Ready keeps mirroring: a late write flows on the next round.
257 h.f.put("data-ie", "instances/"+host+"/generations/g/wal/001", "e3")
258 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
259 t.Fatal(err)
260 }
261 staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
262 if len(staged) != 3 || h.read(t, host).Move.State != provision.MoveReady {
263 t.Fatalf("late write did not flow: %v", staged)
264 }
265}
266
267func TestMovesPassCutSequence(t *testing.T) {
268 h := newMoveHarness(t)
269 host := "a.go.example.com"
270 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
271 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
272 // A local warm cache that must be purged.
273 db := filepath.Join(h.cfg.DataDir, host+".db")
274 if err := writeFileForTest(db, "stale"); err != nil {
275 t.Fatal(err)
276 }
277
278 // Not quiesced yet: the pass sets the status and stops there.
279 h.quiet = false
280 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
281 t.Fatal(err)
282 }
283 if len(h.statuses) != 1 || h.statuses[0] != host+"=moving" {
284 t.Fatalf("statuses = %v", h.statuses)
285 }
286 if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
287 t.Fatalf("cut advanced past an un-quiesced child: %+v", got)
288 }
289 if len(h.f.ops) != 0 {
290 t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
291 }
292
293 // Quiesced: the whole tail runs, in order — the final own-bucket
294 // checkpoint strictly BEFORE the mirror round (the mirror of a stale
295 // replica is faithful and useless).
296 h.quiet = true
297 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
298 t.Fatal(err)
299 }
300 ckIdx, cpIdx := -1, -1
301 for i, op := range h.f.ops {
302 if strings.HasPrefix(op, "checkpoint ") && ckIdx == -1 {
303 ckIdx = i
304 }
305 if strings.HasPrefix(op, "copy ") && cpIdx == -1 {
306 cpIdx = i
307 }
308 }
309 if ckIdx == -1 || cpIdx == -1 || ckIdx > cpIdx {
310 t.Fatalf("cut op order = %v, want checkpoint before the mirror copy", h.f.ops)
311 }
312 got := h.read(t, host)
313 if got.Region != "uk" || got.Move != nil {
314 t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
315 }
316 if fileExistsForTest(db) {
317 t.Fatal("local DB survived the cut — the warm-cache trap")
318 }
319 staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
320 if len(staged) != 1 {
321 t.Fatalf("staging after final round = %v", staged)
322 }
323 // The pass is now inert for this record (region uk ≠ cfg ie).
324 before := len(h.f.ops)
325 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
326 t.Fatal(err)
327 }
328 if len(h.f.ops) != before {
329 t.Fatal("a completed move kept mirroring")
330 }
331}
332
333func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) {
334 h := newMoveHarness(t)
335 host := "a.go.example.com"
336 h.record(t, host, provision.MoveRequest("se", time.Now())) // se publishes nothing
337 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
338 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
339 t.Fatal(err)
340 }
341 if got := h.read(t, host); got.Move.State != provision.MovePreparing {
342 t.Fatalf("state advanced with no target coords: %+v", got.Move)
343 }
344 if len(h.f.ops) != 0 {
345 t.Fatalf("mirrored toward an unknown target: %v", h.f.ops)
346 }
347}
348
349// A cutting move for a host whose layout migration is in flight (the
350// journal at provision.LayoutJournalPath) does nothing this round: its DB
351// may be mid-copy between the flat and nested paths, so neither checkpoint
352// nor the warm-cache purge is safe until the journal clears. It is the same
353// fence Classify applies to a wake, applied here to a cut.
354func TestMovesPassCutFencedByLayoutJournal(t *testing.T) {
355 h := newMoveHarness(t)
356 host := "a.go.example.com"
357 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
358 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
359 journal := provision.LayoutJournalPath(h.cfg.DataDir, host)
360 if err := os.MkdirAll(filepath.Dir(journal), 0o755); err != nil {
361 t.Fatal(err)
362 }
363 if err := writeFileForTest(journal, "{}"); err != nil {
364 t.Fatal(err)
365 }
366 h.quiet = true
367 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
368 t.Fatal(err)
369 }
370 if len(h.f.ops) != 0 {
371 t.Fatalf("cut proceeded past the layout fence: %v", h.f.ops)
372 }
373 if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
374 t.Fatalf("cut state advanced past the layout fence: %+v", got)
375 }
376}
377
378// TestMovesPassCutFencedWhenTheJournalCannotBeRead pins the fail-closed
379// fix: a journal directory os.Stat cannot even resolve (ENOTDIR here, a
380// plain file sitting where "layout" should be a directory) is not proof
381// there is no journal, and must fence the cut exactly as a readable
382// journal does, not fail open and let it proceed.
383func TestMovesPassCutFencedWhenTheJournalCannotBeRead(t *testing.T) {
384 h := newMoveHarness(t)
385 host := "a.go.example.com"
386 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
387 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
388 // A file where the journal's own parent directory should be: any stat
389 // under it fails with ENOTDIR, never ENOENT.
390 if err := writeFileForTest(filepath.Join(h.cfg.DataDir, "layout"), "not a directory"); err != nil {
391 t.Fatal(err)
392 }
393 h.quiet = true
394 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
395 t.Fatal(err)
396 }
397 if len(h.f.ops) != 0 {
398 t.Fatalf("cut proceeded past an unreadable layout fence: %v", h.f.ops)
399 }
400 if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
401 t.Fatalf("cut state advanced past an unreadable layout fence: %+v", got)
402 }
403}
404
405// TestMovesPassCutFencedByDanglingJournalSymlink pins the same fence for a
406// dangling symlink at the journal path: os.Stat follows the link and reports
407// ErrNotExist for a target that is gone, which would read as "no journal"
408// even though something still sits at the path. Lstat sees the link itself
409// and does not resolve it, so the cut must fence here exactly as it does for
410// a regular journal file.
411func TestMovesPassCutFencedByDanglingJournalSymlink(t *testing.T) {
412 h := newMoveHarness(t)
413 host := "a.go.example.com"
414 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
415 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
416 journal := provision.LayoutJournalPath(h.cfg.DataDir, host)
417 if err := os.MkdirAll(filepath.Dir(journal), 0o755); err != nil {
418 t.Fatal(err)
419 }
420 if err := os.Symlink(filepath.Join(filepath.Dir(journal), "gone.json"), journal); err != nil {
421 t.Fatal(err)
422 }
423 h.quiet = true
424 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
425 t.Fatal(err)
426 }
427 if len(h.f.ops) != 0 {
428 t.Fatalf("cut proceeded past a dangling journal symlink: %v", h.f.ops)
429 }
430 if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
431 t.Fatalf("cut state advanced past a dangling journal symlink: %+v", got)
432 }
433}
434
435// A cutting move for a nested row checkpoints the row's OWN DBPath, not the
436// legacy <DataDir>/<host>.db guess: a nested row's DB lives under
437// provision.InstanceDir and the flat guess names a file that was never
438// written for it.
439func TestMovesPassCutUsesRouteDBPath(t *testing.T) {
440 h := newMoveHarness(t)
441 host := "a.go.example.com"
442 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
443 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
444 nested := provision.NestedDBPath(h.cfg.DataDir, host)
445 if err := os.MkdirAll(filepath.Dir(nested), 0o755); err != nil {
446 t.Fatal(err)
447 }
448 if err := writeFileForTest(nested, "stale"); err != nil {
449 t.Fatal(err)
450 }
451 h.cfg.Route = func(string) (*registry.Route, error) {
452 return &registry.Route{Host: host, DBPath: nested}, nil
453 }
454 h.quiet = true
455 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
456 t.Fatal(err)
457 }
458 found := false
459 for _, op := range h.f.ops {
460 if op == "checkpoint "+filepath.Base(nested) {
461 found = true
462 }
463 }
464 if !found {
465 t.Fatalf("checkpoint did not use the route's own DBPath: %v", h.f.ops)
466 }
467 if fileExistsForTest(nested) {
468 t.Fatal("the route's own DB survived the cut. The warm-cache trap did not follow DBPath")
469 }
470 if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
471 t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
472 }
473}
474
475// The cut fences the host against wakes before it trusts Quiesced (G12),
476// holds the fence through the flip while this box still has the row, and
477// lifts it once the row is retired.
478func TestMovesPassCutFencesWakesUntilTheRouteRetires(t *testing.T) {
479 h := newMoveHarness(t)
480 host := "a.go.example.com"
481 rec := h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
482 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
483 fences := map[string]uint64{}
484 h.cfg.Fence = func(_ context.Context, host string, gen uint64) error {
485 fences[host] = gen
486 return nil
487 }
488 h.cfg.LiftFence = func(host string, gen uint64) bool {
489 if fences[host] != gen {
490 return false
491 }
492 delete(fences, host)
493 return true
494 }
495 routeHeld := true
496 h.cfg.Route = func(string) (*registry.Route, error) {
497 if !routeHeld {
498 return nil, nil
499 }
500 return &registry.Route{Host: host, Status: "moving"}, nil
501 }
502 h.cfg.Quiesced = func(string) bool {
503 if fences[host] == 0 {
504 t.Fatal("Quiesced was asked before the wake fence was installed")
505 }
506 return h.quiet
507 }
508 w := h.worker()
509
510 // Not quiesced: fenced, nothing else.
511 h.quiet = false
512 if err := movesPass(context.Background(), h.s, w); err != nil {
513 t.Fatal(err)
514 }
515 want := uint64(rec.Move.RequestedAt.UnixNano())
516 if fences[host] != want {
517 t.Fatalf("fence generation = %d, want the move's request time %d", fences[host], want)
518 }
519 if len(h.f.ops) != 0 {
520 t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
521 }
522
523 // Quiesced: the cut completes and the fence stays, since the row is
524 // still here waiting for reconcile to retire it.
525 h.quiet = true
526 if err := movesPass(context.Background(), h.s, w); err != nil {
527 t.Fatal(err)
528 }
529 if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
530 t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
531 }
532 if fences[host] != want {
533 t.Fatal("the fence was lifted while this box still held the moving row")
534 }
535
536 // Reconcile retires the row: the next pass lifts the fence.
537 routeHeld = false
538 if err := movesPass(context.Background(), h.s, w); err != nil {
539 t.Fatal(err)
540 }
541 if _, ok := fences[host]; ok {
542 t.Fatal("the fence outlived the route")
543 }
544}
545
546// A fence on a host whose record is at rest here again (a move back landed)
547// is lifted even though this box holds the row: the row serves now.
548func TestMovesPassLiftsTheFenceOfARecordAtRestHere(t *testing.T) {
549 h := newMoveHarness(t)
550 host := "a.go.example.com"
551 h.record(t, host) // region ie, no move: at rest on this box
552 lifted := uint64(0)
553 h.cfg.LiftFence = func(_ string, gen uint64) bool { lifted = gen; return true }
554 h.cfg.Route = func(string) (*registry.Route, error) { return &registry.Route{Host: host, Status: "ready"}, nil }
555 w := h.worker()
556 w.fenced.Store(host, uint64(42))
557 if err := movesPass(context.Background(), h.s, w); err != nil {
558 t.Fatal(err)
559 }
560 if lifted != 42 {
561 t.Fatalf("lifted generation %d, want 42", lifted)
562 }
563 if _, ok := w.fenced.Load(host); ok {
564 t.Fatal("the worker still remembers a lifted fence")
565 }
566}
567
568// Passes run one at a time: the fast converge and the instances loop both
569// call runMovesPass, and a pass that lifted fences from a record list read
570// before another pass's cut would reopen wakes mid-cut.
571func TestMovesPassRunsOneAtATime(t *testing.T) {
572 t.Parallel()
573 h := newMoveHarness(t)
574 w := h.worker()
575 w.pass.Lock()
576 done := make(chan error, 1)
577 go func() { done <- movesPass(context.Background(), h.s, w) }()
578 select {
579 case <-done:
580 t.Fatal("a second pass ran while the first held the worker")
581 case <-time.After(50 * time.Millisecond):
582 }
583 w.pass.Unlock()
584 if err := <-done; err != nil {
585 t.Fatal(err)
586 }
587}
588
589// A fence that cannot be installed stops the cut before it looks at the
590// child: nothing is mirrored, purged or flipped.
591func TestMovesPassCutStopsWhenTheFenceFails(t *testing.T) {
592 h := newMoveHarness(t)
593 host := "a.go.example.com"
594 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
595 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
596 h.cfg.Fence = func(context.Context, string, uint64) error { return errors.New("fence not installed") }
597 h.cfg.Quiesced = func(string) bool {
598 t.Fatal("Quiesced was asked with no fence in place")
599 return true
600 }
601 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
602 t.Fatal(err)
603 }
604 if len(h.f.ops) != 0 {
605 t.Fatalf("the cut ran without a fence: %v", h.f.ops)
606 }
607 if got := h.read(t, host); got.Region != "ie" || got.Move == nil || got.Move.State != provision.MoveCutting {
608 t.Fatalf("after a failed fence: region=%q move=%+v, want still cutting in ie", got.Region, got.Move)
609 }
610}
611
612// moveRound runs one pass and returns the record's move state after it.
613func (h *moveHarness) moveRound(t *testing.T, host string) string {
614 t.Helper()
615 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
616 t.Fatal(err)
617 }
618 got := h.read(t, host)
619 if got.Move == nil {
620 t.Fatalf("move cleared unexpectedly: %+v", got)
621 }
622 return got.Move.State
623}
624
625// An AWAKE instance never stops writing: litestream lands a new L0 file
626// between every two listings, so no round ever copies nothing. Production
627// (#175, kpmg.bdn.oncarlos.com) sat in preparing for 80 minutes that way.
628// The bar is bytes, not zero: once the initial bulk has moved, a round
629// that copied only the small new files is ready.
630func TestMovesPassBusySourceReachesReady(t *testing.T) {
631 h := newMoveHarness(t)
632 host := "a.go.example.com"
633 src := "instances/" + host + "/generations/g/"
634 h.record(t, host, provision.MoveRequest("uk", time.Now()))
635 h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)
636 h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200)
637
638 // Round 1 moves the bulk: well over the bar, still preparing.
639 if st := h.moveRound(t, host); st != provision.MovePreparing {
640 t.Fatalf("after the bulk round: %s, want preparing", st)
641 }
642 // Every later round finds a fresh L0 file and a lease renewal: never a
643 // zero-change round, always a small one.
644 st := ""
645 for i := 1; i <= 3 && st != provision.MoveReady; i++ {
646 h.f.putSized("data-ie", src+fmt.Sprint("ltx/0/", i), fmt.Sprint("l0-", i), 8<<10)
647 h.f.putSized("data-ie", "instances/"+host+"/lock.json", fmt.Sprint("renewal-", i), 200)
648 st = h.moveRound(t, host)
649 }
650 if st != provision.MoveReady {
651 t.Fatalf("a busy source with small deltas never reached ready (state %s)", st)
652 }
653}
654
655// A round that copied more than the bar does not promote, however many
656// rounds it takes: compaction rewriting a 138 MiB level file every round
657// (the #175 shape) keeps the move preparing. The bar is inclusive.
658func TestMovesPassLargeDeltaDoesNotPromote(t *testing.T) {
659 h := newMoveHarness(t)
660 host := "a.go.example.com"
661 src := "instances/" + host + "/generations/g/"
662 h.record(t, host, provision.MoveRequest("uk", time.Now()))
663 h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)
664 for i := 0; i < 4; i++ {
665 // Compaction rewrites the level file under a new ETag each round.
666 h.f.putSized("data-ie", src+"ltx/1/compacted", fmt.Sprint("c", i), 138<<20)
667 if st := h.moveRound(t, host); st != provision.MovePreparing {
668 t.Fatalf("round %d copied 138 MiB and promoted to %s", i, st)
669 }
670 }
671
672 // The boundary, with a small bar: one byte over stays preparing, the
673 // bar exactly promotes.
674 h.cfg.ReadyDelta = 1000
675 h.f.putSized("data-ie", src+"ltx/0/a", "a", 1001)
676 if st := h.moveRound(t, host); st != provision.MovePreparing {
677 t.Fatalf("a 1001-byte round under a 1000-byte bar promoted to %s", st)
678 }
679 h.f.putSized("data-ie", src+"ltx/0/b", "b", 1000)
680 if st := h.moveRound(t, host); st != provision.MoveReady {
681 t.Fatalf("a 1000-byte round under a 1000-byte bar left the move %s", st)
682 }
683}
684
685// A zero bar means zero bytes moved: a round that copied (or deferred)
686// even one byte does not promote. A deletion-only round moves no bytes, so
687// it does, which the old Changed == 0 rule did not allow.
688func TestMovesPassZeroBarNeedsZeroCopiedBytes(t *testing.T) {
689 h := newMoveHarness(t)
690 h.cfg.ReadyDelta = 0
691 host := "a.go.example.com"
692 h.record(t, host, provision.MoveRequest("uk", time.Now()))
693 h.f.putSized("data-ie", "instances/"+host+"/generations/g/wal/000", "e1", 1)
694 if st := h.moveRound(t, host); st != provision.MovePreparing {
695 t.Fatalf("a 1-byte round under a zero bar promoted to %s", st)
696 }
697 if st := h.moveRound(t, host); st != provision.MoveReady {
698 t.Fatalf("a quiet round under a zero bar left the move %s", st)
699 }
700
701 // Deletion-only: retention pruned a staged object, nothing to copy.
702 other := "b.go.example.com"
703 h.record(t, other, provision.MoveRequest("uk", time.Now()))
704 h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/000", "e1", 1)
705 h.moveRound(t, other)
706 delete(h.f.buckets["data-ie"], "instances/"+other+"/generations/g/wal/000")
707 h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/001", "e2", 0)
708 if st := h.moveRound(t, other); st != provision.MoveReady {
709 t.Fatalf("a zero-byte round (one delete, one empty copy) under a zero bar left the move %s", st)
710 }
711}
712
713// Ready no longer means "staging matched the source". It means the gap was
714// small. The cut's final round still has to make staging exact: what was
715// written after the last ready round, what compaction rewrote, and what
716// retention pruned all reach the target, and the lease never does.
717func TestMovesPassCutAfterBusyReadyLandsByteExact(t *testing.T) {
718 h := newMoveHarness(t)
719 host := "a.go.example.com"
720 own := "instances/" + host + "/"
721 gen := own + "generations/g/"
722 h.record(t, host, provision.MoveRequest("uk", time.Now()))
723 h.f.putSized("data-ie", gen+"snapshots/0", "s0", 293<<20)
724 h.f.putSized("data-ie", gen+"ltx/1/compacted", "c0", 138<<20)
725 h.f.putSized("data-ie", gen+"ltx/0/0", "l0-0", 4<<10)
726 h.f.putSized("data-ie", own+"lock.json", "renewal-0", 200)
727 if st := h.moveRound(t, host); st != provision.MovePreparing {
728 t.Fatalf("after the bulk round: %s", st)
729 }
730 h.f.putSized("data-ie", gen+"ltx/0/1", "l0-1", 4<<10)
731 if st := h.moveRound(t, host); st != provision.MoveReady {
732 t.Fatalf("a 4 KiB round left the move %s, want ready", st)
733 }
734 staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
735 if _, ok := staged["moves/"+host+"/generations/g/ltx/0/1"]; !ok {
736 t.Fatalf("the ready round's own delta was not staged: %v", staged)
737 }
738
739 // After ready and before the cut, the instance keeps writing: a new L0
740 // file, a compaction rewrite, retention pruning an old L0 file, and a
741 // lease renewal.
742 h.f.putSized("data-ie", gen+"ltx/0/2", "l0-2", 4<<10)
743 h.f.putSized("data-ie", gen+"ltx/1/compacted", "c1", 139<<20)
744 delete(h.f.buckets["data-ie"], gen+"ltx/0/0")
745 h.f.putSized("data-ie", own+"lock.json", "renewal-1", 200)
746 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil {
747 t.Fatal(err)
748 }
749 h.quiet = true
750 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
751 t.Fatal(err)
752 }
753 if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
754 t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
755 }
756
757 assertStagedExact(t, h, host)
758}
759
760// assertStagedExact compares CONTENT, source against staging, lease
761// excluded: a multipart source and its copy carry different ETags by
762// design, so metadata equality would prove nothing.
763func assertStagedExact(t *testing.T, h *moveHarness, host string) {
764 t.Helper()
765 want := h.f.bodies("data-ie", "instances/"+host+"/")
766 delete(want, "lock.json")
767 got := h.f.bodies("data-uk", "moves/"+host+"/")
768 if len(got) != len(want) {
769 t.Fatalf("staged %d objects, want %d:\nstaged %v\nwant %v", len(got), len(want), got, want)
770 }
771 for k, v := range want {
772 if got[k] != v {
773 t.Fatalf("staged %s = %q, want %q (the cut's final round must make staging exact)", k, got[k], v)
774 }
775 }
776}
777
778// kpmgShape stocks a source like kpmg.bdn.oncarlos.com's (#175): a
779// multipart snapshot and six multipart 138 MB level files, all of which
780// land in staging with a DIFFERENT ETag than the source lists.
781func kpmgShape(h *moveHarness, host string) {
782 gen := "instances/" + host + "/ltx/"
783 h.f.putMultipart("data-ie", gen+"9/snapshot.ltx", "snap", 293<<20)
784 for i := 0; i < 6; i++ {
785 h.f.putMultipart("data-ie", fmt.Sprint(gen, "1/", i, ".ltx"), fmt.Sprint("level-", i), 138<<20)
786 }
787 h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200)
788}
789
790// The production failure (#175): multipart sources never compare equal to
791// their copies, so before receipts every round recopied ~1 GB and the
792// move could never reach ready, busy or not. With receipts, a busy
793// instance whose rounds add only small L0 files reaches ready on the
794// first round after the bulk.
795func TestMovesPassBusyMultipartInstanceReachesReady(t *testing.T) {
796 h := newMoveHarness(t)
797 host := "kpmg.bdn.example.com"
798 h.record(t, host, provision.MoveRequest("uk", time.Now()))
799 kpmgShape(h, host)
800 if st := h.moveRound(t, host); st != provision.MovePreparing {
801 t.Fatalf("after the bulk round: %s, want preparing", st)
802 }
803 h.f.putSized("data-ie", "instances/"+host+"/ltx/0/1.ltx", "l0-1", 8<<10)
804 h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-1", 200)
805 before := len(h.f.ops)
806 if st := h.moveRound(t, host); st != provision.MoveReady {
807 t.Fatalf("a busy multipart instance with an 8 KiB delta left the move %s", st)
808 }
809 for _, op := range h.f.ops[before:] {
810 if strings.Contains(op, "/ltx/1/") || strings.Contains(op, "/ltx/9/") {
811 t.Fatalf("an unchanged multipart object was recopied: %s (ops %v)", op, h.f.ops[before:])
812 }
813 }
814 // The receipts are in the SOURCE's bucket, never in staging.
815 if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok {
816 t.Fatal("no receipts in the source's bucket")
817 }
818 for k := range h.f.buckets["data-uk"] {
819 if strings.Contains(k, "receipt") {
820 t.Fatalf("receipts staged as data: %s", k)
821 }
822 }
823}
824
825// A source object overwritten between the listing and the copy defers
826// that object. Its listed bytes still count against the bar, so a round
827// that deferred a large object does not promote.
828func TestMovesPassDeferredBytesCountAgainstTheBar(t *testing.T) {
829 h := newMoveHarness(t)
830 host := "a.go.example.com"
831 h.record(t, host, provision.MoveRequest("uk", time.Now()))
832 kpmgShape(h, host)
833 h.moveRound(t, host)
834 big := "instances/" + host + "/ltx/1/0.ltx"
835 h.f.putMultipart("data-ie", big, "level-0b", 138<<20)
836 var once sync.Once
837 h.f.beforeCopy = func(k string) {
838 if k == big {
839 once.Do(func() { h.f.putMultipart("data-ie", big, "level-0c", 138<<20) })
840 }
841 }
842 if st := h.moveRound(t, host); st != provision.MovePreparing {
843 t.Fatalf("a round that deferred 138 MiB promoted to %s", st)
844 }
845 if st := h.moveRound(t, host); st != provision.MovePreparing {
846 t.Fatalf("the round that copied the deferred 138 MiB promoted to %s", st)
847 }
848 if st := h.moveRound(t, host); st != provision.MoveReady {
849 t.Fatalf("the quiet round after left the move %s", st)
850 }
851}
852
853// The cut's final round with multipart sources lands byte-exact, and the
854// move's receipts are cleared once the record has flipped.
855func TestMovesPassCutWithMultipartSourcesLandsExact(t *testing.T) {
856 h := newMoveHarness(t)
857 host := "kpmg.bdn.example.com"
858 h.record(t, host, provision.MoveRequest("uk", time.Now()))
859 kpmgShape(h, host)
860 h.moveRound(t, host)
861 if st := h.moveRound(t, host); st != provision.MoveReady {
862 t.Fatalf("state %s, want ready", st)
863 }
864 // Compaction after ready: one level file rewritten, one pruned.
865 h.f.putMultipart("data-ie", "instances/"+host+"/ltx/1/2.ltx", "level-2b", 140<<20)
866 delete(h.f.buckets["data-ie"], "instances/"+host+"/ltx/1/5.ltx")
867 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil {
868 t.Fatal(err)
869 }
870 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
871 t.Fatal(err)
872 }
873 if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
874 t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
875 }
876 assertStagedExact(t, h, host)
877 if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok {
878 t.Fatal("the move's receipts outlived the cut")
879 }
880}
881
882// A source that changes under the cut's final round (it must not, the
883// child is down, but nothing else proves it) leaves staging short of
884// exact. The cut stops before the purge and the flip, and the next pass
885// finishes it exactly.
886func TestMovesPassCutStopsOnADeferredCopy(t *testing.T) {
887 h := newMoveHarness(t)
888 host := "a.go.example.com"
889 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
890 kpmgShape(h, host)
891 db := filepath.Join(h.cfg.DataDir, host+".db")
892 if err := writeFileForTest(db, "warm"); err != nil {
893 t.Fatal(err)
894 }
895 big := "instances/" + host + "/ltx/1/3.ltx"
896 var once sync.Once
897 h.f.beforeCopy = func(k string) {
898 if k == big {
899 once.Do(func() { h.f.putMultipart("data-ie", big, "level-3b", 138<<20) })
900 }
901 }
902 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
903 t.Fatal(err)
904 }
905 if got := h.read(t, host); got.Region != "ie" || got.Move == nil || got.Move.State != provision.MoveCutting {
906 t.Fatalf("the cut completed over a deferred copy: region=%q move=%+v", got.Region, got.Move)
907 }
908 if !fileExistsForTest(db) {
909 t.Fatal("the local DB was purged before staging was exact")
910 }
911 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
912 t.Fatal(err)
913 }
914 if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
915 t.Fatalf("the retried cut: region=%q move=%+v", got.Region, got.Move)
916 }
917 assertStagedExact(t, h, host)
918}
919
920func TestMoveReadyDeltaFromEnv(t *testing.T) {
921 for _, tc := range []struct {
922 raw string
923 want int64
924 ok bool
925 }{
926 {"", defaultMoveReadyDelta, true}, // unset: the default, nothing to report
927 {"0", 0, true}, // zero copied bytes: only a round that moved no data promotes
928 {"1048576", 1 << 20, true},
929 {" 67108864 ", 64 << 20, true},
930 {"-1", defaultMoveReadyDelta, false},
931 {"16MiB", defaultMoveReadyDelta, false},
932 {"1.5", defaultMoveReadyDelta, false},
933 {"99999999999999999999", defaultMoveReadyDelta, false},
934 } {
935 got, ok := moveReadyDeltaFromEnv(func(k string) string {
936 if k != "CARLOS_MOVE_READY_DELTA_BYTES" {
937 t.Fatalf("read %s, want CARLOS_MOVE_READY_DELTA_BYTES", k)
938 }
939 return tc.raw
940 })
941 if got != tc.want || ok != tc.ok {
942 t.Errorf("%q: got %d, %v; want %d, %v", tc.raw, got, ok, tc.want, tc.ok)
943 }
944 }
945 if defaultMoveReadyDelta != 16<<20 {
946 t.Fatalf("default bar = %d, want 16 MiB", defaultMoveReadyDelta)
947 }
948}
949
950// armMovesPass is where the bar reaches the worker: the env override when
951// it parses, the default when it does not.
952func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {
953 prev := armedMovesWorker.Load()
954 t.Cleanup(func() { armedMovesWorker.Store(prev) })
955 for _, tc := range []struct {
956 raw string
957 want int64
958 }{
959 {"", defaultMoveReadyDelta},
960 {"4096", 4096},
961 {"-5", defaultMoveReadyDelta},
962 {"lots", defaultMoveReadyDelta},
963 } {
964 t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)
965 armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)
966 if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {
967 t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)
968 }
969 }
970}
971
972// armMovesPass reads the Phase 1 switch into the worker's config: off when
973// the variable is unset or not an on-value, on for "on".
974func TestArmMovesPassReadsTheSwitch(t *testing.T) {
975 prev := armedMovesWorker.Load()
976 t.Cleanup(func() { armedMovesWorker.Store(prev) })
977 for _, tc := range []struct {
978 raw string
979 want bool
980 }{
981 {"", false}, {"on", true}, {"1", true}, {"true", true}, {"off", false}, {"yes", false},
982 } {
983 t.Setenv(moves.SwitchEnv, tc.raw)
984 armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)
985 if got := armedMovesWorker.Load().cfg.Enabled; got != tc.want {
986 t.Errorf("%s=%q armed Enabled=%v, want %v", moves.SwitchEnv, tc.raw, got, tc.want)
987 }
988 }
989}
990
991func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) }
992func fileExistsForTest(path string) bool { _, err := os.Stat(path); return err == nil }
993
994// The rails and the ownership gate (review round 1): pooled, non-hibernate,
995// non-instance, and not-my-route records are skipped untouched.
996func TestMovesPassRailsAndOwnership(t *testing.T) {
997 h := newMoveHarness(t)
998 host := "a.go.example.com"
999 h.record(t, host, provision.MoveRequest("uk", time.Now()))
1000 h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
1001
1002 // Not this box's route (a standby sharing the region): untouched.
1003 h.owns = false
1004 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1005 t.Fatal(err)
1006 }
1007 if len(h.f.ops) != 0 || h.read(t, host).Move.State != provision.MovePreparing {
1008 t.Fatalf("a non-owning box acted: ops=%v", h.f.ops)
1009 }
1010 h.owns = true
1011
1012 // A pooled sibling with a move field: skipped (the platform box holds
1013 // none of its data — a round here would promote ready over nothing).
1014 pooled := "p.go.example.com"
1015 rec := provision.Record{Schema: 1, Host: pooled, Slug: "p", App: "app", Account: "bab",
1016 Region: "ie", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()}
1017 if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
1018 t.Fatal(err)
1019 }
1020 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", pooled, provision.MoveRequest("uk", time.Now())); err != nil {
1021 t.Fatal(err)
1022 }
1023 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1024 t.Fatal(err)
1025 }
1026 if h.read(t, pooled).Move.State != provision.MovePreparing {
1027 t.Fatal("a pooled record's move advanced on the platform box")
1028 }
1029}
1030
1031// Abort's box half: staging deleted, then the field cleared.
1032func TestMovesPassAbortPhase(t *testing.T) {
1033 h := newMoveHarness(t)
1034 host := "a.go.example.com"
1035 h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort())
1036 h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1")
1037 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1038 t.Fatal(err)
1039 }
1040 staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
1041 if len(staged) != 0 {
1042 t.Fatalf("staging survived the abort: %v", staged)
1043 }
1044 if got := h.read(t, host); got.Move != nil || got.Region != "ie" {
1045 t.Fatalf("after abort phase: %+v region=%q", got.Move, got.Region)
1046 }
1047}
1048
1049// Abort clears the move's receipts along with the staged copy.
1050func TestMovesPassAbortClearsReceipts(t *testing.T) {
1051 h := newMoveHarness(t)
1052 host := "a.go.example.com"
1053 h.record(t, host, provision.MoveRequest("uk", time.Now()))
1054 kpmgShape(h, host)
1055 h.moveRound(t, host)
1056 if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok {
1057 t.Fatal("no receipts after the bulk round")
1058 }
1059 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveAbort()); err != nil {
1060 t.Fatal(err)
1061 }
1062 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1063 t.Fatal(err)
1064 }
1065 if got := h.read(t, host); got.Move != nil {
1066 t.Fatalf("abort did not complete: %+v", got.Move)
1067 }
1068 if staged := h.f.bodies("data-uk", "moves/"+host+"/"); len(staged) != 0 {
1069 t.Fatalf("staging survived the abort: %v", staged)
1070 }
1071 if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok {
1072 t.Fatal("the receipts survived the abort")
1073 }
1074}
1075
1076func TestMovesPassSkipsAnUnplacedSeed(t *testing.T) {
1077 logs := captureLog(t)
1078 h := newMoveHarness(t)
1079 host := "a.go.example.com"
1080 id, _ := seeds.NewID()
1081 rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",
1082 CreatedAt: time.Now().UTC(),
1083 Move: &provision.MoveState{To: "uk", State: provision.MovePreparing, RequestedAt: time.Now().UTC()},
1084 Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: time.Now().UTC()}}
1085 if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
1086 t.Fatal(err)
1087 }
1088 h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e1")
1089 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1090 t.Fatal(err)
1091 }
1092 if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 0 {
1093 t.Fatalf("the worker mirrored an unplaced seed: %v", staged)
1094 }
1095 if got := h.read(t, host); got.Move.State != provision.MovePreparing {
1096 t.Fatalf("state = %s", got.Move.State)
1097 }
1098 if len(h.f.ops) != 0 {
1099 t.Fatalf("a skipped round wrote: %v", h.f.ops)
1100 }
1101 if want := "moves: " + host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {
1102 t.Fatalf("log %q lacks %q", logs(), want)
1103 }
1104 // Placed, the same record moves on the next pass.
1105 if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", host, func(r *provision.Record) error {
1106 r.Seed.State = seeds.StatePlaced
1107 return nil
1108 }); err != nil {
1109 t.Fatal(err)
1110 }
1111 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1112 t.Fatal(err)
1113 }
1114 if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {
1115 t.Fatalf("a placed seed did not mirror: %v", staged)
1116 }
1117}
1118
1119// The skip is for preparing only: a legacy move an old console already
1120// took past preparing has had its effects, and the worker carries it on.
1121func TestMovesPassCarriesOnAReadyMoveWithAnUnplacedSeed(t *testing.T) {
1122 h := newMoveHarness(t)
1123 host := "a.go.example.com"
1124 id, _ := seeds.NewID()
1125 now := time.Now().UTC()
1126 rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",
1127 CreatedAt: now,
1128 Move: &provision.MoveState{To: "uk", State: provision.MoveReady, RequestedAt: now, ReadyAt: &now},
1129 Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: now}}
1130 if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
1131 t.Fatal(err)
1132 }
1133 h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e1")
1134 if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
1135 t.Fatal(err)
1136 }
1137 if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {
1138 t.Fatalf("a ready move stopped mirroring: %v", staged)
1139 }
1140}
1141
1142// seedTheMove gives the Phase 1 harness's record a seed after its move
1143// started: a record that slipped past an old console, whose doors did not
1144// refuse it (spec §5.8). The seed writer is the only writer of Seed.
1145func seedTheMove(t *testing.T, h *phase1Harness, kind, state string) {
1146 t.Helper()
1147 id, err := seeds.NewID()
1148 if err != nil {
1149 t.Fatal(err)
1150 }
1151 if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {
1152 r.Seed = &provision.SeedSource{ID: id, Kind: kind, State: state, By: "t", At: h.clock.Now()}
1153 return nil
1154 }); err != nil {
1155 t.Fatal(err)
1156 }
1157}
1158
1159func setSeedState(t *testing.T, h *phase1Harness, state string) {
1160 t.Helper()
1161 if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {
1162 r.Seed.State = state
1163 return nil
1164 }); err != nil {
1165 t.Fatal(err)
1166 }
1167}
1168
1169// The Phase 1 source round refuses an uncommitted import before its first
1170// effect: no staging lease, no claim touch, no staging write, no source
1171// report. Committed, the same move prepares.
1172func TestPhase1SourceSkipsAnUncommittedImport(t *testing.T) {
1173 logs := captureLog(t)
1174 h := newPhase1Harness(t)
1175 h.start(t, "manual")
1176 seedTheMove(t, h, seeds.KindImport, seeds.StateSwapped)
1177 before := h.claim(t)
1178 h.pass(t)
1179 if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
1180 t.Fatalf("state = %s, want preparing", got)
1181 }
1182 if h.lease.acquired != 0 {
1183 t.Fatalf("the skipped round took the staging lease %d times", h.lease.acquired)
1184 }
1185 if calls := h.f.stagingCalls(); len(calls) != 0 {
1186 t.Fatalf("the skipped round staged: %+v", calls)
1187 }
1188 if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); len(keys) != 0 {
1189 t.Fatalf("the lane holds %v", keys)
1190 }
1191 if sr := h.sourceReport(t); sr != nil {
1192 t.Fatalf("the skipped round published a source report: %+v", sr)
1193 }
1194 if after := h.claim(t); fmt.Sprint(after.Touched) != fmt.Sprint(before.Touched) {
1195 t.Fatalf("the skipped round touched the claim: %+v, was %+v", after.Touched, before.Touched)
1196 }
1197 if len(h.probePaths) != 0 {
1198 t.Fatalf("the skipped round probed the instance: %v", h.probePaths)
1199 }
1200 if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUncommitted); !strings.Contains(logs(), want) {
1201 t.Fatalf("log %q lacks %q", logs(), want)
1202 }
1203
1204 setSeedState(t, h, seeds.StateCommitted)
1205 h.pass(t)
1206 if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); !containsString(keys, "move.json") {
1207 t.Fatalf("a committed import did not prepare: %v", keys)
1208 }
1209 if h.sourceReport(t) == nil {
1210 t.Fatal("a committed import published no source report")
1211 }
1212}
1213
1214// The target box writes no report for a move whose seed is unplaced.
1215// Placed, it does.
1216func TestPhase1TargetSkipsAnUnplacedSeed(t *testing.T) {
1217 logs := captureLog(t)
1218 h := newPhase1Harness(t)
1219 h.start(t, "manual")
1220 seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)
1221 h.targetPass(t, h.targetCfg())
1222 if tr := h.targetReport(t); tr != nil {
1223 t.Fatalf("the target reported on an unplaced seed: %+v", tr)
1224 }
1225 if calls := h.f.stagingCalls(); len(calls) != 0 {
1226 t.Fatalf("the target's skipped round staged: %+v", calls)
1227 }
1228 if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
1229 t.Fatalf("state = %s, want preparing", got)
1230 }
1231 if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {
1232 t.Fatalf("log %q lacks %q", logs(), want)
1233 }
1234
1235 setSeedState(t, h, seeds.StatePlaced)
1236 h.targetPass(t, h.targetCfg())
1237 if h.targetReport(t) == nil {
1238 t.Fatal("the target wrote no report for a placed seed")
1239 }
1240}
1241
1242// A box that is neither the move's source nor its target says nothing
1243// about the seed: the log line belongs to the boxes the move waits on.
1244func TestMovesPassSeedNoticeOnlyOnParticipants(t *testing.T) {
1245 logs := captureLog(t)
1246 h := newPhase1Harness(t)
1247 h.start(t, "manual")
1248 seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)
1249 c := h.targetCfg()
1250 c.Region, c.Box = "us", "us-box"
1251 h.targetPass(t, c)
1252 if strings.Contains(logs(), provision.SeedMoveSentence(provision.SeedMoveUnplaced)) {
1253 t.Fatalf("a bystander logged the seed: %q", logs())
1254 }
1255}
1256