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 "errors"
9 "net/http/httptest"
10 "os"
11 "path/filepath"
12 "reflect"
13 "strings"
14 "sync/atomic"
15 "syscall"
16 "testing"
17 "time"
18
19 "github.com/carlosframework/platform/internal/activator"
20 "github.com/carlosframework/platform/internal/cgroup"
21 "github.com/carlosframework/platform/internal/genselect"
22 "github.com/carlosframework/platform/internal/hostbox"
23 "github.com/carlosframework/platform/internal/hostclaim"
24 "github.com/carlosframework/platform/internal/hostguard"
25 "github.com/carlosframework/platform/internal/hostwrite"
26 "github.com/carlosframework/platform/internal/lease"
27 "github.com/carlosframework/platform/internal/lease/leasetest"
28 "github.com/carlosframework/platform/internal/provision"
29 "github.com/carlosframework/platform/internal/registry"
30 "github.com/superfly/ltx"
31 "golang.org/x/sys/unix"
32)
33
34// This fixture ends at Restore: a legacy holder has no C8 evidence yet.
35// It uses the real selector, lease against local S3, retry wrapper, admission,
36// activation, prelaunch journal, selected Work, stop coordinator and retirement
37// transport. Initial installation, kernel checks and launch preflight are
38// explicit seams; it does not claim an installed-unit or privileged witness.
39type servingPreparation struct {
40 assembly ClaimAssembly
41 source hostbox.WorkSource
42 writer hostbox.Writer
43 f *servingRootFixture
44 owners *claimServingOwners
45 wake *ClaimWakeAssembly
46 runtime *activator.ClaimRuntime
47 local claimStopLocal
48 events []string
49 baseline *genselect.Plan
50 enrollFailure error
51 afterEnroll func()
52}
53
54func newServingPreparation(t *testing.T, f *servingRootFixture, boot, controller string, configure ...func(*ClaimWakeAssembly)) *servingPreparation {
55 t.Helper()
56 if boot == "boot" {
57 var err error
58 boot, err = f.p.ledger.fs.(*servingParkFS).currentBoot()
59 if err != nil {
60 t.Fatal(err)
61 }
62 }
63 p := &servingPreparation{f: f, local: claimStopLocal{Actor: f.actor, BootID: boot, ControllerID: controller, PID: 77, ProcessStart: controller + "-start"}}
64 f.f.peer = StopPeer{UID: 1001, PID: p.local.PID, ProcessStart: p.local.ProcessStart, BootID: boot, ControllerID: controller}
65 launchFixture := plan9D5Fixture(t)
66 launchFixture.launch.Args = []string{"--socket", provision.NestedSocket(f.f.hostName()), "--db", f.db}
67 launchFixture.launch.Limit = cgroup.TierDefault
68 launchFixture.launch.Env = []string{"CARLOS_ADMIN_TOKEN=test"}
69 sources := launchFixture.owners.Sources
70 p.owners = &claimServingOwners{Data: f.f.data, Bucket: f.f.bucket, Region: f.actor.Region, Box: f.actor.Box, Sources: sources, Now: func() time.Time { return f.f.now }, NewID: f.f.newID}
71 producers := &ClaimProducerSet{}
72 w := &ClaimWakeAssembly{LeaseRegion: "eu-west-1", Manager: new(cgroup.Manager), Inputs: p.owners, RequireServingEnrollment: true}
73 p.wake = w
74 root := servingHandoverRoot(t, f)
75 hook := coldAdmissionHook(f.f.guard, handoverRecoveryIdentity(t, f.actor), servingRootSocket(t, root))
76 w.AfterAdmission = func(ctx context.Context, prior, admitted hostclaim.Observed) error {
77 p.events = append(p.events, "admit")
78 err := hook(ctx, prior, admitted)
79 if err != nil {
80 old := prior.Claim.Holder
81 receipt, receiptErr := f.f.guard.CompletedColdPredecessor(ctx, prior.Host, hostguard.Captured{Incarnation: old.Incarnation, Stamp: old.Stamp}, old.Session, prior.Claim.Op.ID, prior.Claim.Op.Phase)
82 inventory, invErr := root.Reconcile.Inventory.ReadInstallation(ctx, prior.Host, old.Incarnation, old.Stamp)
83 var st syscall.Stat_t
84 statErr := syscall.Stat(filepath.Dir(f.db), &st)
85 t.Logf("handover diagnosis prior=%+v next=%+v receipt=%+v receiptErr=%v inventory=%+v invErr=%v liveDev=%d liveIno=%d statErr=%v", old, admitted.Claim.Holder, receipt.Target, receiptErr, inventory, invErr, st.Dev, st.Ino, statErr)
86 }
87 return err
88 }
89 w.Prelaunch = func(ctx context.Context, o hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) (*activator.RoleLaunch, error) {
90 p.events = append(p.events, "prelaunch")
91 target := recoveryTuple(o, p.local)
92 err := work.With(ctx, in.Identity, func() error {
93 return f.f.journal.Prelaunch(ctx, hostbox.LaunchInventory{Target: target, ControllerPID: p.local.PID, ProcessStart: p.local.ProcessStart, NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: target.Session}}})
94 })
95 return &activator.RoleLaunch{Target: target, Journal: f.f.journal}, err
96 }
97 roles := &ClaimRoleAssembly{Inventory: f.p.root.Journals.Inventory, ManagedRoots: map[string]string{"data": f.f.guard.DataDir}, Identity: handoverRecoveryIdentity(t, f.actor)}
98 w.Roles = roles
99 roles.Root = fakeLaunchRoot{protocol: hostbox.RoleLaunchProtocol}
100 w.PrepareState = func(ctx context.Context, o hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) error {
101 p.events = append(p.events, "state")
102 return roles.PrepareState(ctx, o, work, in)
103 }
104 w.Launch = func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error) {
105 t.Fatal("tenant launch in restore witness")
106 return nil, hostbox.ErrUnsupported
107 }
108 w.Producers = func(_ context.Context, o hostclaim.Observed) (*ClaimProducers, error) {
109 return producers.Ensure(recoveryTuple(o, p.local), p.local.ProcessStart, f.f.journal.RecordProducer)
110 }
111 w.Retain = func(hostclaim.Observed, activator.Proc, activator.Rep, error) {}
112 w.ObserveLoss = func(hostclaim.Observed, error) {}
113 w.Retries = activator.ServingRetries{Claims: f.f.claims, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: boot, ControllerID: controller, Now: func() time.Time { return f.f.now }, NewID: f.f.newID, Inputs: p.owners.launchDigest}
114 a := ClaimAssembly{Claims: f.f.claims, Launches: f.f.journal, Wake: w, Producers: producers, stopLocal: func(context.Context) (claimStopLocal, error) { return p.local, nil }}
115 a.collectStop = func(ctx context.Context, req StopRequest, drain StopDrainer) (StopResult, error) {
116 f.f.requests = append(f.f.requests, req)
117 coordinator := f.p.root.Stop.Coordinator
118 coordinator.Drainer = drain
119 coordinator.RetainNoProcess = f.p.retainLegacyNoProcess
120 if req.WriterSession != 0 {
121 coordinator.Kernel = f.f.kernel
122 }
123 result, err := coordinator.FenceAndStop(ctx, req)
124 if err != nil {
125 inventory, inventoryErr := coordinator.Inventory.ReadStopInventory(ctx, req.Host)
126 t.Logf("stop diagnosis request=%+v inventory=%+v inventoryErr=%v stopErr=%v", req, inventory, inventoryErr, err)
127 return result, err
128 }
129 return validateTransportStop(req, result)
130 }
131 a.bindClaimWakeStop()
132 retirement := &claimServingArtifactRetirement{Claims: f.f.claims, Guard: f.f.guard, DataDir: f.f.guard.DataDir, Box: f.actor.Box, Region: f.actor.Region, Root: hostbox.ServingParkClient{Call: servingParkSocket(t, servingTransportRoot(t, f))}}
133 w.RetireArtifacts = retirement.Retire
134 w.RecoverArtifacts = func(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
135 p.events = append(p.events, "recover")
136 return retirement.Recover(ctx, o, in)
137 }
138 w.BeforeLegacyInstall = retirement.BeforeLegacyInstall
139 w.BeforeInstallLocked = func(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
140 if err := retirement.EnrollLocked(ctx, o, in); err != nil {
141 return err
142 }
143 p.events = append(p.events, "enrolled")
144 if p.afterEnroll != nil {
145 p.afterEnroll()
146 }
147 return p.enrollFailure
148 }
149 source := hostbox.WorkSource{Guard: f.f.guard, Box: f.actor.Box, Region: f.actor.Region, Now: func() time.Time { return f.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 }}
150 writer := hostbox.Writer{Data: f.f.data, Bucket: f.f.bucket, Claims: f.f.claims, Box: f.actor.Box, Region: f.actor.Region, AdmissionEvidence: p.owners.admissionEvidence}
151 var err error
152 for _, fn := range configure {
153 fn(w)
154 }
155 p.assembly, p.source, p.writer = a, source, writer
156 p.runtime, err = w.Build(t.Context(), source, writer)
157 if err != nil {
158 t.Fatal(err)
159 }
160 t.Cleanup(func() {
161 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
162 defer cancel()
163 if err := w.Close(ctx); err != nil {
164 t.Error(err)
165 }
166 })
167 return p
168}
169
170func (p *servingPreparation) prepare(t *testing.T) (hostclaim.Observed, activator.ClaimStart) {
171 t.Helper()
172 f := p.f
173 rt := registry.Route{Host: f.f.hostName(), DBPath: f.db, Socket: provision.NestedSocket(f.f.hostName())}
174 o := f.f.read()
175 // Prerequisites' isolated launch preflight requires an installed /data
176 // namespace. Drive its selection, stop and real lease in the same order;
177 // Admit and Prepare below are the unmodified production runtime closures.
178 var err error
179 if f.f.guard.DataDir == "/data" {
180 if err = p.runtime.Prerequisites(t.Context(), rt, o, recoveryPolicy); err != nil {
181 t.Fatal("prerequisites", err)
182 }
183 p.events = append(p.events, "prerequisites")
184 pending := p.wake.get(rt.Host)
185 if pending == nil || pending.held == nil {
186 t.Fatal("prerequisites returned without a lease")
187 }
188 _, selected := pending.snapshot()
189 p.baseline = selected.Baseline
190 } else {
191 selected, err := p.owners.Select(t.Context(), rt, o, recoveryPolicy)
192 if err != nil {
193 t.Fatal(err)
194 }
195 p.events = append(p.events, "select")
196 if selected.Empty || selected.Baseline == nil {
197 t.Fatal("baseline restore replaced by empty install")
198 }
199 p.baseline = selected.Baseline
200 if err = p.wake.stopPredecessor(t.Context(), o); err != nil {
201 t.Fatal("stop predecessor", err)
202 }
203 p.events = append(p.events, "stop")
204 held, err := lease.Acquire(t.Context(), f.f.bucket, o.Claim.Holder.Location, p.wake.LeaseRegion)
205 if err != nil {
206 t.Fatal(err)
207 }
208 p.events = append(p.events, "lease")
209 live, cancel := context.WithCancel(t.Context())
210 done := make(chan struct{})
211 go func() { <-live.Done(); close(done) }()
212 p.wake.pending[rt.Host] = &claimWakeAttempt{owner: p.wake, observed: o, inputs: selected, held: held, ctx: live, cancel: cancel, done: done}
213 }
214 admitted, err := p.runtime.Admit(t.Context(), rt, f.f.read(), recoveryPolicy)
215 if err != nil {
216 t.Fatal("admit", err)
217 }
218 r := admitted.Claim.Holder.WakeRetry
219 if r == nil || r.Attempt == nil || r.Attempt.Session != admitted.Claim.Holder.Session || r.Attempt.ControllerID != p.local.ControllerID {
220 t.Fatalf("retry not bound: %+v", r)
221 }
222 start, err := p.runtime.Prepare(t.Context(), rt, admitted, recoveryPolicy)
223 if err != nil {
224 t.Fatal("prepare", err)
225 }
226 return admitted, start
227}
228
229func TestFilesSeedOrdinaryPreparationRestore(t *testing.T) {
230 for _, failure := range []string{"", "enrolled-refusal", "missing-marker", "cancel-after-enrollment", "disk-after-enrollment", "invalid-sqlite", "cancel-after-output", "restore-checksum"} {
231 for _, restart := range []string{"same-controller", "controller", "reboot"} {
232 t.Run(failure+"/"+restart, func(t *testing.T) { runFilesSeedOrdinaryPreparation(t, failure, restart) })
233 }
234 }
235}
236
237func runFilesSeedOrdinaryPreparation(t *testing.T, failure, restart string, mounted ...bool) {
238 runFilesSeedPreparationWithLive(t, failure, restart, mounted, nil)
239}
240func runFilesSeedPreparationWithLive(t *testing.T, failure, restart string, mounted []bool, live []filesSeedLiveWitness) {
241 if failure == "real-seed" {
242 t.Setenv("CARLOS_HOST_CLAIM", "1")
243 t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
244 }
245 t.Setenv("CARLOS_HOST_CLAIM", "1")
246 t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
247 t.Setenv("CARLOS_MOVES_PHASE1", "")
248 var f *servingRootFixture
249 if len(mounted) > 0 && mounted[0] {
250 shared := false
251 for _, witness := range live {
252 shared = shared || witness.shared
253 }
254 if !shared {
255 t.Cleanup(func() {
256 if err := os.RemoveAll("/data"); err != nil {
257 t.Error(err)
258 }
259 })
260 t.Cleanup(func() {
261 if err := os.RemoveAll("/var/lib/carlos-claims"); err != nil {
262 t.Error(err)
263 }
264 })
265 }
266 if failure == "restore-enospc" {
267 if err := os.Mkdir("/data", 0700); err != nil {
268 t.Fatal(err)
269 }
270 if err := unix.Mount("tmpfs", "/data", "tmpfs", unix.MS_NOSUID|unix.MS_NODEV, "size=2g"); err != nil {
271 t.Fatal(err)
272 }
273 t.Cleanup(func() {
274 if err := unix.Unmount("/data", 0); err != nil {
275 t.Error(err)
276 }
277 })
278 }
279 state, err := openClaimServingParkAuthority()
280 if err != nil {
281 t.Fatal("production private store", err)
282 }
283 t.Cleanup(func() {
284 if err := state.Close(); err != nil {
285 t.Error(err)
286 }
287 })
288 host := recoveryHost
289 for _, witness := range live {
290 if witness.host != "" {
291 host = witness.host
292 }
293 }
294 f, _, _ = newServingLegacyRootSetup(t, false, "/data", host)
295 if !shared {
296 instances := filepath.Join(f.f.guard.DataDir, "instances")
297 if err := unix.Mount(instances, instances, "", unix.MS_BIND, ""); err != nil {
298 t.Fatal(err)
299 }
300 t.Cleanup(func() {
301 if err := unix.Unmount(instances, 0); err != nil {
302 t.Error(err)
303 }
304 })
305 if err := unix.Mount("", instances, "", unix.MS_BIND|unix.MS_REMOUNT|unix.MS_NOSUID|unix.MS_NODEV|unix.MS_NOEXEC, ""); err != nil {
306 t.Fatal(err)
307 }
308 if parent, _, err := newServingParkFS(f.f.guard.DataDir, state).parent(); err == nil {
309 parent.Close()
310 t.Fatal("production filesystem accepted a mount without nosymfollow")
311 }
312 if err := unix.Mount("", instances, "", unix.MS_BIND|unix.MS_REMOUNT|unix.MS_NOSYMFOLLOW|unix.MS_NOSUID|unix.MS_NODEV|unix.MS_NOEXEC, ""); err != nil {
313 t.Fatal(err)
314 }
315 link := filepath.Join(instances, ".nosymfollow-witness")
316 if err := os.Symlink(f.db, link); err != nil {
317 t.Fatal(err)
318 }
319 fd, err := unix.Open(link, unix.O_RDONLY|unix.O_CLOEXEC, 0)
320 if fd >= 0 {
321 unix.Close(fd)
322 }
323 if !errors.Is(err, syscall.ELOOP) {
324 t.Fatalf("kernel followed symlink on nosymfollow mount: %v", err)
325 }
326 if err := os.Remove(link); err != nil {
327 t.Fatal(err)
328 }
329 }
330 // Use the production UID and mount checks in this isolated namespace.
331 f.p.ledger.state = state
332 f.p.ledger.fs = newServingParkFS(f.f.guard.DataDir, state)
333 source, err := os.Open(filepath.Dir(f.db))
334 if err != nil {
335 t.Fatal(err)
336 }
337 sourceID, err := servingParkDirID(source)
338 source.Close()
339 if err != nil {
340 t.Fatal(err)
341 }
342 outside, err := os.Open(f.f.guard.DataDir)
343 if err != nil {
344 t.Fatal(err)
345 }
346 outsideID, err := servingParkDirID(outside)
347 outside.Close()
348 if err != nil {
349 t.Fatal(err)
350 }
351 if sourceID.Device != outsideID.Device || sourceID.Mount == outsideID.Mount {
352 t.Fatal("fixture did not create same-device cross-mount boundary")
353 }
354 if err := os.Rename(f.db, filepath.Join(f.f.guard.DataDir, "old-flat-park")); !errors.Is(err, syscall.EXDEV) {
355 t.Fatalf("old destination must fail with EXDEV: %v", err)
356 }
357 } else {
358 f, _, _ = newServingLegacyRootSetup(t, false)
359 }
360 extraBytes := 0
361 if failure == "cancel-after-output" {
362 extraBytes = 2 << 20
363 }
364 selected := plan12RestoreBaseline(t, f.f, extraBytes)
365 raw, err := f.f.data.GetVersion(t.Context(), f.f.bucket, selected.Files[0].Key, selected.Files[0].VersionID)
366 if err != nil {
367 t.Fatal(err)
368 }
369 bucket := f.f.data.(*leasetest.VersionedBucket)
370 key := strings.Replace(selected.Files[0].Key, "/g-src/", "/", 1)
371 bucket.Put(f.f.bucket, key, []byte("older invalid snapshot"))
372 version := bucket.Put(f.f.bucket, key, raw)
373 // Keep a valid LTX envelope so failures occur inside the production restore,
374 // after legacy retirement, rather than at selected-object decoding.
375 if failure == "invalid-sqlite" {
376 var spec ltx.FileSpec
377 if _, err := spec.ReadFrom(bytes.NewReader(raw)); err != nil {
378 t.Fatal(err)
379 }
380 copy(spec.Pages[0].Data, []byte("not a sqlite db!"))
381 var encoded bytes.Buffer
382 if _, err := spec.WriteTo(&encoded); err != nil {
383 t.Fatal(err)
384 }
385 version = bucket.Put(f.f.bucket, key, encoded.Bytes())
386 }
387 var deltaKey, deltaVersion string
388 var deltaGood []byte
389 if failure == "restore-checksum" {
390 var spec ltx.FileSpec
391 if _, err := spec.ReadFrom(bytes.NewReader(raw)); err != nil {
392 t.Fatal(err)
393 }
394 checksum := ltx.ChecksumFlag
395 for _, page := range spec.Pages {
396 checksum ^= ltx.ChecksumPage(page.Header.Pgno, page.Data)
397 }
398 checksum |= ltx.ChecksumFlag
399 // Delta trailers cannot validate the resulting database without applying
400 // the snapshot. Keep the envelope valid and fail only the final check.
401 spec.Header.Flags = 0
402 spec.Trailer.PostApplyChecksum = checksum
403 var snapshot bytes.Buffer
404 if _, err := spec.WriteTo(&snapshot); err != nil {
405 t.Fatal(err)
406 }
407 raw = append([]byte(nil), snapshot.Bytes()...)
408 version = bucket.Put(f.f.bucket, key, raw)
409 spec.Header.MinTXID, spec.Header.MaxTXID = 2, 2
410 spec.Header.PreApplyChecksum = checksum
411 var encoded bytes.Buffer
412 if _, err := spec.WriteTo(&encoded); err != nil {
413 t.Fatal(err)
414 }
415 deltaGood = append([]byte(nil), encoded.Bytes()...)
416 encoded.Reset()
417 spec.Trailer.PostApplyChecksum ^= 1
418 if _, err := spec.WriteTo(&encoded); err != nil {
419 t.Fatal(err)
420 }
421 deltaKey = filepath.Dir(filepath.Dir(key)) + "/0000/" + ltx.FormatFilename(2, 2)
422 deltaVersion = bucket.Put(f.f.bucket, deltaKey, encoded.Bytes())
423 }
424 handler, _ := leasetest.NewS3Server(bucket)
425 server := httptest.NewServer(handler)
426 t.Cleanup(server.Close)
427 for key, value := range map[string]string{"AWS_ACCESS_KEY_ID": "test-key", "AWS_SECRET_ACCESS_KEY": "test-secret", "AWS_SESSION_TOKEN": "", "AWS_EC2_METADATA_DISABLED": "true", "AWS_CONFIG_FILE": filepath.Join(t.TempDir(), "absent"), "AWS_SHARED_CREDENTIALS_FILE": filepath.Join(t.TempDir(), "absent"), "CARLOS_S3_ENDPOINT": server.URL} {
428 t.Setenv(key, value)
429 }
430 sentinels := map[string]string{filepath.Join(filepath.Dir(f.db), "files", "payload"): "live files", filepath.Join(filepath.Dir(f.db), ".env"): "environment"}
431 for path, body := range sentinels {
432 if err = os.MkdirAll(filepath.Dir(path), 0700); err != nil {
433 t.Fatal(err)
434 }
435 if err = os.WriteFile(path, []byte(body), 0600); err != nil {
436 t.Fatal(err)
437 }
438 }
439 if failure == "real-seed" {
440 f.actor.Bucket = f.f.bucket
441 f.actor.Cloud.Region = "eu-west-1"
442 f.actor.HomeRegion = "ie"
443 f.actor.FleetHomeRegion = "ie"
444 }
445 p := newServingPreparation(t, f, "boot", "controller")
446 if failure == "real-seed" {
447 runFilesSeedRealPreparation(t, p, live...)
448 return
449 }
450 var legacy servingParkRecord
451 var baseline *genselect.Plan
452 for i := 0; i < 2; i++ {
453 if i == 1 {
454 if failure == "invalid-sqlite" {
455 version = bucket.Put(f.f.bucket, key, raw)
456 }
457 if failure == "restore-checksum" {
458 deltaVersion = bucket.Put(f.f.bucket, deltaKey, deltaGood)
459 }
460 f.f.now = f.f.now.Add(time.Minute)
461 }
462 p.events = nil
463 admitted, start := p.prepare(t)
464 wantFiles := 1
465 if failure == "restore-checksum" {
466 wantFiles = 2
467 }
468 if len(p.baseline.Files) != wantFiles || p.baseline.Files[0].Key != key || p.baseline.Files[0].VersionID != version {
469 t.Fatalf("wrong exact baseline: %+v", p.baseline)
470 }
471 if failure == "restore-checksum" && (p.baseline.Files[1].Key != deltaKey || p.baseline.Files[1].VersionID != deltaVersion) {
472 t.Fatalf("wrong exact delta: %+v", p.baseline)
473 }
474 if baseline != nil && failure != "restore-checksum" && failure != "invalid-sqlite" && !reflect.DeepEqual(baseline, p.baseline) {
475 t.Fatal("baseline changed")
476 }
477 baseline = p.baseline
478 p.enrollFailure = nil
479 cancelCtx, cancel := context.WithCancel(t.Context())
480 var restoreCtx context.Context = cancelCtx
481 var outputCancel *servingRestoreOutputCancel
482 if i == 0 && failure == "cancel-after-output" {
483 outputCancel = &servingRestoreOutputCancel{Context: cancelCtx, cancel: cancel, directory: filepath.Dir(f.db)}
484 restoreCtx = outputCancel
485 }
486 p.afterEnroll = nil
487 if i == 0 {
488 switch failure {
489 case "enrolled-refusal":
490 p.enrollFailure = os.ErrPermission
491 case "disk-after-enrollment":
492 p.enrollFailure = syscall.ENOSPC
493 case "cancel-after-enrollment":
494 p.afterEnroll = cancel
495 case "restore-enospc":
496 p.afterEnroll = func() {
497 if err := unix.Mount("tmpfs", "/data", "tmpfs", unix.MS_REMOUNT|unix.MS_NOSUID|unix.MS_NODEV, "size=1m"); err != nil {
498 t.Fatal(err)
499 }
500 err := os.WriteFile(filepath.Join(filepath.Dir(f.db), ".disk-full-witness"), make([]byte, 1<<20), 0600)
501 if !errors.Is(err, syscall.ENOSPC) {
502 t.Fatalf("small private tmpfs did not fill: %v", err)
503 }
504 }
505 }
506 }
507 restoreErr := start.Restore(restoreCtx)
508 if i == 0 && failure == "restore-enospc" {
509 if !errors.Is(restoreErr, syscall.ENOSPC) || !strings.Contains(restoreErr.Error(), "restore selected generation") {
510 t.Fatalf("not a production restore ENOSPC: %v", restoreErr)
511 }
512 if err := os.Remove(filepath.Join(filepath.Dir(f.db), ".disk-full-witness")); err != nil {
513 t.Fatal(err)
514 }
515 if err := unix.Mount("tmpfs", "/data", "tmpfs", unix.MS_REMOUNT|unix.MS_NOSUID|unix.MS_NODEV, "size=2g"); err != nil {
516 t.Fatal(err)
517 }
518 }
519 if outputCancel != nil && !outputCancel.observed.Load() {
520 t.Fatalf("cancellation never observed real restore output: %v", restoreErr)
521 }
522 cancel()
523 failed := i == 0 && failure != "" && failure != "missing-marker"
524 if failed {
525 if restoreErr == nil {
526 t.Fatal("missed restore failure")
527 }
528 if p.enrollFailure != nil && !errors.Is(restoreErr, p.enrollFailure) {
529 t.Fatalf("wrong failure: %v", restoreErr)
530 }
531 if (failure == "cancel-after-enrollment" || failure == "cancel-after-output") && !errors.Is(restoreErr, context.Canceled) {
532 t.Fatalf("wrong cancellation: %v", restoreErr)
533 }
534 if failure == "invalid-sqlite" && !strings.Contains(restoreErr.Error(), "check restored database") {
535 t.Fatalf("failure before database check: %v", restoreErr)
536 }
537 if failure == "restore-checksum" && !strings.Contains(restoreErr.Error(), "restored database checksum") {
538 t.Fatalf("failure before restored checksum: %v", restoreErr)
539 }
540 if _, err = os.Stat(f.db); !os.IsNotExist(err) {
541 t.Fatalf("failed restore left database: %v", err)
542 }
543 temps, err := filepath.Glob(filepath.Join(filepath.Dir(f.db), ".restore-*"))
544 if err != nil || len(temps) != 0 {
545 t.Fatalf("restore scratch survived: %v %v", temps, err)
546 }
547 } else {
548 if restoreErr != nil {
549 t.Fatal("restore", restoreErr)
550 }
551 if got := mwReadDB(t, f.db); got != "selected recovery baseline" {
552 t.Fatal(got)
553 }
554 marker, err := hostbox.ReadGenerationMarker(f.db + ".gen")
555 if err != nil || marker != admitted.Claim.Holder.Generation {
556 t.Fatalf("marker %q %v", marker, err)
557 }
558 if i == 0 && failure == "missing-marker" {
559 if err = os.Remove(f.db + ".gen"); err != nil {
560 t.Fatal(err)
561 }
562 }
563 }
564 want := []string{"select", "stop", "lease", "admit", "prelaunch", "recover", "state", "enrolled"}
565 if f.f.guard.DataDir == "/data" {
566 want = append([]string{"prerequisites"}, want[3:]...)
567 }
568 if !reflect.DeepEqual(p.events, want) {
569 t.Fatalf("preparation order: %v", p.events)
570 }
571 current, err := servingWriterIdentity(f.f.read(), f.actor.Box, f.actor.Region)
572 if err != nil {
573 t.Fatal(err)
574 }
575 slot, err := f.p.ledger.slot(servingParkAuthorization{Successor: current})
576 if err != nil {
577 t.Fatal(err)
578 }
579 record, err := f.p.ledger.read(slot.ID)
580 if err != nil || !record.Complete || !record.Authorization.Legacy {
581 t.Fatalf("legacy slot: %+v %v", record, err)
582 }
583 if i == 0 {
584 legacy = record
585 } else if !reflect.DeepEqual(legacy, record) {
586 t.Fatal("never-launched restore displaced legacy park or changed deadline")
587 }
588 if i == 0 && restart != "same-controller" {
589 // Lose only the controller's lease and memory. Leave its retry and
590 // prelaunch evidence unsettled for the next controller's stop path.
591 attempt := p.wake.get(recoveryHost)
592 if err = p.wake.release(t.Context(), recoveryHost, attempt); err != nil {
593 t.Fatal(err)
594 }
595 // This is ordinary retry recovery, not Task 1R's future seed rerun.
596 nextBoot := "boot"
597 if restart == "reboot" {
598 nextBoot = "boot-2"
599 }
600 next := newServingPreparation(t, f, nextBoot, "controller-2")
601 err = next.runtime.Prerequisites(t.Context(), registry.Route{Host: f.f.hostName(), DBPath: f.db}, f.f.read(), recoveryPolicy)
602 if !errors.Is(err, hostbox.ErrChanged) {
603 t.Fatalf("retry settlement: %v", err)
604 }
605 if r := f.f.retry(); r.State != "open" || r.Attempt != nil || r.AttemptsReserved != 1 {
606 t.Fatalf("settlement refunded/stranded: %+v", r)
607 }
608 p = next
609 } else {
610 if err = p.runtime.FailedAdmission(t.Context(), admitted, errors.New("restore-only witness")); err != nil {
611 t.Fatal("settlement", err)
612 }
613 }
614 }
615 for suffix, want := range map[string]string{"": "legacy bytes", "-wal": "legacy WAL", "-shm": "legacy SHM"} {
616 body, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", legacy.Authorization.ID, filepath.Base(f.db)+suffix))
617 if err != nil || string(body) != want {
618 t.Fatalf("park bytes %q %v", body, err)
619 }
620 }
621 for path, want := range sentinels {
622 body, err := os.ReadFile(path)
623 if err != nil || string(body) != want {
624 t.Fatalf("unrelated bytes: %s %v", path, err)
625 }
626 }
627 after, err := f.f.data.GetVersion(t.Context(), f.f.bucket, key, version)
628 if err != nil || string(after) != string(raw) {
629 t.Fatal("remote baseline changed", err)
630 }
631 if r := f.f.retry(); r.State != "open" || r.Attempt != nil || r.AttemptsReserved != 2 {
632 t.Fatalf("final retry: %+v", r)
633 }
634 if restart != "same-controller" && len(f.f.kernel.recovered) == 0 {
635 t.Fatal("restart did not recover original launch tuple")
636 }
637}
638
639func TestFilesSeedLegacyStopRefusesForeignWriterBeforeRoot(t *testing.T) {
640 f, _, _ := newServingLegacyRootSetup(t, false)
641 o, err := hostclaim.Update(t.Context(), f.f.claims, f.f.read(), func(c *hostclaim.Claim) error { c.Holder.WriterBox = "foreign-box"; return nil })
642 if err != nil {
643 t.Fatal(err)
644 }
645 calls := 0
646 a := ClaimAssembly{Claims: f.f.claims, collectStop: func(context.Context, StopRequest, StopDrainer) (StopResult, error) { calls++; return StopResult{}, nil }}
647 local := claimStopLocal{Actor: f.actor, BootID: "boot", ControllerID: "controller", PID: 77, ProcessStart: "start"}
648 before, _ := f.disk.Names()
649 _, err = a.collectCapturedStop(t.Context(), o, local, claimLaunchOwner{BootID: local.BootID, ControllerID: local.ControllerID}, false)
650 if !errors.Is(err, hostbox.ErrChanged) || calls != 0 {
651 t.Fatalf("foreign legacy writer: calls %d error %v", calls, err)
652 }
653 after, _ := f.disk.Names()
654 if !reflect.DeepEqual(before, after) {
655 t.Fatal("foreign legacy stop wrote private evidence")
656 }
657}
658
659func TestServingParkNestedBindMount(t *testing.T) {
660 if !plan14RootFixture(t) {
661 return
662 }
663 for _, failure := range []string{"", "enrolled-refusal", "missing-marker", "cancel-after-enrollment", "disk-after-enrollment", "invalid-sqlite", "cancel-after-output", "restore-checksum", "restore-enospc"} {
664 for _, restart := range []string{"same-controller", "controller", "reboot"} {
665 t.Run(failure+"/"+restart, func(t *testing.T) { runFilesSeedOrdinaryPreparation(t, failure, restart, true) })
666 }
667 }
668}
669
670// Cancel only once the real restore has written a temporary database. This
671// context seam injects a fault without replacing Restore or its filesystem.
672type servingRestoreOutputCancel struct {
673 context.Context
674 cancel context.CancelFunc
675 directory string
676 observed atomic.Bool
677}
678
679func (c *servingRestoreOutputCancel) Err() error {
680 if !c.observed.Load() {
681 names, _ := filepath.Glob(filepath.Join(c.directory, ".restore-*"))
682 for _, name := range names {
683 if !strings.HasSuffix(name, ".db") && !strings.HasSuffix(name, ".db.tmp") {
684 continue
685 }
686 if info, err := os.Stat(name); err == nil && info.Mode().IsRegular() && info.Size() > 0 {
687 c.observed.Store(true)
688 c.cancel()
689 break
690 }
691 }
692 }
693 return c.Context.Err()
694}
695
696func (c *servingRestoreOutputCancel) Done() <-chan struct{} {
697 _ = c.Err()
698 return c.Context.Done()
699}
700