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 "fmt"
8 "strings"
9 "testing"
10 "time"
11
12 "github.com/carlosframework/platform/internal/hostbox"
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/hostfiles"
15 "github.com/carlosframework/platform/internal/lease"
16 "github.com/carlosframework/platform/internal/park"
17 "github.com/carlosframework/platform/internal/provision"
18 "github.com/carlosframework/platform/internal/release"
19)
20
21// The files live set (J2/F4, Paul's #236 ruling; M10): every moves park
22// excludes the objects a live generation's files manifest names, in its own
23// region, and the source's are protected on the source until the record
24// commits. Plan 5 Task 9a.
25
26// filesManifest writes a hostfiles manifest at <location>/files/g-<gen>/
27// naming objects, each "path=key".
28func filesManifest(t *testing.T, put func(key string, body []byte), location, gen string, objects ...string) {
29 t.Helper()
30 m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: int64(len(objects)), Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
31 for i, o := range objects {
32 path, key, _ := strings.Cut(o, "=")
33 m.Entries[path] = hostfiles.Entry{Object: key, SHA256: fmt.Sprintf("%064x", i+1), Size: 4, Mode: 0o644, Seq: int64(i + 1)}
34 }
35 body, err := json.Marshal(m)
36 if err != nil {
37 t.Fatal(err)
38 }
39 put(hostfiles.ManifestKey(location+"/files/g-"+gen+"/"), body)
40}
41
42// TestParkKeepsTheFilesLiveSet (M10): in the authoritative region the view
43// pins the exact versions a live generation's files manifest names, also
44// under an older generation's files prefix, so the sweep never makes that
45// generation a candidate and its objects stay. Without a manifest nothing
46// more is pinned and the older generation is parked. Seen from another
47// region nothing is read or pinned.
48func TestParkKeepsTheFilesLiveSet(t *testing.T) {
49 setup := func(t *testing.T, manifest bool) (*evFixture, park.Engine, []string) {
50 f := newEvFixture(t)
51 body, err := json.Marshal(map[string]any{"op": evMoveID, "incarnation": f.inc, "host": evHost, "box": "box", "installed_at": evCreated, "manifest_sha256": "x", "snapshot_txid": 1})
52 if err != nil {
53 t.Fatal(err)
54 }
55 f.bucket.Put("data", f.location+"/generation.json", body)
56 old := []string{f.location + "/files/g-old/objects/0000000000000001-aa", f.genKey("old")}
57 for _, k := range old {
58 f.bucket.Put("data", k, []byte("old"))
59 }
60 f.bucket.Put("data", f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb", []byte("cur"))
61 if manifest {
62 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID,
63 "kept.txt="+old[0], "new.txt="+f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb")
64 }
65 seq := 0
66 e := park.Engine{Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
67 NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
68 View: movesParkView(f.provider().Records, f.bucket, "data", "ie"),
69 Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
70 return f, e, old
71 }
72 t.Run("a live manifest names an older generation's object", func(t *testing.T) {
73 f, e, old := setup(t, true)
74 v, err := e.View(t.Context(), f.observe(t))
75 if err != nil {
76 t.Fatal(err)
77 }
78 ver := f.bucket.LatestID("data", old[0])
79 if !v.Pinned[park.Ref{Key: old[0], VersionID: ver}] {
80 t.Fatalf("pins %v; want %s@%s", v.Pinned, old[0], ver)
81 }
82 d, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit)
83 if err != nil {
84 t.Fatal(err)
85 }
86 for _, c := range d.Candidates {
87 if c.Generation == "old" {
88 t.Fatalf("the generation a live manifest names is a candidate: %+v", d.Candidates)
89 }
90 }
91 for _, k := range old {
92 if _, ok := f.bucket.Current("data", k); !ok {
93 t.Fatalf("%s was parked", k)
94 }
95 }
96 // Seen from a region that is not authoritative, nothing is read.
97 other, err := movesParkView(f.provider().Records, f.bucket, "data", "uk")(t.Context(), f.observe(t))
98 if err != nil || len(other.Pinned) != 0 || other.MarkerKnown {
99 t.Fatalf("the uk view %+v, %v", other, err)
100 }
101 })
102 t.Run("no manifest", func(t *testing.T) {
103 f, e, old := setup(t, false)
104 if _, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit); err != nil && !errors.Is(err, park.ErrPending) {
105 t.Fatal(err)
106 }
107 for _, k := range old {
108 if _, ok := f.bucket.Current("data", k); ok {
109 t.Fatalf("%s was not parked", k)
110 }
111 }
112 })
113 t.Run("a manifest naming what the location does not hold", func(t *testing.T) {
114 for name, key := range map[string]string{"outside the location": "instances/" + evHost + "/elsewhere/x", "missing": "LOC/files/g-old/objects/0000000000000009-zz"} {
115 f, e, _ := setup(t, false)
116 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID, "x="+strings.Replace(key, "LOC", f.location, 1))
117 if _, err := e.View(t.Context(), f.observe(t)); !errors.Is(err, park.ErrUnsafe) {
118 t.Fatalf("%s: %v", name, err)
119 }
120 }
121 })
122}
123
124// TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit (J2/F4): once the
125// target's probe is admitted the target's region is authoritative, but seen
126// from the source's region the source stays authoritative until the record
127// commits, with its own live manifests pinned; after the commit the source
128// pins nothing (P6-D3: the target's manifests name no source object), and
129// its location park may run.
130func TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit(t *testing.T) {
131 f := newVwFixture(t)
132 f.marker(t, "cur", f.inc)
133 src := f.location + "/files/g-cur/objects/0000000000000001-aa"
134 ver := f.bucket.Put("data", src, []byte("source file"))
135 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "cur", "a.txt="+src)
136 landing := func(phase string) hostclaim.Observed {
137 return vwObs(&hostclaim.Claim{Holder: f.holder(vwMove, "cur@ie", vwMove+"@uk"), Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: vwMove, Phase: phase, Exec: "uk-box"}})
138 }
139 f.move(provision.MoveLanding)
140 v, err := f.view("ie")(t.Context(), landing(provision.LandProbed))
141 if err != nil {
142 t.Fatal(err)
143 }
144 if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: src, VersionID: ver}] {
145 t.Fatalf("before the commit the source's view %+v", v)
146 }
147 if v, err := f.view("uk")(t.Context(), landing(provision.LandProbed)); err != nil || v.AuthoritativeRegion != "uk" {
148 t.Fatalf("before the commit the target's view %+v, %v", v, err)
149 }
150 f.move(provision.MoveCommitted)
151 v, err = f.view("ie")(t.Context(), landing(hostclaim.PhaseCommitted))
152 if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 {
153 t.Fatalf("after the commit the source's view %+v, %v", v, err)
154 }
155}
156
157// TestLandingResidueParkKeepsTheFilesLiveSet (M10, the landing install's
158// park): after the target's probe makes its generation live, a residue
159// chore's acknowledgement re-lists the location under the real view. An
160// object of the residue generation that the live generation's manifest names
161// refuses the acknowledgement and stays; without that reference the same
162// object is parked.
163func TestLandingResidueParkKeepsTheFilesLiveSet(t *testing.T) {
164 t.Parallel()
165 for _, named := range []bool{true, false} {
166 t.Run(fmt.Sprintf("named %v", named), func(t *testing.T) {
167 ctx := context.Background()
168 h := newLandingHarness(t)
169 h.flip(t)
170 h.residue(t)
171 h.land(t)
172 h.installed(t)
173 // The probe's admission: g-<move id> live in uk, its marker
174 // published, a files manifest that names an object under the
175 // earlier stay's generation (written back after the parking).
176 late := cutRoot + "/files/g-old/objects/0000000000000003-cc"
177 if _, err := h.vs.PutIfAbsent(ctx, "data-uk", late, []byte("kept")); err != nil {
178 t.Fatal(err)
179 }
180 if named {
181 // A new version over the prepared manifest the landing's
182 // install built (moves Task 6), as the probe's
183 // establishment writes one.
184 filesManifest(t, func(k string, b []byte) { putNewVersion(h.f, "data-uk", k, b) }, cutRoot, p1ID, "kept.txt="+late)
185 }
186 marker := cutRoot + "/generation.json"
187 markerVer := h.f.nextVersion()
188 body, _ := json.Marshal(map[string]any{"op": p1ID, "incarnation": h.inc, "host": p1Host, "box": "uk-box", "installed_at": h.clock.Now(), "manifest_sha256": "x", "snapshot_txid": 1})
189 h.f.store("data-uk", marker, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + markerVer, Size: int64(len(body))}, string(body)})
190 h.f.setVersion("data-uk", marker, markerVer)
191 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
192 c.Holder.Generation = p1ID
193 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Session: c.Holder.Session + 1, Box: "uk-box", Generation: p1ID, Predecessor: "cur", Mode: "cold"})
194 })
195 releaseOwnLease(h.lw, p1Host)
196 h.clock.Advance(park.SignatureAge + landingRequestLimit + time.Minute)
197 l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
198 eng := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
199 View: movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk"), Authorize: movesArmAuthorize(allowParkAuthorize, l.authorizeObligation)}
200 var runErr error
201 for _, o := range h.claim(t).Obligations {
202 if o.Kind == "park-generation" && o.Generation == "old" {
203 for i := 0; i < 3; i++ {
204 runErr = eng.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID})
205 }
206 }
207 }
208 _, kept := h.f.buckets["data-uk"][late]
209 if named {
210 if !errors.Is(runErr, park.ErrUnsafe) || !kept {
211 t.Fatalf("a named object: run %v, kept %v", runErr, kept)
212 }
213 return
214 }
215 if kept {
216 t.Fatalf("an unnamed residue object stayed (run %v)", runErr)
217 }
218 })
219 }
220}
221
222// TestReturnResidueParkKeepsTheFilesLiveSet (M10, the rollback return's
223// park): once the source's probe admits the return generation the source is
224// authoritative again, and its live manifest's objects are pinned, also
225// under a pre-move generation; before that admission the target is
226// authoritative and the source pins nothing (the dirty return parks the
227// pre-move generations, D6).
228func TestReturnResidueParkKeepsTheFilesLiveSet(t *testing.T) {
229 f := newVwFixture(t)
230 f.rec.Move = &provision.MoveState{ID: vwMove, Incarnation: f.inc, From: "ie", To: "uk", State: provision.MoveRollingBack,
231 Rollback: &provision.MoveRollback{ID: vwRb, Dirty: true}}
232 body, _ := json.Marshal(map[string]any{"op": vwRb, "incarnation": f.inc, "host": vwHost, "box": "ie-box", "installed_at": vwCreated, "manifest_sha256": "x", "snapshot_txid": 1})
233 f.bucket.Put("data", f.location+"/generation.json", body)
234 pre := f.location + "/files/g-cur/objects/0000000000000001-aa"
235 ver := f.bucket.Put("data", pre, []byte("pre-move"))
236 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, vwRb, "a.txt="+pre)
237 restoring := func(gen string, admissions ...string) hostclaim.Observed {
238 return vwObs(&hostclaim.Claim{Holder: f.holder(gen, admissions...), Op: hostclaim.Op{Kind: hostclaim.KindRestoring, ID: vwMove, Phase: hostclaim.PhaseProbing, Exec: "ie-box"}})
239 }
240 v, err := f.view("ie")(t.Context(), restoring(vwRb, "cur@ie", vwMove+"@uk", vwRb+"@ie"))
241 if err != nil {
242 t.Fatal(err)
243 }
244 if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: pre, VersionID: ver}] {
245 t.Fatalf("after the source's admission %+v", v)
246 }
247 v, err = f.view("ie")(t.Context(), restoring(vwMove, "cur@ie", vwMove+"@uk"))
248 if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 {
249 t.Fatalf("before the source's admission %+v, %v", v, err)
250 }
251}
252
253// TestRestoringKeepsTheSourcesFilesLiveSetUntilTheReturn (J2/F4; L18 joint
254// round, finding 1): a park of an older source generation is queued in the
255// source's region before the target's probe is admitted; then the target is
256// admitted and the landing is rolled back. While the target releases
257// (restoring/releasing, released) the newest admission is the target's, but
258// a clean return restores the source's retained generation, so seen from the
259// source's region the source stays authoritative: its marker and files pins
260// hold, the chore is refused and nothing the retained manifest names is
261// retired. A dirty release is protected the same way (a declared loss turns
262// it clean). The target's region still names the target. From the return CAS
263// on the newest admission decides, as before: a clean return is admitted in
264// the source, and a dirty one names the target until the source's probe is
265// admitted, so the source pins nothing and the chore parks the pre-move
266// generation (D6).
267func TestRestoringKeepsTheSourcesFilesLiveSetUntilTheReturn(t *testing.T) {
268 for _, dirty := range []bool{false, true} {
269 t.Run(fmt.Sprintf("dirty %v", dirty), func(t *testing.T) {
270 ctx := t.Context()
271 f := newVwFixture(t)
272 markerVer, snapshot := f.marker(t, "cur", f.inc)
273 snapKey := f.location + "/g-cur/0009/0000000000000001-0000000000000005.ltx"
274 old := map[string]string{}
275 for _, k := range []string{f.location + "/files/g-old/objects/0000000000000001-aa", f.location + "/g-old/0000/db"} {
276 old[k] = f.bucket.Put("data", k, []byte("old"))
277 }
278 named := f.location + "/files/g-old/objects/0000000000000001-aa"
279 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "cur", "a.txt="+named)
280 f.move(provision.MoveLanding)
281
282 // The chore, queued in the source's region at landing, before
283 // the target's probe is admitted.
284 claims := release.NewDirStore(t.TempDir())
285 chore := hostclaim.Obligation{ID: "park-old", Op: vwMove, Kind: "park-generation", Region: "ie", Owner: "region:ie", Generation: "old", Retirement: "ret-1",
286 Park: &hostclaim.ParkScope{Location: f.location, Incarnation: f.inc, EvidenceAt: evAt, RequestLimit: hostbox.RequestLimit}}
287 update := func(step func(*hostclaim.Claim) error) {
288 t.Helper()
289 obs, err := hostclaim.Read(ctx, claims, vwHost)
290 if err != nil {
291 t.Fatal(err)
292 }
293 if _, err := hostclaim.Update(ctx, claims, obs, step); err != nil {
294 t.Fatal(err)
295 }
296 }
297 update(func(c *hostclaim.Claim) error {
298 c.Host, c.Holder = vwHost, f.holder("cur", "cur@ie")
299 c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: vwMove, Phase: provision.LandProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"},
300 Intent: map[string]string{"source": "ie-box", "target": "uk-box", "from": "ie", "to": "uk"},
301 Journal: map[string]string{hostclaim.JournalSourceStamp: "3", hostclaim.JournalSrcGen: "cur"}}
302 c.Obligations = []hostclaim.Obligation{chore}
303 return nil
304 })
305 // The target's probe's admission, then the rollback.
306 update(func(c *hostclaim.Claim) error {
307 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 2, Box: "uk-box", Generation: vwMove, Predecessor: "cur", Mode: "cold", Region: "uk", PredecessorRegion: "ie"})
308 c.Holder.Generation = vwMove
309 return nil
310 })
311 f.rec.Move.State = provision.MoveRollingBack
312 f.rec.Move.Rollback = &provision.MoveRollback{ID: vwRb, TargetPhase: provision.RbTargetReleasing}
313 update(hostclaim.RollBack(vwMove, vwRb))
314
315 seq := 0
316 engine := park.Engine{Store: f.bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
317 NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) }, View: f.view("ie"), Authorize: allowParkAuthorize}
318 observe := func() hostclaim.Observed {
319 t.Helper()
320 obs, err := hostclaim.Read(ctx, claims, vwHost)
321 if err != nil {
322 t.Fatal(err)
323 }
324 return obs
325 }
326 intact := func() bool {
327 for k, ver := range old {
328 if _, ok := f.bucket.Current("data", k); !ok || f.bucket.LatestID("data", k) != ver {
329 return false
330 }
331 }
332 return true
333 }
334 protectedOnSource := func(when string) {
335 t.Helper()
336 if err := engine.Run(ctx, park.Work{Host: vwHost, ID: chore.ID, Op: vwMove}); !errors.Is(err, park.ErrUnsafe) {
337 t.Fatalf("%s: the source's park of a generation its retained manifest names: %v", when, err)
338 }
339 if !intact() {
340 t.Fatalf("%s: an object the retained manifest names was retired", when)
341 }
342 v, err := f.view("ie")(ctx, observe())
343 if err != nil {
344 t.Fatalf("%s: the source's view: %v", when, err)
345 }
346 if v.AuthoritativeRegion != "ie" || v.MarkerGeneration != "cur" || v.Marker != (park.Ref{Key: f.location + "/generation.json", VersionID: markerVer}) ||
347 !v.Pinned[park.Ref{Key: named, VersionID: old[named]}] || !v.Pinned[park.Ref{Key: snapKey, VersionID: snapshot}] {
348 t.Fatalf("%s: the source's view %+v", when, v)
349 }
350 }
351 targetView := func(when, want string) {
352 t.Helper()
353 if v, err := f.view("uk")(ctx, observe()); err != nil || v.AuthoritativeRegion != want {
354 t.Fatalf("%s: the target's view %+v, %v; want %s", when, v, err, want)
355 }
356 }
357
358 protectedOnSource("releasing")
359 targetView("releasing", "uk")
360
361 f.rec.Move.Rollback.TargetPhase, f.rec.Move.Rollback.Dirty = provision.RbTargetReleased, dirty
362 update(hostclaim.TargetReleased(vwMove, "uk-box", "uk", "stopped-target", hostclaim.ParkScope{Location: f.location, Incarnation: f.inc, EvidenceAt: evAt, RequestLimit: hostbox.RequestLimit},
363 func() string { seq++; return fmt.Sprintf("id%d", seq) }))
364 if op := observe().Claim.Op; op.Phase != hostclaim.PhaseReleased || op.Exec != "" {
365 t.Fatalf("the release: %+v", op)
366 }
367 protectedOnSource("released")
368 targetView("released", "uk")
369
370 // The source's return CAS.
371 update(hostclaim.ReturnToSource(vwMove, "ie-box", !dirty, hostclaim.ReturnFence{}))
372 if !dirty {
373 protectedOnSource("after the clean return")
374 targetView("after the clean return", "ie")
375 return
376 }
377 v, err := f.view("ie")(ctx, observe())
378 if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 || v.MarkerKnown {
379 t.Fatalf("after the dirty return the source's view %+v, %v", v, err)
380 }
381 targetView("after the dirty return", "uk")
382 if err := engine.Run(ctx, park.Work{Host: vwHost, ID: chore.ID, Op: vwMove}); err != nil && !errors.Is(err, park.ErrPending) {
383 t.Fatalf("the dirty return's park of the pre-move generation: %v", err)
384 }
385 for k := range old {
386 if _, ok := f.bucket.Current("data", k); ok {
387 t.Fatalf("%s was not parked after the dirty return", k)
388 }
389 }
390 })
391 }
392}
393
394// TestFilesLiveSetFollowsTheLiveChain (fix round 1, I2): with the marker
395// lagging, park.protected treats the whole unfinished chain g3 -> g2 -> g1
396// as live, and the view reads every one of those generations' manifests. An
397// object only g2's manifest names, under the older generation g0, is pinned
398// and survives the sweep's park.
399func TestFilesLiveSetFollowsTheLiveChain(t *testing.T) {
400 f := newEvFixture(t)
401 f.setClaim(t, func(c *hostclaim.Claim) {
402 c.Holder.Generation = "g3"
403 c.Holder.Admissions = []hostclaim.AdmissionRecord{
404 {Session: 1, Box: "box", Generation: "g1", Mode: "cold", Region: "ie"},
405 {Session: 2, Box: "box", Generation: "g2", Predecessor: "g1", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
406 {Session: 3, Box: "box", Generation: "g3", Predecessor: "g2", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
407 }
408 c.Holder.Stamp, c.Holder.Session = 3, 3
409 })
410 body, err := json.Marshal(map[string]any{"op": "g1", "incarnation": f.inc, "host": evHost, "box": "box", "installed_at": evCreated, "manifest_sha256": "x", "snapshot_txid": 1})
411 if err != nil {
412 t.Fatal(err)
413 }
414 f.bucket.Put("data", f.location+"/generation.json", body)
415 only := f.location + "/files/g-g0/objects/0000000000000001-aa"
416 ver := f.bucket.Put("data", only, []byte("named by g2 alone"))
417 other := f.location + "/files/g-g0/objects/0000000000000002-bb"
418 f.bucket.Put("data", other, []byte("named by nobody"))
419 filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "g2", "a.txt="+only)
420 seq := 0
421 e := park.Engine{Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
422 NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
423 View: movesParkView(f.provider().Records, f.bucket, "data", "ie"),
424 Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
425 v, err := e.View(t.Context(), f.observe(t))
426 if err != nil {
427 t.Fatal(err)
428 }
429 if !v.Pinned[park.Ref{Key: only, VersionID: ver}] {
430 t.Fatalf("pins %v; want g2's object %s", v.Pinned, only)
431 }
432 if _, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit); err != nil && !errors.Is(err, park.ErrPending) {
433 t.Fatal(err)
434 }
435 if got, ok := f.bucket.Current("data", only); !ok || string(got) != "named by g2 alone" {
436 t.Fatalf("g2's object did not survive the park: %q %v", got, ok)
437 }
438}
439
440// putFilesGeneration writes, in bucket, a files object of gen at location
441// and gen's manifest naming it, returning the object's key.
442func putFilesGeneration(t *testing.T, vs *fakeVersioned, bucket, location, gen string) string {
443 t.Helper()
444 ctx := context.Background()
445 obj := location + "/files/g-" + gen + "/objects/0000000000000001-aa"
446 if _, err := vs.PutIfAbsent(ctx, bucket, obj, []byte("a live file")); err != nil {
447 t.Fatal(err)
448 }
449 // gen's manifest is a new version: a landing's or a return's install
450 // may already have built gen's prepared manifest (moves Task 6), and the
451 // probe's establishment writes the live one over it.
452 filesManifest(t, func(k string, b []byte) { putNewVersion(vs.f, bucket, k, b) }, location, gen, "a.txt="+obj)
453 return obj
454}
455
456// putNewVersion writes body as a new current version of key in bucket,
457// whether or not the key holds one.
458func putNewVersion(f *fakeMoveMirror, bucket, key string, body []byte) {
459 ver := f.nextVersion()
460 f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
461 f.setVersion(bucket, key, ver)
462}
463
464// archivedUnder reports whether bucket holds a parked copy of key, a key
465// under location (park's destination: parked/<inc>/<host>/<retirement>/
466// instances/<key below the location>@<version>).
467func archivedUnder(f *fakeMoveMirror, bucket, inc, location, key string) bool {
468 tail := strings.TrimPrefix(key, location+"/")
469 for k := range f.buckets[bucket] {
470 if strings.HasPrefix(k, "parked/"+inc+"/") && strings.Contains(k, "/instances/"+tail+"@") {
471 return true
472 }
473 }
474 return false
475}
476
477// admitTarget is the target's probe admission on the claim (as Plan 3's
478// probe makes it): holder.generation is the move's, admitted in uk.
479func admitTarget(t *testing.T, h *landingHarness) {
480 t.Helper()
481 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
482 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Session: c.Holder.Session + 1, Box: "uk-box", Generation: p1ID, Predecessor: c.Holder.Generation, Mode: "cold"})
483 c.Holder.Generation = p1ID
484 })
485}
486
487// TestSourceCleanupKeepsTheFilesLiveSetUntilTheCommit (fix round 1, I3; the
488// source cleanup, J2/F4): the source's location park runs through the real
489// view and engine. Before the record commits, with the target's probe
490// admitted, a park of the source's location in its own region is refused
491// and the source's files stay byte for byte. After the commit the source's
492// finalization runs that park and the files are retired, archived under
493// parked/.
494func TestSourceCleanupKeepsTheFilesLiveSetUntilTheCommit(t *testing.T) {
495 t.Parallel()
496 ctx := context.Background()
497 h := newFinalHarness(t)
498 h.advanceToProbed(t)
499 admitTarget(t, h.landingHarness)
500 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
501 obj := putFilesGeneration(t, h.vs, "data-ie", location, "cur")
502 view := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
503
504 // Before the commit: the same scope the commit queues, run by the engine.
505 rec := h.read(t, p1Host)
506 scope, err := sourceLocationParkScope(rec, *h.claim(t).Holder)
507 if err != nil {
508 t.Fatal(err)
509 }
510 early := hostclaim.Obligation{ID: obligationID(), Op: p1ID, Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: obligationID(), Park: &scope}
511 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Obligations = append(c.Obligations, early) })
512 e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-ie", Region: "ie", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
513 View: view, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
514 if err := e.Run(ctx, park.Work{Host: p1Host, ID: early.ID, Op: p1ID}); !errors.Is(err, park.ErrUnsafe) {
515 t.Fatalf("a park of the source before the commit: %v", err)
516 }
517 if body, ok := h.f.buckets["data-ie"][obj]; !ok || body.body != "a live file" {
518 t.Fatalf("the source's file before the commit: %+v %v", body, ok)
519 }
520 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
521 kept := c.Obligations[:0]
522 for _, o := range c.Obligations {
523 if o.ID != early.ID {
524 kept = append(kept, o)
525 }
526 }
527 c.Obligations = kept
528 })
529
530 // The commit, then the source's finalization through the real view.
531 h.commitRecord(t)
532 h.commitClaim(t)
533 h.cfg.ParkView = view
534 h.sourcePass(t)
535 if m := h.read(t, p1Host).Move; m.SourceParkedAt == nil {
536 t.Fatalf("the source did not finalize: %+v", m)
537 }
538 if _, ok := h.f.buckets["data-ie"][obj]; ok {
539 t.Fatal("the source's file was not retired after the commit")
540 }
541 if !archivedUnder(h.f, "data-ie", h.inc, location, obj) {
542 t.Fatal("the source's file was not archived")
543 }
544}
545
546// TestRollbackReturnCleanupKeepsTheFilesLiveSet (fix round 1, I3; the
547// rollback return's cleanup): the target's region location park, queued
548// when the target released, runs through the real view and engine. While
549// the target's admitted generation is still the holder's (before the return
550// names the source's again) the target region is authoritative and the park
551// is refused, the target's files kept; once the source has taken the host
552// back, the target's cleaning runs it and the files are retired.
553func TestRollbackReturnCleanupKeepsTheFilesLiveSet(t *testing.T) {
554 t.Parallel()
555 ctx := context.Background()
556 h := newRollbackHarness(t)
557 admitTarget(t, h.landingHarness)
558 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
559 body, err := json.Marshal(map[string]any{"op": p1ID, "incarnation": h.inc, "host": p1Host, "box": "uk-box", "installed_at": h.clock.Now(), "manifest_sha256": "x", "snapshot_txid": 1})
560 if err != nil {
561 t.Fatal(err)
562 }
563 if _, err := h.vs.PutIfAbsent(ctx, "data-uk", location+"/generation.json", body); err != nil {
564 t.Fatal(err)
565 }
566 obj := putFilesGeneration(t, h.vs, "data-uk", location, p1ID)
567 view := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk")
568 h.parkView = view
569 h.rollBack(t)
570 h.rollbackPass(t)
571 var chore *hostclaim.Obligation
572 for _, o := range h.claim(t).Obligations {
573 if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" && o.Park != nil && o.Park.Location == location {
574 chore = &o
575 }
576 }
577 if chore == nil {
578 t.Fatalf("the release queued no target location park: %+v", h.claim(t).Obligations)
579 }
580 e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
581 View: view, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
582 if err := e.Run(ctx, park.Work{Host: p1Host, ID: chore.ID, Op: p1ID}); !errors.Is(err, park.ErrUnsafe) {
583 t.Fatalf("a park of the target while its generation is live: %v", err)
584 }
585 if b, ok := h.f.buckets["data-uk"][obj]; !ok || b.body != "a live file" {
586 t.Fatal("the target's file was taken before the return")
587 }
588
589 // The source takes the host back: the return names its generation again
590 // and it recovers; the target's cleaning then runs the park.
591 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = "cur" })
592 h.recovered(t)
593 h.rollbackPass(t)
594 if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned {
595 t.Fatalf("the target did not clean: %+v", rb)
596 }
597 if _, ok := h.f.buckets["data-uk"][obj]; ok {
598 t.Fatal("the target's file was not retired by the cleaning")
599 }
600 if !archivedUnder(h.f, "data-uk", h.inc, location, obj) {
601 t.Fatal("the target's file was not archived")
602 }
603}
604
605// returnPreMove lists the pre-move generation's objects at the source's
606// location, with their bodies.
607func returnPreMove(t *testing.T, h *returnHarness, location string) map[string]string {
608 t.Helper()
609 out := map[string]string{}
610 for k, o := range h.f.buckets["data-ie"] {
611 if strings.HasPrefix(k, location+"/g-cur/") {
612 out[k] = o.body
613 }
614 }
615 if len(out) == 0 {
616 t.Fatal("fixture: no pre-move generation at the source's location")
617 }
618 return out
619}
620
621// TestReturnResidueParkRunsThroughTheInstall (fix round 2, I3; the source's
622// return-residue park, movesinstall.go parking): with the record's evidence
623// that the target admitted this move (its probe's admission of the move's
624// prepared generation), a dirty return's parking runs its residue chores
625// through the installer's own engine with the real view: the pre-move
626// generation is archived under parked/ and retired from the source's
627// location, and the return moves on.
628func TestReturnResidueParkRunsThroughTheInstall(t *testing.T) {
629 t.Parallel()
630 h := newReturnHarness(t, false)
631 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
632 pre := returnPreMove(t, h, location)
633 h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
634 if _, err := h.install(t); err != nil {
635 t.Fatalf("the return with the target's admission: %v", err)
636 }
637 if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase == provision.RbSourceParking {
638 t.Fatalf("the return is still parking: %+v", rb)
639 }
640 for k := range pre {
641 if _, ok := h.f.buckets["data-ie"][k]; ok {
642 t.Fatalf("%s was not retired", k)
643 }
644 if !archivedUnder(h.f, "data-ie", h.inc, location, k) {
645 t.Fatalf("%s was not archived", k)
646 }
647 }
648 if j := h.returnJournal(t); j.Steps[installStepResidueKept] != "" {
649 t.Fatalf("a return with admission evidence journalled the no-op: %q", j.Steps[installStepResidueKept])
650 }
651}
652
653// TestDirtyReturnWithoutAdmissionKeepsThePreMoveGeneration (fix round 3,
654// ruling (a)): the record carries no evidence that the target admitted this
655// move (the landing never reached probing, and the claim holds no admission
656// of the move's generation). The pre-move generation is still the source's
657// live state, so the return's parking is a journalled no-op: through the
658// real view and engine the return completes, the pre-move bytes are intact
659// and no residue park is queued.
660func TestDirtyReturnWithoutAdmissionKeepsThePreMoveGeneration(t *testing.T) {
661 t.Parallel()
662 h := newReturnHarnessWith(t, false, false)
663 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
664 pre := returnPreMove(t, h, location)
665 h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
666 if _, err := h.install(t); err != nil {
667 t.Fatalf("the return without admission evidence: %v", err)
668 }
669 if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceRestored {
670 t.Fatalf("the return did not complete its install: %+v", rb)
671 }
672 for k, body := range pre {
673 if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
674 t.Fatalf("%s changed", k)
675 }
676 }
677 j := h.returnJournal(t)
678 if got := j.Steps[installStepResidueKept]; !strings.Contains(got, "never admitted") {
679 t.Fatalf("the journal's no-op line: %q", got)
680 }
681 if got := j.Steps[landStepResidue]; got != "[]" {
682 t.Fatalf("the residue list %q, want none", got)
683 }
684 for _, o := range h.claim(t).Obligations {
685 if o.Kind == "park-generation" {
686 t.Fatalf("a residue park was queued: %+v", o)
687 }
688 }
689}
690
691// TestReturnResidueParkWaitsOnATransientRefusal (fix round 3): with the
692// target's admission recorded, a view refusal never turns the parking into
693// the no-op. The return waits at parking with the pre-move bytes intact and
694// no no-op line, and parks once the view answers.
695func TestReturnResidueParkWaitsOnATransientRefusal(t *testing.T) {
696 t.Parallel()
697 h := newReturnHarness(t, false)
698 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
699 pre := returnPreMove(t, h, location)
700 real := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
701 down := true
702 h.rcfg.ParkView = func(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
703 if down {
704 return park.View{}, fmt.Errorf("%w: the record read timed out", park.ErrUnsafe)
705 }
706 return real(ctx, obs)
707 }
708 if _, err := h.install(t); err == nil {
709 t.Fatal("the return went on through a view refusal")
710 }
711 if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceParking {
712 t.Fatalf("the return left parking: %+v", rb)
713 }
714 if j := h.returnJournal(t); j.Steps[installStepResidueKept] != "" {
715 t.Fatalf("a view refusal became the no-op: %q", j.Steps[installStepResidueKept])
716 }
717 for k, body := range pre {
718 if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
719 t.Fatalf("%s changed during the refusal", k)
720 }
721 }
722 down = false
723 if _, err := h.install(t); err != nil {
724 t.Fatalf("after the refusal: %v", err)
725 }
726 for k := range pre {
727 if _, ok := h.f.buckets["data-ie"][k]; ok {
728 t.Fatalf("%s was not retired once the view answered", k)
729 }
730 }
731}
732
733// TestDirtyReturnAtProbingWithoutAdmissionKeepsTheResidue (fix round 4): the
734// record enters probing before the probe runs (movesland.go), so a landing
735// that crashed or failed at probing before its admission and was then
736// rolled back dirty is not evidence that the target admitted. With the
737// holder holding no admission of the move's generation, the return keeps
738// the pre-move generation and completes.
739func TestDirtyReturnAtProbingWithoutAdmissionKeepsTheResidue(t *testing.T) {
740 t.Parallel()
741 h := newReturnHarnessWith(t, false, false)
742 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
743 rec.Move.LandingPhase = provision.LandProbing
744 return nil
745 }); err != nil {
746 t.Fatal(err)
747 }
748 location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
749 pre := returnPreMove(t, h, location)
750 h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
751 if _, err := h.install(t); err != nil {
752 t.Fatalf("the return at probing without an admission: %v", err)
753 }
754 if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceRestored {
755 t.Fatalf("the return did not complete its install: %+v", rb)
756 }
757 for k, body := range pre {
758 if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
759 t.Fatalf("%s changed", k)
760 }
761 }
762 if got := h.returnJournal(t).Steps[installStepResidueKept]; !strings.Contains(got, "never admitted") {
763 t.Fatalf("the journal's no-op line: %q", got)
764 }
765}
766