| new file mode 100644 |
| index 000000000..50023ee27 |
| --- /dev/null |
| +++ b/internal/host/movesfileslive_test.go |
| @@ -0,0 +1,250 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostfiles" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/park" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The files live set (J2/F4, Paul's #236 ruling; M10): every moves park |
| +// excludes the objects a live generation's files manifest names, in its own |
| +// region, and the source's are protected on the source until the record |
| +// commits. Plan 5 Task 9a. |
| + |
| +// filesManifest writes a hostfiles manifest at <location>/files/g-<gen>/ |
| +// naming objects, each "path=key". |
| +func filesManifest(t *testing.T, put func(key string, body []byte), location, gen string, objects ...string) { |
| + t.Helper() |
| + m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: int64(len(objects)), Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}} |
| + for i, o := range objects { |
| + path, key, _ := strings.Cut(o, "=") |
| + m.Entries[path] = hostfiles.Entry{Object: key, SHA256: fmt.Sprintf("%064x", i+1), Size: 4, Mode: 0o644, Seq: int64(i + 1)} |
| + } |
| + body, err := json.Marshal(m) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + put(hostfiles.ManifestKey(location+"/files/g-"+gen+"/"), body) |
| +} |
| + |
| +// TestParkKeepsTheFilesLiveSet (M10): in the authoritative region the view |
| +// pins the exact versions a live generation's files manifest names, also |
| +// under an older generation's files prefix, so the sweep never makes that |
| +// generation a candidate and its objects stay. Without a manifest nothing |
| +// more is pinned and the older generation is parked. Seen from another |
| +// region nothing is read or pinned. |
| +func TestParkKeepsTheFilesLiveSet(t *testing.T) { |
| + setup := func(t *testing.T, manifest bool) (*evFixture, park.Engine, []string) { |
| + f := newEvFixture(t) |
| + 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}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f.bucket.Put("data", f.location+"/generation.json", body) |
| + old := []string{f.location + "/files/g-old/objects/0000000000000001-aa", f.genKey("old")} |
| + for _, k := range old { |
| + f.bucket.Put("data", k, []byte("old")) |
| + } |
| + f.bucket.Put("data", f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb", []byte("cur")) |
| + if manifest { |
| + filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID, |
| + "kept.txt="+old[0], "new.txt="+f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb") |
| + } |
| + seq := 0 |
| + e := park.Engine{Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow }, |
| + NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) }, |
| + View: movesParkView(f.provider().Records, f.bucket, "data", "ie"), |
| + Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }} |
| + return f, e, old |
| + } |
| + t.Run("a live manifest names an older generation's object", func(t *testing.T) { |
| + f, e, old := setup(t, true) |
| + v, err := e.View(t.Context(), f.observe(t)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ver := f.bucket.LatestID("data", old[0]) |
| + if !v.Pinned[park.Ref{Key: old[0], VersionID: ver}] { |
| + t.Fatalf("pins %v; want %s@%s", v.Pinned, old[0], ver) |
| + } |
| + d, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, c := range d.Candidates { |
| + if c.Generation == "old" { |
| + t.Fatalf("the generation a live manifest names is a candidate: %+v", d.Candidates) |
| + } |
| + } |
| + for _, k := range old { |
| + if _, ok := f.bucket.Current("data", k); !ok { |
| + t.Fatalf("%s was parked", k) |
| + } |
| + } |
| + // Seen from a region that is not authoritative, nothing is read. |
| + other, err := movesParkView(f.provider().Records, f.bucket, "data", "uk")(t.Context(), f.observe(t)) |
| + if err != nil || len(other.Pinned) != 0 || other.MarkerKnown { |
| + t.Fatalf("the uk view %+v, %v", other, err) |
| + } |
| + }) |
| + t.Run("no manifest", func(t *testing.T) { |
| + f, e, old := setup(t, false) |
| + if _, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit); err != nil && !errors.Is(err, park.ErrPending) { |
| + t.Fatal(err) |
| + } |
| + for _, k := range old { |
| + if _, ok := f.bucket.Current("data", k); ok { |
| + t.Fatalf("%s was not parked", k) |
| + } |
| + } |
| + }) |
| + t.Run("a manifest naming what the location does not hold", func(t *testing.T) { |
| + for name, key := range map[string]string{"outside the location": "instances/" + evHost + "/elsewhere/x", "missing": "LOC/files/g-old/objects/0000000000000009-zz"} { |
| + f, e, _ := setup(t, false) |
| + filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID, "x="+strings.Replace(key, "LOC", f.location, 1)) |
| + if _, err := e.View(t.Context(), f.observe(t)); !errors.Is(err, park.ErrUnsafe) { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + } |
| + }) |
| +} |
| + |
| +// TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit (J2/F4): once the |
| +// target's probe is admitted the target's region is authoritative, but seen |
| +// from the source's region the source stays authoritative until the record |
| +// commits, with its own live manifests pinned; after the commit the source |
| +// pins nothing (P6-D3: the target's manifests name no source object), and |
| +// its location park may run. |
| +func TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit(t *testing.T) { |
| + f := newVwFixture(t) |
| + f.marker(t, "cur", f.inc) |
| + src := f.location + "/files/g-cur/objects/0000000000000001-aa" |
| + ver := f.bucket.Put("data", src, []byte("source file")) |
| + filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "cur", "a.txt="+src) |
| + landing := func(phase string) hostclaim.Observed { |
| + 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"}}) |
| + } |
| + f.move(provision.MoveLanding) |
| + v, err := f.view("ie")(t.Context(), landing(provision.LandProbed)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: src, VersionID: ver}] { |
| + t.Fatalf("before the commit the source's view %+v", v) |
| + } |
| + if v, err := f.view("uk")(t.Context(), landing(provision.LandProbed)); err != nil || v.AuthoritativeRegion != "uk" { |
| + t.Fatalf("before the commit the target's view %+v, %v", v, err) |
| + } |
| + f.move(provision.MoveCommitted) |
| + v, err = f.view("ie")(t.Context(), landing(hostclaim.PhaseCommitted)) |
| + if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 { |
| + t.Fatalf("after the commit the source's view %+v, %v", v, err) |
| + } |
| +} |
| + |
| +// TestLandingResidueParkKeepsTheFilesLiveSet (M10, the landing install's |
| +// park): after the target's probe makes its generation live, a residue |
| +// chore's acknowledgement re-lists the location under the real view. An |
| +// object of the residue generation that the live generation's manifest names |
| +// refuses the acknowledgement and stays; without that reference the same |
| +// object is parked. |
| +func TestLandingResidueParkKeepsTheFilesLiveSet(t *testing.T) { |
| + for _, named := range []bool{true, false} { |
| + t.Run(fmt.Sprintf("named %v", named), func(t *testing.T) { |
| + ctx := context.Background() |
| + h := newLandingHarness(t) |
| + h.flip(t) |
| + h.residue(t) |
| + h.land(t) |
| + h.installed(t) |
| + // The probe's admission: g-<move id> live in uk, its marker |
| + // published, a files manifest that names an object under the |
| + // earlier stay's generation (written back after the parking). |
| + late := cutRoot + "/files/g-old/objects/0000000000000003-cc" |
| + if _, err := h.vs.PutIfAbsent(ctx, "data-uk", late, []byte("kept")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if named { |
| + filesManifest(t, func(k string, b []byte) { |
| + if _, err := h.vs.PutIfAbsent(ctx, "data-uk", k, b); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, cutRoot, p1ID, "kept.txt="+late) |
| + } |
| + marker := cutRoot + "/generation.json" |
| + markerVer := h.f.nextVersion() |
| + 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}) |
| + h.f.store("data-uk", marker, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + markerVer, Size: int64(len(body))}, string(body)}) |
| + h.f.setVersion("data-uk", marker, markerVer) |
| + rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { |
| + c.Holder.Generation = p1ID |
| + 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"}) |
| + }) |
| + releaseOwnLease(h.lw, p1Host) |
| + h.clock.Advance(park.SignatureAge + landingRequestLimit + time.Minute) |
| + l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}} |
| + eng := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID, |
| + View: movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk"), Authorize: movesArmAuthorize(allowParkAuthorize, l.authorizeObligation)} |
| + var runErr error |
| + for _, o := range h.claim(t).Obligations { |
| + if o.Kind == "park-generation" && o.Generation == "old" { |
| + for i := 0; i < 3; i++ { |
| + runErr = eng.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}) |
| + } |
| + } |
| + } |
| + _, kept := h.f.buckets["data-uk"][late] |
| + if named { |
| + if !errors.Is(runErr, park.ErrUnsafe) || !kept { |
| + t.Fatalf("a named object: run %v, kept %v", runErr, kept) |
| + } |
| + return |
| + } |
| + if kept { |
| + t.Fatalf("an unnamed residue object stayed (run %v)", runErr) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// TestReturnResidueParkKeepsTheFilesLiveSet (M10, the rollback return's |
| +// park): once the source's probe admits the return generation the source is |
| +// authoritative again, and its live manifest's objects are pinned, also |
| +// under a pre-move generation; before that admission the target is |
| +// authoritative and the source pins nothing (the dirty return parks the |
| +// pre-move generations, D6). |
| +func TestReturnResidueParkKeepsTheFilesLiveSet(t *testing.T) { |
| + f := newVwFixture(t) |
| + f.rec.Move = &provision.MoveState{ID: vwMove, Incarnation: f.inc, From: "ie", To: "uk", State: provision.MoveRollingBack, |
| + Rollback: &provision.MoveRollback{ID: vwRb, Dirty: true}} |
| + 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}) |
| + f.bucket.Put("data", f.location+"/generation.json", body) |
| + pre := f.location + "/files/g-cur/objects/0000000000000001-aa" |
| + ver := f.bucket.Put("data", pre, []byte("pre-move")) |
| + filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, vwRb, "a.txt="+pre) |
| + restoring := func(gen string, admissions ...string) hostclaim.Observed { |
| + return vwObs(&hostclaim.Claim{Holder: f.holder(gen, admissions...), Op: hostclaim.Op{Kind: hostclaim.KindRestoring, ID: vwMove, Phase: hostclaim.PhaseProbing, Exec: "ie-box"}}) |
| + } |
| + v, err := f.view("ie")(t.Context(), restoring(vwRb, "cur@ie", vwMove+"@uk", vwRb+"@ie")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: pre, VersionID: ver}] { |
| + t.Fatalf("after the source's admission %+v", v) |
| + } |
| + v, err = f.view("ie")(t.Context(), restoring(vwMove, "cur@ie", vwMove+"@uk")) |
| + if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 { |
| + t.Fatalf("before the source's admission %+v, %v", v, err) |
| + } |
| +} |