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 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "path/filepath"
12 "reflect"
13 "strconv"
14 "strings"
15 "testing"
16 "time"
17
18 "github.com/carlosframework/platform/internal/activator"
19 "github.com/carlosframework/platform/internal/genselect"
20 "github.com/carlosframework/platform/internal/hostbox"
21 "github.com/carlosframework/platform/internal/hostclaim"
22 "github.com/carlosframework/platform/internal/hostfiles"
23 "github.com/carlosframework/platform/internal/hostguard"
24 "github.com/carlosframework/platform/internal/hostrelocate"
25 "github.com/carlosframework/platform/internal/hostwrite"
26 "github.com/carlosframework/platform/internal/lease/leasetest"
27 "github.com/carlosframework/platform/internal/moves"
28 "github.com/carlosframework/platform/internal/provision"
29 "github.com/carlosframework/platform/internal/registry"
30 "github.com/carlosframework/platform/internal/release"
31 "github.com/superfly/ltx"
32)
33
34const (
35 roHost = "shop.example"
36 roOp = "rel-1"
37 roP = "P1"
38)
39
40var roProbe = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
41
42// roReleases is the release authority the owners verify a drain's release
43// with; it records every call.
44type roReleases struct {
45 calls int
46 err error
47}
48
49func (r *roReleases) VerifyRelease(context.Context, hostclaim.Claim, hostclaim.Release) error {
50 r.calls++
51 return r.err
52}
53
54// roFiles is moves' C8 provider: it records each Establish under the work
55// it is given.
56type roFiles struct{ established []string }
57
58func (f *roFiles) Establish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
59 return work.With(ctx, in.Identity, func() error {
60 f.established = append(f.established, fmt.Sprintf("%s@%d", o.Host, o.Claim.Holder.Session))
61 return nil
62 })
63}
64
65// roFixture is a relocation target at relocating/probing, P built and not
66// opened: the claim as the participant's ToProbing leaves it, P's snapshot,
67// anchor and install manifest in the versioned bucket, P's prepared files
68// manifest in the data bucket, and the wake inputs bound as RelocationSource
69// binds them.
70type roFixture struct {
71 t *testing.T
72 claims *release.DirStore
73 files *release.DirStore
74 bucket *leasetest.VersionedBucket
75 releases *roReleases
76 c8 *roFiles
77 inputs *RelocationWakeInputs
78 route registry.Route
79 manifest moves.InstallManifest
80 seq int
81 stamp uint64
82}
83
84type roOptions struct {
85 empty bool // an evidenced empty pin: no snapshot, no install manifest
86 failover bool // no journaled release; the source lost and fenced
87}
88
89func newROFixture(t *testing.T, o roOptions) *roFixture {
90 t.Helper()
91 ctx := t.Context()
92 f := &roFixture{t: t, claims: release.NewDirStore(t.TempDir()), files: release.NewDirStore(t.TempDir()), bucket: leasetest.NewVersionedBucket(), releases: &roReleases{}, c8: &roFiles{}}
93 f.bucket.AddBucket("data")
94 created := time.Date(2026, 10, 7, 1, 0, 0, 0, time.UTC)
95 enrollment := strings.Repeat("a", 32)
96 nh := hostclaim.NewHolder{Family: "instance", Key: "apps/acct/app/instances/" + roHost + ".json", Account: "acct", App: "app", Host: roHost, Region: "ie", RecordRegion: "ie", CreatedAt: created}
97 body := []byte(`{"host":"` + roHost + `","account":"acct","app":"app","region":"ie","created_at":"2026-10-07T01:00:00Z"}`)
98 obs, err := hostclaim.Read(ctx, f.claims, roHost)
99 if err != nil {
100 t.Fatal(err)
101 }
102 for _, tr := range []func(*hostclaim.Claim) error{
103 hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment}),
104 hostclaim.CompleteCreate("c1", "src"),
105 hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}),
106 } {
107 if obs, err = hostclaim.Update(ctx, f.claims, obs, tr); err != nil {
108 t.Fatal(err)
109 }
110 }
111 h := obs.Claim.Holder
112 prefix := moves.GenerationPrefix(h.Location, roP)
113 auth := hostclaim.PreparedAuthority{Generation: roP, SourceFilesSeq: 5}
114 ref := hostclaim.RecoveryPin{Key: "control/relocations/" + roHost + "/" + roOp + "/pin-1.json", SHA256: strings.Repeat("e", 64), Generation: "g1", FilesSeq: 5, Release: "v1"}
115 if o.empty {
116 ref.Empty = true
117 } else {
118 ref.MaxTXID = 42
119 snap, anchor := moves.SnapshotKey(prefix, 42), moves.AnchorKey(prefix, 42)
120 f.manifest = moves.InstallManifest{V: 1, Op: roOp, Host: roHost, N: 42, SnapshotKey: snap, SnapshotVersion: f.bucket.Put("data", snap, ltxHeader(t, 1, 42)), SnapshotCRC64: 7,
121 AnchorKey: anchor, AnchorVersion: f.bucket.Put("data", anchor, ltxHeader(t, 42, 42)), IntegrityCheck: "ok", DBSHA256: strings.Repeat("d", 64)}
122 raw, _ := json.Marshal(f.manifest)
123 f.bucket.Put("data", moves.InstallManifestKey(prefix), raw)
124 sum := sha256.Sum256(raw)
125 auth.InstallManifest, auth.SHA256, auth.SnapshotTXID = moves.InstallManifestKey(prefix), hex.EncodeToString(sum[:]), 42
126 }
127 pm, _ := json.Marshal(hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}})
128 if _, err := f.files.Put(ctx, hostfiles.ManifestKey(moves.GenerationFilesPrefix(h.Location, roP)), pm, ""); err != nil {
129 t.Fatal(err)
130 }
131 pmSum := sha256.Sum256(pm)
132 auth.FilesManifestSHA256 = hex.EncodeToString(pmSum[:])
133 if obs, err = hostclaim.Update(ctx, f.claims, obs, func(c *hostclaim.Claim) error {
134 f.stamp = c.Gen + 1
135 c.Holder.Stamp = f.stamp
136 auth.InstallStamp = f.stamp
137 stamp := strconv.FormatUint(f.stamp, 10)
138 journal := map[string]string{"pin": string(mustJSON(t, ref)), "prepared": roP, "install_stamp": stamp, "install_release": "v1",
139 "prepared_authority": string(mustJSON(t, auth)), "probing_since": created.Add(time.Hour).Format(time.RFC3339Nano),
140 "exec_ack.src": `{"op":"rel-1"}`, "exec_ack.dst": `{"op":"rel-1"}`,
141 "local_install": string(mustJSON(t, hostclaim.LocalInstall{Box: "dst", Enrollment: enrollment, InstallStamp: f.stamp, JournalSHA256: strings.Repeat("c", 64)}))}
142 if !o.failover {
143 journal["release"] = string(mustJSON(t, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment, Generation: c.Holder.Generation, Session: c.Holder.Session, FinalTXID: ref.MaxTXID, ManifestSeq: 5, Receipt: "r", ReceiptSHA256: strings.Repeat("f", 64)}))
144 } else {
145 c.Lost = []string{"src"}
146 journal["fence.src"] = "control/hosts/" + roHost + "/fences/src.json"
147 }
148 c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: roOp, Phase: hostclaim.PhaseRelocatingProbing, Exec: "dst", Participants: []string{"src", "dst"},
149 Intent: map[string]string{"region": "ie", "source_box": "src", "target_box": "dst", "target_enrollment": enrollment}, Journal: journal}
150 return nil
151 }); err != nil {
152 t.Fatal(err)
153 }
154 f.route = registry.Route{Host: roHost, Kind: "instance", App: "app", Account: "acct", Status: "relocating", HoldOp: roOp, DBPath: provision.NestedDBPath("/data", roHost), Socket: provision.NestedSocket(roHost)}
155 f.inputs = &RelocationWakeInputs{
156 Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
157 return ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:config", Release: "sha256:release"}, Executable: "/opt/carlos/live/" + roHost, ExecutableSHA256: strings.Repeat("b", 64), Version: "v1"}, nil
158 },
159 FilesManifest: f.c8, Markers: plan9D5Markers{f.bucket}, Claims: f.claims, FilesStore: f.files, Data: f.bucket, Bucket: "data", Region: "ie", Box: "dst",
160 NewID: func() string { f.seq++; return fmt.Sprintf("fresh%d", f.seq) }, Now: func() time.Time { return created.Add(2 * time.Hour) }, Releases: f.releases}
161 return f
162}
163
164func ltxHeader(t *testing.T, min, max uint64) []byte {
165 t.Helper()
166 body, err := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}).MarshalBinary()
167 if err != nil {
168 t.Fatal(err)
169 }
170 return body
171}
172
173func (f *roFixture) read() hostclaim.Observed {
174 f.t.Helper()
175 o, err := hostclaim.Read(f.t.Context(), f.claims, roHost)
176 if err != nil {
177 f.t.Fatal(err)
178 }
179 return o
180}
181
182func (f *roFixture) update(tr func(*hostclaim.Claim) error) hostclaim.Observed {
183 f.t.Helper()
184 o, err := hostclaim.Update(f.t.Context(), f.claims, f.read(), tr)
185 if err != nil {
186 f.t.Fatal(err)
187 }
188 return o
189}
190
191// admit counts the attempt and applies the selected admission with the
192// claim's own CAS, as the participant and the wake do.
193func (f *roFixture) admit(o hostclaim.Observed, in ClaimLaunchInputs) hostclaim.Observed {
194 f.t.Helper()
195 counted, err := hostclaim.Update(f.t.Context(), f.claims, o, hostclaim.BeginProbeAttempt(roOp, "dst"))
196 if err != nil {
197 f.t.Fatal(err)
198 }
199 in.Admission.ObservedSession, in.Admission.ObservedGeneration = counted.Claim.Holder.Session, counted.Claim.Holder.Generation
200 admitted, err := hostclaim.Update(f.t.Context(), f.claims, counted, hostclaim.AdmitWriter(in.Admission))
201 if err != nil {
202 f.t.Fatal(err)
203 }
204 return admitted
205}
206
207// C4, ruling 47: P's first opening is cold into P, keeping the install
208// stamp, prepared from P's snapshot, and certified only by this
209// operation's verified release; a failover's opening is not certified; an
210// empty pin's opening is empty, with no baseline.
211func TestRelocationSelectCertifiesOnlyPsOpeningAfterItsRelease(t *testing.T) {
212 ctx := t.Context()
213 t.Run("a drain's release", func(t *testing.T) {
214 f := newROFixture(t, roOptions{})
215 o := f.read()
216 in, err := f.inputs.Select(ctx, f.route, o, roProbe)
217 if err != nil {
218 t.Fatal(err)
219 }
220 h := o.Claim.Holder
221 want := hostclaim.Admission{ObservedOpID: roOp, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation,
222 Box: "dst", Mode: "cold", NewGeneration: roP, KeepStamp: true, PredecessorCertified: true, Region: "ie", PredecessorRegion: "ie", RemoteMaxTXID: 42}
223 if !reflect.DeepEqual(in.Admission, want) {
224 t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
225 }
226 if !in.Prepared || in.Empty || in.Baseline == nil || in.Baseline.Generation != roP || in.Baseline.MaxTXID != 42 || f.releases.calls != 1 {
227 t.Fatalf("prepared=%v empty=%v baseline=%+v release checks=%d", in.Prepared, in.Empty, in.Baseline, f.releases.calls)
228 }
229 if in.DBPath != f.route.DBPath || in.Version != "v1" || len(in.Roles) != 1 || in.Roles[0] != "main" || in.ReadyPath != "/" {
230 t.Fatalf("launch inputs %+v", in)
231 }
232 f.releases.err = errors.New("the receipt does not verify")
233 if _, err := f.inputs.Select(ctx, f.route, o, roProbe); !errors.Is(err, f.releases.err) {
234 t.Fatalf("an unverified release certified: %v", err)
235 }
236 })
237 t.Run("a failover", func(t *testing.T) {
238 f := newROFixture(t, roOptions{failover: true})
239 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
240 if err != nil {
241 t.Fatal(err)
242 }
243 if in.Admission.PredecessorCertified || !in.Admission.KeepStamp || in.Admission.NewGeneration != roP || f.releases.calls != 0 {
244 t.Fatalf("failover admission %+v (release checks %d)", in.Admission, f.releases.calls)
245 }
246 })
247 t.Run("an empty pin", func(t *testing.T) {
248 f := newROFixture(t, roOptions{empty: true})
249 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
250 if err != nil {
251 t.Fatal(err)
252 }
253 if !in.Empty || in.Prepared || in.Baseline != nil || in.Admission.NewGeneration != roP || !in.Admission.KeepStamp || in.Admission.RemoteMaxTXID != 0 {
254 t.Fatalf("empty opening %+v", in)
255 }
256 if err := f.inputs.EmptyAuthority(ctx, f.read(), in); err != nil {
257 t.Fatalf("the evidenced empty pin: %v", err)
258 }
259 })
260}
261
262// P is opened once. After its opening every attempt is a fresh generation
263// from the predecessor, keeping no stamp, never certified, and evidence for
264// P's opening is refused on the opened claim.
265func TestRelocationSelectNeverReopensP(t *testing.T) {
266 f := newROFixture(t, roOptions{})
267 ctx := t.Context()
268 first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
269 if err != nil {
270 t.Fatal(err)
271 }
272 opened := f.admit(f.read(), first)
273 if !hostclaim.PreparedOpened(opened.Claim) || opened.Claim.Holder.Stamp != f.stamp {
274 t.Fatalf("P's opening: %+v", opened.Claim.Holder)
275 }
276 again, err := f.inputs.Select(ctx, f.route, opened, roProbe)
277 if err != nil {
278 t.Fatal(err)
279 }
280 a := again.Admission
281 if a.NewGeneration != "fresh1" || a.KeepStamp || a.PredecessorCertified || again.Prepared || a.ObservedGeneration != roP || a.ObservedSession != opened.Claim.Holder.Session {
282 t.Fatalf("the later attempt %+v prepared=%v", a, again.Prepared)
283 }
284 if err := f.inputs.AdmissionEvidence(ctx, opened, again.Admission, again.Baseline); err != nil {
285 t.Fatalf("the later attempt's evidence: %v", err)
286 }
287 reopen := again.Admission
288 reopen.NewGeneration, reopen.KeepStamp = roP, true
289 if err := f.inputs.AdmissionEvidence(ctx, opened, reopen, first.Baseline); !errors.Is(err, hostbox.ErrChanged) {
290 t.Fatalf("P reopened: %v", err)
291 }
292}
293
294// The inputs serve only this box's relocation probe, of the pinned
295// release, at a nested route.
296func TestRelocationSelectRefusesEverythingButThisBoxsProbe(t *testing.T) {
297 ctx := t.Context()
298 for name, tc := range map[string]struct {
299 claim func(*hostclaim.Claim) error
300 policy activator.StartPolicy
301 route func(*registry.Route)
302 launch string
303 want error
304 }{
305 "installed": {claim: func(c *hostclaim.Claim) error { c.Op.Phase = hostclaim.PhaseInstalled; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
306 "another executor": {claim: func(c *hostclaim.Claim) error { c.Op.Exec, c.Op.Intent["target_box"] = "src", "src"; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
307 "stuck": {claim: func(c *hostclaim.Claim) error { c.Op.Journal["stuck"] = "an operator"; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
308 "not a probe": {policy: activator.StartPolicy{MaxStarts: 5, Window: time.Minute}, want: hostrelocate.ErrRefused},
309 "another release": {policy: roProbe, launch: "v2", want: hostrelocate.ErrRefused},
310 "a relative database": {policy: roProbe, route: func(rt *registry.Route) { rt.DBPath = "shop.db" }, want: hostbox.ErrUnsupported},
311 "another host": {policy: roProbe, route: func(rt *registry.Route) { rt.Host = "other.example" }, want: hostbox.ErrUnsupported},
312 } {
313 t.Run(name, func(t *testing.T) {
314 f := newROFixture(t, roOptions{})
315 if tc.claim != nil {
316 f.update(tc.claim)
317 }
318 rt := f.route
319 if tc.route != nil {
320 tc.route(&rt)
321 }
322 if tc.launch != "" {
323 launch := f.inputs.Launch
324 f.inputs.Launch = func(ctx context.Context, rt registry.Route) (ClaimServingLaunch, error) {
325 l, err := launch(ctx, rt)
326 l.Version = tc.launch
327 return l, err
328 }
329 }
330 if _, err := f.inputs.Select(ctx, rt, f.read(), tc.policy); !errors.Is(err, tc.want) {
331 t.Fatalf("got %v, want %v", err, tc.want)
332 }
333 })
334 }
335}
336
337// Ruling 47: P's opening is bound to its observation, its certified
338// baseline (P's snapshot at the pin's N) and P's prepared files manifest by
339// the digest prepared_authority journals, read again.
340func TestRelocationAdmissionEvidenceBindsTheBaselineAndThePreparedManifest(t *testing.T) {
341 ctx := t.Context()
342 f := newROFixture(t, roOptions{})
343 o := f.read()
344 in, err := f.inputs.Select(ctx, f.route, o, roProbe)
345 if err != nil {
346 t.Fatal(err)
347 }
348 if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
349 t.Fatal(err)
350 }
351 for name, mutate := range map[string]func(*hostclaim.Admission, *genselect.Plan){
352 "a stale baseline": func(a *hostclaim.Admission, p *genselect.Plan) { p.MaxTXID, a.RemoteMaxTXID = 41, 41 },
353 "another generation": func(_ *hostclaim.Admission, p *genselect.Plan) { p.Generation = "g1" },
354 "stamp not kept": func(a *hostclaim.Admission, _ *genselect.Plan) { a.KeepStamp = false },
355 "not certified": func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorCertified = false },
356 "another box": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Box = "src" },
357 "sealed coverage": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Coverage = "sealed" },
358 "another region": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Region = "uk" },
359 "another session": func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedSession++ },
360 "warm": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Mode = "warm" },
361 "lost writes claimed": func(a *hostclaim.Admission, _ *genselect.Plan) { a.AdditionalWritesLost = []hostclaim.WritesLost{{}} },
362 } {
363 a, p := in.Admission, *in.Baseline
364 mutate(&a, &p)
365 if err := f.inputs.AdmissionEvidence(ctx, o, a, &p); !errors.Is(err, hostbox.ErrChanged) {
366 t.Fatalf("%s: %v", name, err)
367 }
368 }
369 // P's files manifest replaced after the build: the digest no longer
370 // matches the authority.
371 key := hostfiles.ManifestKey(moves.GenerationFilesPrefix(o.Claim.Holder.Location, roP))
372 _, etag, err := f.files.Get(ctx, key)
373 if err != nil {
374 t.Fatal(err)
375 }
376 other, _ := json.Marshal(hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 0, Entries: map[string]hostfiles.Entry{"x": {Seq: 1, Deleted: true}}, Platform: map[string]hostfiles.Entry{}})
377 if _, err := f.files.Put(ctx, key, other, etag); err != nil {
378 t.Fatal(err)
379 }
380 if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
381 t.Fatalf("another files manifest: %v", err)
382 }
383 // An empty pin's opening has no baseline; a non-empty one must.
384 e := newROFixture(t, roOptions{empty: true})
385 eo := e.read()
386 ein, err := e.inputs.Select(ctx, e.route, eo, roProbe)
387 if err != nil {
388 t.Fatal(err)
389 }
390 if err := e.inputs.AdmissionEvidence(ctx, eo, ein.Admission, nil); err != nil {
391 t.Fatalf("the empty opening: %v", err)
392 }
393 if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, nil); !errors.Is(err, hostbox.ErrChanged) {
394 t.Fatalf("a non-empty opening without its baseline: %v", err)
395 }
396}
397
398// Ruling 41: one admitted session is bound to one record attempt k (the
399// claim's probe_attempts.<box>); the attempt's first app start is kept
400// across restarts; a start is reserved, settled and re-reserved under the
401// same session; a session bound to k is refused once the record moves on.
402// The journal is keyed by op (ruling 44).
403func TestRelocationProbeAttemptsBindOneSessionPerRecordAttempt(t *testing.T) {
404 ctx := t.Context()
405 f := newROFixture(t, roOptions{})
406 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
407 if err != nil {
408 t.Fatal(err)
409 }
410 admitted := f.admit(f.read(), in)
411 h := admitted.Claim.Holder
412 dataDir := t.TempDir()
413 now := time.Date(2026, 10, 7, 4, 0, 0, 0, time.UTC)
414 p := &relocationProbeAttempts{Claims: f.claims, DataDir: dataDir, Box: "dst", NewID: f.inputs.NewID, Now: func() time.Time { return now }}
415 owner := hostguard.FenceTarget{Region: "ie", Host: roHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: strings.Repeat("a", 32), BootID: "boot", ControllerID: "ctl"}
416 s := OperationStart{Admitted: admitted, Policy: roProbe, Inputs: in, Owner: owner}
417 opened, err := p.OpenProbe(ctx, s, now)
418 if err != nil || !opened.Equal(now) {
419 t.Fatalf("open %v %v", opened, err)
420 }
421 if again, err := p.OpenProbe(ctx, s, now.Add(time.Minute)); err != nil || !again.Equal(now) {
422 t.Fatalf("a restart moved the attempt's first start: %v %v", again, err)
423 }
424 if err := p.Reserve(ctx, s, 0); err != nil {
425 t.Fatal(err)
426 }
427 a := p.CurrentAttempt(s)
428 if a.ID == "" || a.Session != h.Session || a.Box != "dst" || a.ControllerID != "ctl" {
429 t.Fatalf("the reservation %+v", a)
430 }
431 if got, err := (relocationAttemptReader{DataDir: dataDir}).ReadAttempt(ctx, admitted); err != nil || got != a {
432 t.Fatalf("root reads %+v %v", got, err)
433 }
434 if err := p.Reserve(ctx, s, 0); err != nil || p.CurrentAttempt(s) != a {
435 t.Fatalf("a repeated reservation: %v", err)
436 }
437 if err := p.Reserve(ctx, s, 1); !errors.Is(err, hostbox.ErrChanged) {
438 t.Fatalf("a restart before the first start settled: %v", err)
439 }
440 if err := p.Failed(ctx, s, 0); err != nil {
441 t.Fatal(err)
442 }
443 if err := p.Reserve(ctx, s, 1); err != nil {
444 t.Fatalf("the restart after a failed start: %v", err)
445 }
446 if b := p.CurrentAttempt(s); b.Session != h.Session || b.LaunchInputs != a.LaunchInputs {
447 t.Fatalf("the restart %+v", b)
448 }
449 path, _ := relocationAttemptsPath(dataDir, roOp)
450 if strings.Contains(path, roHost) || filepath.Base(filepath.Dir(path)) != roOp {
451 t.Fatalf("the journal path %s", path)
452 }
453 // The record attempt moves on with the same session admitted: the
454 // session is bound to attempt 1 and refused for attempt 2.
455 moved := f.update(func(c *hostclaim.Claim) error { c.Op.Journal["probe_attempts.dst"] = "2"; return nil })
456 s.Admitted = moved
457 if err := p.Reserve(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
458 t.Fatalf("one session, two record attempts: %v", err)
459 }
460 // Another session than the admitted writer is refused before any journal.
461 other := OperationStart{Admitted: admitted, Policy: roProbe, Inputs: in, Owner: owner}
462 other.Owner.Session++
463 if _, err := p.OpenProbe(ctx, other, now); !errors.Is(err, hostbox.ErrChanged) {
464 t.Fatalf("another session: %v", err)
465 }
466}
467
468// The admission hooks: before P's first admission the source's release is
469// verified at its receipt (a drain) or its fence is journaled with the
470// source lost (a failover); after it, this box's install journal is the
471// receipt's and P keeps the install stamp.
472func TestRelocationAdmissionHooksVerifyReleaseAndInstall(t *testing.T) {
473 ctx := t.Context()
474 releases := &roReleases{}
475 before := relocationBeforeAdmission("dst", releases)
476 f := newROFixture(t, roOptions{})
477 if err := before(ctx, f.read()); err != nil || releases.calls != 1 {
478 t.Fatalf("a drain's release: %v (%d checks)", err, releases.calls)
479 }
480 releases.err = errors.New("no receipt")
481 if err := before(ctx, f.read()); !errors.Is(err, releases.err) {
482 t.Fatalf("an unverified release: %v", err)
483 }
484 g := newROFixture(t, roOptions{failover: true})
485 if err := before(ctx, g.read()); err != nil {
486 t.Fatalf("a fenced failover: %v", err)
487 }
488 unfenced := g.update(func(c *hostclaim.Claim) error { delete(c.Op.Journal, "fence.src"); return nil })
489 if err := before(ctx, unfenced); !errors.Is(err, hostrelocate.ErrRefused) {
490 t.Fatalf("neither release nor fence: %v", err)
491 }
492 if err := relocationBeforeAdmission("other", releases)(ctx, f.read()); !errors.Is(err, hostbox.ErrChanged) {
493 t.Fatalf("another box: %v", err)
494 }
495
496 // After: the install journal at the receipt's digest.
497 dataDir := t.TempDir()
498 after := relocationAfterAdmission(dataDir, "dst")
499 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
500 if err != nil {
501 t.Fatal(err)
502 }
503 prior := f.read()
504 path, _ := relocationInstallPath(dataDir, roOp)
505 if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
506 t.Fatal(err)
507 }
508 journal := []byte(`{"v":1,"host":"shop.example","op":"rel-1"}`)
509 if err := os.WriteFile(path, journal, 0o600); err != nil {
510 t.Fatal(err)
511 }
512 sum := sha256.Sum256(journal)
513 f.update(func(c *hostclaim.Claim) error {
514 var l hostclaim.LocalInstall
515 _ = json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l)
516 l.JournalSHA256 = hex.EncodeToString(sum[:])
517 c.Op.Journal["local_install"] = string(mustJSON(t, l))
518 return nil
519 })
520 admitted := f.admit(f.read(), in)
521 if err := after(ctx, prior, admitted); err != nil {
522 t.Fatalf("after P's opening: %v", err)
523 }
524 if err := os.WriteFile(path, []byte(`{"v":1}`), 0o600); err != nil {
525 t.Fatal(err)
526 }
527 if err := after(ctx, prior, admitted); !errors.Is(err, hostbox.ErrChanged) {
528 t.Fatalf("another install journal: %v", err)
529 }
530 if err := relocationAfterAdmission(dataDir, "src")(ctx, prior, admitted); !errors.Is(err, hostbox.ErrChanged) {
531 t.Fatalf("another box's admission: %v", err)
532 }
533}
534
535// relocationProbeActivator is the activator's probe and fence, recorded.
536type relocationProbeActivator struct {
537 probes []activator.ProbeConfig
538 result activator.ProbeResult
539 err error
540}
541
542func (a *relocationProbeActivator) ProbeWake(_ context.Context, pc activator.ProbeConfig) (activator.ProbeResult, error) {
543 a.probes = append(a.probes, pc)
544 return a.result, a.err
545}
546
547// The probe attempt never fences: a fence asked of this double fails the
548// test's call with an error naming it, and lifts nothing.
549func (a *relocationProbeActivator) FenceAndStopAs(context.Context, string, uint64, string) (*activator.StopEvidence, error) {
550 return nil, errors.New("the probe attempt asked for a fence")
551}
552
553func (a *relocationProbeActivator) LiftFenceAs(string, uint64, string) bool { return false }
554
555// One probe attempt is one ProbeWake, bounded by the probe deadline,
556// carrying P's install manifest on P's opening; a failed or overdue
557// attempt is not passed, and any other error is returned.
558func TestRelocationAdmitRunsOneProbeAttempt(t *testing.T) {
559 ctx := t.Context()
560 f := newROFixture(t, roOptions{})
561 act := &relocationProbeActivator{result: activator.ProbeResult{Session: 99}}
562 admit := ClaimAssembly{}.relocationAdmit(RelocationAssemblyConfig{Activator: act, Versions: f.bucket, Bucket: "data"})
563 o := f.read()
564 session, passed, err := admit(ctx, o, true)
565 if err != nil || !passed || session != 99 || len(act.probes) != 1 {
566 t.Fatalf("passed: %d %v %v", session, passed, err)
567 }
568 pc := act.probes[0]
569 since, _ := time.Parse(time.RFC3339Nano, o.Claim.Op.Journal["probing_since"])
570 if pc.Host != roHost || pc.Op != roOp || pc.HoldOp != roOp || !pc.Deadline.Equal(since.Add(hostrelocate.ProbeDeadline)) || pc.Prepared == nil || *pc.Prepared != f.manifest {
571 t.Fatalf("probe config %+v", pc)
572 }
573 for _, failed := range []error{activator.ErrProbeFailed, activator.ErrAttemptOver} {
574 act.err = failed
575 if _, passed, err := admit(ctx, o, true); passed || err != nil {
576 t.Fatalf("%v: passed=%v %v", failed, passed, err)
577 }
578 }
579 act.err = errors.New("the activator is closing")
580 if _, _, err := admit(ctx, o, true); !errors.Is(err, act.err) {
581 t.Fatalf("another error: %v", err)
582 }
583 if _, _, err := admit(ctx, o, false); !errors.Is(err, hostrelocate.ErrRefused) {
584 t.Fatalf("certification disagreed: %v", err)
585 }
586 if n := len(act.probes); n != 4 {
587 t.Fatalf("%d probes, want 4", n)
588 }
589}
590
591// Rulings 37 and 39: with moves' C8 provider bound, Open returns the
592// complete owner set and the two refusals; the relocating dispatch binds.
593// Without the region's bucket or a store constructor it is unavailable,
594// and a nil source is refused before it is locked.
595func TestRelocationSourceOpensACompleteOwnerSet(t *testing.T) {
596 ctx := t.Context()
597 var opened []string
598 src := &RelocationSource{DataDir: t.TempDir(), FilesStore: func(_ context.Context, bucket string) (release.Store, error) {
599 opened = append(opened, bucket)
600 return release.NewDirStore(t.TempDir()), nil
601 }}
602 identityErr := errors.New("the roster entry has moved")
603 var identity error
604 sources := ClaimServingSources{
605 Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) { return ClaimServingLaunch{}, nil },
606 Markers: plan9D5Markers{leasetest.NewVersionedBucket()},
607 Files: &roFiles{},
608 Identity: func(context.Context) error { return identity },
609 }
610 src.SetServingSources(sources)
611 actor := hostbox.Actor{Box: "dst", Bucket: "data-ie", Region: "ie"}
612 owners, refusals, err := src.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, actor)
613 if err != nil {
614 t.Fatal(err)
615 }
616 if owners.zero() || !owners.complete() || owners.Operations == nil || owners.Policy == nil || owners.BeforeAdmission == nil || owners.AfterAdmission == nil || len(refusals) != 2 {
617 t.Fatalf("owners %+v refusals %d", owners, len(refusals))
618 }
619 if len(opened) != 1 || opened[0] != "data-ie" {
620 t.Fatalf("the files store was opened for %v", opened)
621 }
622 if bound, err := claimRelocationC1(plan11Env("1", "1"), true, owners); !bound || err != nil {
623 t.Fatalf("the relocating dispatch did not bind: %v %v", bound, err)
624 }
625 // M2: every data-bucket request the inputs make runs the box's own
626 // identity check first, as main's serving owners do.
627 identity = identityErr
628 if _, err := owners.Inputs.(*RelocationWakeInputs).Data.ListVersions(ctx, "data-ie", "instances/"); !errors.Is(err, identityErr) {
629 t.Fatalf("a read without the identity check: %v", err)
630 }
631 identity = nil
632 if _, _, err := src.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, hostbox.Actor{Box: "dst", Region: "ie"}); !errors.Is(err, hostwrite.ErrUnavailable) {
633 t.Fatalf("no bucket: %v", err)
634 }
635 for name, without := range map[string]func(*ClaimServingSources){
636 "no launch": func(s *ClaimServingSources) { s.Launch = nil },
637 "no markers": func(s *ClaimServingSources) { s.Markers = nil },
638 "no identity": func(s *ClaimServingSources) { s.Identity = nil },
639 } {
640 partial := sources
641 without(&partial)
642 src.SetServingSources(partial)
643 if _, _, err := src.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, actor); !errors.Is(err, hostwrite.ErrUnavailable) {
644 t.Fatalf("%s: %v", name, err)
645 }
646 }
647 src.SetServingSources(sources)
648 var none *RelocationSource
649 if _, _, err := none.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, actor); !errors.Is(err, hostwrite.ErrUnavailable) {
650 t.Fatalf("a nil source: %v", err)
651 }
652}
653
654// Finding 11 (C8): without moves' files-manifest provider the source opens
655// no owners, and the dispatcher keeps relocating refused.
656func TestRelocationStaysUnboundWithoutTheC8Provider(t *testing.T) {
657 src := &RelocationSource{DataDir: t.TempDir(), FilesStore: func(context.Context, string) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }}
658 src.SetServingSources(ClaimServingSources{Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) { return ClaimServingLaunch{}, nil }, Markers: nil})
659 owners, refusals, err := src.Open(context.Background(), release.NewDirStore(t.TempDir()), &fakeVersioned{}, hostbox.Actor{Box: "dst", Bucket: "data-ie", Region: "ie"})
660 if err != nil || !owners.zero() || refusals != nil {
661 t.Fatalf("opened without C8: %+v %v %v", owners, refusals, err)
662 }
663 bound, err := claimRelocationC1(plan11Env("1", "1"), true, owners)
664 if bound || err != nil {
665 t.Fatalf("relocating bound without C8: %v %v", bound, err)
666 }
667}
668
669// authority rewrites the journaled prepared authority in place.
670func (f *roFixture) authority(mutate func(*hostclaim.PreparedAuthority)) hostclaim.Observed {
671 f.t.Helper()
672 return f.update(func(c *hostclaim.Claim) error {
673 var a hostclaim.PreparedAuthority
674 if err := json.Unmarshal([]byte(c.Op.Journal["prepared_authority"]), &a); err != nil {
675 return err
676 }
677 mutate(&a)
678 c.Op.Journal["prepared_authority"] = string(mustJSON(f.t, a))
679 return nil
680 })
681}
682
683// Absence is never evidence: a journaled release that is present but
684// unreadable, or that names another writer than the predecessor, is not
685// "no release" (fix round 1, I3). Before the fix a damaged release on a
686// claim whose source is lost and fenced passed the admission hook as a
687// failover, and a mismatched one opened P uncertified, declaring the
688// source's writes lost with no failover evidence.
689func TestRelocationAnUnreadableReleaseIsNeverAFailover(t *testing.T) {
690 ctx := t.Context()
691 for name, damage := range map[string]func(*hostclaim.Claim){
692 "unparseable": func(c *hostclaim.Claim) { c.Op.Journal["release"] = `{"released_by":` },
693 "no releasing box": func(c *hostclaim.Claim) { c.Op.Journal["release"] = `{}` },
694 "another session": func(c *hostclaim.Claim) {
695 var rel hostclaim.Release
696 _ = json.Unmarshal([]byte(c.Op.Journal["release"]), &rel)
697 rel.Session++
698 c.Op.Journal["release"] = string(mustJSON(t, rel))
699 },
700 "another box": func(c *hostclaim.Claim) {
701 var rel hostclaim.Release
702 _ = json.Unmarshal([]byte(c.Op.Journal["release"]), &rel)
703 rel.ReleasedBy = "other"
704 c.Op.Journal["release"] = string(mustJSON(t, rel))
705 },
706 } {
707 t.Run(name, func(t *testing.T) {
708 // A failover shape (the source lost and fenced) for the two
709 // unreadable cases, a drain for the two mismatched ones.
710 f := newROFixture(t, roOptions{failover: strings.HasPrefix(name, "unparseable") || strings.HasPrefix(name, "no ")})
711 good, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
712 if err != nil {
713 t.Fatal(err)
714 }
715 damaged := f.update(func(c *hostclaim.Claim) error { damage(c); return nil })
716 releases := &roReleases{}
717 if err := relocationBeforeAdmission("dst", releases)(ctx, damaged); !errors.Is(err, hostrelocate.ErrRelease) || releases.calls != 0 {
718 t.Fatalf("the hook took a damaged release: %v (%d checks)", err, releases.calls)
719 }
720 checked := f.releases.calls
721 if _, err := f.inputs.Select(ctx, f.route, damaged, roProbe); !errors.Is(err, hostrelocate.ErrRelease) || f.releases.calls != checked {
722 t.Fatalf("Select opened P over a damaged release: %v (%d checks)", err, f.releases.calls-checked)
723 }
724 if err := f.inputs.AdmissionEvidence(ctx, damaged, good.Admission, good.Baseline); !errors.Is(err, hostrelocate.ErrRelease) {
725 t.Fatalf("AdmissionEvidence passed over a damaged release: %v", err)
726 }
727 act := &relocationProbeActivator{}
728 probe := ClaimAssembly{}.relocationAdmit(RelocationAssemblyConfig{Activator: act, Versions: f.bucket, Bucket: "data"})
729 if _, _, err := probe(ctx, damaged, false); !errors.Is(err, hostrelocate.ErrRelease) || len(act.probes) != 0 {
730 t.Fatalf("the probe attempt ran over a damaged release: %v (%d probes)", err, len(act.probes))
731 }
732 // M1: the preflight finds it before any attempt is counted,
733 // before root is asked.
734 if err := (ClaimAssembly{Wake: &ClaimWakeAssembly{Roles: &ClaimRoleAssembly{}}}).relocationPreflight(RelocationAssemblyConfig{})(ctx, damaged); !errors.Is(err, hostrelocate.ErrRelease) {
735 t.Fatalf("the preflight passed a damaged release: %v", err)
736 }
737 })
738 }
739}
740
741// A drain's P opening is certified only by a release naming the exact
742// predecessor; the hook's PreparedOpened short-circuit stands on its own:
743// once P is open no release or fence is asked for.
744func TestRelocationBeforeAdmissionIgnoresTheReleaseOncePIsOpen(t *testing.T) {
745 ctx := t.Context()
746 f := newROFixture(t, roOptions{})
747 first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
748 if err != nil {
749 t.Fatal(err)
750 }
751 f.admit(f.read(), first)
752 opened := f.update(func(c *hostclaim.Claim) error { delete(c.Op.Journal, "release"); return nil })
753 releases := &roReleases{err: errors.New("never asked")}
754 if err := relocationBeforeAdmission("dst", releases)(ctx, opened); err != nil || releases.calls != 0 {
755 t.Fatalf("after P's opening: %v (%d checks)", err, releases.calls)
756 }
757}
758
759// Absence is never evidence: a later attempt follows P's admitted
760// session, which ran, so a listing that finds nothing is nothing
761// replicated or not yet visible, never an empty predecessor. Main's
762// serving rule admits an empty cold admission only for a holder that
763// never admitted a writer, which a later attempt never is (fix round 1,
764// I2). Over an empty P as over a pinned snapshot, Select refuses, no
765// EmptyAuthority proves it, and no evidence passes a nil baseline.
766func TestRelocationLaterAttemptIsNeverEmpty(t *testing.T) {
767 ctx := t.Context()
768 for name, o := range map[string]roOptions{"a pinned snapshot": {}, "an empty P": {empty: true}} {
769 t.Run(name, func(t *testing.T) {
770 f := newROFixture(t, o)
771 first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
772 if err != nil {
773 t.Fatal(err)
774 }
775 opened := f.admit(f.read(), first)
776 // The same inputs over a bucket that lists nothing for the location.
777 blank := leasetest.NewVersionedBucket()
778 blank.AddBucket("data")
779 elsewhere := &RelocationWakeInputs{Launch: f.inputs.Launch, FilesManifest: f.c8, Markers: plan9D5Markers{blank}, Claims: f.claims, FilesStore: f.files, Data: blank, Bucket: "data", Region: "ie", Box: "dst",
780 NewID: f.inputs.NewID, Now: f.inputs.Now, Releases: f.releases}
781 for _, inputs := range []*RelocationWakeInputs{f.inputs, elsewhere} {
782 // Over the pinned snapshot the real bucket lists P's root, a
783 // baseline; the blank bucket, and an empty P's own bucket, list
784 // nothing, which is never an empty admission.
785 if inputs == elsewhere || o.empty {
786 if in, err := inputs.Select(ctx, f.route, opened, roProbe); !errors.Is(err, hostbox.ErrChanged) {
787 t.Fatalf("an empty listing after P's session selected empty=%v: %v", in.Empty, err)
788 }
789 }
790 h := opened.Claim.Holder
791 fresh := hostclaim.Admission{ObservedOpID: roOp, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "dst", Mode: "cold", NewGeneration: "fresh9", Region: "ie", PredecessorRegion: "ie"}
792 if err := inputs.EmptyAuthority(ctx, opened, ClaimLaunchInputs{Empty: true, Admission: fresh}); !errors.Is(err, hostbox.ErrChanged) {
793 t.Fatalf("EmptyAuthority proved an empty later attempt: %v", err)
794 }
795 if err := inputs.AdmissionEvidence(ctx, opened, fresh, nil); !errors.Is(err, hostbox.ErrChanged) {
796 t.Fatalf("AdmissionEvidence passed an empty later attempt: %v", err)
797 }
798 }
799 })
800 }
801}
802
803// Absence is never evidence: P opens empty only on the pin's word
804// (RecoveryPin.Empty), which the prepared authority must agree with. A
805// missing install manifest alone, an authority with a manifest under an
806// empty pin, or an unreadable pin, refuses at Select and at the probe
807// attempt, before any admission.
808func TestRelocationEmptyPNeedsThePinAndTheAuthorityToAgree(t *testing.T) {
809 ctx := t.Context()
810 admit := func(f *roFixture) (func(context.Context, hostclaim.Observed, bool) (uint64, bool, error), *relocationProbeActivator) {
811 act := &relocationProbeActivator{}
812 return ClaimAssembly{}.relocationAdmit(RelocationAssemblyConfig{Activator: act, Versions: f.bucket, Bucket: "data"}), act
813 }
814 t.Run("no manifest under a non-empty pin", func(t *testing.T) {
815 f := newROFixture(t, roOptions{})
816 o := f.authority(func(a *hostclaim.PreparedAuthority) { a.InstallManifest, a.SHA256, a.SnapshotTXID = "", "", 0 })
817 if in, err := f.inputs.Select(ctx, f.route, o, roProbe); !errors.Is(err, hostbox.ErrChanged) {
818 t.Fatalf("selected empty=%v: %v", in.Empty, err)
819 }
820 probe, act := admit(f)
821 if _, _, err := probe(ctx, o, true); !errors.Is(err, hostbox.ErrChanged) || len(act.probes) != 0 {
822 t.Fatalf("the probe ran without P's manifest: %v (%d probes)", err, len(act.probes))
823 }
824 })
825 t.Run("a manifest under an empty pin", func(t *testing.T) {
826 f := newROFixture(t, roOptions{empty: true})
827 o := f.authority(func(a *hostclaim.PreparedAuthority) { a.InstallManifest = "somewhere/install.json" })
828 if _, err := f.inputs.Select(ctx, f.route, o, roProbe); !errors.Is(err, hostbox.ErrChanged) {
829 t.Fatalf("selected: %v", err)
830 }
831 probe, act := admit(f)
832 if _, _, err := probe(ctx, o, true); !errors.Is(err, hostbox.ErrChanged) || len(act.probes) != 0 {
833 t.Fatalf("the probe ran: %v (%d probes)", err, len(act.probes))
834 }
835 })
836 t.Run("an unreadable pin", func(t *testing.T) {
837 f := newROFixture(t, roOptions{})
838 o := f.update(func(c *hostclaim.Claim) error { c.Op.Journal["pin"] = `{"empty":`; return nil })
839 if _, err := f.inputs.Select(ctx, f.route, o, roProbe); !errors.Is(err, hostbox.ErrChanged) {
840 t.Fatalf("selected: %v", err)
841 }
842 probe, act := admit(f)
843 if _, _, err := probe(ctx, o, true); !errors.Is(err, hostbox.ErrChanged) || len(act.probes) != 0 {
844 t.Fatalf("the probe ran: %v (%d probes)", err, len(act.probes))
845 }
846 })
847}
848
849// roWork is a spawn selection whose fence always holds.
850type roWork struct{}
851
852func (roWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error { return fn() }
853
854// putManifest replaces P's install.json in the bucket and the authority's
855// digest of it, so only the mutated field disagrees.
856func (f *roFixture) putManifest(mutate func(*moves.InstallManifest)) hostclaim.Observed {
857 f.t.Helper()
858 im := f.manifest
859 mutate(&im)
860 raw, _ := json.Marshal(im)
861 f.bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix(f.read().Claim.Holder.Location, roP)), raw)
862 sum := sha256.Sum256(raw)
863 return f.authority(func(a *hostclaim.PreparedAuthority) { a.SHA256 = hex.EncodeToString(sum[:]) })
864}
865
866// atP is the observation as the admission that opens P leaves the holder,
867// without writing the claim.
868func atP(o hostclaim.Observed) hostclaim.Observed {
869 c := *o.Claim
870 h := *c.Holder
871 h.Generation = roP
872 c.Holder = &h
873 o.Claim = &c
874 return o
875}
876
877// Rulings 37 and 39, by behaviour: the two policy refusals refuse a
878// relocating selection at another installation than the holder's, and
879// any selection by a box the claim lists as lost; they pass everything
880// else and never grant.
881func TestRelocationPolicyRefusalsRefuseByBehaviour(t *testing.T) {
882 ctx := t.Context()
883 f := newROFixture(t, roOptions{})
884 c := f.read().Claim
885 h := c.Holder
886 at := hostbox.Selection{Host: roHost, Action: "probe", Box: "dst", Region: "ie", Captured: hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}}
887 if err := refuseStaleStamp(ctx, c, at, hostwrite.Identity{}); err != nil {
888 t.Fatalf("the holder's installation: %v", err)
889 }
890 stale := at
891 stale.Captured.Stamp++
892 if err := refuseStaleStamp(ctx, c, stale, hostwrite.Identity{}); !errors.Is(err, hostbox.ErrChanged) {
893 t.Fatalf("a retired stamp: %v", err)
894 }
895 other := at
896 other.Captured.Incarnation = strings.Repeat("9", 64)
897 if err := refuseStaleStamp(ctx, c, other, hostwrite.Identity{}); !errors.Is(err, hostbox.ErrChanged) {
898 t.Fatalf("another incarnation: %v", err)
899 }
900 serving := *c
901 serving.Op = hostclaim.Op{Kind: hostclaim.KindServing}
902 if err := refuseStaleStamp(ctx, &serving, stale, hostwrite.Identity{}); err != nil {
903 t.Fatalf("the stamp refusal reaches beyond relocating: %v", err)
904 }
905 if err := refuseLostBox(ctx, c, at, hostwrite.Identity{}); err != nil {
906 t.Fatalf("a box not lost: %v", err)
907 }
908 lost := *c
909 lost.Lost = []string{"dst"}
910 if err := refuseLostBox(ctx, &lost, at, hostwrite.Identity{}); !errors.Is(err, hostbox.ErrChanged) {
911 t.Fatalf("a lost box: %v", err)
912 }
913 if err := refuseLostBox(ctx, &lost, hostbox.Selection{Box: "src"}, hostwrite.Identity{}); err != nil {
914 t.Fatalf("another box while dst is lost: %v", err)
915 }
916}
917
918// preparedMarker binds install.json to the prepared authority (its host,
919// its snapshot position, the holder at P) and takes an existing marker
920// for P only when it is this box's, this incarnation's, this host's, with
921// the authority's digest; another op's marker is superseded by its ETag.
922func TestRelocationPreparedMarkerBindsTheManifestAndTheMarker(t *testing.T) {
923 ctx := t.Context()
924 fence := func(context.Context) error { return nil }
925 t.Run("install.json", func(t *testing.T) {
926 for name, mutate := range map[string]func(*moves.InstallManifest){
927 "another host": func(im *moves.InstallManifest) { im.Host = "other.example" },
928 "another position": func(im *moves.InstallManifest) { im.N = 41 },
929 } {
930 f := newROFixture(t, roOptions{})
931 o := f.putManifest(mutate)
932 if _, err := f.inputs.preparedMarker(ctx, atP(o), fence); !errors.Is(err, hostbox.ErrChanged) {
933 t.Fatalf("%s: %v", name, err)
934 }
935 }
936 f := newROFixture(t, roOptions{})
937 if _, err := f.inputs.preparedMarker(ctx, f.read(), fence); !errors.Is(err, hostbox.ErrChanged) {
938 t.Fatalf("the holder not at P: %v", err)
939 }
940 if _, err := f.inputs.preparedMarker(ctx, atP(f.read()), fence); err != nil {
941 t.Fatalf("the authority's own manifest: %v", err)
942 }
943 })
944 t.Run("an existing marker", func(t *testing.T) {
945 f := newROFixture(t, roOptions{})
946 o := atP(f.read())
947 h := o.Claim.Holder
948 auth, _ := preparedAuthority(o.Claim)
949 mine := genselect.Marker{Op: roP, Incarnation: h.Incarnation, Host: roHost, Box: "dst", ManifestSHA256: auth.SHA256}
950 for name, mutate := range map[string]func(*genselect.Marker){
951 "another box": func(m *genselect.Marker) { m.Box = "src" },
952 "another incarnation": func(m *genselect.Marker) { m.Incarnation = strings.Repeat("9", 64) },
953 "another host": func(m *genselect.Marker) { m.Host = "other.example" },
954 "another manifest": func(m *genselect.Marker) { m.ManifestSHA256 = strings.Repeat("0", 64) },
955 } {
956 m := mine
957 mutate(&m)
958 f.bucket.Put("data", markerKey(h), mustJSON(t, m))
959 if _, err := f.inputs.preparedMarker(ctx, o, fence); !errors.Is(err, hostbox.ErrChanged) {
960 t.Fatalf("%s: %v", name, err)
961 }
962 }
963 written := f.bucket.Put("data", markerKey(h), mustJSON(t, mine))
964 if r, err := f.inputs.preparedMarker(ctx, o, fence); err != nil || r.MarkerVersion != written {
965 t.Fatalf("P's own marker: %+v %v", r, err)
966 }
967 earlier := f.bucket.Put("data", markerKey(h), mustJSON(t, genselect.Marker{Op: "g1", Incarnation: h.Incarnation, Host: roHost, Box: "src"}))
968 r, err := f.inputs.preparedMarker(ctx, o, fence)
969 if err != nil || r.MarkerVersion == earlier || r.MarkerVersion != f.bucket.LatestID("data", markerKey(h)) {
970 t.Fatalf("an earlier generation's marker superseded: %+v %v", r, err)
971 }
972 })
973}
974
975// PublicationEvidence for P's prepared opening requires the holder still
976// at P, with the marker exactly the recorded version.
977func TestRelocationPublicationEvidenceNeedsTheHolderAtP(t *testing.T) {
978 ctx := t.Context()
979 f := newROFixture(t, roOptions{})
980 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
981 if err != nil {
982 t.Fatal(err)
983 }
984 admitted := f.admit(f.read(), in)
985 if err := f.inputs.Materialize(ctx, admitted, in, roWork{}); err != nil {
986 t.Fatal(err)
987 }
988 if err := f.inputs.PublicationEvidence(ctx, admitted); err != nil {
989 t.Fatalf("P's receipt: %v", err)
990 }
991 moved := admitted
992 c := *moved.Claim
993 h := *c.Holder
994 h.Generation = "g9"
995 c.Holder = &h
996 moved.Claim = &c
997 if err := f.inputs.PublicationEvidence(ctx, moved); !errors.Is(err, hostbox.ErrChanged) {
998 t.Fatalf("the holder moved off P: %v", err)
999 }
1000 f.bucket.Put("data", markerKey(admitted.Claim.Holder), mustJSON(t, genselect.Marker{Op: "g9", Box: "dst"}))
1001 if err := f.inputs.PublicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
1002 t.Fatalf("another marker version: %v", err)
1003 }
1004}
1005
1006// A later attempt's evidence re-selects the predecessor's baseline and
1007// binds the admission to it: its position, the declared remote position,
1008// and the exact files.
1009func TestRelocationLaterAttemptEvidenceReselectsTheBaseline(t *testing.T) {
1010 ctx := t.Context()
1011 f := newROFixture(t, roOptions{})
1012 first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
1013 if err != nil {
1014 t.Fatal(err)
1015 }
1016 opened := f.admit(f.read(), first)
1017 again, err := f.inputs.Select(ctx, f.route, opened, roProbe)
1018 if err != nil || again.Baseline == nil {
1019 t.Fatalf("the later attempt: %+v %v", again, err)
1020 }
1021 if err := f.inputs.AdmissionEvidence(ctx, opened, again.Admission, again.Baseline); err != nil {
1022 t.Fatal(err)
1023 }
1024 for name, mutate := range map[string]func(*hostclaim.Admission, *genselect.Plan){
1025 "a stale position": func(a *hostclaim.Admission, p *genselect.Plan) { p.MaxTXID, a.RemoteMaxTXID = 41, 41 },
1026 "another declared loss": func(a *hostclaim.Admission, _ *genselect.Plan) { a.RemoteMaxTXID++ },
1027 "other files": func(_ *hostclaim.Admission, p *genselect.Plan) {
1028 p.Files = append([]genselect.File{{Key: "x"}}, p.Files...)
1029 },
1030 "the predecessor's name": func(_ *hostclaim.Admission, p *genselect.Plan) { p.Generation = "g1" },
1031 } {
1032 a, p := again.Admission, *again.Baseline
1033 mutate(&a, &p)
1034 if err := f.inputs.AdmissionEvidence(ctx, opened, a, &p); !errors.Is(err, hostbox.ErrChanged) {
1035 t.Fatalf("%s: %v", name, err)
1036 }
1037 }
1038}
1039
1040// After the admission, P's opening keeps the install stamp: an admitted
1041// holder at P with another stamp is refused even with the receipt whole.
1042func TestRelocationAfterAdmissionNeedsTheInstallStampAtP(t *testing.T) {
1043 ctx := t.Context()
1044 f := newROFixture(t, roOptions{})
1045 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
1046 if err != nil {
1047 t.Fatal(err)
1048 }
1049 dataDir := t.TempDir()
1050 path, _ := relocationInstallPath(dataDir, roOp)
1051 if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
1052 t.Fatal(err)
1053 }
1054 journal := []byte(`{"v":1,"host":"shop.example","op":"rel-1"}`)
1055 if err := os.WriteFile(path, journal, 0o600); err != nil {
1056 t.Fatal(err)
1057 }
1058 sum := sha256.Sum256(journal)
1059 f.update(func(c *hostclaim.Claim) error {
1060 var l hostclaim.LocalInstall
1061 _ = json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l)
1062 l.JournalSHA256 = hex.EncodeToString(sum[:])
1063 c.Op.Journal["local_install"] = string(mustJSON(t, l))
1064 return nil
1065 })
1066 prior := f.read()
1067 admitted := f.admit(prior, in)
1068 after := relocationAfterAdmission(dataDir, "dst")
1069 if err := after(ctx, prior, admitted); err != nil {
1070 t.Fatal(err)
1071 }
1072 moved := admitted
1073 c := *moved.Claim
1074 h := *c.Holder
1075 h.Stamp++
1076 c.Holder = &h
1077 moved.Claim = &c
1078 if err := after(ctx, prior, moved); !errors.Is(err, hostbox.ErrChanged) {
1079 t.Fatalf("P at another stamp: %v", err)
1080 }
1081}
1082
1083// The journal refuses once the claim no longer names this box the
1084// executor at probing, unstuck; a reserved start is bound to the
1085// admission's generation and stamp; a settle binds the exact reservation
1086// this process made.
1087func TestRelocationProbeAttemptsRefuseAMovedClaimOrJournal(t *testing.T) {
1088 ctx := t.Context()
1089 now := time.Date(2026, 10, 7, 4, 0, 0, 0, time.UTC)
1090 start := func(t *testing.T, f *roFixture, dataDir string) (*relocationProbeAttempts, OperationStart) {
1091 t.Helper()
1092 in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
1093 if err != nil {
1094 t.Fatal(err)
1095 }
1096 admitted := f.admit(f.read(), in)
1097 h := admitted.Claim.Holder
1098 p := &relocationProbeAttempts{Claims: f.claims, DataDir: dataDir, Box: "dst", NewID: f.inputs.NewID, Now: func() time.Time { return now }}
1099 owner := hostguard.FenceTarget{Region: "ie", Host: roHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: strings.Repeat("a", 32), BootID: "boot", ControllerID: "ctl"}
1100 s := OperationStart{Admitted: admitted, Policy: roProbe, Inputs: in, Owner: owner}
1101 if err := p.Reserve(ctx, s, 0); err != nil {
1102 t.Fatal(err)
1103 }
1104 return p, s
1105 }
1106 t.Run("stuck", func(t *testing.T) {
1107 f := newROFixture(t, roOptions{})
1108 p, s := start(t, f, t.TempDir())
1109 f.update(func(c *hostclaim.Claim) error { c.Op.Journal["stuck"] = "an operator"; return nil })
1110 if err := p.Reserve(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
1111 t.Fatalf("a stuck operation: %v", err)
1112 }
1113 })
1114 t.Run("another executor", func(t *testing.T) {
1115 // A journal bound for src, with src's record attempt counted, over
1116 // a claim whose executor is dst: the admitted writer still matches,
1117 // only the executor does not.
1118 f := newROFixture(t, roOptions{})
1119 _, s := start(t, f, t.TempDir())
1120 s.Admitted = f.update(func(c *hostclaim.Claim) error { c.Op.Journal["probe_attempts.src"] = "1"; return nil })
1121 s.Owner.Box = "src"
1122 src := &relocationProbeAttempts{Claims: f.claims, DataDir: t.TempDir(), Box: "src", NewID: f.inputs.NewID, Now: func() time.Time { return now }}
1123 if err := src.Reserve(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
1124 t.Fatalf("another executor: %v", err)
1125 }
1126 })
1127 for name, rewrite := range map[string]func(*relocationStart){
1128 "another generation": func(st *relocationStart) { st.Generation = "g9" },
1129 "another stamp": func(st *relocationStart) { st.Stamp++ },
1130 } {
1131 t.Run("a journaled start under "+name, func(t *testing.T) {
1132 f := newROFixture(t, roOptions{})
1133 dd := t.TempDir()
1134 p, s := start(t, f, dd)
1135 j, err := readRelocationAttempts(dd, roHost, roOp)
1136 if err != nil {
1137 t.Fatal(err)
1138 }
1139 st := j.Starts["1"]
1140 rewrite(&st)
1141 j.Starts["1"] = st
1142 if err := j.write(dd); err != nil {
1143 t.Fatal(err)
1144 }
1145 if err := p.Reserve(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
1146 t.Fatalf("%s: %v", name, err)
1147 }
1148 })
1149 }
1150 t.Run("a settle of another reservation", func(t *testing.T) {
1151 f := newROFixture(t, roOptions{})
1152 dd := t.TempDir()
1153 p, s := start(t, f, dd)
1154 j, _ := readRelocationAttempts(dd, roHost, roOp)
1155 st := j.Starts["1"]
1156 st.Attempt.ID = "someone-else"
1157 j.Starts["1"] = st
1158 if err := j.write(dd); err != nil {
1159 t.Fatal(err)
1160 }
1161 if err := p.Failed(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
1162 t.Fatalf("settled a reservation this process did not make: %v", err)
1163 }
1164 })
1165}
1166
1167// M1: the preflight refuses a claim whose pin, authority or probing start
1168// is damaged before root is asked and before any attempt is counted; a
1169// sound claim goes on to root.
1170func TestRelocationPreflightRefusesADamagedClaimBeforeCounting(t *testing.T) {
1171 ctx := t.Context()
1172 preflight := (ClaimAssembly{Wake: &ClaimWakeAssembly{Roles: &ClaimRoleAssembly{}}}).relocationPreflight(RelocationAssemblyConfig{})
1173 for name, damage := range map[string]func(*hostclaim.Claim){
1174 "an unreadable pin": func(c *hostclaim.Claim) { c.Op.Journal["pin"] = `{"empty":` },
1175 "an unreadable authority": func(c *hostclaim.Claim) { c.Op.Journal["prepared_authority"] = `{` },
1176 "no probing start": func(c *hostclaim.Claim) { delete(c.Op.Journal, "probing_since") },
1177 } {
1178 f := newROFixture(t, roOptions{})
1179 damaged := f.update(func(c *hostclaim.Claim) error { damage(c); return nil })
1180 if err := preflight(ctx, damaged); !errors.Is(err, hostbox.ErrChanged) {
1181 t.Fatalf("%s: %v", name, err)
1182 }
1183 }
1184 f := newROFixture(t, roOptions{})
1185 if err := preflight(ctx, f.read()); !errors.Is(err, activator.ErrClaimIsolationUnavailable) {
1186 t.Fatalf("a sound claim went to root: %v", err)
1187 }
1188}
1189