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 "io/fs"
8 "os"
9 "path/filepath"
10 "sort"
11 "strconv"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/park"
18
19 "github.com/carlosframework/platform/internal/lease"
20 "github.com/carlosframework/platform/internal/moves"
21 "github.com/carlosframework/platform/internal/provision"
22 "github.com/carlosframework/platform/internal/release"
23)
24
25const mvEvMoveID = "mv_abcdefghijklmnopqrstuvwxyz"
26
27// evRec is a record whose move is in the given state, edited by mut.
28func mvEvRec(state string, mut func(*provision.MoveState)) provision.Record {
29 m := &provision.MoveState{ID: mvEvMoveID, Incarnation: strings.Repeat("ab", 32), From: "ie", To: "uk", Cut: "auto", State: state, RequestedAt: time.Unix(1_800_000_000, 0).UTC()}
30 if mut != nil {
31 mut(m)
32 }
33 return provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie", Move: m}
34}
35
36func mvEvAt(sec int64) *time.Time {
37 t := time.Unix(1_800_000_000+sec, 0).UTC()
38 return &t
39}
40
41// The events a full move that ends done goes through, each record the one a
42// writer leaves, in order. The console's request is seen by the source's
43// pass (requested); every later record is a box writer's.
44func doneFlow() []provision.Record {
45 cut := func(phase string) provision.Record {
46 return mvEvRec(provision.MoveCutting, func(m *provision.MoveState) {
47 m.CutPhase, m.CutAt, m.CutBudgetS = phase, mvEvAt(10), 45
48 })
49 }
50 land := func(phase string, attempts int) provision.Record {
51 r := mvEvRec(provision.MoveLanding, func(m *provision.MoveState) {
52 m.CutPhase, m.CutAt, m.CutBudgetS, m.FlippedAt, m.LandingPhase = provision.CutFlipped, mvEvAt(10), 45, mvEvAt(20), phase
53 if attempts > 0 {
54 m.Landing = &provision.MoveLandingJournal{Attempts: attempts}
55 }
56 })
57 r.Region = "uk"
58 return r
59 }
60 committed := func(mut func(*provision.MoveState)) provision.Record {
61 r := mvEvRec(provision.MoveCommitted, func(m *provision.MoveState) {
62 m.CutPhase, m.CutAt, m.FlippedAt, m.CommittedAt = provision.CutFlipped, mvEvAt(10), mvEvAt(20), mvEvAt(40)
63 m.Landing = &provision.MoveLandingJournal{Attempts: 1}
64 if mut != nil {
65 mut(m)
66 }
67 })
68 r.Region = "uk"
69 return r
70 }
71 closed := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "uk",
72 LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "done", At: *mvEvAt(60), OfflineS: 41.2}}
73 return []provision.Record{
74 mvEvRec(provision.MovePreparing, nil),
75 mvEvRec(provision.MoveReady, nil),
76 cut(provision.CutRequested),
77 cut(provision.CutAdmitted), cut(provision.CutStopping), cut(provision.CutStopped), cut(provision.CutStaged), cut(provision.CutArchived),
78 land(provision.LandChecking, 0),
79 land(provision.LandParking, 0), land(provision.LandBuilding, 0), land(provision.LandPublished, 0), land(provision.LandLocal, 0),
80 land(provision.LandConfigured, 0), land(provision.LandRestored, 0), land(provision.LandProbing, 0), land(provision.LandProbing, 1),
81 land(provision.LandProbed, 1),
82 committed(nil),
83 committed(func(m *provision.MoveState) { m.OpenedAt = mvEvAt(51) }),
84 committed(func(m *provision.MoveState) { m.OpenedAt, m.SourceParkedAt = mvEvAt(51), mvEvAt(55) }),
85 closed,
86 }
87}
88
89var doneFlowEvents = []string{
90 "requested", "ready", "cut_requested", "admitted", "stopping", "stopped", "staged", "archived", "flipped",
91 "parking", "building", "published", "local", "configured", "restored", "probing", "probe_attempt", "probed",
92 "committed", "opened", "source_parked", "done",
93}
94
95// eventWorker is a worker whose LogEvent hook records every event it is
96// handed.
97func eventWorker(got *[]moves.LogEvent) *movesWorker {
98 w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
99 w.cfg.Box = "box-test"
100 w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() }
101 w.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
102 *got = append(*got, ev)
103 return nil
104 }
105 return w
106}
107
108// Review Focus 3: across a full done flow, every transition gives one event,
109// in order, with its identity; the record seen again (a status poll, a pass
110// that changes nothing, a replay of the same write) gives none.
111func TestEveryTransitionEmitsOneEvent(t *testing.T) {
112 ctx := context.Background()
113 var got []moves.LogEvent
114 w := eventWorker(&got)
115 flow := doneFlow()
116 w.observeMove(ctx, nil, flow[0])
117 for i := 1; i < len(flow); i++ {
118 w.logTransition(ctx, nil, flow[i-1], flow[i])
119 // A status poll, a pass that changes nothing, a replayed write. A
120 // closed move's identities are forgotten (create-only storage refuses
121 // a replay of the close), so the close is not replayed here.
122 w.observeMove(ctx, nil, flow[i])
123 w.logTransition(ctx, nil, flow[i], flow[i])
124 if i < len(flow)-1 {
125 w.logTransition(ctx, nil, flow[i-1], flow[i])
126 }
127 }
128 var events []string
129 for _, ev := range got {
130 events = append(events, ev.Event)
131 if ev.Op != mvEvMoveID || ev.Host != "acme.go.tito.io" || ev.Actor != "box-test" || ev.V != 1 || ev.At.IsZero() {
132 t.Errorf("event %+v", ev)
133 }
134 }
135 if strings.Join(events, " ") != strings.Join(doneFlowEvents, " ") {
136 t.Fatalf("events\n got %v\nwant %v", events, doneFlowEvents)
137 }
138 ids := map[string]bool{}
139 for _, ev := range got {
140 if ids[ev.Identity] {
141 t.Fatalf("identity %q twice", ev.Identity)
142 }
143 ids[ev.Identity] = true
144 }
145 for _, want := range []string{"requested", "ready", "cutting-requested", "cutting-stopped", "cutting-flipped", "landing-probing", "probe_attempt-1", "committed", "committed-opened", "done"} {
146 if !ids[want] {
147 t.Errorf("no event with identity %q in %v", want, ids)
148 }
149 }
150 if f := got[len(got)-1].Fields; f["offline_s"] != "41.2" {
151 t.Errorf("done carries %v, want offline_s", f)
152 }
153 if f := got[19].Fields; got[19].Event != "opened" || f["offline_s"] != "41" {
154 t.Errorf("opened carries %v, want offline_s from cut_at to the open", f)
155 }
156}
157
158// A rollback, a stuck and a retry: the rollback's phases are named for the
159// half that moved; stuck and retry are keyed by the recovery retry epoch
160// (claim Plan 10's RecoveryRetryEpoch, preflight L11).
161func TestRollbackStuckAndRetryEvents(t *testing.T) {
162 ctx := context.Background()
163 var got []moves.LogEvent
164 w := eventWorker(&got)
165 rb := func(target, source string, mut func(*provision.MoveState)) provision.Record {
166 return mvEvRec(provision.MoveRollingBack, func(m *provision.MoveState) {
167 m.CutAt = mvEvAt(10)
168 m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline, TargetPhase: target, SourcePhase: source}
169 if mut != nil {
170 mut(m)
171 }
172 })
173 }
174 stuck := func(epoch uint64) provision.Record {
175 return mvEvRec(provision.MoveStuck, func(m *provision.MoveState) {
176 m.CutAt = mvEvAt(10)
177 m.Rollback = &provision.MoveRollback{ID: "rb_x", TargetPhase: provision.RbTargetCleaned, SourcePhase: provision.RbSourceVerified, StuckAt: provision.RbSourceVerified}
178 m.StuckReason, m.RecoveryRetryEpoch = provision.StuckBaselineMismatch, epoch
179 })
180 }
181 landing := mvEvRec(provision.MoveLanding, func(m *provision.MoveState) { m.LandingPhase = provision.LandProbing })
182 seq := []provision.Record{
183 landing,
184 rb(provision.RbTargetReleasing, provision.RbSourceWaiting, nil),
185 rb(provision.RbTargetReleased, provision.RbSourceWaiting, nil),
186 rb(provision.RbTargetReleased, provision.RbSourceClaimed, nil),
187 rb(provision.RbTargetCleaned, provision.RbSourceVerified, nil),
188 stuck(0),
189 }
190 for i := 1; i < len(seq); i++ {
191 w.logTransition(ctx, nil, seq[i-1], seq[i])
192 }
193 // The operator's retry (the console's write, seen by the box) advances
194 // the epoch and leaves stuck; the second stuck is a new entry.
195 retried := rb(provision.RbTargetCleaned, provision.RbSourceVerified, func(m *provision.MoveState) { m.RecoveryRetryEpoch = 1 })
196 w.observeMove(ctx, nil, retried)
197 w.observeMove(ctx, nil, retried)
198 w.logTransition(ctx, nil, retried, stuck(1))
199 w.logTransition(ctx, nil, stuck(1), provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
200 LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "rolled_back", At: *mvEvAt(90), OfflineS: 80}})
201 var events, ids []string
202 for _, ev := range got {
203 events = append(events, ev.Event)
204 ids = append(ids, ev.Identity)
205 }
206 want := "rolling_back released claimed cleaned stuck retry stuck rolled_back"
207 if strings.Join(events, " ") != want {
208 t.Fatalf("events %v, want %s", events, want)
209 }
210 wantIDs := "rolling_back rolling_back-target-released rolling_back-source-claimed rolling_back-target-cleaned stuck-0 retry-1 stuck-1 rolled_back"
211 if strings.Join(ids, " ") != wantIDs {
212 t.Fatalf("identities %v, want %s", ids, wantIDs)
213 }
214}
215
216type memberSink struct{ events []string }
217
218func (s *memberSink) App(acct, app, host, line string) {}
219func (s *memberSink) Event(acct, app, host, level, msg string) {
220 s.events = append(s.events, acct+"/"+app+"/"+host+" "+level+" "+msg)
221}
222
223// Preflight L9 and L27: the production LogEvent writes the object
224// create-only, and only the writer that created it tells members; a second
225// writer, or a replay, does neither. Steps members do not need (stopping)
226// are logged and not shown to them.
227func TestMoveLogWriterTellsMembersOnce(t *testing.T) {
228 ctx := context.Background()
229 s := release.NewDirStore(t.TempDir())
230 sink := &memberSink{}
231 write := moveLogWriter(sink)
232 flow := doneFlow()
233 ev := func(event, identity string, fields map[string]string) moves.LogEvent {
234 return moves.LogEvent{V: 1, Op: mvEvMoveID, Host: "acme.go.tito.io", Event: event, Identity: identity, Actor: "box-uk", At: time.Now().UTC(), Fields: fields}
235 }
236 done := flow[len(flow)-1]
237 write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
238 write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
239 write(ctx, s, flow[4], ev("stopping", "cutting-stopping", map[string]string{"from": "ie", "to": "uk"}))
240 if len(sink.events) != 1 || sink.events[0] != "bab/titogo/acme.go.tito.io info Moved. acme.go.tito.io is served from uk. It was offline for 41 seconds." {
241 t.Fatalf("member events %q", sink.events)
242 }
243 prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", mvEvMoveID)
244 logged, err := moves.ReadLog(ctx, s, prefix)
245 if err != nil || len(logged) != 2 || logged[0].Actor != "box-uk" {
246 t.Fatalf("log %+v %v", logged, err)
247 }
248}
249
250// The member lines (preflight L9, Paul's choice): M-C1, M-C5, M-C6, M-C7,
251// M-C8, M-C9, and for stuck a line with no operator command in it.
252func TestMemberLines(t *testing.T) {
253 flow := doneFlow()
254 rolling := mvEvRec(provision.MoveRollingBack, func(m *provision.MoveState) {
255 m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline}
256 })
257 stuck := mvEvRec(provision.MoveStuck, nil)
258 rolledBack := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
259 LastMove: &provision.MoveOutcome{ID: mvEvMoveID, Outcome: "rolled_back"}}
260 for _, c := range []struct {
261 event string
262 rec provision.Record
263 level string
264 line string
265 }{
266 {"requested", flow[0], "info", "Moving acme.go.tito.io from ie to uk."},
267 {"admitted", flow[3], "info", "uk is ready. Starting the cut. acme.go.tito.io will be offline for about 45 seconds."},
268 {"flipped", flow[8], "info", "Stopped in ie. Starting in uk."},
269 {"rolling_back", rolling, "warn", "uk could not start acme.go.tito.io: " + provision.RollbackLandingDeadline + ". Moving it back to ie."},
270 {"rolled_back", rolledBack, "warn", "acme.go.tito.io is back in ie and serving. Nothing was lost."},
271 {"stuck", stuck, "error", "acme.go.tito.io could not start in uk or in ie. It is offline. An operator has been paged."},
272 {"stopping", flow[4], "", ""},
273 } {
274 // The entry's own fields, as the writer names the transition (the
275 // admitted cut's budget, the rollback's reason): the line needs
276 // nothing else, so an entry the outbox writes later says the same.
277 ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": "ie", "to": "uk"}}
278 if before := (provision.Record{Host: c.rec.Host}); c.rec.Move != nil {
279 before.Move = &provision.MoveState{ID: c.rec.Move.ID, From: "ie", To: "uk", State: provision.MoveLanding}
280 if c.event == "admitted" {
281 before = flow[2]
282 }
283 if tr, ok := moves.TransitionOf(before, c.rec); ok && tr.Event == c.event {
284 ev.Fields = tr.Fields
285 }
286 }
287 line, ok := memberLine(ev, c.rec.Host)
288 level := ""
289 if ok {
290 level = memberLevel(c.event)
291 }
292 if ok != (c.line != "") || level != c.level || line != c.line {
293 t.Errorf("%s: %q %q %v, want %q %q", c.event, level, line, ok, c.level, c.line)
294 }
295 }
296}
297
298// ---- Task 3 fix round 1 ----
299
300// Ruling 4: the box knows the home region only as "", so the member line
301// names it with the reviewed phrase (MV-N381), never dropping the line.
302func TestHomeRegionMemberLines(t *testing.T) {
303 fromHome := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.From = "" })
304 toHome := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.To = "" })
305 for _, c := range []struct {
306 rec provision.Record
307 event string
308 want string
309 }{
310 {fromHome, "requested", "Moving acme.go.tito.io from the home region to uk."},
311 {toHome, "requested", "Moving acme.go.tito.io from ie to the home region."},
312 {toHome, "flipped", "Stopped in ie. Starting in the home region."},
313 } {
314 ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": c.rec.Move.From, "to": c.rec.Move.To}}
315 if line, ok := memberLine(ev, c.rec.Host); !ok || line != c.want {
316 t.Errorf("%s: %q %v, want %q", c.event, line, ok, c.want)
317 }
318 }
319}
320
321// Ruling 5: the events carry the transition's evidence. The cut's stopped
322// carries the source's digests, the landing's probed the installed
323// database, a stuck its role, and the source's recovered the recovery's
324// timing.
325func TestEventsCarryTheirEvidence(t *testing.T) {
326 ctx := context.Background()
327 var got []moves.LogEvent
328 w := eventWorker(&got)
329 cut := func(phase string, src *provision.MoveSource) provision.Record {
330 return mvEvRec(provision.MoveCutting, func(m *provision.MoveState) { m.CutPhase, m.Source = phase, src })
331 }
332 src := &provision.MoveSource{FinalTXID: 42, DBSHA256: strings.Repeat("d", 64), TimersSHA256: strings.Repeat("e", 64),
333 Files: &provision.MoveFilesDigest{Plain: strings.Repeat("f", 64)}, Evidence: "moves/x/stopped-source.json"}
334 w.logTransition(ctx, nil, cut(provision.CutStopping, nil), cut(provision.CutStopped, src))
335 land := func(phase, db string) provision.Record {
336 return mvEvRec(provision.MoveLanding, func(m *provision.MoveState) {
337 m.LandingPhase = phase
338 m.Landing = &provision.MoveLandingJournal{InstalledDBSHA256: db, Attempts: 1}
339 })
340 }
341 w.logTransition(ctx, nil, land(provision.LandProbing, strings.Repeat("a", 64)), land(provision.LandProbed, strings.Repeat("a", 64)))
342 rb := func(target, source string, stuck bool) provision.Record {
343 state := provision.MoveRollingBack
344 if stuck {
345 state = provision.MoveStuck
346 }
347 return mvEvRec(state, func(m *provision.MoveState) {
348 m.CutAt = mvEvAt(10)
349 m.Rollback = &provision.MoveRollback{ID: "rb_x", RequestedAt: mvEvAt(30), TargetPhase: target, SourcePhase: source}
350 if source == provision.RbSourceRecovered {
351 m.Rollback.RecoveredAt = mvEvAt(70)
352 }
353 })
354 }
355 w.logTransition(ctx, nil, rb(provision.RbTargetReturning, provision.RbSourceWaiting, false), rb(provision.RbTargetReturning, provision.RbSourceWaiting, true))
356 w.logTransition(ctx, nil, rb(provision.RbTargetCleaned, provision.RbSourceProbed, false), rb(provision.RbTargetCleaned, provision.RbSourceRecovered, false))
357 // The source's stuck, on another box's worker (the target's stuck above
358 // is stuck-0 in this one).
359 eventWorker(&got).logTransition(ctx, nil, rb(provision.RbTargetCleaned, provision.RbSourceVerified, false), rb(provision.RbTargetCleaned, provision.RbSourceVerified, true))
360 by := map[string]map[string]string{}
361 for _, ev := range got {
362 by[ev.Identity] = ev.Fields
363 }
364 if f := by["cutting-stopped"]; f["db_sha256"] != src.DBSHA256 || f["final_txid"] != "42" || f["files_plain"] != src.Files.Plain || f["timers_sha256"] != src.TimersSHA256 {
365 t.Errorf("stopped carries %v", f)
366 }
367 if f := by["landing-probed"]; f["installed_db_sha256"] != strings.Repeat("a", 64) {
368 t.Errorf("probed carries %v", f)
369 }
370 if f := by["rolling_back-source-recovered"]; f["recovery_s"] != "40" || f["offline_s"] != "60" {
371 t.Errorf("recovered carries %v, want recovery_s 40 (rollback request to recovered) and offline_s 60 (cut to recovered)", f)
372 }
373 var roles []string
374 for _, ev := range got {
375 if ev.Event == "stuck" {
376 roles = append(roles, ev.Fields["role"])
377 }
378 }
379 if strings.Join(roles, " ") != "target source" {
380 t.Errorf("stuck roles %v, want the target's (at returning), then the source's", roles)
381 }
382}
383
384// Ruling 7: the real writers log, through the move harness. The landing
385// harness drives the source's readiness, the console's cut, the cut and the
386// flip, then the landing to installed; the close harness the probe, the
387// commit, the open, the source's park and the close. A writer whose log
388// call went missing drops out of the list.
389func TestRealWritersLogEveryTransition(t *testing.T) {
390 t.Parallel()
391 t.Run("readiness to installed", func(t *testing.T) {
392 h := newLandingHarness(t)
393 var got []moves.LogEvent
394 h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
395 got = append(got, ev)
396 return nil
397 }
398 h.flip(t)
399 h.land(t)
400 h.installed(t)
401 wantSubsequence(t, got, "ready", "cut_requested", "admitted", "stopping", "stopped", "staged", "archived", "flipped", "parking", "building", "published", "local", "configured", "restored")
402 })
403 t.Run("probe to done", func(t *testing.T) {
404 h := newCloseHarness(t)
405 var got []moves.LogEvent
406 h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
407 got = append(got, ev)
408 return nil
409 }
410 h.land(t)
411 h.committedAndOpen(t)
412 h.sourcePass(t)
413 h.land(t)
414 lane := h.chore(t, "uk", true)
415 h.clockPast(lane.Park.EvidenceAt.Add(park.SignatureAge + lane.Park.RequestLimit))
416 h.land(t)
417 src := h.chore(t, "ie", false)
418 h.clockPast(src.Park.EvidenceAt.Add(park.SignatureAge + src.Park.RequestLimit))
419 if err := h.regionalAck(t, src); err != nil {
420 t.Fatal(err)
421 }
422 h.land(t)
423 h.done(t)
424 wantSubsequence(t, got, "probing", "probe_attempt", "probed", "committed", "opened", "source_parked", "done")
425 })
426}
427
428// wantSubsequence requires want's events, in order, among got's, and no
429// identity twice.
430func wantSubsequence(t *testing.T, got []moves.LogEvent, want ...string) {
431 t.Helper()
432 var events []string
433 seen := map[string]bool{}
434 for _, ev := range got {
435 events = append(events, ev.Event)
436 if seen[ev.Identity] {
437 t.Errorf("identity %q twice", ev.Identity)
438 }
439 seen[ev.Identity] = true
440 }
441 i := 0
442 for _, e := range events {
443 if i < len(want) && e == want[i] {
444 i++
445 }
446 }
447 if i != len(want) {
448 t.Fatalf("events %v lack %v in order (matched %d)", events, want, i)
449 }
450}
451
452// flakyLogStore fails every write under a move's log while fail is set.
453type flakyLogStore struct {
454 release.Store
455 fail bool
456}
457
458func (f *flakyLogStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
459 if f.fail && strings.Contains(key, "/log/") {
460 return "", errors.New("the bucket is down")
461 }
462 return f.Store.Put(ctx, key, body, ifMatch)
463}
464
465// outboxWorker is a worker with the production log writer over s and an
466// outbox under dataDir.
467func outboxWorker(dataDir string) *movesWorker {
468 w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
469 w.cfg.Box = "box-test"
470 w.cfg.DataDir = dataDir
471 w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() }
472 w.cfg.LogEvent = moveLogWriter(nil)
473 return w
474}
475
476func outboxFiles(t *testing.T, dataDir string) []string {
477 t.Helper()
478 var out []string
479 root := moveLogOutboxDir(dataDir)
480 err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
481 if errors.Is(err, fs.ErrNotExist) {
482 return fs.SkipAll
483 }
484 if err != nil {
485 return err
486 }
487 if !d.IsDir() {
488 rel, _ := filepath.Rel(root, p)
489 out = append(out, filepath.ToSlash(rel))
490 }
491 return nil
492 })
493 if err != nil {
494 t.Fatal(err)
495 }
496 return out
497}
498
499// Fix round 2, ruling 1: a log write that fails is kept, with the entry's
500// bytes, in the worker's outbox on disk, and every pass retries the outbox,
501// whatever the record now says. Here the write fails at the request and
502// again at the close; the move is gone from the record by the time the
503// bucket answers, and a fresh worker (an agent restart) writes both.
504func TestLogOutboxOutlivesTheMove(t *testing.T) {
505 ctx := context.Background()
506 dataDir := t.TempDir()
507 s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
508 w := outboxWorker(dataDir)
509 flow := doneFlow()
510 w.observeMove(ctx, s, flow[0])
511 last := flow[len(flow)-1]
512 w.logTransition(ctx, s, flow[len(flow)-2], last)
513 if last.Move != nil {
514 t.Fatal("the fixture's last record still has its move")
515 }
516 files := outboxFiles(t, dataDir)
517 if strings.Join(files, " ") != "acme.go.tito.io/"+mvEvMoveID+"/done.json acme.go.tito.io/"+mvEvMoveID+"/requested.json" {
518 t.Fatalf("outbox %v", files)
519 }
520 w.flushLogOutbox(ctx, s) // still failing: kept
521 if len(outboxFiles(t, dataDir)) != 2 {
522 t.Fatal("a failed retry dropped an entry")
523 }
524 pending, err := os.ReadFile(filepath.Join(moveLogOutboxDir(dataDir), "acme.go.tito.io", mvEvMoveID, "requested.json"))
525 if err != nil {
526 t.Fatal(err)
527 }
528 var kept moveLogPending
529 if err := json.Unmarshal(pending, &kept); err != nil {
530 t.Fatal(err)
531 }
532 s.fail = false
533 outboxWorker(dataDir).flushLogOutbox(ctx, s)
534 if f := outboxFiles(t, dataDir); len(f) != 0 {
535 t.Fatalf("written entries left in the outbox: %v", f)
536 }
537 prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", mvEvMoveID)
538 got, err := moves.ReadLog(ctx, s, prefix)
539 if err != nil || len(got) != 2 {
540 t.Fatalf("log %+v %v", got, err)
541 }
542 body, _, err := s.Get(ctx, moves.LogKeyFor(prefix, "requested"))
543 if err != nil || string(body) != string(kept.Entry) {
544 t.Fatalf("the object is not the kept bytes:\n%s\n%s", body, kept.Entry)
545 }
546 // Another sighting of the closed move writes nothing more.
547 w.observeMove(ctx, s, last)
548 if f := outboxFiles(t, dataDir); len(f) != 0 {
549 t.Fatalf("outbox %v", f)
550 }
551}
552
553// Fix round 2, ruling 2: recovery probe starts are logged through the
554// journal as a box builds it (movesWorker.probeAttempts), and a start whose
555// log write fails is kept in the outbox.
556func TestRecoveryProbeAttemptsAreLogged(t *testing.T) {
557 ctx := context.Background()
558 f := recoveryPAFixture(t, hostclaim.KindRestoring)
559 s := &flakyLogStore{Store: f.claims, fail: true}
560 w := outboxWorker(f.dataDir)
561 ids := 0
562 ops := w.probeAttempts(s, func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil },
563 func() string { ids++; return "id-" + strings.Repeat("x", ids) })
564 for n := 0; n < 3; n++ {
565 if _, err := ops.OpenProbe(ctx, f.start, time.Now()); err != nil {
566 t.Fatal(err)
567 }
568 if err := ops.Reserve(ctx, f.start, n); err != nil {
569 t.Fatalf("start %d: %v", n, err)
570 }
571 if err := ops.Failed(ctx, f.start, n); err != nil {
572 t.Fatal(err)
573 }
574 s.fail = false // only the first start's write fails
575 }
576 if files := outboxFiles(t, f.dataDir); len(files) != 1 || !strings.HasSuffix(files[0], "/probe_attempt-0-1.json") {
577 t.Fatalf("outbox %v, want the first start's entry", files)
578 }
579 w.flushLogOutbox(ctx, s)
580 got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.rec.Move.ID))
581 if err != nil {
582 t.Fatal(err)
583 }
584 var names []string
585 for _, ev := range got {
586 if ev.Event != "probe_attempt" {
587 t.Fatalf("event %+v", ev)
588 }
589 names = append(names, ev.Identity)
590 }
591 sort.Strings(names)
592 if strings.Join(names, " ") != "probe_attempt-0-1 probe_attempt-0-2 probe_attempt-0-3" {
593 t.Fatalf("logged %v", names)
594 }
595}
596
597// Fix round 2, ruling 3: the memo forgets a move as soon as a pass sees its
598// record with no move or another move: here a source box whose move was
599// closed on the target, then the same box when the next move replaced it.
600// The memo is also bounded, dropping its oldest identities.
601func TestLogMemoRetiresAMoveClosedElsewhere(t *testing.T) {
602 ctx := context.Background()
603 var got []moves.LogEvent
604 w := eventWorker(&got)
605 flow := doneFlow()
606 w.observeMove(ctx, nil, flow[0])
607 w.logTransition(ctx, nil, flow[0], flow[1])
608 if w.logged.len() == 0 {
609 t.Fatal("nothing remembered")
610 }
611 w.cfg.Enabled = true // the pass forgets only with the switch on (I1)
612 s := release.NewDirStore(t.TempDir())
613 closed := flow[len(flow)-1]
614 if err := provision.CreateRecord(ctx, s, closed); err != nil {
615 t.Fatal(err)
616 }
617 if err := movesPass(ctx, s, w); err != nil {
618 t.Fatal(err)
619 }
620 if n := w.logged.len(); n != 0 {
621 t.Fatalf("%d identities remembered after the pass saw the move closed elsewhere", n)
622 }
623 w.observeMove(ctx, nil, flow[0])
624 next := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.ID = "mv_next" })
625 s = release.NewDirStore(t.TempDir())
626 if err := provision.CreateRecord(ctx, s, next); err != nil {
627 t.Fatal(err)
628 }
629 if err := movesPass(ctx, s, w); err != nil {
630 t.Fatal(err)
631 }
632 if w.logged.seen(next.Host + "/" + mvEvMoveID + "/requested") {
633 t.Fatal("the replaced move is still remembered")
634 }
635}
636
637func TestLogMemoIsBounded(t *testing.T) {
638 var m moveLogMemo
639 for i := 0; i < moveLogMemoCap+10; i++ {
640 m.add("h", "op", "h/op/"+strconv.Itoa(i))
641 }
642 if m.len() != moveLogMemoCap {
643 t.Fatalf("%d entries, want %d", m.len(), moveLogMemoCap)
644 }
645 if m.seen("h/op/0") || !m.seen("h/op/"+strconv.Itoa(moveLogMemoCap+9)) {
646 t.Fatal("not the oldest dropped")
647 }
648 m.forget("h", "op")
649 if m.len() != 0 {
650 t.Fatalf("%d entries after forget", m.len())
651 }
652}
653
654// Fix round 3, ruling 4: the outbox keeps the first entry queued for an
655// identity. A restart during the outage forgets what was handed off, so the
656// next pass sees the transition again and builds it again, later; that never
657// replaces the kept bytes or their time.
658func TestLogOutboxKeepsTheFirstEntry(t *testing.T) {
659 ctx := context.Background()
660 dataDir := t.TempDir()
661 s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
662 rec := doneFlow()[0]
663 w := outboxWorker(dataDir)
664 w.observeMove(ctx, s, rec)
665 kept := filepath.Join(moveLogOutboxDir(dataDir), rec.Host, mvEvMoveID, "requested.json")
666 first, err := os.ReadFile(kept)
667 if err != nil {
668 t.Fatal(err)
669 }
670 restarted := outboxWorker(dataDir)
671 restarted.cfg.Now = func() time.Time { return time.Unix(1_800_000_900, 0).UTC() }
672 restarted.flushLogOutbox(ctx, s) // still failing
673 restarted.observeMove(ctx, s, rec)
674 again, err := os.ReadFile(kept)
675 if err != nil {
676 t.Fatal(err)
677 }
678 if string(again) != string(first) {
679 t.Fatalf("the kept entry was replaced:\n%s\n%s", first, again)
680 }
681 s.fail = false
682 restarted.flushLogOutbox(ctx, s)
683 got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, mvEvMoveID))
684 if err != nil || len(got) != 1 || !got[0].At.Equal(time.Unix(1_800_000_100, 0)) {
685 t.Fatalf("log %+v %v, want the first entry's time", got, err)
686 }
687}
688
689// syncSpy stands in for outboxSync: it records every path synced, and
690// fails the paths in fail.
691type syncSpy struct {
692 fail map[string]bool
693 synced map[string]int
694}
695
696func spyOutboxSync(t *testing.T) *syncSpy {
697 t.Helper()
698 spy := &syncSpy{fail: map[string]bool{}, synced: map[string]int{}}
699 saved := outboxSync
700 outboxSync = func(f *os.File) error {
701 // Keyed by the absolute path, so a relative data directory's
702 // syncs are named the same way.
703 name := f.Name()
704 if abs, err := filepath.Abs(name); err == nil {
705 name = abs
706 }
707 if spy.fail[name] {
708 return errors.New("injected sync failure")
709 }
710 spy.synced[name]++
711 return f.Sync()
712 }
713 t.Cleanup(func() { outboxSync = saved })
714 return spy
715}
716
717// reset forgets what was synced and fails nothing.
718func (s *syncSpy) reset() {
719 s.fail, s.synced = map[string]bool{}, map[string]int{}
720}
721
722// Final review I2: the log outbox acknowledges an entry only once it is
723// durable. Every directory the write created is synced into its parent, a
724// failed directory sync is an error, and an entry already there (an earlier
725// try whose sync failed) is synced, file and directory, before it counts.
726// The caller does not remember an entry the outbox could not make durable,
727// so the next sighting tries again.
728func TestLogOutboxIsDurableBeforeItAcknowledges(t *testing.T) {
729 ctx := context.Background()
730 spy := spyOutboxSync(t)
731 rec := doneFlow()[0]
732 ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"}
733
734 t.Run("new directories are synced into their parents", func(t *testing.T) {
735 spy.reset()
736 dd := t.TempDir()
737 if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
738 t.Fatal(err)
739 }
740 root := moveLogOutboxDir(dd)
741 for _, d := range []string{dd, filepath.Join(dd, ".moves"), root, filepath.Join(root, rec.Host), filepath.Join(root, rec.Host, mvEvMoveID)} {
742 if spy.synced[d] == 0 {
743 t.Errorf("%s was not synced; synced %v", d, spy.synced)
744 }
745 }
746 })
747
748 t.Run("a failed directory sync is an error, and the retry syncs the existing entry", func(t *testing.T) {
749 spy.reset()
750 dd := t.TempDir()
751 dir := filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)
752 spy.fail[dir] = true
753 if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
754 t.Fatal("an entry whose directory sync failed was acknowledged")
755 }
756 file := filepath.Join(dir, "requested.json")
757 if _, err := os.Lstat(file); err != nil {
758 t.Fatalf("the fixture wants the entry linked before the failed sync: %v", err)
759 }
760 spy.reset()
761 spy.fail[file] = true
762 if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
763 t.Fatal("an existing entry whose file sync failed was acknowledged")
764 }
765 spy.reset()
766 if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
767 t.Fatal(err)
768 }
769 if spy.synced[file] == 0 || spy.synced[dir] == 0 {
770 t.Fatalf("an existing entry was acknowledged without syncing it and its directory: %v", spy.synced)
771 }
772 })
773
774 t.Run("the caller retries an entry the outbox could not make durable", func(t *testing.T) {
775 spy.reset()
776 dd := t.TempDir()
777 s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
778 w := outboxWorker(dd)
779 spy.fail[filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)] = true
780 w.observeMove(ctx, s, rec)
781 key := rec.Host + "/" + mvEvMoveID + "/requested"
782 if w.logged.seen(key) {
783 t.Fatal("an entry the outbox could not make durable counts as handed off")
784 }
785 spy.reset()
786 w.observeMove(ctx, s, rec)
787 if !w.logged.seen(key) {
788 t.Fatal("the next sighting did not hand the entry off")
789 }
790 })
791}
792
793// Final re-review 1, I2: the worker is built with its data directory made
794// absolute (movesDataDir), so the outbox's ancestor walk always ends at that
795// root. With CARLOS_DATA_DIR=".", an ancestor sync that fails leaves the new
796// directories behind; the retry finds them there and still syncs every one
797// of them, up to the data directory, before it acknowledges the entry.
798func TestLogOutboxUnderARelativeDataDir(t *testing.T) {
799 spy := spyOutboxSync(t)
800 t.Chdir(t.TempDir())
801 t.Setenv("CARLOS_DATA_DIR", ".")
802 root, err := filepath.Abs(".")
803 if err != nil {
804 t.Fatal(err)
805 }
806 w := outboxWorker(movesDataDir(dataDir()))
807 rec := doneFlow()[0]
808 ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"}
809 spy.fail[filepath.Join(root, ".moves")] = true
810 if err := w.queueLog(rec, ev); err == nil {
811 t.Fatal("an entry whose new directory's sync failed was acknowledged")
812 }
813 spy.reset()
814 if err := w.queueLog(rec, ev); err != nil {
815 t.Fatal(err)
816 }
817 out := moveLogOutboxDir(root)
818 for _, d := range []string{root, filepath.Join(root, ".moves"), out, filepath.Join(out, rec.Host), filepath.Join(out, rec.Host, mvEvMoveID)} {
819 if spy.synced[d] == 0 {
820 t.Errorf("the retry acknowledged the entry without syncing %s; synced %v", d, spy.synced)
821 }
822 }
823}
824