amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "crypto/ed25519"
9 "crypto/sha256"
10 "database/sql"
11 "encoding/binary"
12 "encoding/hex"
13 "encoding/json"
14 "errors"
15 "fmt"
16 "os"
17 "os/exec"
18 "path/filepath"
19 "reflect"
20 "strconv"
21 "strings"
22 "testing"
23 "time"
24
25 "github.com/carlosframework/platform/internal/activator"
26 "github.com/carlosframework/platform/internal/cgroup"
27 "github.com/carlosframework/platform/internal/hostactors"
28 "github.com/carlosframework/platform/internal/hostbox"
29 "github.com/carlosframework/platform/internal/hostclaim"
30 "github.com/carlosframework/platform/internal/hostfence"
31 "github.com/carlosframework/platform/internal/hostguard"
32 "github.com/carlosframework/platform/internal/hostops"
33 "github.com/carlosframework/platform/internal/hostwrite"
34 "github.com/carlosframework/platform/internal/moves"
35 "github.com/carlosframework/platform/internal/park"
36 "github.com/carlosframework/platform/internal/provision"
37 "github.com/carlosframework/platform/internal/registry"
38 "github.com/carlosframework/platform/internal/release"
39 "github.com/superfly/ltx"
40)
41
42// Plan 5 Task 9b's integrated acceptance tests (ruling Q12): claim Plan 12
43// proved its half with claim fixtures (TestPlan12RecoveryAdmissionAfter
44// TargetSession, TestPlan12RecoveryRestoreSourceDatabase); these run the
45// same paths with the moves owners the registered wake source opens: the
46// claim's wake runtime, the cross-box handover, the claim's writer with
47// the moves owners' evidence and loss owner, and, where the database is
48// restored, ColdRestore and the source installation's retirement.
49
50// mwAccess is the claim's move read access over a poFixture: its route and
51// launch as they are when asked, and the held works the fixture selects.
52type mwAccess struct{ f *poFixture }
53
54func (a mwAccess) HeldWork(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error) {
55 return a.f.owners.Works(ctx, host, action, captured)
56}
57func (a mwAccess) Route(_ context.Context, host string) (*registry.Route, error) {
58 if host != a.f.route.Host {
59 return nil, hostbox.ErrChanged
60 }
61 rt := a.f.route
62 return &rt, nil
63}
64func (a mwAccess) Launch(_ context.Context, host string) (ClaimServingLaunch, moves.EnvReport, error) {
65 if host != a.f.route.Host {
66 return ClaimServingLaunch{}, moves.EnvReport{}, hostbox.ErrChanged
67 }
68 // The env report of the same launch env, as claimMoveLaunchSource makes
69 // it: its digest is the launch's move-portable pin.
70 key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
71 if err != nil {
72 return ClaimServingLaunch{}, moves.EnvReport{}, err
73 }
74 return a.f.launch, moves.ReportLaunch(key, launchVars(a.f.launch.Env), nil), nil
75}
76
77// mwWorld is one box's registered moves wake source over a poFixture's
78// world, and the claim's wake runtime that box builds around it.
79type mwWorld struct {
80 t *testing.T
81 f *poFixture
82 box, region string
83 local claimStopLocal
84 owners ClaimMovesOwners
85 m *movesProbeOwners
86 files ClaimFilesManifest
87 guard hostguard.Guard
88 journal *hostbox.LaunchJournal
89 handover claimCrossBoxHandover
90 wake *ClaimWakeAssembly
91 runtime *activator.ClaimRuntime
92}
93
94// mwRegister registers the moves wake source as the agent does (the parks,
95// then the wake source, both switches on), opens it as the claim's
96// OpenClaimOwners does for box in region (the source's Open, then the
97// claim's C8 provider through openClaimFilesManifest), and repins the
98// fixture's recorded environment reports under the registered source's key
99// (CARLOS_SECRETS_KEY's). Only the records and the marker store are the
100// fixture's: the fixture keeps its record in memory and its bucket in a
101// fake.
102func mwRegister(t *testing.T, f *poFixture, box, region string, files ClaimFilesManifest) *mwWorld {
103 t.Helper()
104 ctx := t.Context()
105 withMovesWakeSource(t, false)
106 withClaimFilesSource(t, nil)
107 parks, err := newMovesParkSources(f.owners.Records, f.bucket, "data", fixedRegion(region), movesRetirementEvidence{Records: f.owners.Records, Now: time.Now})
108 if err != nil {
109 t.Fatal(err)
110 }
111 if err := registerMovesParks(parks); err != nil {
112 t.Fatal(err)
113 }
114 env := mwEnv(true, true)
115 if err := registerProductionMovesWakeSource(env, parks, f.dataDir); err != nil {
116 t.Fatal(err)
117 }
118 if err := movesWakeSourceRegistered(); err != nil {
119 t.Fatal(err)
120 }
121 // The app's manifest, which the registered owners read for the
122 // readiness path, and the files directory where production keeps it
123 // (movesFilesDirFor: <data>/<host>); a rename keeps its identity.
124 if err := provision.PutManifest(ctx, f.claims, "bab", "jam", provision.Manifest{Moves: &provision.MovesManifest{ReadyPath: "/healthz"}}); err != nil {
125 t.Fatal(err)
126 }
127 if dir, _ := movesFilesDirFor(f.dataDir)(poHost); f.files != dir {
128 if err := os.Rename(f.files, dir); err != nil {
129 t.Fatal(err)
130 }
131 f.files = dir
132 }
133 src := claimMoveSources.Source.(*movesWakeSourceRegistration).ClaimProductionMoveSources.(*movesWakeSources)
134 records := f.owners.Records
135 src.records = func(release.Store) MovesRecordReader { return records }
136 src.markers = func(context.Context, string, string) (ClaimMarkerStore, error) { return plan9D5Markers{f.bucket}, nil }
137 src.now, src.newID = f.owners.Now, f.owners.NewID
138 w := &mwWorld{t: t, f: f, box: box, region: region}
139 // The box's local guard and parked area under the fixture's data
140 // directory.
141 for _, dir := range []string{"hostguard", ".parked"} {
142 if err := os.MkdirAll(filepath.Join(f.dataDir, dir), 0o770); err != nil {
143 t.Fatal(err)
144 }
145 }
146 w.guard = hostguard.Guard{DataDir: f.dataDir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: f.claims}}
147 w.local = claimStopLocal{Actor: hostbox.Actor{Box: box, Enrollment: "enrollment-1", Revision: 7, Binding: "7", Region: region}, BootID: "boot-1", ControllerID: "controller-1", PID: 4101, ProcessStart: "start-1"}
148 actor := hostbox.Actor{Box: box, Region: region, Bucket: "data", Enrollment: "enrollment-1", Cloud: hostbox.CloudIdentity{Region: "eu-west-1"}}
149 out := ClaimOwnerBindings{Data: f.bucket}
150 var opened ClaimFilesManifest
151 // The claim hands Open its bucket through the request store
152 // (openProductionClaimMoveWake), with the box's identity check.
153 data := hostbox.RequestStore{Store: f.bucket, Check: func(context.Context) error { return nil }}
154 out.Moves, opened, err = claimMoveSources.Source.Open(ctx, f.claims, data, actor, mwAccess{f})
155 if err != nil || opened != nil {
156 t.Fatalf("the registered source's Open: %v (files %v)", err, opened)
157 }
158 if _, err := openClaimFilesManifest(ctx, env, &fakeFilesSource{files: files}, f.claims, f.bucket, actor, &out, func(context.Context) error { return nil }); err != nil {
159 t.Fatal(err)
160 }
161 w.owners, w.files = out.Moves, out.Serving.Files
162 w.m = out.Moves.Inputs.(*movesProbeOwners)
163 if w.m.Files != files || w.files != files {
164 t.Fatalf("the recovery owners' C8 provider %v is not the claim's %v", w.m.Files, w.files)
165 }
166 key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
167 if err != nil {
168 t.Fatal(err)
169 }
170 if mv := f.rec.Move; mv != nil {
171 if mv.Inputs != nil {
172 mv.Inputs.EnvReport = envDigest(key, f.launch.Env)
173 }
174 if mv.RecoveryInputs != nil {
175 mv.RecoveryInputs.EnvReport = localEnvDigest(key, f.launch.Env)
176 }
177 if mv.Rollback != nil && mv.Rollback.RecoveryInputs != nil {
178 mv.Rollback.RecoveryInputs.EnvReport = localEnvDigest(key, f.launch.Env)
179 }
180 }
181 return w
182}
183
184// build is the claim's wake runtime on this box (claimconstruction.go's
185// shape): the moves owners behind ClaimOwnerDispatch, the claim's writer
186// with their evidence and the moves loss owner (pub is the loss records'
187// root key), the cross-box handover over real registry rows and stop
188// inventory, and the source installation's retirement through park.Local.
189func (w *mwWorld) build(pub ed25519.PublicKey, inventory *hostbox.StopInventoryDB, reg *registry.Registry) {
190 t, f := w.t, w.f
191 journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db"))
192 if err != nil {
193 t.Fatal(err)
194 }
195 t.Cleanup(func() { journal.Close() })
196 w.journal = journal
197 dispatch := ClaimOwnerDispatch{Moves: &w.owners.ClaimKindOwners}
198 local, now := w.local, f.owners.Now
199 producers := &ClaimProducerSet{}
200 retirement := claimArtifactRetirement{Local: park.Local{Guard: w.guard, Claims: f.claims, DataDir: f.dataDir, Box: w.box}, Continue: hostclaim.SessionContinues}
201 wk := &ClaimWakeAssembly{LeaseRegion: "eu-west-1", Inputs: dispatch, Manager: new(cgroup.Manager), Operations: w.owners.Operations, Continue: hostclaim.SessionContinues}
202 wk.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
203 wk.AfterAdmission = func(context.Context, hostclaim.Observed, hostclaim.Observed) error { return nil }
204 // Every predecessor here is a remote one (the cross-box handover) or
205 // none; root's local stop is never asked.
206 wk.Stop = func(context.Context, hostclaim.Observed) error { return hostbox.ErrChanged }
207 wk.PrepareState = func(ctx context.Context, _ hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) error {
208 return work.With(ctx, in.Identity, func() error { return os.MkdirAll(filepath.Dir(in.DBPath), 0o700) })
209 }
210 wk.RetireArtifacts, wk.RecoverArtifacts = retirement.Retire, retirement.Recover
211 wk.Retain = func(hostclaim.Observed, activator.Proc, activator.Rep, error) {}
212 wk.ObserveLoss = func(hostclaim.Observed, error) {}
213 wk.Producers = func(_ context.Context, o hostclaim.Observed) (*ClaimProducers, error) {
214 return producers.Ensure(recoveryTuple(o, local), local.ProcessStart, journal.RecordProducer)
215 }
216 wk.Launch = func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error) {
217 return nil, hostwrite.ErrUnavailable
218 }
219 wk.Prelaunch = func(ctx context.Context, o hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) (*activator.RoleLaunch, error) {
220 target := recoveryTuple(o, local)
221 err := work.With(ctx, in.Identity, func() error {
222 return journal.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: local.PID, ProcessStart: local.ProcessStart, Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: target.Session}}, NoState: true})
223 })
224 if err != nil {
225 return nil, err
226 }
227 return &activator.RoleLaunch{Target: target, Journal: journal}, nil
228 }
229 wk.Retries = activator.ServingRetries{Claims: f.claims, Box: w.box, Enrollment: local.Actor.Enrollment, BootID: local.BootID, ControllerID: local.ControllerID, Now: now, NewID: f.owners.NewID,
230 Inputs: func(context.Context, registry.Route) (string, error) { return "inputs", nil },
231 StopAttempt: func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error { return hostbox.ErrChanged }}
232 w.handover = claimCrossBoxHandover{Guard: w.guard, Evidence: f.claims, Box: w.box, PublicKey: pub, FenceAuthority: plan12FenceAuthority{}, Identity: handoverRecoveryIdentity(t, local.Actor), Inventory: inventory, Registry: reg}
233 wireClaimAdmission(wk, w.box, wk.AfterAdmission, w.handover, nil, nil)
234 source := hostbox.WorkSource{Guard: w.guard, Box: w.box, Region: w.region, Now: now,
235 Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil },
236 Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
237 writer := hostbox.Writer{Data: f.bucket, Bucket: "data", Claims: f.claims, Box: w.box, Region: w.region, AdmissionEvidence: dispatch.AdmissionEvidence,
238 Loss: &hostbox.LossAdmission{PublicKey: pub, Owner: w.owners.LossRecovery, Sources: w.owners.LossSources}}
239 runtime, err := wk.Build(t.Context(), source, writer)
240 if err != nil {
241 t.Fatal(err)
242 }
243 w.wake, w.runtime = wk, runtime
244}
245
246// admit stands in for Prerequisites' S3 lease, then admits through the
247// runtime: the moves owners' Select, the handover's recovery intent, the
248// claim's writer and the moves owners' evidence, the handover after the
249// CAS, the guard's activation and the pre-launch record.
250func (w *mwWorld) admit(o hostclaim.Observed, p activator.StartPolicy) (hostclaim.Observed, error) {
251 w.t.Helper()
252 live, cancel := context.WithCancel(context.Background())
253 w.t.Cleanup(cancel)
254 done := make(chan struct{})
255 go func() { <-live.Done(); close(done) }()
256 w.wake.mu.Lock()
257 w.wake.pending[o.Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
258 w.wake.mu.Unlock()
259 return w.runtime.Admit(w.t.Context(), w.f.route, o, p)
260}
261
262// mwSnapshot is a real Litestream snapshot (one level-9 LTX file, TXIDs 1
263// to maxTXID) of a SQLite database holding value.
264func mwSnapshot(t *testing.T, maxTXID uint64, value string) []byte {
265 t.Helper()
266 path := filepath.Join(t.TempDir(), "snapshot.db")
267 db, err := sql.Open("sqlite", path)
268 if err != nil {
269 t.Fatal(err)
270 }
271 if _, err := db.Exec("CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES (?)", value); err != nil {
272 t.Fatal(err)
273 }
274 if err := db.Close(); err != nil {
275 t.Fatal(err)
276 }
277 raw, err := os.ReadFile(path)
278 if err != nil {
279 t.Fatal(err)
280 }
281 size := int(binary.BigEndian.Uint16(raw[16:18]))
282 if size == 1 {
283 size = 65536
284 }
285 spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(size), Commit: uint32(len(raw) / size), MinTXID: 1, MaxTXID: ltx.TXID(maxTXID)}}
286 for off := 0; off < len(raw); off += size {
287 pg := uint32(off/size + 1)
288 if pg != ltx.LockPgno(uint32(size)) {
289 spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pg}, Data: raw[off : off+size]})
290 }
291 }
292 var out bytes.Buffer
293 if _, err := spec.WriteTo(&out); err != nil {
294 t.Fatal(err)
295 }
296 return out.Bytes()
297}
298
299// mwReadDB reads the fixture value of the database at path.
300func mwReadDB(t *testing.T, path string) string {
301 t.Helper()
302 db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
303 if err != nil {
304 t.Fatal(err)
305 }
306 defer db.Close()
307 var value string
308 if err := db.QueryRow("SELECT value FROM fixture").Scan(&value); err != nil {
309 t.Fatalf("read %s: %v", path, err)
310 }
311 return value
312}
313
314// mwSourceInstallation is the source's installation at S on ie-box as the
315// cut and the return leave it, for the handover and the retirement: the
316// source's stopped report (the cut's evidence), the guard's local record at
317// S, the database and its marker on disk tagged S in the claim's inventory,
318// and the registry row (with a timer) and complete stop inventory at S.
319type mwSourceInstallation struct {
320 inventory *hostbox.StopInventoryDB
321 registry *registry.Registry
322 dbRel string
323}
324
325func (w *mwWorld) sourceInstallation() mwSourceInstallation { return w.sourceInstallationRow(rcS) }
326
327// sourceInstallationRow is sourceInstallation with the registry's route row
328// owned by the installation at rowStamp.
329func (w *mwWorld) sourceInstallationRow(rowStamp uint64) mwSourceInstallation {
330 t, f := w.t, w.f
331 ctx := t.Context()
332 o := f.read()
333 h := o.Claim.Holder
334 report := moves.StoppedReport{V: 1, Role: "source", Box: w.box, Host: poHost, OpID: poOp, Incarnation: h.Incarnation, ClaimGen: o.Claim.Gen, CheckedAt: time.Now().UTC(),
335 Steps: map[string]moves.Check{moves.StepTree: {OK: true}, moves.StepStray: {OK: true}, moves.StepFDs: {OK: true}, moves.StepLease: {OK: true}, moves.StepReplica: {OK: true}}}
336 raw, _ := json.Marshal(report)
337 if _, err := f.claims.Put(ctx, moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "source"), raw, ""); err != nil {
338 t.Fatal(err)
339 }
340 if err := w.guard.Activate(ctx, poHost, w.box, nil); err != nil {
341 t.Fatal(err)
342 }
343 if err := os.WriteFile(f.route.DBPath, []byte("source database"), 0o600); err != nil {
344 t.Fatal(err)
345 }
346 dbRel, err := filepath.Rel(f.dataDir, f.route.DBPath)
347 if err != nil {
348 t.Fatal(err)
349 }
350 local := park.Local{Guard: w.guard, Claims: f.claims, DataDir: f.dataDir, Box: w.box}
351 if err := local.RecordInventory(ctx, poHost, hostguard.Captured{Incarnation: h.Incarnation, Stamp: rcS}, "flat", []string{dbRel, dbRel + ".gen"}, func(*hostclaim.Claim) error { return nil }); err != nil {
352 t.Fatal(err)
353 }
354 inv, err := openLegacyInventory(t, filepath.Join(t.TempDir(), "inventory.db"))
355 if err != nil {
356 t.Fatal(err)
357 }
358 t.Cleanup(func() { inv.Close() })
359 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
360 if err != nil {
361 t.Fatal(err)
362 }
363 t.Cleanup(func() { reg.Close() })
364 row := registry.Route{Host: poHost, Kind: "instance", DBPath: f.route.DBPath, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: h.Incarnation, ClaimStamp: rowStamp, ClaimOwner: "old-owner", ClaimRowID: "old-row"}}
365 if err := reg.Upsert(row); err != nil {
366 t.Fatal(err)
367 }
368 // A timer takes its route's ownership, under a row id of its own.
369 timer, err := reg.UpsertTimerRegistration(registry.Timer{Host: poHost, Name: "nightly", Kind: "cron", Spec: "0 3 * * *"}, "")
370 if err != nil {
371 t.Fatal(err)
372 }
373 if err := inv.WriteStopInventory(ctx, hostbox.StopInventory{Complete: true, Host: poHost, Incarnation: h.Incarnation, InstallationStamp: rcS, Box: w.box, Enrollment: w.local.Actor.Enrollment, DirectoryRevision: "7",
374 Entries: []hostbox.StopInventoryEntry{
375 {Kind: "route", ArtifactID: "route", ManagedRoot: "registry", RelativePath: poHost, RowIdentity: "old-row", Ownership: "old-owner", Materialization: "materialized"},
376 {Kind: "timer", ArtifactID: "timer", ManagedRoot: "registry", RelativePath: poHost + "/nightly", RowIdentity: timer.ClaimRowID, Ownership: "old-owner", Materialization: "materialized"},
377 }}); err != nil {
378 t.Fatal(err)
379 }
380 return mwSourceInstallation{inventory: inv, registry: reg, dbRel: dbRel}
381}
382
383// targetStopped journals the target's stop evidence for its last session
384// (the claim's target journal, J3): the session, the generation, and the
385// target's stopped report.
386func (w *mwWorld) targetStopped() {
387 t, f := w.t, w.f
388 o := f.read()
389 h := o.Claim.Holder
390 last := h.Admissions[len(h.Admissions)-1]
391 key := moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "target")
392 report := moves.StoppedReport{V: 1, Role: "target", Box: last.Box, Host: poHost, OpID: poOp, Incarnation: h.Incarnation, ClaimGen: o.Claim.Gen, CheckedAt: time.Now().UTC(),
393 Steps: map[string]moves.Check{moves.StepTree: {OK: true}, moves.StepStray: {OK: true}, moves.StepFDs: {OK: true}, moves.StepLease: {OK: true}, moves.StepReplica: {OK: true}}}
394 raw, _ := json.Marshal(report)
395 if _, err := f.claims.Put(t.Context(), key, raw, ""); err != nil {
396 t.Fatal(err)
397 }
398 c := *o.Claim
399 c.Op.Journal[hostclaim.JournalEvidenceTarget] = key
400 c.Op.Journal[hostclaim.JournalTargetSession] = strconv.FormatUint(last.Session, 10)
401 c.Op.Journal[hostclaim.JournalTargetGeneration] = last.Generation
402 f.putClaim(&c)
403}
404
405// srcSnapshot replaces src_gen's root with a real snapshot holding value at
406// the source's maximum (30), the clean return's restore point.
407func (w *mwWorld) srcSnapshot(value string) {
408 w.f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), mwSnapshot(w.t, 30, value))
409}
410
411// restore prepares the admitted start and runs its restore.
412func (w *mwWorld) restore(admitted hostclaim.Observed, p activator.StartPolicy) activator.ClaimStart {
413 w.t.Helper()
414 s, err := w.runtime.Prepare(w.t.Context(), w.f.route, admitted, p)
415 if err != nil {
416 w.t.Fatal(err)
417 }
418 if err := s.Restore(w.t.Context()); err != nil {
419 w.t.Fatal(err)
420 }
421 return s
422}
423
424// TestRecoveryAdmissionAfterTargetSession (Q12; claim Plan 12's
425// TestPlan12RecoveryAdmissionAfterTargetSession with the registered moves
426// owners): the source's first recovery admission after a target session,
427// on each of the three authorities a remote predecessor can have, the
428// target's journalled stop, a declared loss (J4) and a consumed
429// termination fence, goes through the handover, the claim's writer under
430// the moves owners' evidence and loss owner, and the restore: the source
431// executes a new session over the restored source baseline.
432func TestRecoveryAdmissionAfterTargetSession(t *testing.T) {
433 for _, mode := range []string{"journal", "loss", "fence"} {
434 t.Run(mode, func(t *testing.T) {
435 f := newRCFixture(t, rcClean)
436 var pub ed25519.PublicKey
437 switch mode {
438 case "loss":
439 pub = f.declare(false, []hostclaim.WriteSource{rcTargetSource(poOp, 20), rcCutSource()}, nil)
440 case "fence":
441 f.fence()
442 }
443 w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
444 w.srcSnapshot("the source baseline")
445 src := w.sourceInstallation()
446 w.build(pub, src.inventory, src.registry)
447 if mode == "journal" {
448 w.targetStopped()
449 }
450 o := f.read()
451 before := o.Claim.Holder
452 if !claimRemotePredecessor(o, "ie-box") || before.WriterBox != "uk-box" || before.Session != 20 {
453 t.Fatalf("the predecessor is not the target's session: %+v", before)
454 }
455 if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
456 t.Fatalf("the predecessor's authority: %v", err)
457 }
458 admitted, err := w.admit(o, f.policy())
459 if err != nil {
460 t.Fatal(err)
461 }
462 h := admitted.Claim.Holder
463 a := h.Admissions[len(h.Admissions)-1]
464 if h.WriterBox != "ie-box" || h.Session <= 20 || h.Stamp != h.Session || h.Generation != "fresh1" || a.Box != "ie-box" || a.Region != "ie" || a.Mode != "cold" || a.Predecessor != poOp || a.PredecessorRegion != "uk" {
465 t.Fatalf("the admission: holder %+v, last %+v", h, a)
466 }
467 switch mode {
468 case "journal":
469 if a.Coverage != recoveryCoverageClean || a.LossID != "" || a.WritesLost != nil {
470 t.Fatalf("a stopped target's clean return: %+v", a)
471 }
472 case "loss":
473 if a.Coverage != "" || a.LossID != rcLossID || a.WritesLost == nil || a.WritesLost.Box != "uk-box" || a.WritesLost.Session != 20 || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) {
474 t.Fatalf("a declared loss's admission: %+v", a)
475 }
476 case "fence":
477 if a.Coverage != "" || a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) {
478 t.Fatalf("a fenced target's admission: %+v", a)
479 }
480 }
481 w.restore(admitted, f.policy())
482 if got, err := os.ReadFile(f.route.DBPath + ".gen"); err != nil || string(got) != h.Generation+"\n" {
483 t.Fatalf("the restored marker: %q %v", got, err)
484 }
485 if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
486 t.Fatalf("restored %q", got)
487 }
488 })
489 }
490}
491
492// TestCleanReturnAdmissionRetiresTheSTaggedDatabase (Q12): the clean
493// return's first admission restores src_gen over the source's own
494// installation, whose database and marker the claim's inventory tags with
495// the source stamp S. The restore retires them, on the cut's stop
496// evidence, into the superseded installation's parked artifacts, and the
497// claim's inventory then tags the paths with the admitted stamp; a restart
498// restores the same pinned baseline and retires nothing further.
499func TestCleanReturnAdmissionRetiresTheSTaggedDatabase(t *testing.T) {
500 f := newRCFixture(t, rcClean)
501 w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
502 w.srcSnapshot("the source baseline")
503 src := w.sourceInstallation()
504 w.build(nil, src.inventory, src.registry)
505 w.targetStopped()
506 o := f.read()
507 if tag := o.Claim.Inventory["ie-box"].Paths[src.dbRel]; tag != rcS {
508 t.Fatalf("the source database is tagged %d, not S", tag)
509 }
510 if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
511 t.Fatal(err)
512 }
513 admitted, err := w.admit(o, f.policy())
514 if err != nil {
515 t.Fatal(err)
516 }
517 h := admitted.Claim.Holder
518 w.restore(admitted, f.policy())
519 retired := filepath.Join(f.dataDir, ".parked", h.Incarnation, poHost, fmt.Sprintf("superseded-%d-%d", h.Session, rcS), "artifacts", src.dbRel)
520 if got, err := os.ReadFile(retired); err != nil || string(got) != "source database" {
521 t.Fatalf("the S-tagged database was not retired: %q %v", got, err)
522 }
523 if got, err := os.ReadFile(retired + ".gen"); err != nil || string(got) != "src\n" {
524 t.Fatalf("the S-tagged marker was not retired: %q %v", got, err)
525 }
526 if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
527 t.Fatalf("restored %q", got)
528 }
529 after := f.read()
530 for _, p := range []string{src.dbRel, src.dbRel + ".gen"} {
531 if tag := after.Claim.Inventory["ie-box"].Paths[p]; tag != h.Stamp {
532 t.Fatalf("%s is tagged %d after the restore, want the admitted stamp %d", p, tag, h.Stamp)
533 }
534 }
535 // The recorded source baseline pins the restore: a newer version of
536 // src_gen's root is refused, and nothing more is retired.
537 w.f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), mwSnapshot(t, 30, "a newer write"))
538 if _, err := w.m.Select(t.Context(), f.route, after, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
539 t.Fatalf("a restart over another version of src_gen: %v", err)
540 }
541}
542
543// TestDirtyReturnThroughAfterRecoveryOnRealRows (Q12, the rev 3 recheck
544// §2.3): a dirty return's first admission, into the return generation,
545// goes through the handover's afterRecovery over the box's real registry
546// rows (the route and a timer) and its stop inventory at S: the rows and
547// the inventory move to the admitted installation's ownership, the
548// source's inventory stays for retirement and replay, and a row another
549// installation owns refuses the admission's handover.
550func TestDirtyReturnThroughAfterRecoveryOnRealRows(t *testing.T) {
551 t.Run("the rows move to the admitted installation", func(t *testing.T) {
552 f := newRCFixture(t, rcDirty)
553 w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
554 src := w.sourceInstallation()
555 w.build(nil, src.inventory, src.registry)
556 w.targetStopped()
557 o := f.read()
558 if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
559 t.Fatal(err)
560 }
561 admitted, err := w.admit(o, f.policy())
562 if err != nil {
563 t.Fatal(err)
564 }
565 h := admitted.Claim.Holder
566 a := h.Admissions[len(h.Admissions)-1]
567 if h.Generation != rpRB || h.Stamp == rcS || h.Stamp != h.Session || a.Coverage != "sealed" || a.Predecessor != poOp {
568 t.Fatalf("the dirty return's admission: holder %+v, last %+v", h, a)
569 }
570 sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", poHost, h.Incarnation, h.Stamp, "ie-box")))
571 owner := hex.EncodeToString(sum[:])
572 row, err := src.registry.Get(poHost)
573 if err != nil || row.ClaimStamp != h.Stamp || row.ClaimOwner != owner || row.ClaimIncarnation != h.Incarnation || row.ClaimRowID == "old-row" {
574 t.Fatalf("the route row: %+v %v", row, err)
575 }
576 timers, err := src.registry.TimersFor(poHost)
577 if err != nil || len(timers) != 1 || timers[0].ClaimStamp != h.Stamp || timers[0].ClaimOwner != owner || timers[0].ClaimRowID == "" || timers[0].ClaimRowID == row.ClaimRowID {
578 t.Fatalf("the timer row: %+v %v", timers, err)
579 }
580 inv, err := src.inventory.ReadInstallation(t.Context(), poHost, h.Incarnation, h.Stamp)
581 if err != nil || !inv.Complete || inv.Box != "ie-box" || len(inv.Entries) != 2 {
582 t.Fatalf("the admitted installation's inventory: %+v %v", inv, err)
583 }
584 for _, e := range inv.Entries {
585 if e.Ownership != owner || (e.Kind == "route" && e.RowIdentity != row.ClaimRowID) || (e.Kind == "timer" && e.RowIdentity != timers[0].ClaimRowID) {
586 t.Fatalf("an inventory entry: %+v", e)
587 }
588 }
589 if old, err := src.inventory.ReadInstallation(t.Context(), poHost, h.Incarnation, rcS); err != nil || !old.Complete {
590 t.Fatalf("the source's inventory at S is gone: %v", err)
591 }
592 })
593 t.Run("a row another installation owns refuses the handover", func(t *testing.T) {
594 f := newRCFixture(t, rcDirty)
595 w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
596 src := w.sourceInstallationRow(99)
597 w.build(nil, src.inventory, src.registry)
598 w.targetStopped()
599 o := f.read()
600 if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
601 t.Fatal(err)
602 }
603 if _, err := w.admit(o, f.policy()); !errors.Is(err, hostbox.ErrTypedInventory) {
604 t.Fatalf("a foreign row: %v", err)
605 }
606 if got, err := src.registry.Get(poHost); err != nil || got.ClaimStamp != 99 || got.ClaimRowID != "old-row" {
607 t.Fatalf("the foreign row moved: %+v %v", got, err)
608 }
609 })
610}
611
612// TestCommittedColdWakeThroughTheRegisteredSource (Q12): after the
613// landing's probe passed and the move committed, a wake on the target is a
614// cold admission into a fresh generation from the certified remote, through
615// the registered source's owners and the claim's runtime: the moves
616// owners' evidence admits it, and the start runs the committed wake's
617// ordinary launch, no probe variable, bound to the moves attempt journal.
618func TestCommittedColdWakeThroughTheRegisteredSource(t *testing.T) {
619 f, o := committedAfterProbe(t)
620 w := mwRegister(t, f, "uk-box", "uk", &rcFiles{f: f})
621 w.build(nil, nil, nil)
622 if err := w.guard.Activate(t.Context(), poHost, "uk-box", nil); err != nil {
623 t.Fatal(err)
624 }
625 h := o.Claim.Holder
626 if h.WriterBox != "uk-box" || h.Generation != poOp {
627 t.Fatalf("not the committed landing: %+v", h)
628 }
629 admitted, err := w.admit(o, poWake)
630 if err != nil {
631 t.Fatal(err)
632 }
633 ah := admitted.Claim.Holder
634 a := ah.Admissions[len(ah.Admissions)-1]
635 if ah.Generation == poOp || ah.Generation != a.Generation || a.Mode != "cold" || a.Coverage != "" || a.Predecessor != poOp || a.Region != "uk" || a.PredecessorRegion != "uk" || ah.WriterBox != "uk-box" || ah.Session <= h.Session {
636 t.Fatalf("the committed admission: holder %+v, last %+v", ah, a)
637 }
638 s, err := w.runtime.Prepare(t.Context(), f.route, admitted, poWake)
639 if err != nil {
640 t.Fatal(err)
641 }
642 if s.OpenProbe != nil || s.Identity != committedPinsIdentity(f.rec.Move.Inputs) {
643 t.Fatalf("the committed start: probe %v, identity %+v", s.OpenProbe != nil, s.Identity)
644 }
645 cmd := s.Build()
646 for _, kv := range cmd.Env {
647 if len(kv) >= len("CARLOS_MOVE_PROBE=") && kv[:len("CARLOS_MOVE_PROBE=")] == "CARLOS_MOVE_PROBE=" {
648 t.Fatalf("the committed wake carries %s", kv)
649 }
650 }
651 if err := s.Reserve(t.Context(), 0); err != nil {
652 t.Fatal(err)
653 }
654 j, err := moves.ReadJournal(f.dataDir, poHost, poOp)
655 if err != nil {
656 t.Fatal(err)
657 }
658 if _, ok := j.Steps[probeStepStart+"committed-"+strconv.FormatUint(ah.Session, 10)]; !ok {
659 t.Fatalf("the committed start is not in the moves attempt journal: %v", j.Steps)
660 }
661}
662
663// mwDeclare turns f into the landing whose target uk-box is lost, and
664// declares it through the console's real path: the record and the claim in
665// the deployment store (the target admitted session 20 into the move id
666// after the cut at 15, whose seal did not cover the source's replica), a
667// signed roster naming both boxes, the move's admission registration, then
668// hostops' InspectMoveLoss and DeclareMoveLoss, which run hostresolve's
669// declaration with the real moves loss owner. It returns the root's public
670// key, the declaration's id and the rollback id the declaration chose.
671func mwDeclare(t *testing.T, f *rcFixture) (ed25519.PublicKey, string, string) {
672 t.Helper()
673 ctx := t.Context()
674 pub, priv, err := ed25519.GenerateKey(nil)
675 if err != nil {
676 t.Fatal(err)
677 }
678 signer := release.RootSigner(priv)
679 clock := time.Now().UTC().Truncate(time.Second)
680 control, err := hostops.NewControlWriter(f.claims, pub, func() time.Time { return clock })
681 if err != nil {
682 t.Fatal(err)
683 }
684 const bucket = "control"
685 d := hostactors.Directory{Store: f.claims, PublicKey: pub, Bucket: bucket}
686 if err := d.Initialize(ctx, signer); err != nil {
687 t.Fatal(err)
688 }
689 ro, err := d.Read(ctx)
690 if err != nil {
691 t.Fatal(err)
692 }
693 r := hostactors.Clone(ro.Roster)
694 r.Boxes["ie-box"] = hostactors.Box{Enrollment: strings.Repeat("a", 32), Namespace: "platform", BoxID: "ie-box", Label: "ie-label", Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-ie"}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}
695 r.Boxes["uk-box"] = hostactors.Box{Enrollment: strings.Repeat("b", 32), Namespace: "platform", BoxID: "uk-box", Label: "uk-label", Region: "uk", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-2", InstanceID: "i-uk"}, EvidenceSHA256: strings.Repeat("d", 64), State: "active"}
696 if _, err := d.Replace(ctx, signer, ro, r); err != nil {
697 t.Fatal(err)
698 }
699 at := clock.Add(-time.Hour)
700 seal := rcCutSeal(false)
701 c := f.recoveryClaim()
702 c.Holder.Acquisition = "acq"
703 c.Holder.Stamp = 18
704 c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: poOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"},
705 Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box", "app_id": "canonical", "admission_id": poOp, "source_enrollment": strings.Repeat("a", 32), "target_enrollment": strings.Repeat("b", 32)},
706 Journal: map[string]string{hostclaim.JournalSeal: seal.Encode(), hostclaim.JournalCutSession: "15", hostclaim.JournalSourceStamp: "15", hostclaim.JournalSrcGen: "src",
707 hostclaim.JournalEvidenceSource: moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "source")}}
708 if err := c.Validate(); err != nil {
709 t.Fatal(err)
710 }
711 f.putClaim(c)
712 flipped, deadline := at, at.Add(2*time.Hour)
713 rec := provision.Record{Schema: 1, Host: poHost, Slug: "jam", Account: "bab", App: "jam", Acquisition: "acq", Channel: "edge", Region: "uk", CreatedAt: c.Holder.CreatedAt,
714 Move: &provision.MoveState{ID: poOp, Incarnation: f.inc, From: "ie", To: "uk", Cut: "auto", State: provision.MoveLanding, CutPhase: provision.CutFlipped,
715 FlippedAt: &flipped, LandingDeadline: &deadline, LandingPhase: provision.LandProbing, Inputs: mpPins(), Source: f.rec.Move.Source}}
716 raw, _ := json.Marshal(rec)
717 if _, err := f.claims.Put(ctx, provision.RecordKey("bab", "jam", poHost), raw, ""); err != nil {
718 t.Fatal(err)
719 }
720 owner := hostops.RegistrationOwner{Console: "console", Enrollment: strings.Repeat("e", 32), Controller: "coordinator"}
721 adm := hostops.AppAdmission{V: 1, AppID: "canonical", Epoch: 1, Revision: 1, State: "open", Effects: map[string]hostops.Effect{},
722 Registrations: map[string]hostops.Registration{poOp: {Kind: hostops.RegistrationMove, Host: poHost, Owner: owner, AppID: "canonical", Epoch: 1, Claim: hostops.SavedClaim{ETag: "serving", Gen: 1}, Expires: clock.Add(hostops.RegistrationTTL), CreatedAt: clock, Incarnation: f.inc}}}
723 key, err := hostops.AdmissionKey("bab", "jam")
724 if err != nil {
725 t.Fatal(err)
726 }
727 raw, _ = json.Marshal(adm)
728 if _, err := f.claims.Put(ctx, key, raw, ""); err != nil {
729 t.Fatal(err)
730 }
731 in, err := control.InspectMoveLoss(ctx, signer, bucket, "operator", "bab", "jam", poHost, "uk-label")
732 if err != nil {
733 t.Fatal(err)
734 }
735 clock = clock.Add(2 * time.Second)
736 sub := hostfence.Submission{Challenge: in.Challenge, Confirmation: in.Details.ID,
737 Observation: hostfence.CloudObservation{ObservedAt: in.Details.IssuedAt.Add(time.Second), Account: "123", Region: "eu-west-2", InstanceID: "i-uk", State: "stopped", RequestID: "request"},
738 Attestation: hostfence.OperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true, Disks: in.Details.Consequences.DestroyedStorageAssertion}}
739 declared, err := control.DeclareMoveLoss(ctx, signer, bucket, "operator", "bab", "jam", poHost, "uk-box", sub)
740 if err != nil {
741 t.Fatal(err)
742 }
743 return pub, in.Details.ID, declared.Claim.Op.Intent["rollback"]
744}
745
746// TestLostTargetRecoveryRunsAgainstARealDeclaration (Q12; Plan 4 Task
747// 10): the source's recovery runs against the claim and the signed loss
748// record a real declaration produced (hostresolve with the real moves loss
749// owner, through the console's typed writer), never a hand-written one.
750// The source's return then takes the claim back (ReturnToSource, probing)
751// and its record to probing with the recorded inputs, as the source pass
752// does, and the registered source's owners, reading the record from the
753// store, admit through the claim's writer with the moves loss owner: the
754// declaration is consumed, the target's session is declared lost at 0 (the
755// restore does not contain the move id's generation) and the uncovered cut
756// at the seal's maximum; the source restores its own baseline.
757func TestLostTargetRecoveryRunsAgainstARealDeclaration(t *testing.T) {
758 ctx := t.Context()
759 f := newRCFixture(t, rcClean)
760 pub, lossID, rb := mwDeclare(t, f)
761 o := f.read()
762 if o.Claim.Op.Kind != hostclaim.KindRestoring || o.Claim.Op.Phase != hostclaim.PhaseReleased || !provision.ValidRollbackID(rb) || !hostclaim.LostTargetReleased(o.Claim, lossID) {
763 t.Fatalf("the declaration's claim: %+v", o.Claim.Op)
764 }
765 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, lossID)
766 if err != nil || loss.Record.State != "committed" {
767 t.Fatalf("the loss record: %+v %v", loss.Record, err)
768 }
769 wantSources := []hostclaim.WriteSource{
770 {Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp},
771 {Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: "seal:" + poOp},
772 }
773 if !reflect.DeepEqual(loss.Record.WriteSources, wantSources) {
774 t.Fatalf("the declaration froze %+v", loss.Record.WriteSources)
775 }
776
777 // The source's pass: the claim back to this box at probing, the record
778 // through the clean branch to probing, the inputs it renders recorded.
779 if _, err := hostclaim.Update(ctx, f.claims, o, hostclaim.ReturnToSource(poOp, "ie-box", true, hostclaim.ReturnFence{})); err != nil {
780 t.Fatal(err)
781 }
782 if _, err := hostclaim.Update(ctx, f.claims, f.read(), hostclaim.SetRestoringPhase(poOp, hostclaim.PhaseClaimed, hostclaim.PhaseProbing)); err != nil {
783 t.Fatal(err)
784 }
785 key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
786 if err != nil {
787 t.Fatal(err)
788 }
789 inputs := *f.inputs
790 inputs.EnvReport = localEnvDigest(key, f.launch.Env)
791 *f.inputs = inputs
792 g := provision.MoveGuard{ID: poOp, Incarnation: f.inc}
793 for _, step := range [][2]string{{provision.RbSourceWaiting, provision.RbSourceClaimed}, {provision.RbSourceClaimed, provision.RbSourceRestoring}, {provision.RbSourceRestoring, provision.RbSourceVerified}, {provision.RbSourceVerified, provision.RbSourceProbing}} {
794 if _, err := provision.SetRecordMove(ctx, f.claims, "bab", "jam", poHost, provision.MoveRollbackSourcePhase(g, step[0], step[1], nil)); err != nil {
795 t.Fatalf("%s to %s: %v", step[0], step[1], err)
796 }
797 }
798 if _, err := provision.SetRecordMove(ctx, f.claims, "bab", "jam", poHost, provision.MoveRecordRecoveryInputs(g, rb, inputs)); err != nil {
799 t.Fatal(err)
800 }
801 f.route.HoldOp = rb
802
803 // The registered source over the store's records.
804 f.owners.Records = MovesRecordsFrom(f.claims)
805 w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
806 w.srcSnapshot("the source baseline")
807 src := w.sourceInstallation()
808 w.build(pub, src.inventory, src.registry)
809 o = f.read()
810 if err := w.wake.stopPredecessor(ctx, o); err != nil {
811 t.Fatalf("the declaration as the predecessor's authority: %v", err)
812 }
813 admitted, err := w.admit(o, f.policy())
814 if err != nil {
815 t.Fatal(err)
816 }
817 h := admitted.Claim.Holder
818 a := h.Admissions[len(h.Admissions)-1]
819 if h.WriterBox != "ie-box" || a.Coverage != "" || a.LossID != lossID || !hostclaim.OperationLossConsumed(admitted.Claim, lossID) {
820 t.Fatalf("the loss admission: holder %+v, last %+v", h, a)
821 }
822 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) ||
823 !reflect.DeepEqual(a.AdditionalWritesLost, []hostclaim.WritesLost{{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}) {
824 t.Fatalf("the declared writes: %+v, additional %+v", a.WritesLost, a.AdditionalWritesLost)
825 }
826 w.restore(admitted, f.policy())
827 if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
828 t.Fatalf("restored %q", got)
829 }
830}
831