amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package activator
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "os"
8 "os/exec"
9 "path/filepath"
10 "reflect"
11 "runtime"
12 "strings"
13 "sync"
14 "sync/atomic"
15 "testing"
16 "time"
17
18 _ "modernc.org/sqlite"
19
20 "github.com/carlosframework/platform/internal/cgroup"
21 "github.com/carlosframework/platform/internal/edge"
22 "github.com/carlosframework/platform/internal/lease"
23 "github.com/carlosframework/platform/internal/registry"
24)
25
26// TestExecBackendTierSeam: a nil Tier means the default tier, and a set
27// Tier is consulted with the route. The cgroup manager is nil here (no
28// delegation in a test), so this pins the seam's contract, not the write.
29func TestExecBackendTierSeam(t *testing.T) {
30 var b ExecBackend
31 rt := registry.Route{Host: "h.example.com"}
32 if got := b.tier(rt); got != cgroup.TierDefault {
33 t.Fatalf("nil Tier: %v, want TierDefault", got)
34 }
35 b.Tier = func(registry.Route) cgroup.Limit { return cgroup.TierChildren }
36 if got := b.tier(rt); got != cgroup.TierChildren {
37 t.Fatalf("set Tier: %v, want TierChildren", got)
38 }
39}
40
41// TestSpawnInCgroupAppliesTheTier: the seam every spawn site uses really
42// writes the route's tier into the host cgroup — a regression that swapped
43// a site back to a bare Cgroups.Start(…, cgroup.Limit{}, …) compiles fine
44// and this is what catches the class. Real cgroup.SetupAt against a
45// temp-dir fixture; the spawn is /bin/true.
46func TestSpawnInCgroupAppliesTheTier(t *testing.T) {
47 root := t.TempDir()
48 for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
49 if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
50 t.Fatal(err)
51 }
52 }
53 m, err := cgroup.SetupAt(root, t.Logf)
54 if err != nil {
55 t.Fatal(err)
56 }
57 b := ExecBackend{Cgroups: m, Tier: func(registry.Route) cgroup.Limit { return cgroup.TierChildren }}
58 rt := registry.Route{Host: "h.example.com"}
59 cmd, _, err := b.spawnInCgroup(rt, func() *exec.Cmd { return exec.Command("/bin/true") })
60 if err != nil {
61 t.Fatal(err)
62 }
63 _ = cmd.Wait()
64 got, err := os.ReadFile(filepath.Join(root, "carlos-exec@h.example.com", "memory.max"))
65 if err != nil {
66 t.Fatalf("memory.max not written: %v", err)
67 }
68 if string(got) != "1073741824" {
69 t.Errorf("memory.max = %q, want the children tier", got)
70 }
71}
72
73// The box's placement code reaches every exec child as CARLOS_REGION (issue
74// #329) — agent-owned in both directions: set from ExecBackend.Region, and
75// STRIPPED when the backend has none, even when a tenant's own env file
76// tries to supply it. Same rule and same reason as CARLOS_CONTROL_SOCKET /
77// CARLOS_PURGE: an app must not be told it runs somewhere the agent never
78// vouched for.
79func TestExecBackendStartChildRegion(t *testing.T) {
80 for _, tc := range []struct {
81 name, backendRegion, want string
82 wantSet bool
83 }{
84 {"set from the backend", "uk", "uk", true},
85 {"stripped when the backend has none", "", "", false},
86 } {
87 t.Run(tc.name, func(t *testing.T) {
88 b := testExecBackend(t)
89 b.Region = tc.backendRegion
90 host := "region-env.example.com"
91 _, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")
92 // A tenant-supplied value that must never survive as-is.
93 if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte("CARLOS_REGION=forged\n"), 0o600); err != nil {
94 t.Fatal(err)
95 }
96 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "instance.sock")}
97 ctx, cancel := recordedCtx(2*time.Second, envPath)
98 defer cancel()
99 proc, _, _ := b.Start(ctx, rt) // fake binary never binds the socket; health-wait error expected
100 if proc != nil {
101 defer proc.Stop(context.Background())
102 }
103 child := map[string]string{}
104 for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
105 if k, v, ok := strings.Cut(line, "="); ok {
106 child[k] = v
107 }
108 }
109 got, ok := child["CARLOS_REGION"]
110 if ok != tc.wantSet || got != tc.want {
111 t.Fatalf("child CARLOS_REGION = %q (set=%v), want %q (set=%v)", got, ok, tc.want, tc.wantSet)
112 }
113 })
114 }
115}
116
117// Every exec child is told which host it is, in its environment, from the
118// row (discussion #256): an app that used to read its hostname off its
119// socket's file name has nothing to read nested (app.sock). The variable
120// is agent-owned: a tenant's own env file cannot make an app another host.
121func TestExecBackendStartChildHandsTheHost(t *testing.T) {
122 b := testExecBackend(t)
123 host := "host-env.example.com"
124 _, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")
125 if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte("CARLOS_HOST=forged.example.com\n"), 0o600); err != nil {
126 t.Fatal(err)
127 }
128 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "instance.sock")}
129 ctx, cancel := recordedCtx(2*time.Second, envPath)
130 defer cancel()
131 proc, _, _ := b.Start(ctx, rt) // fake binary never binds the socket; health-wait error expected
132 if proc != nil {
133 defer proc.Stop(context.Background())
134 }
135 child := map[string]string{}
136 for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
137 if k, v, ok := strings.Cut(line, "="); ok {
138 child[k] = v
139 }
140 }
141 if got := child["CARLOS_HOST"]; got != host {
142 t.Fatalf("child CARLOS_HOST = %q, want %q", got, host)
143 }
144}
145
146// The staging fallback (issue #331): a host whose ordinary prefix is empty
147// restores from a move's staged copy, promoted server-side and dropped
148// after a clean restore. A staged copy present beside an own lineage wins:
149// it is promoted over the ordinary prefix (the boomerang case below;
150// ruling R-STAGEDREAD-2).
151func TestRestoreStagingFallback(t *testing.T) {
152 ctx := context.Background()
153 b := testExecBackend(t)
154 f := newFakeMirrorStore()
155 b.Mirror = f
156
157 host := "moved.example.com"
158 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
159
160 // restoreReplica is stubbed for the WHOLE test (the real one dials S3):
161 // it restores — materializing a genuinely valid SQLite file, so CheckDB
162 // passes — exactly when the fake bucket holds objects under the prefix
163 // it was pointed at, which is the promotion's observable effect.
164 var restoredFrom []string
165 oldRestore := restoreReplica
166 restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) {
167 objs, _ := f.List(context.Background(), bucket, prefix)
168 if len(objs) == 0 {
169 return false, nil
170 }
171 restoredFrom = append(restoredFrom, bucket+"/"+prefix)
172 return true, writeMinimalSQLite(dbPath)
173 }
174 defer func() { restoreReplica = oldRestore }()
175
176 // Nothing staged, nothing own: plain fresh boot (restored=false→fresh=true).
177 fresh, err := b.Restore(ctx, rt)
178 if err != nil || !fresh {
179 t.Fatalf("empty everything: fresh=%v err=%v, want fresh boot", fresh, err)
180 }
181
182 // Staged copy present, own prefix empty: promoted then restored.
183 f.put(b.Cfg.Bucket, "moves/"+host+"/generations/a/snapshot", "e1")
184 _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
185 fresh, err = b.Restore(ctx, rt)
186 if err != nil {
187 t.Fatalf("staged restore: %v", err)
188 }
189 if fresh {
190 t.Fatal("staged restore reported a fresh boot")
191 }
192 if len(restoredFrom) != 1 || restoredFrom[0] != b.Cfg.Bucket+"/"+b.Cfg.hibernatePrefix(host) {
193 t.Fatalf("restored from %v, want the promoted ordinary prefix %s", restoredFrom, b.Cfg.hibernatePrefix(host))
194 }
195 own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
196 if len(own) != 1 {
197 t.Fatalf("promotion left own prefix = %v", own)
198 }
199 staged, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
200 if len(staged) != 0 {
201 t.Fatalf("staging not dropped after a clean restore: %v", staged)
202 }
203
204 // STAGING WINS (review round 1): a fresh staged copy beside a stale
205 // own lineage is the boomerang landing — the staged bytes replace the
206 // own prefix and are what gets restored.
207 f.put(b.Cfg.Bucket, "moves/"+host+"/generations/b/newer-snapshot", "e9")
208 _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
209 restoredFrom = nil
210 if _, err := b.Restore(ctx, rt); err != nil {
211 t.Fatal(err)
212 }
213 own, _ = f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
214 if len(own) != 1 {
215 t.Fatalf("own prefix after boomerang promote = %v, want exactly the staged copy", own)
216 }
217 for k := range own {
218 if !strings.Contains(k, "newer-snapshot") {
219 t.Fatalf("own prefix kept the stale lineage: %v", own)
220 }
221 }
222 staged, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
223 if len(staged) != 0 {
224 t.Fatalf("staging not dropped after the boomerang restore: %v", staged)
225 }
226
227 // A staging-fallback ERROR fails the wake — never a fresh boot + seed
228 // that would shadow the staged data forever.
229 f.put(b.Cfg.Bucket, "moves/"+host+"/generations/c/snapshot", "eA")
230 f.failList = true
231 _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
232 if _, err := b.Restore(ctx, rt); err == nil {
233 t.Fatal("List failure in the staging fallback did not fail the restore")
234 }
235 f.failList = false
236}
237
238// A move's copy receipts live in the SOURCE box's own bucket at
239// MoveReceiptsKey, and this box may be that source for an earlier move of
240// the same host (a move back). Landing must never take them for data:
241// they are not under MovesPrefix, so they are never promoted, and not
242// under the hibernate prefix, so they are never restored from or swept.
243func TestRestoreIgnoresMoveReceipts(t *testing.T) {
244 ctx := context.Background()
245 b := testExecBackend(t)
246 f := newFakeMirrorStore()
247 b.Mirror = f
248 host := "moved.example.com"
249 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
250
251 receipts := MoveReceiptsKey(host)
252 if strings.HasPrefix(receipts, MovesPrefix(host)) || strings.HasPrefix(receipts, b.Cfg.hibernatePrefix(host)+"/") ||
253 strings.HasPrefix(receipts, b.Cfg.Prefix+"/") {
254 t.Fatalf("receipts key %q sits under a data prefix", receipts)
255 }
256
257 var restoredFrom []string
258 oldRestore := restoreReplica
259 restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) {
260 objs, _ := f.List(context.Background(), bucket, prefix)
261 if len(objs) == 0 {
262 return false, nil
263 }
264 restoredFrom = append(restoredFrom, bucket+"/"+prefix)
265 return true, writeMinimalSQLite(dbPath)
266 }
267 defer func() { restoreReplica = oldRestore }()
268
269 // Receipts alone: a fresh boot, nothing promoted, receipts untouched.
270 f.put(b.Cfg.Bucket, receipts, "r1")
271 fresh, err := b.Restore(ctx, rt)
272 if err != nil || !fresh || len(restoredFrom) != 0 {
273 t.Fatalf("receipts alone: fresh=%v err=%v restored=%v, want a fresh boot", fresh, err, restoredFrom)
274 }
275 // Beside a staged copy: only the staged data is promoted.
276 f.put(b.Cfg.Bucket, "moves/"+host+"/ltx/0/a.ltx", "e1")
277 _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
278 if _, err := b.Restore(ctx, rt); err != nil {
279 t.Fatal(err)
280 }
281 own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
282 if len(own) != 1 {
283 t.Fatalf("own prefix after promotion = %v, want only the staged object", own)
284 }
285 if _, ok := f.buckets[b.Cfg.Bucket][receipts]; !ok {
286 t.Fatal("landing deleted the receipts of a move it did not make")
287 }
288}
289
290// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→object.
291// A mirror round copies and deletes side by side, so every method holds mu.
292type fakeMirrorStore struct {
293 mu sync.Mutex
294 buckets map[string]map[string]lease.MirrorObject
295 failList bool
296}
297
298func newFakeMirrorStore() *fakeMirrorStore {
299 return &fakeMirrorStore{buckets: map[string]map[string]lease.MirrorObject{}}
300}
301func (f *fakeMirrorStore) put(bucket, key, etag string) {
302 f.mu.Lock()
303 defer f.mu.Unlock()
304 f.putLocked(bucket, key, etag)
305}
306func (f *fakeMirrorStore) putLocked(bucket, key, etag string) {
307 if f.buckets[bucket] == nil {
308 f.buckets[bucket] = map[string]lease.MirrorObject{}
309 }
310 f.buckets[bucket][key] = lease.MirrorObject{ETag: etag}
311}
312func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) {
313 f.mu.Lock()
314 defer f.mu.Unlock()
315 if f.failList {
316 return nil, context.DeadlineExceeded
317 }
318 out := map[string]lease.MirrorObject{}
319 for k, v := range f.buckets[bucket] {
320 if strings.HasPrefix(k, prefix) {
321 out[k] = v
322 }
323 }
324 return out, nil
325}
326func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
327 f.mu.Lock()
328 defer f.mu.Unlock()
329 o, ok := f.buckets[srcBucket][srcKey]
330 if !ok || (srcETag != "" && o.ETag != srcETag) {
331 return "", lease.ErrCopySourceChanged
332 }
333 if f.buckets[dstBucket] == nil {
334 f.buckets[dstBucket] = map[string]lease.MirrorObject{}
335 }
336 f.buckets[dstBucket][dstKey] = o
337 return o.ETag, nil
338}
339func (f *fakeMirrorStore) Get(_ context.Context, bucket, key string) ([]byte, error) {
340 f.mu.Lock()
341 defer f.mu.Unlock()
342 if _, ok := f.buckets[bucket][key]; !ok {
343 return nil, lease.ErrMirrorNotFound
344 }
345 return []byte("{}"), nil
346}
347func (f *fakeMirrorStore) Put(_ context.Context, bucket, key string, _ []byte) error {
348 f.mu.Lock()
349 defer f.mu.Unlock()
350 f.putLocked(bucket, key, "receipts")
351 return nil
352}
353func (f *fakeMirrorStore) Delete(_ context.Context, bucket, key string) error {
354 f.mu.Lock()
355 defer f.mu.Unlock()
356 delete(f.buckets[bucket], key)
357 return nil
358}
359
360// writeMinimalSQLite creates a genuinely valid empty SQLite database so
361// lease.CheckDB passes on the stubbed restore's output.
362func writeMinimalSQLite(dbPath string) error {
363 db, err := sql.Open("sqlite", dbPath)
364 if err != nil {
365 return err
366 }
367 if _, err := db.Exec("create table t(x)"); err != nil {
368 return err
369 }
370 return db.Close()
371}
372
373func TestLaunchEnvironmentComposesTheSpawnEnvironment(t *testing.T) {
374 b := &ExecBackend{DataDir: t.TempDir(), Region: "ie"} // Cache nil: no purge channel on this box
375 rt := registry.Route{Host: "a.go.example.com", Socket: filepath.Join(b.DataDir, "a.sock")}
376 if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".env"), []byte("MINE=1\nCARLOS_ADMIN_TOKEN=t\nCARLOS_PURGE=header\nCARLOS_REGION=fake\n"), 0o600); err != nil {
377 t.Fatal(err)
378 }
379 t.Setenv("CARLOS_SECRETS_KEY", "never")
380 t.Setenv("UNRELATED", "yes")
381 // The rendered configuration file (the path instanceEnv reads first)
382 // renders MINE=0; the instance's own .env overrides it, so MINE is
383 // reported as shadowed (spec §7.3 T3).
384 if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+".config.env"), []byte("MINE=0\nFROM_CONFIG=c\n"), 0o600); err != nil {
385 t.Fatal(err)
386 }
387 env, shadowed, err := b.LaunchEnvironment(rt)
388 if err != nil {
389 t.Fatal(err)
390 }
391 if len(shadowed) != 1 || shadowed[0] != "MINE" || env["FROM_CONFIG"] != "c" {
392 t.Fatalf("shadowed names %v, want [MINE]; FROM_CONFIG: want set from the config file, got set=%v", shadowed, env["FROM_CONFIG"] != "")
393 }
394 if env["MINE"] != "1" || env["UNRELATED"] != "yes" || env["CARLOS_SECRETS_KEY"] != "" || env["CARLOS_ADMIN_TOKEN"] != "t" {
395 t.Fatalf("launch env: MINE matches the own file: %v; UNRELATED passed through: %v; CARLOS_SECRETS_KEY: want unset, got set=%v; CARLOS_ADMIN_TOKEN matches the own file: %v",
396 env["MINE"] == "1", env["UNRELATED"] == "yes", env["CARLOS_SECRETS_KEY"] != "", env["CARLOS_ADMIN_TOKEN"] == "t")
397 }
398 // The agent-owned layer, exactly as agentOwnedVars composes it: the
399 // tenant's region and purge values are overridden or removed, and the
400 // host is the row's (#256).
401 if env["CARLOS_HOST"] != rt.Host {
402 t.Fatalf("CARLOS_HOST = %q, want the row's host %q", env["CARLOS_HOST"], rt.Host)
403 }
404 if env["CARLOS_REGION"] != "ie" || env["CARLOS_CONTROL_SOCKET"] != rt.Socket+".control" {
405 t.Fatalf("agent-owned vars: CARLOS_REGION is the backend's: %v; CARLOS_CONTROL_SOCKET is the route's control path: %v",
406 env["CARLOS_REGION"] == "ie", env["CARLOS_CONTROL_SOCKET"] == rt.Socket+".control")
407 }
408 if _, ok := env["CARLOS_PURGE"]; ok {
409 t.Fatal("CARLOS_PURGE survived on a box without a cache")
410 }
411 b.Cache = edge.NewCache(16, 1<<20)
412 if env, _, _ = b.LaunchEnvironment(rt); env["CARLOS_PURGE"] != "header" {
413 t.Fatalf("CARLOS_PURGE on a box with a cache: want header, got set=%v", env["CARLOS_PURGE"] != "")
414 }
415 // A box with no region strips the tenant's value rather than keeping it.
416 b.Region = ""
417 if env, _, _ = b.LaunchEnvironment(rt); env["CARLOS_REGION"] != "" {
418 t.Fatal("CARLOS_REGION: want unset on a box with no region, got set")
419 }
420}
421
422// An empty CARLOS_INSTANCE_KEY in the instance's own file supplies no key
423// (instance-key plan ruling 13): the merge keeps the configured key, and
424// the launch report does not list the empty one as shadowing it. A
425// non-empty own key still wins and is still reported.
426func TestEmptyOwnInstanceKeyShadowsNothing(t *testing.T) {
427 cfg := map[string]string{"CARLOS_INSTANCE_KEY": "cfg"}
428 if got := mergeEnv(cfg, map[string]string{"CARLOS_INSTANCE_KEY": ""}); got["CARLOS_INSTANCE_KEY"] != "cfg" {
429 t.Fatalf("mergeEnv with an empty own key: CARLOS_INSTANCE_KEY = %q, want the configured key", got["CARLOS_INSTANCE_KEY"])
430 }
431 b := &ExecBackend{DataDir: t.TempDir(), Region: "ie"}
432 rt := registry.Route{Host: "k.go.example.com", Socket: filepath.Join(b.DataDir, "k.sock")}
433 write := func(name, body string) {
434 t.Helper()
435 if err := os.WriteFile(filepath.Join(b.DataDir, rt.Host+name), []byte(body), 0o600); err != nil {
436 t.Fatal(err)
437 }
438 }
439 write(".config.env", "CARLOS_INSTANCE_KEY=cfg\nMINE=0\n")
440 write(".env", "CARLOS_INSTANCE_KEY=\nMINE=1\n")
441 env, shadowed, err := b.LaunchEnvironment(rt)
442 if err != nil {
443 t.Fatal(err)
444 }
445 if env["CARLOS_INSTANCE_KEY"] != "cfg" || len(shadowed) != 1 || shadowed[0] != "MINE" {
446 t.Fatalf("empty own key: CARLOS_INSTANCE_KEY is the configured key: %v; shadowed %v, want [MINE]", env["CARLOS_INSTANCE_KEY"] == "cfg", shadowed)
447 }
448 write(".env", "CARLOS_INSTANCE_KEY=own\nMINE=1\n")
449 if env, shadowed, err = b.LaunchEnvironment(rt); err != nil || env["CARLOS_INSTANCE_KEY"] != "own" || strings.Join(shadowed, ",") != "CARLOS_INSTANCE_KEY,MINE" {
450 t.Fatalf("non-empty own key: CARLOS_INSTANCE_KEY is the own key: %v; shadowed %v, want [CARLOS_INSTANCE_KEY MINE]; err %v", env["CARLOS_INSTANCE_KEY"] == "own", shadowed, err)
451 }
452}
453
454// With neither file present the composition is the agent-owned layer over
455// the agent's own environment, no error and nothing shadowed.
456func TestLaunchEnvironmentNoFiles(t *testing.T) {
457 b := &ExecBackend{DataDir: t.TempDir(), Region: "ie"}
458 rt := registry.Route{Host: "b.go.example.com", Socket: filepath.Join(b.DataDir, "b.sock")}
459 env, shadowed, err := b.LaunchEnvironment(rt)
460 if err != nil || len(shadowed) != 0 || env["CARLOS_REGION"] != "ie" {
461 t.Fatalf("CARLOS_REGION is the backend's: %v; shadowed %v; err %v", env["CARLOS_REGION"] == "ie", shadowed, err)
462 }
463}
464
465// recordingStrict is a strictProc that starts the command for real and
466// records each call in order, so a test can see the spawn went through the
467// strict handle and what happened to it on exit.
468type recordingStrict struct {
469 mu sync.Mutex
470 calls []string
471 deadline bool
472 started chan struct{}
473 closed chan struct{}
474}
475
476func newRecordingStrict() *recordingStrict {
477 return &recordingStrict{started: make(chan struct{}), closed: make(chan struct{})}
478}
479
480func (r *recordingStrict) note(s string) { r.mu.Lock(); r.calls = append(r.calls, s); r.mu.Unlock() }
481func (r *recordingStrict) Start(cmd *exec.Cmd) error {
482 r.note("start")
483 defer close(r.started)
484 return cmd.Start()
485}
486func (r *recordingStrict) KillAndEmpty(ctx context.Context) error {
487 _, ok := ctx.Deadline()
488 r.mu.Lock()
489 r.deadline = ok
490 r.mu.Unlock()
491 r.note("kill-and-empty")
492 return nil
493}
494func (r *recordingStrict) Close() error {
495 r.note("close")
496 close(r.closed)
497 return nil
498}
499func (r *recordingStrict) seen() []string {
500 r.mu.Lock()
501 defer r.mu.Unlock()
502 return append([]string(nil), r.calls...)
503}
504
505// TestSpawnUnderAMoveIsStrictOrRefused (moves design §5.3): a host Strict
506// names spawns only through the strict handle. A handle that cannot be
507// prepared refuses the spawn with ErrStrictRequired and nothing runs; a
508// prepared one starts the child and, on exit, is drained (KillAndEmpty
509// under a deadline) and then closed. With Strict false or unset the plain
510// spawn runs exactly as before and the seam is never asked.
511func TestSpawnUnderAMoveIsStrictOrRefused(t *testing.T) {
512 old := spawnHealthTimeout
513 spawnHealthTimeout = 300 * time.Millisecond
514 t.Cleanup(func() { spawnHealthTimeout = old })
515 host := "strict.example.com"
516 // argvPath is the recorder's first file: once it exists and the child
517 // has exited, the health wait the recorder never answers is over.
518 start := func(t *testing.T, b *ExecBackend, argvPath string) error {
519 t.Helper()
520 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
521 ctx, cancel := recordedCtx(2*time.Second, argvPath)
522 defer cancel()
523 proc, _, err := b.Start(ctx, rt) // the fake binary never binds: a health error is expected
524 if proc != nil {
525 _ = proc.Stop(context.Background())
526 }
527 return err
528 }
529 noSpawn := func(t *testing.T, argvPath string) {
530 t.Helper()
531 awaitChildFile(argvPath, 200*time.Millisecond) // until no child is left to write it
532 if _, err := os.Stat(argvPath); !os.IsNotExist(err) {
533 t.Fatalf("the child ran: %v", err)
534 }
535 }
536
537 t.Run("a handle that cannot be prepared refuses the spawn", func(t *testing.T) {
538 b := testExecBackend(t)
539 argvPath, _ := stageRecordingLive(t, b, host, "v1")
540 b.Strict = func(h string) bool { return h == host }
541 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) { return nil, cgroup.ErrStrictRefused }
542 if err := start(t, b, argvPath); !errors.Is(err, ErrStrictRequired) {
543 t.Fatalf("Start: %v, want ErrStrictRequired", err)
544 }
545 noSpawn(t, argvPath)
546 })
547
548 t.Run("no delegation and no seam refuses the spawn", func(t *testing.T) {
549 b := testExecBackend(t) // Cgroups nil
550 argvPath, _ := stageRecordingLive(t, b, host, "v1")
551 b.Strict = func(string) bool { return true }
552 if err := start(t, b, argvPath); !errors.Is(err, ErrStrictRequired) {
553 t.Fatalf("Start: %v, want ErrStrictRequired", err)
554 }
555 noSpawn(t, argvPath)
556 })
557
558 t.Run("a handle whose Start fails refuses the spawn and is closed", func(t *testing.T) {
559 b := testExecBackend(t)
560 argvPath, _ := stageRecordingLive(t, b, host, "v1")
561 b.Strict = func(string) bool { return true }
562 rec := newRecordingStrict()
563 failing := &failingStrict{recordingStrict: rec}
564 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) { return failing, nil }
565 if err := start(t, b, argvPath); !errors.Is(err, ErrStrictRequired) {
566 t.Fatalf("Start: %v, want ErrStrictRequired", err)
567 }
568 noSpawn(t, argvPath)
569 if got := strings.Join(rec.seen(), ","); got != "start,kill-and-empty,close" {
570 t.Fatalf("handle calls = %s", got)
571 }
572 })
573
574 t.Run("a prepared handle starts the child and is drained then closed on exit", func(t *testing.T) {
575 b := testExecBackend(t)
576 argvPath, _ := stageRecordingLive(t, b, host, "v1")
577 b.Strict = func(string) bool { return true }
578 rec := newRecordingStrict()
579 var gotHost string
580 var gotLimit cgroup.Limit
581 b.Tier = func(registry.Route) cgroup.Limit { return cgroup.TierChildren }
582 b.prepareStrict = func(h string, l cgroup.Limit) (strictProc, error) { gotHost, gotLimit = h, l; return rec, nil }
583 _ = start(t, b, argvPath)
584 waitForFile(t, argvPath, time.Second)
585 select {
586 case <-rec.closed:
587 case <-time.After(5 * time.Second):
588 t.Fatalf("the handle was never closed: %v", rec.seen())
589 }
590 if got := strings.Join(rec.seen(), ","); got != "start,kill-and-empty,close" {
591 t.Fatalf("handle calls = %s, want start, kill-and-empty, close", got)
592 }
593 if !rec.deadline {
594 t.Fatal("KillAndEmpty was called without a deadline")
595 }
596 if gotHost != host || gotLimit != cgroup.TierChildren {
597 t.Fatalf("prepared %q at %v, want %q at the route's tier", gotHost, gotLimit, host)
598 }
599 })
600
601 for name, strict := range map[string]func(string) bool{
602 "Strict false": func(string) bool { return false },
603 "Strict unset": nil,
604 } {
605 t.Run(name+" runs the plain spawn", func(t *testing.T) {
606 b := testExecBackend(t)
607 argvPath, _ := stageRecordingLive(t, b, host, "v1")
608 b.Strict = strict
609 asked := false
610 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
611 asked = true
612 return nil, errors.New("must not be asked")
613 }
614 _ = start(t, b, argvPath)
615 waitForFile(t, argvPath, time.Second)
616 if asked {
617 t.Fatal("the strict seam was asked for a host not in a move")
618 }
619 })
620 }
621}
622
623// failingStrict is a recordingStrict whose Start fails without starting
624// anything, the way Prepared.Start refuses a populated group.
625type failingStrict struct{ *recordingStrict }
626
627func (f *failingStrict) Start(*exec.Cmd) error {
628 f.note("start")
629 return cgroup.ErrStrictRefused
630}
631
632// delegatedManager is a cgroup manager over a temp-dir fixture: delegation
633// "exists" (Root() is set), but the test process is not placed in it.
634func delegatedManager(t *testing.T) *cgroup.Manager {
635 t.Helper()
636 root := t.TempDir()
637 for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
638 if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
639 t.Fatal(err)
640 }
641 }
642 m, err := cgroup.SetupAt(root, t.Logf)
643 if err != nil {
644 t.Fatal(err)
645 }
646 return m
647}
648
649// TestNextSpawnStrict is S2's input: OK only when the host is in a move
650// (Strict), the box has a delegated cgroup subtree, PrepareStrict's own
651// runtime preflight passes (cgroup.Manager.StrictPreflight, the same
652// function PrepareStrict runs, so S2 cannot approve a box PrepareStrict
653// refuses), and no earlier strict group of the host awaits cleanup.
654func TestNextSpawnStrict(t *testing.T) {
655 m := delegatedManager(t)
656 host := "s2.example.com"
657 yes := func(string) bool { return true }
658 ok := func() error { return nil }
659 for name, tc := range map[string]struct {
660 strict func(string) bool
661 cg *cgroup.Manager
662 preflight func() error
663 ok bool
664 inWords string
665 }{
666 "in a move, delegated, preflight passes": {yes, m, ok, true, ""},
667 "not in a move": {func(string) bool { return false }, m, ok, false, "not strict"},
668 "Strict unset": {nil, m, ok, false, "not strict"},
669 "no delegation": {yes, nil, ok, false, "delegated"},
670 // The real preflight, no seam: delegation exists, but this test
671 // process is not placed in the fixture's subtree.
672 "delegated, agent misplaced": {yes, m, nil, false, "not placed"},
673 "preflight says unsupported": {yes, m, func() error { return cgroup.ErrStrictUnsupported }, false, "cannot spawn strictly"},
674 } {
675 t.Run(name, func(t *testing.T) {
676 if tc.preflight == nil && runtime.GOOS != "linux" {
677 t.Skip("the real preflight answers unsupported off Linux")
678 }
679 b := &ExecBackend{Strict: tc.strict, Cgroups: tc.cg, strictPreflight: tc.preflight}
680 c := b.NextSpawnStrict(host)
681 if c.OK != tc.ok || (!c.OK && !strings.Contains(c.Reason, tc.inWords)) || (c.OK && c.Reason != "") {
682 t.Fatalf("NextSpawnStrict = %+v, want ok=%v naming %q", c, tc.ok, tc.inWords)
683 }
684 })
685 }
686}
687
688// flakyStrict is a recordingStrict whose KillAndEmpty fails its first
689// killFails calls and whose Close fails its first closeFails calls, the
690// way a drain that outlives its deadline or a group still settling does.
691type flakyStrict struct {
692 *recordingStrict
693 killFails, closeFails int
694}
695
696func (f *flakyStrict) KillAndEmpty(ctx context.Context) error {
697 f.mu.Lock()
698 fail := f.killFails > 0
699 if fail {
700 f.killFails--
701 }
702 f.mu.Unlock()
703 if fail {
704 f.note("kill-and-empty failed")
705 return cgroup.ErrStrictDeadline
706 }
707 return f.recordingStrict.KillAndEmpty(ctx)
708}
709
710func (f *flakyStrict) Close() error {
711 f.mu.Lock()
712 fail := f.closeFails > 0
713 if fail {
714 f.closeFails--
715 }
716 f.mu.Unlock()
717 if fail {
718 f.note("close failed")
719 return cgroup.ErrStrictRefused
720 }
721 return f.recordingStrict.Close()
722}
723
724// TestStrictCleanupIsRetriedAndRefusesUntilItSucceeds: a strict group whose
725// drain or close fails on exit keeps its handle; cleanup is retried on a
726// bounded schedule, and until it succeeds the host's strict spawns are
727// refused (and S2 says so) without preparing a new group. Once it
728// succeeds, strict spawns work again. A drain that succeeded is not
729// repeated when only the close failed.
730func TestStrictCleanupIsRetriedAndRefusesUntilItSucceeds(t *testing.T) {
731 old := spawnHealthTimeout
732 spawnHealthTimeout = 300 * time.Millisecond
733 t.Cleanup(func() { spawnHealthTimeout = old })
734 host := "retry.example.com"
735 for name, tc := range map[string]struct {
736 killFails, closeFails int
737 want string
738 }{
739 "the drain fails once": {1, 0, "start,kill-and-empty failed,kill-and-empty,close"},
740 "the close fails once": {0, 1, "start,kill-and-empty,close failed,close"},
741 } {
742 t.Run(name, func(t *testing.T) {
743 oldBackoff := strictCleanupBackoff
744 strictCleanupBackoff = []time.Duration{400 * time.Millisecond, 400 * time.Millisecond}
745 t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
746 b := testExecBackend(t)
747 b.Cgroups = delegatedManager(t)
748 b.strictPreflight = func() error { return nil }
749 argvPath, _ := stageRecordingLive(t, b, host, "v1")
750 b.Strict = func(string) bool { return true }
751 var mu sync.Mutex
752 var handles []*flakyStrict
753 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
754 mu.Lock()
755 defer mu.Unlock()
756 h := &flakyStrict{recordingStrict: newRecordingStrict()}
757 if len(handles) == 0 {
758 h.killFails, h.closeFails = tc.killFails, tc.closeFails
759 }
760 handles = append(handles, h)
761 return h, nil
762 }
763 prepared := func() int { mu.Lock(); defer mu.Unlock(); return len(handles) }
764 startOnce := func() error {
765 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
766 ctx, cancel := recordedCtx(2*time.Second, argvPath) // the recorder ran and exited
767 defer cancel()
768 proc, _, err := b.Start(ctx, rt)
769 if proc != nil {
770 _ = proc.Stop(context.Background())
771 }
772 return err
773 }
774
775 _ = startOnce()
776 waitForFile(t, argvPath, time.Second)
777 deadline := time.Now().Add(2 * time.Second)
778 for !b.strictCleanupPending(host) {
779 if time.Now().After(deadline) {
780 t.Fatalf("no cleanup pending after a failed drain: %v", handles[0].seen())
781 }
782 time.Sleep(5 * time.Millisecond)
783 }
784 if c := b.NextSpawnStrict(host); c.OK || !strings.Contains(c.Reason, "cleaned up") {
785 t.Fatalf("S2 while cleanup is pending: %+v", c)
786 }
787 if err := startOnce(); !errors.Is(err, ErrStrictRequired) {
788 t.Fatalf("spawn while cleanup is pending: %v, want ErrStrictRequired", err)
789 }
790 if n := prepared(); n != 1 {
791 t.Fatalf("a new strict group was prepared while the old one awaited cleanup (%d)", n)
792 }
793
794 deadline = time.Now().Add(3 * time.Second)
795 for b.strictCleanupPending(host) {
796 if time.Now().After(deadline) {
797 t.Fatalf("cleanup never succeeded on retry: %v", handles[0].seen())
798 }
799 time.Sleep(5 * time.Millisecond)
800 }
801 if got := strings.Join(handles[0].seen(), ","); got != tc.want {
802 t.Fatalf("first handle's calls = %s, want %s", got, tc.want)
803 }
804 if c := b.NextSpawnStrict(host); !c.OK {
805 t.Fatalf("S2 after cleanup: %+v", c)
806 }
807 if err := os.Remove(argvPath); err != nil {
808 t.Fatal(err)
809 }
810 _ = startOnce()
811 waitForFile(t, argvPath, time.Second)
812 if n := prepared(); n != 2 {
813 t.Fatalf("strict groups prepared = %d, want a second one after cleanup", n)
814 }
815 })
816 }
817}
818
819// TestStrictCleanupGivesUpButStaysRefused: the retry schedule is bounded.
820// When it runs out the handle is kept and the host's strict spawns stay
821// refused: a group that never drained may still hold a process.
822func TestStrictCleanupGivesUpButStaysRefused(t *testing.T) {
823 oldBackoff := strictCleanupBackoff
824 strictCleanupBackoff = []time.Duration{10 * time.Millisecond, 10 * time.Millisecond}
825 t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
826 host := "stuck.example.com"
827 b := testExecBackend(t)
828 h := &flakyStrict{recordingStrict: newRecordingStrict(), killFails: 100}
829 b.Strict = func(string) bool { return true }
830 b.settleStrict(host, b.ownStrictForTest(host, h))
831 time.Sleep(200 * time.Millisecond)
832 if got := strings.Count(strings.Join(h.seen(), ","), "kill-and-empty failed"); got != 3 {
833 t.Fatalf("drain attempts = %d, want the first and two retries", got)
834 }
835 if !b.strictCleanupPending(host) {
836 t.Fatal("the handle was dropped after the schedule ran out")
837 }
838 b.Strict = func(string) bool { return true }
839 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
840 t.Fatal("a strict group was prepared for a host awaiting cleanup")
841 return nil, nil
842 }
843 if _, _, err := b.spawnInCgroup(registry.Route{Host: host}, func() *exec.Cmd { return exec.Command("/bin/true") }); !errors.Is(err, ErrStrictRequired) {
844 t.Fatalf("spawn: %v, want ErrStrictRequired", err)
845 }
846}
847
848// ownStrictForTest registers sp as host's strict owner, as a strict spawn
849// does, so a test can drive settleStrict directly.
850func (b *ExecBackend) ownStrictForTest(host string, sp strictProc) *strictOwner {
851 g := b.gate(host)
852 g.mu.Lock()
853 defer g.mu.Unlock()
854 g.owner = &strictOwner{sp: sp}
855 return g.owner
856}
857
858// dirStrict is a strict handle whose identity is a real directory: the
859// host's group directory under a temp-dir cgroup fixture. Like
860// cgroup.Prepared's identityOK, once that directory is removed (or
861// replaced) the handle can neither drain nor close again, for ever. It
862// models what an exit-path CleanupHost would do to a retained handle.
863type dirStrict struct {
864 *recordingStrict
865 dir string
866 id os.FileInfo
867 killFails, closeFails int
868}
869
870func newDirStrict(t *testing.T, dir string) *dirStrict {
871 t.Helper()
872 if err := os.MkdirAll(dir, 0o755); err != nil {
873 t.Fatal(err)
874 }
875 id, err := os.Stat(dir)
876 if err != nil {
877 t.Fatal(err)
878 }
879 return &dirStrict{recordingStrict: newRecordingStrict(), dir: dir, id: id}
880}
881
882func (d *dirStrict) identityOK() bool {
883 now, err := os.Stat(d.dir)
884 return err == nil && os.SameFile(now, d.id)
885}
886
887func (d *dirStrict) KillAndEmpty(ctx context.Context) error {
888 if !d.identityOK() {
889 d.note("kill refused: group gone")
890 return cgroup.ErrStrictRefused
891 }
892 d.mu.Lock()
893 fail := d.killFails > 0
894 if fail {
895 d.killFails--
896 }
897 d.mu.Unlock()
898 if fail {
899 d.note("kill-and-empty timed out")
900 return cgroup.ErrStrictDeadline
901 }
902 return d.recordingStrict.KillAndEmpty(ctx)
903}
904
905func (d *dirStrict) Close() error {
906 if !d.identityOK() {
907 d.note("close refused: group gone")
908 return cgroup.ErrStrictRefused
909 }
910 d.mu.Lock()
911 fail := d.closeFails > 0
912 if fail {
913 d.closeFails--
914 }
915 d.mu.Unlock()
916 if fail {
917 d.note("close failed")
918 return cgroup.ErrStrictRefused
919 }
920 return d.recordingStrict.Close()
921}
922
923// TestPendingStrictCleanupOwnsTheGroup is the invariant: while a host has
924// pending strict cleanup, nothing else may use or remove that host's
925// cgroup. The exit path does not remove the group directory under a
926// retained handle (with a handle whose identity is that directory, a
927// removal would strand it for ever), so a drain that timed out before the
928// group emptied, and a transient Close failure on an empty group, both
929// recover on retry; the directory goes only once the handle is closed, and
930// strict spawns work afterwards. Meanwhile any spawn of the host, an
931// ordinary one included (Strict switched off), is refused and never runs.
932func TestPendingStrictCleanupOwnsTheGroup(t *testing.T) {
933 old := spawnHealthTimeout
934 spawnHealthTimeout = 300 * time.Millisecond
935 t.Cleanup(func() { spawnHealthTimeout = old })
936 host := "owned.example.com"
937 for name, tc := range map[string]struct {
938 killFails, closeFails int
939 want string
940 }{
941 "a drain timeout, then the group empties": {1, 0, "start,kill-and-empty timed out,kill-and-empty,close"},
942 "a transient close failure on an empty group": {0, 1, "start,kill-and-empty,close failed,close"},
943 } {
944 t.Run(name, func(t *testing.T) {
945 oldBackoff := strictCleanupBackoff
946 strictCleanupBackoff = []time.Duration{500 * time.Millisecond, 500 * time.Millisecond}
947 t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
948 b := testExecBackend(t)
949 b.Cgroups = delegatedManager(t)
950 b.strictPreflight = func() error { return nil }
951 groupDir := filepath.Join(b.Cgroups.Root(), cgroup.ExecName(host))
952 argvPath, _ := stageRecordingLive(t, b, host, "v1")
953 var strict atomic.Bool
954 strict.Store(true)
955 b.Strict = func(string) bool { return strict.Load() }
956 var mu sync.Mutex
957 var handles []*dirStrict
958 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
959 mu.Lock()
960 defer mu.Unlock()
961 h := newDirStrict(t, groupDir)
962 if len(handles) == 0 {
963 h.killFails, h.closeFails = tc.killFails, tc.closeFails
964 }
965 handles = append(handles, h)
966 return h, nil
967 }
968 startOnce := func() error {
969 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
970 ctx, cancel := recordedCtx(2*time.Second, argvPath) // the recorder ran and exited
971 defer cancel()
972 proc, _, err := b.Start(ctx, rt)
973 if proc != nil {
974 _ = proc.Stop(context.Background())
975 }
976 return err
977 }
978
979 _ = startOnce()
980 waitForFile(t, argvPath, time.Second)
981 deadline := time.Now().Add(2 * time.Second)
982 for !b.strictCleanupPending(host) {
983 if time.Now().After(deadline) {
984 t.Fatalf("no cleanup pending: %v", handles[0].seen())
985 }
986 time.Sleep(5 * time.Millisecond)
987 }
988 // The child has exited and its exit path has run: the group
989 // directory is still there, owned by the retained handle.
990 time.Sleep(50 * time.Millisecond)
991 if _, err := os.Stat(groupDir); err != nil {
992 t.Fatalf("the exit path removed the group under a pending handle: %v", err)
993 }
994 // An ordinary spawn of the host is refused while cleanup is
995 // pending, and nothing runs.
996 strict.Store(false)
997 if err := os.Remove(argvPath); err != nil {
998 t.Fatal(err)
999 }
1000 if err := startOnce(); !errors.Is(err, ErrStrictCleanupPending) {
1001 t.Fatalf("ordinary spawn during pending cleanup: %v, want ErrStrictCleanupPending", err)
1002 }
1003 awaitChildFile(argvPath, 100*time.Millisecond) // until no child is left to write it
1004 if _, err := os.Stat(argvPath); !os.IsNotExist(err) {
1005 t.Fatalf("an ordinary child ran in a group awaiting cleanup: %v", err)
1006 }
1007 strict.Store(true)
1008
1009 deadline = time.Now().Add(4 * time.Second)
1010 for b.strictCleanupPending(host) {
1011 if time.Now().After(deadline) {
1012 t.Fatalf("cleanup never succeeded: %v", handles[0].seen())
1013 }
1014 time.Sleep(5 * time.Millisecond)
1015 }
1016 if got := strings.Join(handles[0].seen(), ","); got != tc.want {
1017 t.Fatalf("first handle's calls = %s, want %s", got, tc.want)
1018 }
1019 if _, err := os.Stat(groupDir); !os.IsNotExist(err) {
1020 t.Fatalf("the group directory outlived its handle's cleanup: %v", err)
1021 }
1022 _ = startOnce()
1023 waitForFile(t, argvPath, time.Second)
1024 mu.Lock()
1025 n := len(handles)
1026 mu.Unlock()
1027 if n != 2 {
1028 t.Fatalf("strict groups prepared = %d, want a second one after cleanup", n)
1029 }
1030 })
1031 }
1032}
1033
1034// TestALiveStrictGroupRefusesOrdinarySpawns: while a strict child of a
1035// host is alive, an ordinary spawn of the same host (Strict switched off
1036// meanwhile) is refused rather than placed in the strict group, where the
1037// strict child's exit drain would kill it.
1038func TestALiveStrictGroupRefusesOrdinarySpawns(t *testing.T) {
1039 host := "live.example.com"
1040 b := testExecBackend(t)
1041 var strict atomic.Bool
1042 strict.Store(true)
1043 b.Strict = func(string) bool { return strict.Load() }
1044 rec := newRecordingStrict()
1045 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) { return rec, nil }
1046 rt := registry.Route{Host: host}
1047 cmd, settle, err := b.spawnInCgroup(rt, func() *exec.Cmd { return exec.Command("sleep", "5") })
1048 if err != nil {
1049 t.Fatalf("strict spawn: %v", err)
1050 }
1051 strict.Store(false)
1052 ran := false
1053 if _, _, err := b.spawnInCgroup(rt, func() *exec.Cmd { ran = true; return exec.Command("/bin/true") }); !errors.Is(err, ErrStrictGroupInUse) || ran {
1054 t.Fatalf("ordinary spawn beside a live strict child: %v (ran %v), want ErrStrictGroupInUse", err, ran)
1055 }
1056 _ = cmd.Process.Kill()
1057 _ = cmd.Wait()
1058 settle()
1059 if _, settle2, err := b.spawnInCgroup(rt, func() *exec.Cmd { return exec.Command("/bin/true") }); err != nil {
1060 t.Fatalf("ordinary spawn after the strict child settled: %v", err)
1061 } else {
1062 settle2()
1063 }
1064}
1065
1066// hookAt returns a strictHook that calls fn at point and does nothing at
1067// any other point.
1068func hookAt(point string, fn func()) func(string) {
1069 return func(p string) {
1070 if p == point {
1071 fn()
1072 }
1073 }
1074}
1075
1076// TestAnOlderExitCannotRemoveANewOwnersGroup: the exit path's removal and
1077// a strict spawn's ownership are decided under the same host gate. An exit
1078// that has checked "no owner" holds the gate through its removal, so a
1079// strict spawn arriving meanwhile waits, then owns a group it created
1080// itself; an exit that runs once any owner exists, live or pending, removes
1081// nothing. A strict child whose cleanup then fails keeps its directory.
1082func TestAnOlderExitCannotRemoveANewOwnersGroup(t *testing.T) {
1083 oldBackoff := strictCleanupBackoff
1084 strictCleanupBackoff = []time.Duration{time.Hour}
1085 t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
1086 host := "race.example.com"
1087 b := testExecBackend(t)
1088 b.Cgroups = delegatedManager(t)
1089 groupDir := filepath.Join(b.Cgroups.Root(), cgroup.ExecName(host))
1090 b.Strict = func(string) bool { return true }
1091 var handle *dirStrict
1092 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) {
1093 handle = newDirStrict(t, groupDir)
1094 handle.killFails = 100
1095 return handle, nil
1096 }
1097 if err := os.MkdirAll(groupDir, 0o755); err != nil {
1098 t.Fatal(err)
1099 }
1100
1101 // The older exit has found no owner and is about to remove the group.
1102 // A strict spawn starts right then and must wait for the gate.
1103 spawned := make(chan error, 1)
1104 var settle func()
1105 var cmd *exec.Cmd
1106 b.strictHook = hookAt("exit-cleanup-checked", func() {
1107 go func() {
1108 var err error
1109 cmd, settle, err = b.spawnInCgroup(registry.Route{Host: host}, func() *exec.Cmd { return exec.Command("/bin/true") })
1110 spawned <- err
1111 }()
1112 select {
1113 case err := <-spawned:
1114 t.Errorf("a strict spawn took the host while an exit held its gate: %v", err)
1115 spawned <- err
1116 case <-time.After(150 * time.Millisecond):
1117 }
1118 })
1119 b.cleanupHost(host)
1120 b.strictHook = nil
1121 if err := <-spawned; err != nil {
1122 t.Fatalf("strict spawn after the exit: %v", err)
1123 }
1124 if _, err := os.Stat(groupDir); err != nil {
1125 t.Fatalf("the new owner has no group directory: %v", err)
1126 }
1127 // Any owner, live as here, keeps the directory from an exit.
1128 b.cleanupHost(host)
1129 if _, err := os.Stat(groupDir); err != nil {
1130 t.Fatalf("an exit removed a live owner's group: %v", err)
1131 }
1132 _ = cmd.Wait()
1133 settle() // the drain fails: the owner turns pending
1134 if !b.strictCleanupPending(host) {
1135 t.Fatal("cleanup is not pending after a failed drain")
1136 }
1137 b.cleanupHost(host)
1138 if _, err := os.Stat(groupDir); err != nil {
1139 t.Fatalf("an exit removed a pending owner's group: %v", err)
1140 }
1141 if !handle.identityOK() {
1142 t.Fatal("the retained handle lost its group")
1143 }
1144}
1145
1146// TestAPendingTransitionRacesARefusal: a spawn refused while the owner is
1147// between its failed cleanup attempt and turning pending sees a live owner
1148// (ErrStrictGroupInUse); once pending, ErrStrictCleanupPending. pending is
1149// only ever read and written under the host gate, so -race stays quiet
1150// however the two interleave.
1151func TestAPendingTransitionRacesARefusal(t *testing.T) {
1152 oldBackoff := strictCleanupBackoff
1153 strictCleanupBackoff = []time.Duration{time.Hour}
1154 t.Cleanup(func() { strictCleanupBackoff = oldBackoff })
1155 host := "transition.example.com"
1156 b := testExecBackend(t)
1157 b.Strict = func(string) bool { return false }
1158 o := b.ownStrictForTest(host, &flakyStrict{recordingStrict: newRecordingStrict(), killFails: 100})
1159 spawn := func() error {
1160 _, _, err := b.spawnInCgroup(registry.Route{Host: host}, func() *exec.Cmd { return exec.Command("/bin/true") })
1161 return err
1162 }
1163 var during error
1164 b.strictHook = hookAt("cleanup-failed", func() { during = spawn() })
1165 b.settleStrict(host, o)
1166 b.strictHook = nil
1167 if !errors.Is(during, ErrStrictGroupInUse) {
1168 t.Fatalf("refusal before the owner turned pending: %v, want ErrStrictGroupInUse", during)
1169 }
1170 if err := spawn(); !errors.Is(err, ErrStrictCleanupPending) {
1171 t.Fatalf("refusal after: %v, want ErrStrictCleanupPending", err)
1172 }
1173
1174 // The same transition against concurrent refusals, for -race.
1175 host2 := "transition2.example.com"
1176 o2 := b.ownStrictForTest(host2, &flakyStrict{recordingStrict: newRecordingStrict(), killFails: 100})
1177 var wg sync.WaitGroup
1178 stop := make(chan struct{})
1179 var looping sync.WaitGroup
1180 for i := 0; i < 8; i++ {
1181 wg.Add(1)
1182 looping.Add(1)
1183 go func() {
1184 defer wg.Done()
1185 first := true
1186 for {
1187 _, _, err := b.spawnInCgroup(registry.Route{Host: host2}, func() *exec.Cmd { return exec.Command("/bin/true") })
1188 if !errors.Is(err, ErrStrictGroupInUse) && !errors.Is(err, ErrStrictCleanupPending) {
1189 t.Errorf("spawn beside an owner: %v", err)
1190 }
1191 _ = b.NextSpawnStrict(host2)
1192 if first {
1193 looping.Done()
1194 first = false
1195 }
1196 select {
1197 case <-stop:
1198 return
1199 default:
1200 }
1201 }
1202 }()
1203 }
1204 looping.Wait()
1205 b.settleStrict(host2, o2) // turns pending while the refusals run
1206 time.Sleep(20 * time.Millisecond)
1207 close(stop)
1208 wg.Wait()
1209}
1210
1211// TestAStalledStartBlocksOnlyItsHost: host A's ordinary start stalls inside
1212// Cgroups.Start, holding A's gate. Host B's ordinary and strict spawns and
1213// its NextSpawnStrict proceed; the global map lock is never held across a
1214// start.
1215func TestAStalledStartBlocksOnlyItsHost(t *testing.T) {
1216 b := testExecBackend(t)
1217 b.Cgroups = delegatedManager(t)
1218 b.strictPreflight = func() error { return nil }
1219 b.Strict = func(h string) bool { return h == "b-strict.example.com" }
1220 b.prepareStrict = func(string, cgroup.Limit) (strictProc, error) { return newRecordingStrict(), nil }
1221 release := make(chan struct{})
1222 entered := make(chan struct{})
1223 var once sync.Once
1224 doneA := make(chan error, 1)
1225 go func() {
1226 cmd, settle, err := b.spawnInCgroup(registry.Route{Host: "a.example.com"}, func() *exec.Cmd {
1227 once.Do(func() { close(entered); <-release })
1228 return exec.Command("/bin/true")
1229 })
1230 if err == nil {
1231 _ = cmd.Wait()
1232 settle()
1233 }
1234 doneA <- err
1235 }()
1236 <-entered
1237 within := func(what string, fn func()) {
1238 t.Helper()
1239 done := make(chan struct{})
1240 go func() { fn(); close(done) }()
1241 select {
1242 case <-done:
1243 case <-time.After(2 * time.Second):
1244 t.Fatalf("%s was blocked by host A's stalled start", what)
1245 }
1246 }
1247 within("host B's ordinary spawn", func() {
1248 cmd, settle, err := b.spawnInCgroup(registry.Route{Host: "b.example.com"}, func() *exec.Cmd { return exec.Command("/bin/true") })
1249 if err != nil {
1250 t.Errorf("host B: %v", err)
1251 return
1252 }
1253 _ = cmd.Wait()
1254 settle()
1255 })
1256 within("host B's strict spawn", func() {
1257 cmd, settle, err := b.spawnInCgroup(registry.Route{Host: "b-strict.example.com"}, func() *exec.Cmd { return exec.Command("/bin/true") })
1258 if err != nil {
1259 t.Errorf("host B strict: %v", err)
1260 return
1261 }
1262 _ = cmd.Wait()
1263 settle()
1264 })
1265 within("host B's NextSpawnStrict", func() {
1266 if c := b.NextSpawnStrict("b-strict.example.com"); !c.OK {
1267 t.Errorf("S2 for host B: %+v", c)
1268 }
1269 })
1270 within("host B's exit cleanup", func() { b.cleanupHost("b.example.com") })
1271 close(release)
1272 if err := <-doneA; err != nil {
1273 t.Fatalf("host A: %v", err)
1274 }
1275}
1276
1277// TestSwitchedOffSpawnsTakeNoNewLock: with Strict nil, a spawn and its exit
1278// path are main's: no host gate is created and the global map lock is
1279// never taken (held by the test throughout, a spawn that wanted it would
1280// hang).
1281func TestSwitchedOffSpawnsTakeNoNewLock(t *testing.T) {
1282 old := spawnHealthTimeout
1283 spawnHealthTimeout = 300 * time.Millisecond
1284 t.Cleanup(func() { spawnHealthTimeout = old })
1285 b := testExecBackend(t)
1286 b.Cgroups = delegatedManager(t)
1287 host := "off.example.com"
1288 argvPath, _ := stageRecordingLive(t, b, host, "v1")
1289 b.strictMu.Lock()
1290 defer b.strictMu.Unlock()
1291 done := make(chan struct{})
1292 go func() {
1293 defer close(done)
1294 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
1295 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
1296 defer cancel()
1297 proc, _, _ := b.Start(ctx, rt)
1298 if proc != nil {
1299 _ = proc.Stop(context.Background())
1300 }
1301 }()
1302 select {
1303 case <-done:
1304 case <-time.After(5 * time.Second):
1305 t.Fatal("a switched-off spawn waited on the strict map lock")
1306 }
1307 waitForFile(t, argvPath, time.Second)
1308 // The exit path's call, again, with the lock held.
1309 cleaned := make(chan struct{})
1310 go func() { b.cleanupHost(host); close(cleaned) }()
1311 select {
1312 case <-cleaned:
1313 case <-time.After(2 * time.Second):
1314 t.Fatal("a switched-off exit cleanup waited on the strict map lock")
1315 }
1316 if b.strictGates != nil {
1317 t.Fatalf("a switched-off spawn created host gates: %v", b.strictGates)
1318 }
1319}
1320
1321// Under Phase 1 the staging-wins promotion is gone (spec §5.3, M15): a
1322// staging tree beside an empty own prefix is not promoted, the restore is a
1323// fresh boot, and the tree is left exactly as found. Off, the legacy
1324// promotion runs as TestRestoreStagingFallback pins.
1325func TestRestoreDoesNotPromoteStagingUnderPhase1(t *testing.T) {
1326 ctx := context.Background()
1327 b := testExecBackend(t)
1328 f := newFakeMirrorStore()
1329 b.Mirror = f
1330 b.Phase1 = true
1331 host := "moved.example.com"
1332 rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
1333
1334 oldRestore := restoreReplica
1335 restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) {
1336 objs, _ := f.List(context.Background(), bucket, prefix)
1337 if len(objs) == 0 {
1338 return false, nil
1339 }
1340 return true, writeMinimalSQLite(dbPath)
1341 }
1342 defer func() { restoreReplica = oldRestore }()
1343
1344 // The legacy shape and the Phase 1 shape (moves/<host>/<op>/...).
1345 f.put(b.Cfg.Bucket, "moves/"+host+"/generations/a/snapshot", "e1")
1346 f.put(b.Cfg.Bucket, "moves/"+host+"/mv_abcdefghijklmnopqrstuvwxyz/g-cur/0000/0000000000000001-0000000000000002.ltx", "e2")
1347 before, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
1348 wantStaging := map[string]lease.MirrorObject{
1349 "moves/" + host + "/generations/a/snapshot": {ETag: "e1"},
1350 "moves/" + host + "/mv_abcdefghijklmnopqrstuvwxyz/g-cur/0000/0000000000000001-0000000000000002.ltx": {ETag: "e2"},
1351 }
1352 if !reflect.DeepEqual(before, wantStaging) {
1353 t.Fatalf("staging fixture: %v, want %v", before, wantStaging)
1354 }
1355
1356 fresh, err := b.Restore(ctx, rt)
1357 if err != nil || !fresh {
1358 t.Fatalf("Phase 1 restore with only staging: fresh=%v err=%v, want a fresh boot", fresh, err)
1359 }
1360 if own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)); len(own) != 0 {
1361 t.Fatalf("staging was promoted into the own prefix: %v", own)
1362 }
1363 after, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
1364 if !reflect.DeepEqual(after, before) {
1365 t.Fatalf("staging changed: before %v after %v", before, after)
1366 }
1367
1368 // An own lineage restores as ever, and still leaves staging alone.
1369 f.put(b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)+"/snapshot", "own")
1370 _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
1371 fresh, err = b.Restore(ctx, rt)
1372 if err != nil || fresh {
1373 t.Fatalf("own lineage: fresh=%v err=%v, want a restore", fresh, err)
1374 }
1375 if after, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/"); !reflect.DeepEqual(after, before) {
1376 t.Fatalf("staging touched by an own-lineage restore: %v", after)
1377 }
1378}
1379