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/ed25519"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "os"
10 "path/filepath"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/activator"
16 "github.com/carlosframework/platform/internal/cgroup"
17 "github.com/carlosframework/platform/internal/genselect"
18 "github.com/carlosframework/platform/internal/hostbox"
19 "github.com/carlosframework/platform/internal/hostclaim"
20 "github.com/carlosframework/platform/internal/hostguard"
21 "github.com/carlosframework/platform/internal/hostwrite"
22 "github.com/carlosframework/platform/internal/lease"
23 "github.com/carlosframework/platform/internal/registry"
24 "github.com/carlosframework/platform/internal/release"
25)
26
27// These tests drive the real wake assembly, serving retries, launch journal,
28// guard and stop coordinator. The kernel below is a fixture that reads the
29// real launch journal; it is not privileged cgroup or /proc verification.
30
31const recoveryHost = "shop.bab.oncarlos.com"
32
33var recoveryPolicy = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second}
34
35type recoveryHeld struct{}
36
37func (recoveryHeld) Err() error { return nil }
38func (recoveryHeld) Lost() <-chan struct{} { return nil }
39func (recoveryHeld) Release(context.Context) error { return nil }
40func (recoveryHeld) BindGeneration(string, func(context.Context) error) error {
41 return hostwrite.ErrUnavailable
42}
43func (recoveryHeld) Replicate(context.Context, string, time.Duration) (*lease.Replicator, error) {
44 return nil, hostwrite.ErrUnavailable
45}
46
47type recoveryInputs struct{ f *recoveryFixture }
48
49func (r recoveryInputs) Select(_ context.Context, _ registry.Route, o hostclaim.Observed, _ activator.StartPolicy) (ClaimLaunchInputs, error) {
50 h := o.Claim.Holder
51 r.f.seq++
52 return ClaimLaunchInputs{Roles: []string{"main"}, Identity: hostwrite.Identity{Config: "config", Release: "release"}, Admission: hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-1", Mode: "cold", NewGeneration: fmt.Sprintf("gen-%d", r.f.seq), Region: "ie", PredecessorRegion: "ie"}, Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), DBPath: filepath.Join(r.f.guard.DataDir, "state", "data.db"), Empty: true}, nil
53}
54func (r recoveryInputs) Materialize(ctx context.Context, o hostclaim.Observed, _ ClaimLaunchInputs, _ hostwrite.Work) error {
55 if r.f.materialize == nil {
56 return nil
57 }
58 return r.f.materialize(ctx, o)
59}
60func (r recoveryInputs) EmptyAuthority(context.Context, hostclaim.Observed, ClaimLaunchInputs) error {
61 return r.f.restoreFailure
62}
63func (recoveryInputs) Publish(context.Context, hostclaim.Observed, ClaimLaunchInputs) error {
64 return nil
65}
66
67type recoveryKernel struct {
68 noProcess bool
69 noProcessCalls int
70 journal *hostbox.LaunchJournal
71 alive bool
72 verified []hostguard.FenceTarget
73 recovered []hostguard.FenceTarget
74}
75
76func (k *recoveryKernel) evidence(inv StopInventory, j hostbox.StopKernelJournal) StopKernelEvidence {
77 groups := []string{}
78 for _, g := range j.Cgroups {
79 groups = append(groups, fmt.Sprintf("%s:%d:%d", g.RelativePath, g.Device, g.Inode))
80 }
81 return StopKernelEvidence{ObservedBootID: j.Target.BootID, InventorySHA256: digestStop(inv), LocalJournalSHA256: digestStop(j), CgroupIdentities: groups, UnitInvocationIDs: []string{}}
82}
83func (k *recoveryKernel) VerifyStopped(ctx context.Context, t hostguard.FenceTarget, inv StopInventory, ack DrainAcknowledgement) (StopKernelEvidence, error) {
84 k.verified = append(k.verified, t)
85 j, err := k.journal.ReadStopKernelJournal(ctx, t)
86 if err != nil {
87 return StopKernelEvidence{}, err
88 }
89 if !j.Settled {
90 return StopKernelEvidence{}, ErrClaimStopInventory
91 }
92 if err := validateClaimDrainProducers(j.ProducerIDs, ack); err != nil {
93 return StopKernelEvidence{}, err
94 }
95 return k.evidence(inv, j), nil
96}
97func (k *recoveryKernel) RecoverStopped(ctx context.Context, t hostguard.FenceTarget, inv StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
98 k.recovered = append(k.recovered, t)
99 j, err := k.journal.ReadStopKernelJournal(ctx, t)
100 if err != nil {
101 return StopKernelEvidence{}, DrainAcknowledgement{}, err
102 }
103 if k.alive {
104 return StopKernelEvidence{}, DrainAcknowledgement{}, ErrClaimStopControllerAlive
105 }
106 if err := k.journal.SettleRecovered(ctx, t); err != nil {
107 return StopKernelEvidence{}, DrainAcknowledgement{}, err
108 }
109 j.Settled = true
110 return k.evidence(inv, j), DrainClaimAcknowledgement(t.FenceID, t.ControllerID, j.ProcessStart, j.ProducerIDs), nil
111}
112
113type recoveryFixture struct {
114 host string
115 journalPath string
116 data genselect.Reader
117 bucket string
118 t *testing.T
119 claims release.Store
120 evidence release.Store
121 signer ed25519.PrivateKey
122 guard hostguard.Guard
123 journal *hostbox.LaunchJournal
124 kernel *recoveryKernel
125 now time.Time
126 seq int
127 peer StopPeer
128 requests []StopRequest
129 materialize func(context.Context, hostclaim.Observed) error
130 restoreFailure error
131 recordPrelaunch bool
132 revision uint64
133 // box, enrollment, region and leaseRegion are the controller's box
134 // identity: box-1, enrollment-1, ie and eu-west-1 when unset. A moves
135 // fixture sets its target box's (home part 2: home-box in its roster
136 // region ie).
137 box, enrollment, region, leaseRegion string
138 // homeRegion is the writer's verified home alias (hostbox.Writer's
139 // HomeRegion), "" for none. admissionEvidence, when set, is the
140 // writer's AdmissionEvidence: the claim's owner dispatch in production
141 // (claimassembly.go), bound before the runtime is built because Build
142 // captures the writer. Unset, every admission's evidence passes.
143 homeRegion string
144 admissionEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
145 // fleetHomeRegion is the fleet's signed home designation the box's
146 // actor carries; with homeRegion it is the controller's own stop
147 // identity's home binding (park.VerifiedHome). "" for none.
148 fleetHomeRegion string
149}
150
151// identity is the controller's box identity, the defaults filled in.
152func (f *recoveryFixture) identity() (box, enrollment, region, leaseRegion string) {
153 box, enrollment, region, leaseRegion = "box-1", "enrollment-1", "ie", "eu-west-1"
154 if f.box != "" {
155 box = f.box
156 }
157 if f.enrollment != "" {
158 enrollment = f.enrollment
159 }
160 if f.region != "" {
161 region = f.region
162 }
163 if f.leaseRegion != "" {
164 leaseRegion = f.leaseRegion
165 }
166 return box, enrollment, region, leaseRegion
167}
168
169type recoveryController struct {
170 local claimStopLocal
171 wake *ClaimWakeAssembly
172 runtime *activator.ClaimRuntime
173 assembly ClaimAssembly
174}
175
176func newRecoveryFixture(t *testing.T, hosts ...string) *recoveryFixture {
177 host := recoveryHost
178 if len(hosts) > 0 {
179 host = hosts[0]
180 }
181 t.Helper()
182 ctx := t.Context()
183 st := release.NewDirStore(t.TempDir())
184 now := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC)
185 in := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: now}
186 body := []byte(strings.ReplaceAll(`{"schema":1,"host":"shop.bab.oncarlos.com","account":"bab","app":"jam","region":"ie","created_at":"2026-10-06T12:00:00Z"}`, recoveryHost, host))
187 absent, err := hostclaim.Read(ctx, st, host)
188 if err != nil {
189 t.Fatal(err)
190 }
191 o, err := hostclaim.Update(ctx, st, absent, hostclaim.BeginCreate(in, body, "create", "console", now))
192 if err != nil {
193 t.Fatal(err)
194 }
195 if _, err = hostclaim.Update(ctx, st, o, hostclaim.CompleteCreate("create", "box-1")); err != nil {
196 t.Fatal(err)
197 }
198 data := t.TempDir()
199 if err = os.Mkdir(filepath.Join(data, "hostguard"), 0770); err != nil {
200 t.Fatal(err)
201 }
202 journalPath := filepath.Join(t.TempDir(), "launches.db")
203 journal, err := hostbox.OpenLaunchJournal(journalPath)
204 if err != nil {
205 t.Fatal(err)
206 }
207 t.Cleanup(func() { journal.Close() })
208 return &recoveryFixture{host: host, journalPath: journalPath, t: t, claims: st, evidence: release.NewDirStore(t.TempDir()), signer: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)), guard: hostguard.Guard{DataDir: data, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: st}}, journal: journal, kernel: &recoveryKernel{journal: journal}, now: now, recordPrelaunch: true}
209}
210
211func (f *recoveryFixture) newID() string { f.seq++; return fmt.Sprintf("id-%d", f.seq) }
212
213func (f *recoveryFixture) hostName() string {
214 if f.host != "" {
215 return f.host
216 }
217 return recoveryHost
218}
219
220func (f *recoveryFixture) read() hostclaim.Observed {
221 f.t.Helper()
222 o, err := hostclaim.Read(f.t.Context(), f.claims, f.hostName())
223 if err != nil {
224 f.t.Fatal(err)
225 }
226 return o
227}
228
229func (f *recoveryFixture) retry() hostclaim.WakeRetry {
230 f.t.Helper()
231 o := f.read()
232 if o.Claim.Holder.WakeRetry == nil {
233 f.t.Fatal("retry episode missing")
234 }
235 return *o.Claim.Holder.WakeRetry
236}
237
238func recoveryTuple(o hostclaim.Observed, l claimStopLocal) hostguard.FenceTarget {
239 h := o.Claim.Holder
240 // Fixture actors without a binding stamp their numeral (the shorthand of
241 // handoverRecoveryIdentity); a production actor stamps its binding.
242 stamp := l.Actor.Binding
243 if stamp == "" {
244 stamp = fmt.Sprint(l.Actor.Revision)
245 }
246 return hostguard.FenceTarget{Region: l.Actor.Region, Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: l.Actor.Box, Enrollment: l.Actor.Enrollment, BootID: l.BootID, ControllerID: l.ControllerID, ClaimETag: o.ETag, DirectoryRevision: stamp, RosterRevision: l.Actor.Revision}
247}
248
249// rosterRevision is root's verified roster read in this fixture: the roster
250// the controllers' actors read, at f.revision (7 when unset).
251func (f *recoveryFixture) rosterRevision(context.Context) (uint64, error) {
252 if f.revision == 0 {
253 return 7, nil
254 }
255 return f.revision, nil
256}
257
258// root is instance-control's coordinator for one request. The live peer is
259// whichever controller process the fixture most recently started.
260func (f *recoveryFixture) root(drain StopDrainer) ClaimStopCoordinator {
261 return ClaimStopCoordinator{VerifiedRosterRevision: f.rosterRevision, EvidenceSigner: f.signer, RequestLimit: hostbox.RequestLimit, Guard: f.guard, Claims: hostguard.StoreReader{Store: f.claims}, CarlosUID: 1001, Evidence: f.evidence, Drainer: drain, Kernel: f.kernel, NewID: f.newID, Now: func() time.Time { return f.now },
262 Peers: stopPeerFunc(func(context.Context) (StopPeer, error) { return f.peer, nil }),
263 Inventory: stopInventoryFunc(func(ctx context.Context, host string) (StopInventory, error) {
264 o, err := hostclaim.Read(ctx, f.claims, host)
265 if err != nil {
266 return StopInventory{}, err
267 }
268 h := o.Claim.Holder
269 box, enrollment, _, _ := f.identity()
270 return StopInventory{Complete: true, Host: host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: box, Enrollment: enrollment, DirectoryRevision: "7"}, nil
271 })}
272}
273
274// controller starts one agent process: its own wake assembly, retries and stop
275// path, sharing the durable claim, guard and launch journal of the box.
276// recoveryFixtureBinding is the fixture box's binding (shorthand "7"). A
277// controller started after a roster write (f.revision) keeps it: roster
278// writes move the counter, not the box's binding.
279const recoveryFixtureBinding = "7"
280
281func (f *recoveryFixture) controller(boot, controller string, pid int, start string) *recoveryController {
282 t := f.t
283 revision := f.revision
284 if revision == 0 {
285 revision = 7
286 }
287 box, enrollment, region, leaseRegion := f.identity()
288 local := claimStopLocal{Actor: hostbox.Actor{Box: box, Enrollment: enrollment, Revision: revision, Binding: recoveryFixtureBinding, Region: region, HomeRegion: f.homeRegion, FleetHomeRegion: f.fleetHomeRegion}, BootID: boot, ControllerID: controller, PID: pid, ProcessStart: start}
289 f.peer = StopPeer{UID: 1001, PID: pid, ProcessStart: start, BootID: boot, ControllerID: controller}
290 producers := &ClaimProducerSet{}
291 w := &ClaimWakeAssembly{LeaseRegion: leaseRegion, Inputs: recoveryInputs{f}, Manager: new(cgroup.Manager)}
292 w.AfterAdmission = func(context.Context, hostclaim.Observed, hostclaim.Observed) error { return nil }
293 // The unprivileged state-directory fixture runs under the real selected
294 // work; ColdRestore and its generation-marker writes remain unchanged.
295 w.PrepareState = func(ctx context.Context, _ hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) error {
296 return work.With(ctx, in.Identity, func() error { return os.MkdirAll(filepath.Dir(in.DBPath), 0700) })
297 }
298 w.RetireArtifacts = func(context.Context, hostclaim.Observed, ClaimLaunchInputs, string) error { return hostbox.ErrChanged }
299 w.Retain = func(hostclaim.Observed, activator.Proc, activator.Rep, error) {}
300 w.ObserveLoss = func(hostclaim.Observed, error) {}
301 w.Producers = func(_ context.Context, o hostclaim.Observed) (*ClaimProducers, error) {
302 return producers.Ensure(recoveryTuple(o, local), start, f.journal.RecordProducer)
303 }
304 w.Launch = func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error) {
305 return nil, hostwrite.ErrUnavailable
306 }
307 w.Prelaunch = func(ctx context.Context, o hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) (*activator.RoleLaunch, error) {
308 target := recoveryTuple(o, local)
309 if !f.recordPrelaunch {
310 // rev1 placement: nothing is journaled until the launch itself.
311 return &activator.RoleLaunch{Target: target, Journal: f.journal}, nil
312 }
313 err := work.With(ctx, in.Identity, func() error {
314 return f.journal.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: pid, ProcessStart: start, Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: target.Session}}, NoState: true})
315 })
316 if err != nil {
317 return nil, err
318 }
319 return &activator.RoleLaunch{Target: target, Journal: f.journal}, nil
320 }
321 w.Retries = activator.ServingRetries{Claims: f.claims, Box: box, Enrollment: enrollment, BootID: boot, ControllerID: controller, Now: func() time.Time { return f.now }, NewID: f.newID, Inputs: func(context.Context, registry.Route) (string, error) { return "inputs", nil }}
322 a := ClaimAssembly{Claims: f.claims, Launches: f.journal, Wake: w, Producers: producers, stopLocal: func(context.Context) (claimStopLocal, error) { return local, nil }}
323 a.collectStop = func(ctx context.Context, req StopRequest, drain StopDrainer) (StopResult, error) {
324 f.requests = append(f.requests, req)
325 result, err := f.root(drain).FenceAndStop(ctx, req)
326 if err != nil {
327 return result, err
328 }
329 return validateTransportStop(req, result)
330 }
331 a.bindClaimWakeStop()
332 source := hostbox.WorkSource{Guard: f.guard, Box: box, Region: region, Now: func() time.Time { return f.now }, Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil }, Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
333 evidence := f.admissionEvidence
334 if evidence == nil {
335 evidence = func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error { return nil }
336 }
337 writer := hostbox.Writer{HomeRegion: f.homeRegion, Data: f.data, Bucket: f.bucket, Claims: f.claims, Box: box, Region: region, AdmissionEvidence: evidence}
338 runtime, err := w.Build(t.Context(), source, writer)
339 if err != nil {
340 t.Fatal(err)
341 }
342 return &recoveryController{local: local, wake: w, runtime: runtime, assembly: a}
343}
344
345// admit stands in for Prerequisites' S3 lease, then reserves, admits, journals
346// the pre-launch record and binds the attempt through the real runtime.
347func (c *recoveryController) admit(t *testing.T, f *recoveryFixture) hostclaim.Observed {
348 t.Helper()
349 live, cancel := context.WithCancel(context.Background())
350 t.Cleanup(cancel)
351 done := make(chan struct{})
352 go func() { <-live.Done(); close(done) }()
353 o := f.read()
354 c.wake.mu.Lock()
355 c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
356 c.wake.mu.Unlock()
357 admitted, err := c.runtime.Admit(t.Context(), registry.Route{Host: recoveryHost}, o, recoveryPolicy)
358 if err != nil {
359 t.Fatal(err)
360 }
361 if r := admitted.Claim.Holder.WakeRetry; r == nil || r.Attempt == nil || r.Attempt.Session != admitted.Claim.Holder.Session || r.Attempt.ControllerID != c.local.ControllerID {
362 t.Fatalf("attempt not bound to admitted session: %+v", r)
363 }
364 return admitted
365}
366
367func (f *recoveryFixture) receipts() []StopReceipt {
368 f.t.Helper()
369 keys, err := f.evidence.List(f.t.Context(), "control/hosts/"+recoveryHost+"/serving-evidence/")
370 if err != nil {
371 f.t.Fatal(err)
372 }
373 out := []StopReceipt{}
374 for _, key := range keys {
375 if strings.HasSuffix(key, ".sig") {
376 continue
377 }
378 body, _, err := f.evidence.Get(f.t.Context(), key)
379 if err != nil {
380 f.t.Fatal(err)
381 }
382 var r StopReceipt
383 if err = json.Unmarshal(body, &r); err != nil {
384 f.t.Fatal(err)
385 }
386 out = append(out, r)
387 }
388 return out
389}
390
391func TestClaimWakeRestoreFailureSettlesFromPrelaunchRecord(t *testing.T) {
392 f := newRecoveryFixture(t)
393 ctx := t.Context()
394 c := f.controller("boot-1", "controller-1", 4101, "start-1")
395 admitted := c.admit(t, f)
396 // The record exists before Prepare can hand out a Restore closure.
397 record, err := f.journal.Read(ctx, recoveryTuple(admitted, c.local))
398 if err != nil || record.ControllerPID != 4101 || record.ProcessStart != "start-1" || len(record.Groups) != 1 || len(record.Reservations) != 0 || !record.NoState {
399 t.Fatalf("no recoverable pre-launch record before restore: %+v %v", record, err)
400 }
401 s, err := c.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, recoveryPolicy)
402 if err != nil {
403 t.Fatal(err)
404 }
405 // Fail inside the production restore closure after PrepareState, at
406 // ColdRestore's empty-authority check. Do not replace that closure.
407 failure := errors.New("initial-empty authority unavailable")
408 f.restoreFailure = failure
409 restoreErr := s.Restore(ctx)
410 if !errors.Is(restoreErr, failure) {
411 t.Fatalf("restore failure = %v", restoreErr)
412 }
413 if _, err := os.Stat(filepath.Join(f.guard.DataDir, "state")); err != nil {
414 t.Fatalf("PrepareState did not run: %v", err)
415 }
416 if err = c.runtime.FailedAdmission(ctx, admitted, restoreErr); err != nil {
417 t.Fatalf("restore failure stranded the attempt: %v", err)
418 }
419 retry := f.retry()
420 if retry.State != "open" || retry.AttemptsReserved != 1 || retry.Attempt != nil {
421 t.Fatalf("restore failure not settled without refund: %+v", retry)
422 }
423 if c.wake.get(recoveryHost) != nil {
424 t.Fatal("failed attempt still pending")
425 }
426 if len(f.kernel.verified) == 0 || len(f.kernel.recovered) != 0 {
427 t.Fatalf("verified=%d recovered=%d", len(f.kernel.verified), len(f.kernel.recovered))
428 }
429 for _, req := range f.requests {
430 if req.Recover || req.ControllerID != "controller-1" || req.WriterSession != admitted.Claim.Holder.Session {
431 t.Fatalf("live controller stop used another tuple: %+v", req)
432 }
433 }
434 if got := f.receipts(); len(got) != 1 || got[0].ControllerID != "controller-1" || got[0].WriterSession != admitted.Claim.Holder.Session {
435 t.Fatalf("receipts = %+v", got)
436 }
437}
438
439// Control for the test above: with rev1's placement, nothing is journaled
440// before launch, root cannot verify the stop and the attempt stays reserved.
441func TestClaimWakeRestoreFailureWithoutPrelaunchRecordStrands(t *testing.T) {
442 f := newRecoveryFixture(t)
443 f.restoreFailure = errors.New("initial-empty authority unavailable")
444 ctx := t.Context()
445 f.recordPrelaunch = false
446 c := f.controller("boot-1", "controller-1", 4101, "start-1")
447 admitted := c.admit(t, f)
448 s, err := c.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, recoveryPolicy)
449 if err != nil {
450 t.Fatal(err)
451 }
452 restoreErr := s.Restore(ctx)
453 if err = c.runtime.FailedAdmission(ctx, admitted, restoreErr); err == nil {
454 t.Fatal("stop verified without a launch record")
455 }
456 if retry := f.retry(); retry.State != "attempting" || retry.AttemptsReserved != 1 {
457 t.Fatalf("retry = %+v", retry)
458 }
459 if got := f.receipts(); len(got) != 0 {
460 t.Fatalf("evidence without a launch record: %+v", got)
461 }
462}
463
464func TestClaimWakeMaterializeFailureSettlesFromPrelaunchRecord(t *testing.T) {
465 f := newRecoveryFixture(t)
466 ctx := t.Context()
467 c := f.controller("boot-1", "controller-1", 4101, "start-1")
468 admitted := c.admit(t, f)
469 failure := errors.New("materialization failed")
470 recorded := false
471 f.materialize = func(ctx context.Context, o hostclaim.Observed) error {
472 _, err := f.journal.Read(ctx, recoveryTuple(o, c.local))
473 recorded = err == nil
474 return failure
475 }
476 s, err := c.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, recoveryPolicy)
477 if err != nil {
478 t.Fatal(err)
479 }
480 if err := s.Restore(ctx); err != nil {
481 t.Fatal(err)
482 }
483 // startAdmitted's loop: validate the reservation, materialize, then Failed.
484 if err = s.Reserve(ctx, 0); err != nil {
485 t.Fatal(err)
486 }
487 materializeErr := s.Materialize(ctx)
488 if !errors.Is(materializeErr, failure) || !recorded {
489 t.Fatalf("materialization ran without the pre-launch record: recorded=%v err=%v", recorded, materializeErr)
490 }
491 if err = s.Failed(ctx, 0); err != nil {
492 t.Fatalf("materialization failure stranded the attempt: %v", err)
493 }
494 if err = c.runtime.FailedAdmission(ctx, admitted, materializeErr); err != nil {
495 t.Fatal(err)
496 }
497 retry := f.retry()
498 if retry.State != "open" || retry.AttemptsReserved != 1 || retry.Attempt != nil {
499 t.Fatalf("materialization failure not settled without refund: %+v", retry)
500 }
501 if len(f.kernel.verified) != 1 {
502 t.Fatalf("the repeated stop recollected evidence: %d", len(f.kernel.verified))
503 }
504}
505
506func TestClaimRetryCrashBeforeLaunchRecoversOriginalTupleAfterRestart(t *testing.T) {
507 f := newRecoveryFixture(t)
508 ctx := t.Context()
509 former := f.controller("boot-1", "controller-1", 4101, "start-1")
510 admitted := former.admit(t, f)
511 prepared, err := former.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, recoveryPolicy)
512 if err != nil {
513 t.Fatal(err)
514 }
515 // Run the assembly's real ColdRestore.InstallEmpty, then discard the
516 // ClaimStart before materialization or launch.
517 materializations := 0
518 f.materialize = func(context.Context, hostclaim.Observed) error { materializations++; return nil }
519 if err := prepared.Restore(ctx); err != nil {
520 t.Fatal(err)
521 }
522 marker := filepath.Join(f.guard.DataDir, "state", "data.db.gen")
523 if raw, err := os.ReadFile(marker); err != nil || strings.TrimSpace(string(raw)) != admitted.Claim.Holder.Generation {
524 t.Fatalf("real restore boundary: %q %v", raw, err)
525 }
526 t.Cleanup(func() {
527 if materializations != 0 {
528 t.Fatalf("materialization ran across the crash: %d", materializations)
529 }
530 })
531 attempt := *admitted.Claim.Holder.WakeRetry.Attempt
532 // The controller crashes after restore and before materialization. A later record
533 // for another session must not be touched by recovering this one.
534 successor := recoveryTuple(admitted, claimStopLocal{Actor: former.local.Actor, BootID: "boot-1", ControllerID: "controller-2"})
535 successor.Session++
536 successor.Stamp++
537 if err := f.journal.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: 4202, ProcessStart: "start-2", Target: successor, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/successor", Device: 1, Inode: 99}}, NoState: true}); err != nil {
538 t.Fatal(err)
539 }
540 reservation := hostbox.LaunchReservation{ID: "successor:main", Role: "main", InvocationID: "successor-invocation", Inputs: "inputs", Target: successor, Group: hostbox.LaunchGroup{RelativePath: "carlos/successor/role", Device: 1, Inode: 100, InvocationID: "successor-invocation"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 101}}
541 if err := f.journal.Reserve(ctx, reservation); err != nil {
542 t.Fatal(err)
543 }
544
545 restarted := f.controller("boot-1", "controller-2", 4202, "start-2")
546 rt := registry.Route{Host: recoveryHost}
547 f.kernel.alive = true
548 err = restarted.runtime.Prerequisites(ctx, rt, f.read(), recoveryPolicy)
549 if !errors.Is(err, ErrClaimStopControllerAlive) {
550 t.Fatalf("live former controller = %v", err)
551 }
552 if retry := f.retry(); retry.State != "attempting" || retry.Attempt == nil || *retry.Attempt != attempt || retry.AttemptsReserved != 1 {
553 t.Fatalf("settled before the former controller was proved dead: %+v", retry)
554 }
555 if got := f.receipts(); len(got) != 0 {
556 t.Fatalf("evidence before proof: %+v", got)
557 }
558
559 f.kernel.alive = false
560 if err = restarted.runtime.Prerequisites(ctx, rt, f.read(), recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
561 t.Fatalf("settlement after restart = %v", err)
562 }
563 for _, req := range f.requests {
564 if !req.Recover || req.BootID != attempt.BootID || req.ControllerID != attempt.ControllerID || req.WriterSession != attempt.Session {
565 t.Fatalf("restarted controller substituted its own tuple: %+v", req)
566 }
567 }
568 if len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 2 || f.kernel.recovered[0].FenceID != f.kernel.recovered[1].FenceID {
569 t.Fatalf("verified=%v recovered=%v", f.kernel.verified, f.kernel.recovered)
570 }
571 for _, target := range f.kernel.recovered {
572 if target.ControllerID != "controller-1" || target.BootID != "boot-1" || target.Session != attempt.Session {
573 t.Fatalf("recovery target = %+v", target)
574 }
575 }
576 retry := f.retry()
577 if retry.State != "open" || retry.AttemptsReserved != 1 || retry.Attempt != nil {
578 t.Fatalf("crash attempt not settled without refund: %+v", retry)
579 }
580 if got := f.receipts(); len(got) != 1 || got[0].ControllerID != "controller-1" || got[0].BootID != "boot-1" {
581 t.Fatalf("receipts = %+v", got)
582 }
583 other, err := f.journal.Read(ctx, successor)
584 if err != nil || len(other.Reservations) != 1 || other.Reservations[0].State != "reserved" {
585 t.Fatalf("recovery changed another session's record: %+v %v", other, err)
586 }
587 // The next wake stops its predecessor through the same journaled tuple and
588 // reuses the durable evidence instead of being refused for controller-2.
589 if err = restarted.wake.Stop(ctx, f.read()); err != nil {
590 t.Fatalf("predecessor stop after restart = %v", err)
591 }
592 if len(f.kernel.recovered) != 2 || len(f.kernel.verified) != 0 {
593 t.Fatalf("predecessor stop recollected: recovered=%d verified=%d", len(f.kernel.recovered), len(f.kernel.verified))
594 }
595}
596
597func TestClaimRetryRecoveryRefusesMissingLaunchRecord(t *testing.T) {
598 f := newRecoveryFixture(t)
599 ctx := t.Context()
600 f.recordPrelaunch = false
601 former := f.controller("boot-1", "controller-1", 4101, "start-1")
602 admitted := former.admit(t, f)
603 attempt := *admitted.Claim.Holder.WakeRetry.Attempt
604 restarted := f.controller("boot-2", "controller-2", 4202, "start-2")
605 err := restarted.runtime.Prerequisites(ctx, registry.Route{Host: recoveryHost}, f.read(), recoveryPolicy)
606 if err == nil || errors.Is(err, hostbox.ErrChanged) {
607 t.Fatalf("missing launch record settled: %v", err)
608 }
609 if len(f.requests) != 1 || !f.requests[0].Recover || f.requests[0].ControllerID != "controller-1" || f.requests[0].BootID != "boot-1" {
610 t.Fatalf("stop requests = %+v", f.requests)
611 }
612 if retry := f.retry(); retry.State != "attempting" || retry.Attempt == nil || *retry.Attempt != attempt || retry.AttemptsReserved != 1 {
613 t.Fatalf("retry = %+v", retry)
614 }
615 if got := f.receipts(); len(got) != 0 {
616 t.Fatalf("evidence without a launch record: %+v", got)
617 }
618}
619
620type recoveryOnlyKernel struct {
621 stopKernelFunc
622 recover func(context.Context, hostguard.FenceTarget, StopInventory) (StopKernelEvidence, DrainAcknowledgement, error)
623}
624
625func (k recoveryOnlyKernel) RecoverStopped(ctx context.Context, t hostguard.FenceTarget, in StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
626 return k.recover(ctx, t, in)
627}
628
629func TestClaimStopRecoveryModeRefusesWithoutRootProof(t *testing.T) {
630 c, req, _ := claimStopFixture(t)
631 seq := 0
632 c.NewID = func() string { seq++; return fmt.Sprintf("id-%d", seq) }
633 drained := false
634 c.Drainer = stopDrainFunc(func(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
635 drained = true
636 return DrainAcknowledgement{}, nil
637 })
638 c.Peers = stopPeerFunc(func(context.Context) (StopPeer, error) {
639 return StopPeer{UID: 1001, PID: 43, ProcessStart: "start-43", BootID: "boot-1", ControllerID: "controller-2"}, nil
640 })
641 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopIdentity) {
642 t.Fatalf("another tuple without recovery = %v", err)
643 }
644 req.Recover = true
645 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopUnavailable) || drained || seq != 0 {
646 t.Fatalf("recovery without a recovery kernel = %v drained=%v ids=%d", err, drained, seq)
647 }
648 if pending, err := c.Guard.PendingFences(t.Context()); err != nil || len(pending) != 0 {
649 t.Fatalf("refused recovery installed a fence: %+v %v", pending, err)
650 }
651 cleanup := req
652 cleanup.Cleanup = true
653 if _, err := c.FenceAndStop(t.Context(), cleanup); !errors.Is(err, ErrClaimStopUnavailable) {
654 t.Fatalf("recovery combined with cleanup = %v", err)
655 }
656 evidence := StopKernelEvidence{InventorySHA256: strings.Repeat("b", 64), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"cg-1"}}
657 c.Kernel = recoveryOnlyKernel{recover: func(_ context.Context, target hostguard.FenceTarget, _ StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
658 return StopKernelEvidence{}, DrainAcknowledgement{}, ErrClaimStopControllerAlive
659 }}
660 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopControllerAlive) {
661 t.Fatalf("live former controller = %v", err)
662 }
663 // The same IDs and process start as the live peer are that peer, not a
664 // former controller, whatever the recovery kernel reports.
665 c.Peers = stopPeerFunc(func(context.Context) (StopPeer, error) {
666 return StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: req.BootID, ControllerID: req.ControllerID}, nil
667 })
668 c.Kernel = recoveryOnlyKernel{recover: func(_ context.Context, target hostguard.FenceTarget, _ StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
669 return evidence, DrainClaimAcknowledgement(target.FenceID, target.ControllerID, "start-42", nil), nil
670 }}
671 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopIdentity) {
672 t.Fatalf("live peer recovered as former controller = %v", err)
673 }
674 c.Kernel = recoveryOnlyKernel{recover: func(_ context.Context, target hostguard.FenceTarget, _ StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
675 return evidence, DrainClaimAcknowledgement(target.FenceID, target.ControllerID, "start-41", nil), nil
676 }}
677 result, err := c.FenceAndStop(t.Context(), req)
678 if err != nil || drained {
679 t.Fatalf("restarted controller recovery = %v drained=%v", err, drained)
680 }
681 if result.Target.ControllerID != req.ControllerID || result.Target.BootID != req.BootID {
682 t.Fatalf("recovery receipt names %+v", result.Target)
683 }
684}
685
686func TestClaimRetryRecoveryRetainsRecordedDirectoryRevision(t *testing.T) {
687 f := newRecoveryFixture(t)
688 former := f.controller("boot-1", "controller-1", 4101, "start-1")
689 former.admit(t, f)
690 f.revision = 8
691 restarted := f.controller("boot-1", "controller-2", 4202, "start-2")
692 err := restarted.runtime.Prerequisites(t.Context(), registry.Route{Host: recoveryHost}, f.read(), recoveryPolicy)
693 if !errors.Is(err, hostbox.ErrChanged) {
694 t.Fatalf("same-enrollment roster change stranded recovery: %v", err)
695 }
696 if len(f.requests) != 1 || f.requests[0].ExpectedDirectoryRevision != "7" {
697 t.Fatalf("recovery substituted current directory: %+v", f.requests)
698 }
699 if retry := f.retry(); retry.State != "open" || retry.Attempt != nil || retry.AttemptsReserved != 1 {
700 t.Fatalf("retry = %+v", retry)
701 }
702}
703
704func (k *recoveryKernel) VerifyNoProcess(ctx context.Context, target hostguard.FenceTarget, inv StopInventory) (StopKernelEvidence, error) {
705 k.noProcessCalls++
706 if !k.noProcess {
707 return StopKernelEvidence{}, ErrClaimStopInventory
708 }
709 has, err := k.journal.HasHost(ctx, target.Host)
710 if err != nil || has {
711 return StopKernelEvidence{}, ErrClaimStopInventory
712 }
713 return StopKernelEvidence{InventorySHA256: digestStop(inv), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"absent:carlos-exec@shop.example"}, UnitInvocationIDs: []string{}}, nil
714}
715