amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostrelocate
2
3import (
4 "context"
5 "crypto/ed25519"
6 "crypto/rand"
7 "crypto/sha256"
8 "encoding/hex"
9 "encoding/json"
10 "errors"
11 "fmt"
12 "slices"
13 "strings"
14 "testing"
15 "time"
16
17 "github.com/carlosframework/platform/internal/hostactors"
18 "github.com/carlosframework/platform/internal/hostclaim"
19 "github.com/carlosframework/platform/internal/hostfence"
20 "github.com/carlosframework/platform/internal/park"
21 "github.com/carlosframework/platform/internal/release"
22 "github.com/carlosframework/platform/internal/termfence"
23)
24
25const host = "shop.bab.oncarlos.com"
26
27var boxes = map[string]string{"wrk": "a", "src": "b", "dst": "c", "spa": "d", "sp2": "e", "sp3": "f"}
28
29func enrollment(box string) string { return strings.Repeat(boxes[box], 32) }
30
31type fixture struct {
32 pinErr error
33 t *testing.T
34 s release.Store
35 w Worker
36 now time.Time
37 notified int
38 sign release.Signer
39 roster hostactors.Roster
40 // leases records the lease provider's calls in order with the pinner's.
41 leases []string
42 loseLease bool
43 ineligible map[string]bool
44}
45
46// fakeLease is one held replica lease; lost is closed when the test takes
47// it away before the pin is journaled.
48type fakeLease struct {
49 f *fixture
50 lost chan struct{}
51}
52
53func (l *fakeLease) Lost() <-chan struct{} { return l.lost }
54func (l *fakeLease) Release(context.Context) error {
55 l.f.leases = append(l.f.leases, "release")
56 return nil
57}
58
59type okReleases struct{}
60
61func (okReleases) VerifyRelease(context.Context, hostclaim.Claim, hostclaim.Release) error {
62 return nil
63}
64
65type admission struct{ registered []string }
66
67func (a *admission) Register(_ context.Context, _ hostclaim.Observed, o Owner, id string) (string, error) {
68 if o.Box != "wrk" || o.Enrollment != enrollment("wrk") {
69 return "", errors.New("wrong owner")
70 }
71 a.registered = append(a.registered, id)
72 return "adm-" + id, nil
73}
74func (a *admission) Recheck(context.Context, string) error { return nil }
75
76// built reports P built once the target has said so (the participant's
77// half), as moves' install routine would leave it.
78type built struct{}
79
80func (built) Verify(_ context.Context, o hostclaim.Observed) (hostclaim.PreparedAuthority, bool, error) {
81 c := o.Claim
82 if c.Op.Journal["test.built"] != c.Op.Journal["prepared"] {
83 return hostclaim.PreparedAuthority{}, false, nil
84 }
85 var pin hostclaim.RecoveryPin
86 _ = json.Unmarshal([]byte(c.Op.Journal["pin"]), &pin)
87 return hostclaim.PreparedAuthority{Generation: c.Op.Journal["prepared"], InstallManifest: "m", SHA256: strings.Repeat("c", 64), SnapshotTXID: pin.MaxTXID,
88 SourceFilesSeq: pin.FilesSeq, FilesManifestSHA256: strings.Repeat("f", 64)}, true, nil
89}
90
91type pins struct{ f *fixture }
92
93func (p pins) Pin(_ context.Context, o hostclaim.Observed, now time.Time) (hostclaim.RecoveryPin, error) {
94 p.f.leases = append(p.f.leases, "pin")
95 if p.f.pinErr != nil {
96 return hostclaim.RecoveryPin{}, p.f.pinErr
97 }
98 var r hostclaim.Release
99 _ = json.Unmarshal([]byte(o.Claim.Op.Journal["release"]), &r)
100 return hostclaim.RecoveryPin{Key: PinKey(o.Claim.Host, o.Claim.Op.ID, o.Claim.Gen), SHA256: strings.Repeat("a", 64), Generation: o.Claim.Holder.Generation,
101 MaxTXID: max(r.FinalTXID, 40), FilesManifest: "m", FilesSeq: max(r.ManifestSeq, 7), PinnedAt: now, Release: "v1.4.2"}, nil
102}
103
104func newFixture(t *testing.T) *fixture {
105 t.Helper()
106 ctx := context.Background()
107 f := &fixture{t: t, s: release.NewDirStore(t.TempDir()), now: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)}
108 pub, key, err := ed25519.GenerateKey(rand.Reader)
109 if err != nil {
110 t.Fatal(err)
111 }
112 f.sign = release.RootSigner(key)
113 f.roster = hostactors.Roster{V: 1, DeploymentBucket: "bucket", Revision: 1, Regions: map[string]hostactors.Region{"ie": {Account: "123", AWSRegion: "eu-west-1"}},
114 Boxes: map[string]hostactors.Box{}, Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
115 for ref := range boxes {
116 f.roster.Boxes[ref] = hostactors.Box{Enrollment: enrollment(ref), Namespace: "platform", BoxID: ref, Label: ref, Region: "ie", Pool: "default",
117 Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-" + ref}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}
118 }
119 f.writeRoster()
120 d := hostactors.Directory{Store: f.s, PublicKey: pub, Bucket: "bucket"}
121 clock := func() time.Time { return f.now }
122 n := 0
123 f.w = Worker{Claims: f.s, Directory: d, Fences: hostfence.ClaimFenceAuthority{Directory: d, Now: clock}, Releases: okReleases{}, Admission: &admission{},
124 Census: func(context.Context, string, []string) error { return nil },
125 Choose: func(_ context.Context, _ hostclaim.Observed, exclude []string) (hostclaim.Target, error) {
126 for _, b := range []string{"dst", "spa", "sp2", "sp3"} {
127 if !slices.Contains(exclude, b) {
128 return hostclaim.Target{Box: b, Enrollment: enrollment(b)}, nil
129 }
130 }
131 return hostclaim.Target{}, errors.New("no eligible box")
132 },
133 Eligible: func(_ context.Context, _ hostclaim.Observed, box string) error {
134 if f.ineligible[box] {
135 return errors.New("not eligible: " + box)
136 }
137 return nil
138 },
139 Relocatable: func(context.Context, hostclaim.Observed) error { return nil },
140 Leases: func(context.Context, string) (Lease, error) {
141 f.leases = append(f.leases, "acquire")
142 l := &fakeLease{f: f, lost: make(chan struct{})}
143 if f.loseLease {
144 close(l.lost)
145 }
146 return l, nil
147 },
148 Pins: pins{f},
149 Prepared: built{},
150 Discharge: func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge { return nil },
151 Notify: func(context.Context) { f.notified++ },
152 Self: termfence.WrittenBy{Box: "wrk", Enrollment: enrollment("wrk")}, Region: "ie", Controller: "ctl",
153 NewID: func() string { n++; return "id" + string(rune('0'+n)) }, Now: clock}
154 f.ineligible = map[string]bool{}
155 // A host serving on src, whose source admitted one cold session.
156 created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
157 nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
158 body := []byte(`{"host":"` + host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
159 o, err := hostclaim.Read(ctx, f.s, host)
160 if err != nil {
161 t.Fatal(err)
162 }
163 o = f.step(o, hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment("src")}))
164 o = f.step(o, hostclaim.CompleteCreate("c1", "src"))
165 f.step(o, hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
166 return f
167}
168
169func (f *fixture) writeRoster() {
170 f.t.Helper()
171 raw, err := release.SignRootEnvelope(f.sign, f.roster)
172 if err != nil {
173 f.t.Fatal(err)
174 }
175 _, etag, err := f.s.Get(context.Background(), hostactors.Key)
176 if err != nil {
177 etag = ""
178 }
179 if _, err := f.s.Put(context.Background(), hostactors.Key, raw, etag); err != nil {
180 f.t.Fatal(err)
181 }
182}
183
184func (f *fixture) step(o hostclaim.Observed, tr func(*hostclaim.Claim) error) hostclaim.Observed {
185 f.t.Helper()
186 next, err := hostclaim.Update(context.Background(), f.s, o, tr)
187 if err != nil {
188 f.t.Fatal(err)
189 }
190 return next
191}
192
193func (f *fixture) claim() hostclaim.Observed {
194 f.t.Helper()
195 o, err := hostclaim.Read(context.Background(), f.s, host)
196 if err != nil {
197 f.t.Fatal(err)
198 }
199 return o
200}
201
202// participant applies a box-side transition (Plan 6b's half) to the claim.
203func (f *fixture) participant(tr func(*hostclaim.Claim) error) {
204 f.t.Helper()
205 f.step(f.claim(), tr)
206}
207
208func (f *fixture) ack(box string) {
209 f.participant(hostclaim.AcknowledgeExecution(box, hostclaim.ExecutionAcknowledgement{Op: f.claim().Claim.Op.ID, Enrollment: enrollment(box), BootID: "boot", ControllerID: "ctl", LeaseETag: "lease", Expires: f.now.Add(time.Hour)}))
210}
211
212// terminate writes box's termination fence as hostfence.TerminationWorker
213// would, observed a minute ago by wrk.
214func (f *fixture) terminate(box string) {
215 f.t.Helper()
216 b := f.roster.Boxes[box]
217 rec := termfence.Record{V: 1, Box: box, Enrollment: b.Enrollment, Cloud: b.Cloud, State: "terminated", ObservedAt: f.now.Add(-time.Minute).Format(time.RFC3339Nano),
218 Via: termfence.ViaDescribe, RequestID: "req", WrittenBy: termfence.WrittenBy{Box: "wrk", Enrollment: enrollment("wrk")}}
219 raw, _ := json.Marshal(rec)
220 if _, err := f.s.Put(context.Background(), hostfence.FenceKey(b.Enrollment), raw, ""); err != nil {
221 f.t.Fatal(err)
222 }
223}
224
225func (f *fixture) worker(want error) hostclaim.Observed {
226 f.t.Helper()
227 o, err := f.w.Step(context.Background(), host)
228 if !errors.Is(err, want) {
229 f.t.Fatalf("step: %v, want %v", err, want)
230 }
231 return o
232}
233
234func (f *fixture) installAndProbe(target string) {
235 f.t.Helper()
236 c := f.claim().Claim
237 id, p := c.Op.ID, c.Op.Journal["prepared"]
238 f.worker(ErrNotYet) // P is not built yet
239 f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
240 if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalled {
241 f.t.Fatalf("the worker records installed once P verifies: %s", o.Claim.Op.Phase)
242 }
243 stamp := f.claim().Claim.Op.Journal["install_stamp"]
244 var n uint64
245 _ = json.Unmarshal([]byte(stamp), &n)
246 f.participant(hostclaim.RecordLocalInstall(id, hostclaim.LocalInstall{Box: target, Enrollment: enrollment(target), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("d", 64)}))
247 f.participant(hostclaim.ToProbing(id, f.now))
248 o := f.claim()
249 f.participant(hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: id, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
250 Box: target, Mode: "cold", NewGeneration: p, KeepStamp: true, Region: "ie", PredecessorRegion: "ie"}))
251}
252
253func TestWorkerDrivesADrain(t *testing.T) {
254 f := newFixture(t)
255 o, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"})
256 if err != nil {
257 t.Fatal(err)
258 }
259 if o.Claim.Op.Intent["admission_id"] == "" || o.Claim.Op.Intent["directory_revision"] != "1" {
260 t.Fatalf("intent: %+v", o.Claim.Op.Intent)
261 }
262 f.worker(ErrNotYet) // no acknowledgements yet
263 f.ack("src")
264 f.ack("dst")
265 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseReleasing {
266 t.Fatalf("phase %s", o.Claim.Op.Phase)
267 }
268 f.worker(ErrNotYet) // the source has not released
269 h := f.claim().Claim.Holder
270 f.participant(hostclaim.JournalRelease(o.Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
271 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Exec != "none" {
272 t.Fatalf("pinning: %+v", o.Claim.Op)
273 }
274 if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
275 t.Fatal("no pin journaled")
276 }
277 if want := []string{"acquire", "pin", "release"}; !slices.Equal(f.leases, want) {
278 t.Fatalf("the pin was not selected and journaled under the location lease: %v", f.leases)
279 }
280 before := f.notified
281 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalling || o.Claim.Op.Exec != "dst" || f.notified != before+1 {
282 t.Fatalf("installing: %+v notified %d", o.Claim.Op, f.notified-before)
283 }
284 f.installAndProbe("dst")
285 f.worker(ErrNotYet) // probing, no probe result, within the deadline
286 o = f.claim()
287 f.participant(hostclaim.RecordProbe(o.Claim.Op.ID, hostclaim.Probe{Box: "dst", Session: o.Claim.Holder.Session}))
288 f.worker(ErrNotYet) // the probe passed; its session is not stopped yet (ruling 49)
289 f.participant(hostclaim.RecordProbeStop(o.Claim.Op.ID, hostclaim.ProbeStop{Box: "dst", Session: o.Claim.Holder.Session, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}))
290 if o = f.worker(nil); o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.Exec != "dst" {
291 t.Fatalf("serving: %+v", o.Claim.Op)
292 }
293}
294
295func TestWorkerFailoverWaitsForThePinWindow(t *testing.T) {
296 f := newFixture(t)
297 ctx := context.Background()
298 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
299 t.Fatalf("failover with no fence: %v", err)
300 }
301 f.terminate("src")
302 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
303 t.Fatalf("drain of a fenced source: %v", err)
304 }
305 o, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"})
306 if err != nil {
307 t.Fatal(err)
308 }
309 if o.Claim.Op.Exec != "none" || !slices.Contains(o.Claim.Lost, "src") {
310 t.Fatalf("failover start: %+v", o.Claim.Op)
311 }
312 f.ack("dst")
313 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning {
314 t.Fatalf("phase %s", o.Claim.Op.Phase)
315 }
316 // The fence was observed a minute ago, so the 30-second window is open.
317 if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
318 t.Fatal("not pinned")
319 }
320}
321
322func TestWorkerPinsNoEarlierThanThirtySecondsAfterTheFence(t *testing.T) {
323 f := newFixture(t)
324 ctx := context.Background()
325 f.terminate("src")
326 f.now = f.now.Add(-time.Minute + 10*time.Second) // ten seconds after the observation
327 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
328 t.Fatal(err)
329 }
330 f.ack("dst")
331 f.worker(nil) // pinning
332 f.worker(ErrNotYet)
333 f.now = f.now.Add(20 * time.Second)
334 if o := f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
335 t.Fatal("not pinned at observation plus thirty seconds")
336 }
337}
338
339func TestWorkerConvertsADrainWhoseSourceIsTerminated(t *testing.T) {
340 f := newFixture(t)
341 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
342 t.Fatal(err)
343 }
344 f.ack("dst")
345 f.terminate("src")
346 o := f.worker(nil)
347 if o.Claim.Op.Phase != hostclaim.PhaseStarting || o.Claim.Op.Exec != "none" || slices.Contains(o.Claim.Op.Participants, "src") || !slices.Contains(o.Claim.Lost, "src") {
348 t.Fatalf("conversion: %+v lost %v", o.Claim.Op, o.Claim.Lost)
349 }
350 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning {
351 t.Fatalf("the barrier now needs only the target: %s", o.Claim.Op.Phase)
352 }
353}
354
355func TestWorkerReplacesATerminatedTargetTwiceThenStops(t *testing.T) {
356 f := newFixture(t)
357 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
358 t.Fatal(err)
359 }
360 for _, tc := range []struct{ dead, next string }{{"dst", "spa"}, {"spa", "sp2"}} {
361 f.terminate(tc.dead)
362 o := f.worker(nil)
363 if o.Claim.Op.Intent["target_box"] != tc.next || !slices.Contains(o.Claim.Lost, tc.dead) {
364 t.Fatalf("replacement of %s: %+v", tc.dead, o.Claim.Op.Intent)
365 }
366 }
367 f.terminate("sp2")
368 o := f.worker(nil)
369 if o.Claim.Op.Journal["stuck"] == "" {
370 t.Fatalf("a third replacement must stop the operation: %+v", o.Claim.Op.Journal)
371 }
372 before := o.Claim.Gen
373 if o = f.worker(nil); o.Claim.Gen != before {
374 t.Fatal("a worker moved a stuck operation")
375 }
376}
377
378func TestWorkerFailsAProbePastItsDeadline(t *testing.T) {
379 f := newFixture(t)
380 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
381 t.Fatal(err)
382 }
383 f.ack("src")
384 f.ack("dst")
385 f.worker(nil)
386 h := f.claim().Claim.Holder
387 f.participant(hostclaim.JournalRelease(f.claim().Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
388 f.worker(nil)
389 f.worker(nil)
390 f.worker(nil)
391 f.installAndProbe("dst")
392 f.now = f.now.Add(ProbeDeadline + time.Second)
393 if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
394 t.Fatalf("phase %s", o.Claim.Op.Phase)
395 }
396 f.worker(ErrNotYet) // the failed target has not released
397}
398
399func TestWorkerRefusesWhenItIsNotAnActiveWorker(t *testing.T) {
400 f := newFixture(t)
401 b := f.roster.Boxes["wrk"]
402 b.Region = "uk"
403 f.roster.Boxes["wrk"] = b
404 f.roster.Revision++
405 f.writeRoster()
406 before := f.claim().ETag
407 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrNotWorker) {
408 t.Fatalf("start: %v", err)
409 }
410 if f.claim().ETag != before {
411 t.Fatal("a refused start wrote the claim")
412 }
413}
414
415func TestWorkerStartRefusesAnUpdateReservedForAParticipant(t *testing.T) {
416 f := newFixture(t)
417 f.roster.Activations["upd-1"] = hostactors.Reservation{ID: "upd-1", FloorSHA256: strings.Repeat("f", 64),
418 Actors: map[string]string{"box:dst": enrollment("dst")}, Releases: map[string]hostactors.Report{"box:dst": {Enrollment: enrollment("dst")}}}
419 f.roster.Revision++
420 f.writeRoster()
421 before := f.claim().ETag
422 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err == nil {
423 t.Fatal("started while an executable update is reserved for the target (H14)")
424 }
425 if f.claim().ETag != before {
426 t.Fatal("a refused start wrote the claim")
427 }
428}
429
430func TestWorkerTreatsAnUnreadableFenceAsAnErrorNeverAsNoFence(t *testing.T) {
431 f := newFixture(t)
432 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
433 t.Fatal(err)
434 }
435 f.ack("src")
436 f.ack("dst")
437 if _, err := f.s.Put(context.Background(), hostfence.FenceKey(enrollment("src")), []byte("{"), ""); err != nil {
438 t.Fatal(err)
439 }
440 before := f.claim().ETag
441 if _, err := f.w.Step(context.Background(), host); err == nil || errors.Is(err, ErrNotYet) {
442 t.Fatalf("an unreadable fence let the drain proceed: %v", err)
443 }
444 if f.claim().ETag != before {
445 t.Fatal("the claim moved")
446 }
447}
448
449func TestWorkerRefusesToJournalAPinAfterTheLeaseIsLost(t *testing.T) {
450 f := newFixture(t)
451 f.terminate("src")
452 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
453 t.Fatal(err)
454 }
455 f.ack("dst")
456 f.worker(nil) // pinning
457 f.loseLease = true
458 o := f.worker(ErrNotYet)
459 if o.Claim.Op.Journal["pin"] != "" {
460 t.Fatal("a pin was journaled after its lease was lost: a park could have retired its versions")
461 }
462}
463
464func TestWorkerStartRefusesAnIneligibleTargetAndANonRelocatableHost(t *testing.T) {
465 f := newFixture(t)
466 before := f.claim().ETag
467 f.ineligible["dst"] = true
468 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err == nil {
469 t.Fatal("started on a target the eligible set does not hold")
470 }
471 f.ineligible["dst"] = false
472 shape := errors.New("unit-backed route")
473 f.w.Relocatable = func(context.Context, hostclaim.Observed) error { return shape }
474 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, shape) {
475 t.Fatalf("started a non-relocatable host: %v", err)
476 }
477 if f.claim().ETag != before {
478 t.Fatal("a refused start wrote the claim")
479 }
480}
481
482func TestWorkerRefusesToDrainAHostThatNeverHadAWriter(t *testing.T) {
483 f := newFixture(t)
484 // A second host created on src and never woken: session 0.
485 ctx := context.Background()
486 const quiet = "quiet.bab.oncarlos.com"
487 nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + quiet + ".json", Account: "bab", App: "jam", Host: quiet, Region: "ie", RecordRegion: "ie", CreatedAt: time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)}
488 body := []byte(`{"host":"` + quiet + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
489 o, err := hostclaim.Read(ctx, f.s, quiet)
490 if err != nil {
491 t.Fatal(err)
492 }
493 o = f.step(o, hostclaim.BeginCreateOn(nh, body, "c2", "console", f.now, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment("src")}))
494 f.step(o, hostclaim.CompleteCreate("c2", "src"))
495 if _, err := f.w.Start(ctx, quiet, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
496 t.Fatalf("drained a never-woken host: %v", err)
497 }
498 f.terminate("src")
499 if _, err := f.w.Start(ctx, quiet, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
500 t.Fatalf("a failover of a never-woken host: %v", err)
501 }
502}
503
504func TestWorkerMarksALiveTargetsThirdReplacementStuck(t *testing.T) {
505 f := newFixture(t)
506 ctx := context.Background()
507 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
508 t.Fatal(err)
509 }
510 // Two terminated targets use the replacements; the third target is
511 // alive but fails its probe, and its release names a fourth box: stuck.
512 for _, dead := range []string{"dst", "spa"} {
513 f.terminate(dead)
514 f.worker(nil)
515 }
516 f.ack("src")
517 f.ack("sp2")
518 f.worker(nil) // releasing
519 h := f.claim().Claim.Holder
520 id := f.claim().Claim.Op.ID
521 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
522 f.worker(nil) // pinning
523 f.worker(nil) // pin
524 f.worker(nil) // installing on sp2
525 f.installAndProbe("sp2")
526 f.now = f.now.Add(ProbeDeadline + time.Second)
527 if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
528 t.Fatalf("phase %s", o.Claim.Op.Phase)
529 }
530 h = f.claim().Claim.Holder
531 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "sp2", Enrollment: enrollment("sp2"), Generation: h.Generation, Session: h.Session, FinalTXID: 41, ManifestSeq: 7, Receipt: "r2", ReceiptSHA256: strings.Repeat("d", 64)}))
532 if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
533 t.Fatalf("a live target's third replacement must stop the operation: %+v", o.Claim.Op)
534 }
535}
536
537func TestWorkerMarksExhaustionStuckBeforeLookingForASpare(t *testing.T) {
538 // Two replacements used; no box is eligible for a third. The
539 // operation is stuck at once, not retried forever on "no eligible box".
540 f := newFixture(t)
541 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
542 t.Fatal(err)
543 }
544 for _, dead := range []string{"dst", "spa"} {
545 f.terminate(dead)
546 f.worker(nil)
547 }
548 f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
549 return hostclaim.Target{}, errors.New("no eligible box")
550 }
551 f.terminate("sp2")
552 if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" {
553 t.Fatalf("exhaustion with no spare was not stuck: %+v", o.Claim.Op.Journal)
554 }
555}
556
557// Final review: a malformed replacements count is an error, never 0. The
558// worker refuses before it looks for a spare, and the claim is unchanged.
559func TestWorkerRefusesAMalformedReplacementsCount(t *testing.T) {
560 f := newFixture(t)
561 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
562 t.Fatal(err)
563 }
564 f.participant(func(c *hostclaim.Claim) error {
565 if c.Op.Journal == nil {
566 c.Op.Journal = map[string]string{}
567 }
568 c.Op.Journal["replacements"] = "two"
569 return nil
570 })
571 chose := 0
572 f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
573 chose++
574 return hostclaim.Target{}, errors.New("no eligible box")
575 }
576 f.now = f.now.Add(ReleaseDeadline + time.Minute)
577 f.terminate("dst")
578 before := f.claim()
579 if _, err := f.w.Step(context.Background(), host); !errors.Is(err, hostclaim.ErrRelocation) {
580 t.Fatalf("step over a malformed count: %v", err)
581 }
582 if after := f.claim(); after.ETag != before.ETag || chose != 0 {
583 t.Fatalf("the claim moved (%v) or a spare was looked for (%d)", after.ETag != before.ETag, chose)
584 }
585}
586
587func TestWorkerFencesAnExhaustedTargetThatDiedAfterItsLocalInstall(t *testing.T) {
588 f := newFixture(t)
589 ctx := context.Background()
590 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
591 t.Fatal(err)
592 }
593 for _, dead := range []string{"dst", "spa"} {
594 f.terminate(dead)
595 f.worker(nil)
596 }
597 f.ack("src")
598 f.ack("sp2")
599 f.worker(nil) // releasing
600 h := f.claim().Claim.Holder
601 id := f.claim().Claim.Op.ID
602 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
603 f.worker(nil) // pinning
604 f.worker(nil) // pin
605 f.worker(nil) // installing on sp2
606 c := f.claim().Claim
607 f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = c.Op.Journal["prepared"]; return nil })
608 f.worker(nil) // installed
609 var n uint64
610 _ = json.Unmarshal([]byte(f.claim().Claim.Op.Journal["install_stamp"]), &n)
611 f.participant(hostclaim.RecordLocalInstall(c.Op.ID, hostclaim.LocalInstall{Box: "sp2", Enrollment: enrollment("sp2"), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("d", 64)}))
612 f.terminate("sp2") // dead between installed and its probe
613 o := f.worker(nil)
614 if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Exec != "none" || !slices.Contains(o.Claim.Lost, "sp2") {
615 t.Fatalf("an exhausted dead target was not fenced into a stuck pinning: %+v", o.Claim.Op)
616 }
617 before := o.Claim.Gen
618 if o = f.worker(nil); o.Claim.Gen != before {
619 t.Fatal("a worker moved a stuck operation")
620 }
621}
622
623func TestWorkerMarksAnOverdueReleaseStuck(t *testing.T) {
624 f := newFixture(t)
625 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
626 t.Fatal(err)
627 }
628 f.ack("src")
629 f.ack("dst")
630 f.worker(nil) // releasing
631 f.now = f.now.Add(ReleaseDeadline)
632 f.worker(ErrNotYet) // at the deadline: still waiting
633 f.now = f.now.Add(time.Second)
634 o := f.worker(nil)
635 if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseReleasing || o.Claim.Op.Exec != "src" {
636 t.Fatalf("a source that never released must leave the operation stuck on the source: %+v", o.Claim.Op)
637 }
638}
639
640// Final review Minor 3 (Task 7's ruling): a source removed from the roster
641// makes every pass refuse at its fence read. The release deadline is
642// checked before that refusal stands, so the operation goes stuck at the
643// deadline, released or not, instead of refusing for ever.
644func TestWorkerMarksARosterRemovedSourceStuckAtTheReleaseDeadline(t *testing.T) {
645 for _, journaled := range []bool{false, true} {
646 t.Run(fmt.Sprintf("release journaled %v", journaled), func(t *testing.T) {
647 f := newFixture(t)
648 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
649 t.Fatal(err)
650 }
651 f.ack("src")
652 f.ack("dst")
653 f.worker(nil) // releasing
654 if journaled {
655 h, id := f.claim().Claim.Holder, f.claim().Claim.Op.ID
656 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
657 }
658 delete(f.roster.Boxes, "src")
659 f.roster.Revision++
660 f.writeRoster()
661 f.now = f.now.Add(ReleaseDeadline)
662 if o := f.worker(ErrRefused); o.Claim.Op.Journal["stuck"] != "" {
663 t.Fatal("stuck before the deadline")
664 }
665 f.now = f.now.Add(time.Second)
666 o := f.worker(nil)
667 if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseReleasing || o.Claim.Op.Exec != "src" {
668 t.Fatalf("a roster-removed source must leave the operation stuck at the release deadline: %+v", o.Claim.Op)
669 }
670 })
671 }
672 // Outside releasing and probe-failed there is no release deadline: the
673 // refusal stands.
674 f := newFixture(t)
675 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
676 t.Fatal(err)
677 }
678 delete(f.roster.Boxes, "src")
679 f.roster.Revision++
680 f.writeRoster()
681 f.now = f.now.Add(ReleaseDeadline + time.Hour)
682 if o := f.worker(ErrRefused); o.Claim.Op.Journal["stuck"] != "" || o.Claim.Op.Phase != hostclaim.PhaseStarting {
683 t.Fatalf("starting: %+v", o.Claim.Op)
684 }
685}
686
687// Round 5 finding 1: a spare that owes an older local retirement for the
688// host is never offered to the chooser; the worker replaces the dead
689// target with a clean spare (the claim refuses the other in its CAS).
690func TestWorkerNeverReplacesATargetWithABoxOwingALocalRetirement(t *testing.T) {
691 f := newFixture(t)
692 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
693 t.Fatal(err)
694 }
695 f.participant(func(c *hostclaim.Claim) error {
696 c.Obligations = append(c.Obligations, hostclaim.Obligation{ID: "old", Op: "old-op", Kind: "park-local", Box: "spa", Retirement: "r", Owner: "box:spa"})
697 return nil
698 })
699 f.terminate("dst")
700 o := f.worker(nil)
701 if o.Claim.Op.Intent["target_box"] != "sp2" || slices.Contains(o.Claim.Op.Participants, "spa") {
702 t.Fatalf("replacement: %+v participants %v", o.Claim.Op.Intent, o.Claim.Op.Participants)
703 }
704}
705
706// Two-box model finding F2: a pin that cannot show the platform files
707// never existed (StorePinner's ErrFilesUnevidenced) stops the operation
708// for an operator instead of retrying or pinning empty.
709func TestWorkerMarksAnUnevidencedFilesPinStuck(t *testing.T) {
710 f := newFixture(t)
711 if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
712 t.Fatal(err)
713 }
714 f.ack("src")
715 f.ack("dst")
716 f.worker(nil) // releasing
717 h := f.claim().Claim.Holder
718 f.participant(hostclaim.JournalRelease(f.claim().Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
719 f.worker(nil) // pinning
720 f.pinErr = fmt.Errorf("%w: test", ErrFilesUnevidenced)
721 if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Journal["pin"] != "" {
722 t.Fatalf("an unevidenced files pin: %+v", o.Claim.Op)
723 }
724}
725
726// Two-box model finding F1: a drain's target fails its probe and
727// releases; with no spare (two boxes: the source is excluded) the worker
728// waits, then, ReleaseDeadline after the target failed, marks the
729// operation stuck for an operator.
730func TestWorkerBoundsAProbeFailedWithNoSpare(t *testing.T) {
731 for _, terminated := range []bool{false, true} {
732 t.Run(fmt.Sprintf("target terminated after its release %v", terminated), func(t *testing.T) { boundsAProbeFailed(t, terminated) })
733 }
734}
735
736func boundsAProbeFailed(t *testing.T, terminated bool) {
737 f := newFixture(t)
738 ctx := context.Background()
739 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
740 t.Fatal(err)
741 }
742 f.ack("src")
743 f.ack("dst")
744 f.worker(nil) // releasing
745 h := f.claim().Claim.Holder
746 id := f.claim().Claim.Op.ID
747 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
748 f.worker(nil) // pinning
749 f.worker(nil) // pin
750 f.worker(nil) // installing on dst
751 f.installAndProbe("dst")
752 f.now = f.now.Add(ProbeDeadline + time.Second)
753 if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
754 t.Fatalf("phase %s", o.Claim.Op.Phase)
755 }
756 h = f.claim().Claim.Holder
757 f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "dst", Enrollment: enrollment("dst"), Generation: h.Generation, Session: h.Session, FinalTXID: 41, ManifestSeq: 7, Receipt: "r2", ReceiptSHA256: strings.Repeat("d", 64)}))
758 noSpare := errors.New("no eligible box")
759 f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
760 return hostclaim.Target{}, noSpare
761 }
762 if _, err := f.w.Step(ctx, host); !errors.Is(err, noSpare) || f.claim().Claim.Op.Journal["stuck"] != "" {
763 t.Fatalf("before the deadline: %v %+v", err, f.claim().Claim.Op.Journal)
764 }
765 if terminated {
766 // Round 17 finding 2: the released target is terminated. Its
767 // fence goes to the replacement path first; with no spare that
768 // path is bounded too, and consumes the fence as it stops.
769 f.terminate("dst")
770 if _, err := f.w.Step(ctx, host); !errors.Is(err, noSpare) || f.claim().Claim.Op.Journal["stuck"] != "" {
771 t.Fatalf("a dead target, before the deadline: %v %+v", err, f.claim().Claim.Op.Journal)
772 }
773 }
774 f.now = f.now.Add(ReleaseDeadline + time.Second)
775 o := f.worker(nil)
776 if o.Claim.Op.Journal["stuck"] == "" {
777 t.Fatalf("no spare after the deadline: %+v", o.Claim.Op)
778 }
779 if terminated && (!slices.Contains(o.Claim.Lost, "dst") || o.Claim.Op.Journal["fence.dst"] == "" || o.Claim.Op.Exec != "none") {
780 t.Fatalf("the dead target's fence was not consumed: %+v lost %v", o.Claim.Op, o.Claim.Lost)
781 }
782 if !terminated && o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
783 t.Fatalf("a live failed target: %+v", o.Claim.Op)
784 }
785}
786
787// preFlip is the moves door's durable pre-flip decision as
788// ConvertMovingToRelocating reads it: the record is still in ie.
789type preFlip struct{}
790
791func (preFlip) CheckMovingBeforeFlip(context.Context, hostclaim.Observed, string, hostclaim.FenceRef) (string, error) {
792 return "ie", nil
793}
794
795// convertedMove begins an admitted move of host from src to "far", a box
796// of another region that is not in this fixture's roster, terminates src
797// (observed ten seconds before f.now), and converts the move to
798// relocating as the moves door would. farAcks says whether the move's
799// barrier had heard from far before the source died.
800func convertedMove(f *fixture, farAcks bool) hostclaim.Observed {
801 f.t.Helper()
802 ctx := context.Background()
803 far := strings.Repeat("9", 32) // the move target's enrollment
804 f.step(f.claim(), hostclaim.BeginAdmittedMove("mv1", "member", "src", "far", "uk", f.now,
805 hostclaim.MoveAdmission{AppID: "app", AdmissionID: "mv1", SourceEnrollment: enrollment("src"), TargetEnrollment: far}))
806 f.ack("src")
807 if farAcks {
808 f.participant(hostclaim.AcknowledgeExecution("far", hostclaim.ExecutionAcknowledgement{Op: "mv1", Enrollment: far, BootID: "boot", ControllerID: "ctl", LeaseETag: "lease", Expires: f.now.Add(time.Hour)}))
809 }
810 f.terminate("src")
811 f.now = f.now.Add(-time.Minute + 10*time.Second) // ten seconds after the observation
812 ro, err := f.w.Directory.Read(ctx)
813 if err != nil {
814 f.t.Fatal(err)
815 }
816 ref, _, err := f.w.fence(ctx, ro, "src")
817 if err != nil || ref == nil {
818 f.t.Fatalf("fence: %v %v", ref, err)
819 }
820 o, err := hostclaim.ConvertMovingToRelocating(ctx, f.s, f.claim(), "mv1", *ref, f.w.Fences, preFlip{})
821 if err != nil {
822 f.t.Fatal(err)
823 }
824 if o.Claim.Op.Journal["fence.src"] == "" || o.Claim.Op.Journal["fence.src.observed_at"] != "" || slices.Contains(o.Claim.Lost, "src") || !slices.Contains(o.Claim.Op.Participants, "far") {
825 f.t.Fatalf("not the converted shape this test is about: %+v lost %v", o.Claim.Op, o.Claim.Lost)
826 }
827 return o
828}
829
830// Task 1's fix: a move converted to relocating binds fence.<source> with
831// no observation time, and JournalPin refuses until FenceSource has
832// consumed it. The worker reads the converted source as not yet lost, so
833// its first CAS is FenceSource; the pin then waits FailoverPinDelay past
834// the fence's observed_at, under the location lease. Task 5 fix round 1:
835// the move's own target leaves the barrier when the same-region target is
836// named, so one that never acknowledged (or died) holds nothing up.
837func TestWorkerFencesAConvertedMovesSourceBeforeItPins(t *testing.T) {
838 for _, farAcks := range []bool{true, false} {
839 t.Run(fmt.Sprintf("the move target acknowledged %v", farAcks), func(t *testing.T) {
840 f := newFixture(t)
841 convertedMove(f, farAcks)
842 o := f.worker(nil)
843 if o.Claim.Op.Journal["fence.src.observed_at"] == "" || !slices.Contains(o.Claim.Lost, "src") || o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Journal["pin"] != "" {
844 t.Fatalf("the worker's first CAS on a converted move is FenceSource: %+v lost %v", o.Claim.Op, o.Claim.Lost)
845 }
846 if o = f.worker(nil); o.Claim.Op.Intent["target_box"] != "dst" || !slices.Equal(o.Claim.Op.Participants, []string{"dst"}) {
847 t.Fatalf("the converted move names its same-region target alone: %+v", o.Claim.Op)
848 }
849 f.worker(ErrNotYet) // inside the 30-second hold
850 if len(f.leases) != 0 {
851 t.Fatalf("a lease was taken inside the hold: %v", f.leases)
852 }
853 f.now = f.now.Add(20 * time.Second)
854 if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
855 t.Fatal("not pinned at observation plus thirty seconds")
856 }
857 if want := []string{"acquire", "pin", "release"}; !slices.Equal(f.leases, want) {
858 t.Fatalf("the pin was not selected and journaled under the location lease: %v", f.leases)
859 }
860 f.worker(ErrNotYet) // the named target has not acknowledged
861 f.ack("dst")
862 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalling || o.Claim.Op.Exec != "dst" {
863 t.Fatalf("installing: %+v", o.Claim.Op)
864 }
865 })
866 }
867}
868
869// Task 5 fix round 1, absence is never evidence: a converted move's
870// fence.<source> is bound in the journal, so a fence object that is no
871// longer there is an error, never "not fenced". The worker refuses the
872// pass: it names no target and never calls the pinner.
873func TestWorkerRefusesAConvertedMoveWhoseBoundFenceIsMissing(t *testing.T) {
874 f := newFixture(t)
875 convertedMove(f, false)
876 if err := f.s.Delete(context.Background(), hostfence.FenceKey(enrollment("src"))); err != nil {
877 t.Fatal(err)
878 }
879 before := f.claim().ETag
880 for range 3 {
881 f.now = f.now.Add(time.Minute)
882 if _, err := f.w.Step(context.Background(), host); err == nil || errors.Is(err, ErrNotYet) {
883 t.Fatalf("a missing bound fence read as no fence: %v", err)
884 }
885 }
886 if o := f.claim(); o.ETag != before || o.Claim.Op.Intent["target_box"] != "" || len(f.leases) != 0 {
887 t.Fatalf("the worker acted on a missing fence: %+v leases %v", o.Claim.Op, f.leases)
888 }
889}
890
891// committedGuard is moves' record for the landing: committed, from uk
892// (the move's source region) to ie, this worker's region.
893type committedGuard struct{ op string }
894
895func (g committedGuard) CheckLandingCommitted(_ context.Context, o hostclaim.Observed, id string) (hostclaim.CommittedLanding, error) {
896 if id != g.op || o.Claim.Op.ID != g.op {
897 return hostclaim.CommittedLanding{}, hostclaim.ErrWrongState
898 }
899 h := o.Claim.Holder
900 at := time.Date(2026, 10, 7, 8, 0, 0, 0, time.UTC)
901 return hostclaim.CommittedLanding{FromRegion: "uk", ToRegion: "ie",
902 SourceScope: hostclaim.ParkScope{Location: h.Location, Incarnation: h.Incarnation, EvidenceAt: at, RequestLimit: 120 * time.Second},
903 StagingScope: hostclaim.ParkScope{Location: "moves/" + host + "/" + g.op, Incarnation: h.Incarnation, EvidenceAt: at, RequestLimit: 120 * time.Second}}, nil
904}
905
906// convertedLanding drives the host through an admitted move from src (in
907// uk, for the move) to dst through the claim's own transitions to a probed
908// landing, terminates dst, and converts it with ConvertCommittedLanding:
909// the claim the worker then finds is the conversion's real output.
910func (f *fixture) convertedLanding() hostclaim.Observed {
911 f.t.Helper()
912 ctx := context.Background()
913 const id = "mv_abcdefghijklmnopqrstuvwxyz"
914 o := f.claim()
915 o = f.step(o, hostclaim.BeginAdmittedMove(id, "ops", "src", "dst", "ie", f.now, hostclaim.MoveAdmission{AppID: "app-canonical", AdmissionID: id, SourceEnrollment: enrollment("src"), TargetEnrollment: enrollment("dst")}))
916 o = f.step(o, hostclaim.ExecToNone(id, "src", "moves/"+host+"/"+id+"/stopped-source.json", hostclaim.JournalEvidenceSource))
917 o = f.step(o, hostclaim.SetJournal(id, map[string]string{hostclaim.JournalSeal: `{"n":7}`, hostclaim.JournalCutSession: "1"}))
918 o = f.step(o, hostclaim.SetMovePhase(id, hostclaim.KindMoving, hostclaim.PhaseMoving, hostclaim.PhaseSealed))
919 o = f.step(o, hostclaim.Flip(id, "dst", o.Claim.Holder.Stamp, o.Claim.Holder.Generation))
920 for _, p := range [][2]string{{"checking", "parking"}, {"parking", "building"}, {"building", "published"}, {"published", "local"}, {"local", "configured"}, {"configured", "restored"}, {"restored", "probing"}, {"probing", "probed"}} {
921 o = f.step(o, hostclaim.SetMovePhase(id, hostclaim.KindLanding, p[0], p[1]))
922 }
923 // The probe's cold admission on dst (claim rev 6.9): its generation is
924 // holder.generation and dst the writer.
925 o = f.step(o, func(c *hostclaim.Claim) error {
926 h := c.Holder
927 h.Session++
928 h.Admissions = append(h.Admissions, hostclaim.AdmissionRecord{Region: "ie", PredecessorRegion: "uk", Session: h.Session, Box: "dst", Generation: "gen-landed", Predecessor: h.Generation, Mode: "cold", Coverage: "sealed"})
929 h.Generation, h.WriterBox = "gen-landed", "dst"
930 return nil
931 })
932 f.terminate("dst")
933 enr := enrollment("dst")
934 raw, _, err := f.s.Get(ctx, hostfence.FenceKey(enr))
935 if err != nil {
936 f.t.Fatal(err)
937 }
938 sum := sha256.Sum256(raw)
939 ref := hostclaim.FenceRef{Enrollment: enr, Key: hostfence.FenceKey(enr), SHA256: hex.EncodeToString(sum[:])}
940 n := 0
941 out, err := hostclaim.ConvertCommittedLanding(ctx, f.s, o, id, "dst", ref, f.w.Fences, committedGuard{op: id}, nil, func() string { n++; return fmt.Sprintf("mv-chore-%d", n) })
942 if err != nil {
943 f.t.Fatal(err)
944 }
945 return out
946}
947
948// J3: a committed landing converted to a failover (ConvertCommittedLanding)
949// starts with no target. The worker names one in starting; it acknowledges
950// before PinFailover, and the pin is selected after that, as an initial
951// failover's is; installing then copies the pin's release (R9). The move's
952// cleanup obligations ride through untouched.
953func TestWorkerNamesAConvertedLandingsTargetBeforeItsPin(t *testing.T) {
954 f := newFixture(t)
955 converted := f.convertedLanding()
956 c := converted.Claim
957 if c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseStarting || c.Op.Exec != "none" || c.Op.Intent["source_box"] != "dst" ||
958 c.Op.Intent["move_source_box"] != "src" || c.Op.Journal["conversion"] != "committed-landing" || len(c.Op.Participants) != 0 || len(c.Obligations) == 0 {
959 t.Fatalf("not the conversion's output: %+v obligations %d", c.Op, len(c.Obligations))
960 }
961 chores, _ := json.Marshal(c.Obligations)
962 o := f.worker(nil)
963 if o.Claim.Op.Phase != hostclaim.PhaseStarting || o.Claim.Op.Intent["target_box"] != "spa" || !slices.Equal(o.Claim.Op.Participants, []string{"spa"}) || o.Claim.Op.Journal["pin"] != "" {
964 t.Fatalf("named in starting: %+v", o.Claim.Op)
965 }
966 f.worker(ErrNotYet) // the target has not acknowledged: no pin yet
967 f.ack("spa")
968 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning {
969 t.Fatalf("phase %s", o.Claim.Op.Phase)
970 }
971 if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
972 t.Fatal("not pinned")
973 }
974 if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalling || o.Claim.Op.Exec != "spa" || o.Claim.Op.Journal["install_release"] != "v1.4.2" {
975 t.Fatalf("installing: %+v", o.Claim.Op)
976 }
977 if after, _ := json.Marshal(o.Claim.Obligations); string(after) != string(chores) {
978 t.Fatalf("the failover changed an obligation:\n%s\n%s", after, chores)
979 }
980}
981
982// Main's home-region parks: a worker in the home holds the box's verified
983// home binding. It starts a relocation under it (the op's "home_region",
984// which lets the claim see the home's chores under park.HomeRegion too),
985// and only when the signed roster designates that home and the worker's
986// region is it.
987func TestWorkerStartsUnderItsHomeBinding(t *testing.T) {
988 ctx := context.Background()
989 t.Run("designated", func(t *testing.T) {
990 f := newFixture(t)
991 f.roster.HomeRegion = "ie"
992 f.writeRoster()
993 f.w.HomeRegion = "ie"
994 o, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"})
995 if err != nil {
996 t.Fatal(err)
997 }
998 if o.Claim.Op.Intent["home_region"] != "ie" || o.Claim.Op.Intent["region"] != "ie" {
999 t.Fatalf("intent %v", o.Claim.Op.Intent)
1000 }
1001 })
1002 t.Run("not designated", func(t *testing.T) {
1003 f := newFixture(t)
1004 f.w.HomeRegion = "ie"
1005 before := f.claim()
1006 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrNotWorker) {
1007 t.Fatalf("got %v", err)
1008 }
1009 if after := f.claim(); after.ETag != before.ETag {
1010 t.Fatal("a refused start wrote the claim")
1011 }
1012 })
1013 t.Run("an op bound to another home is not stepped", func(t *testing.T) {
1014 f := newFixture(t)
1015 o, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"})
1016 if err != nil {
1017 t.Fatal(err)
1018 }
1019 f.ack("src")
1020 f.ack("dst")
1021 o, err = hostclaim.Update(ctx, f.s, f.claim(), func(c *hostclaim.Claim) error {
1022 c.Op.Intent["home_region"] = "uk"
1023 return nil
1024 })
1025 if err != nil {
1026 t.Fatal(err)
1027 }
1028 if _, err := f.w.Step(ctx, host); !errors.Is(err, park.ErrUnsafe) {
1029 t.Fatalf("got %v", err)
1030 }
1031 if after := f.claim(); after.ETag != o.ETag {
1032 t.Fatal("the step wrote the claim")
1033 }
1034 })
1035}
1036