| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "net" |
| 8 | "net/http" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "golang.org/x/sys/unix" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/instancectl" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/registry" |
| 21 | ) |
| 22 | |
| 23 | // fakeControl records verbs and lets a test drive the unit's state. |
| 24 | type fakeControl struct { |
| 25 | mu sync.Mutex |
| 26 | calls []string |
| 27 | inv string |
| 28 | seed instancectl.SeedResponse |
| 29 | startErr error |
| 30 | // stopGenErr is what fakeGenControl.StopGen answers. |
| 31 | stopGenErr error |
| 32 | stopHook func(host string, role instancectl.Role) |
| 33 | readyErr error |
| 34 | // onStart runs inside Start, in the fake control's own goroutine, the |
| 35 | // same place the real control service's trampoline would exec the app |
| 36 | // and the app would bind its socket: prepare's stale-socket removal |
| 37 | // races a test's OWN pre-bound fixture socket when that bind happens |
| 38 | // before Start is even called, so tests move it here instead. |
| 39 | onStart func() |
| 40 | // onSeed runs inside Seed, standing in for whatever a real seed process |
| 41 | // might have written to disk before it failed. A test uses it to give |
| 42 | // seedOwnsDB cleanup something real to (not) remove. |
| 43 | onSeed func() |
| 44 | |
| 45 | cmdInv string |
| 46 | cmdErr error |
| 47 | cmdStatus instancectl.CmdStatus |
| 48 | } |
| 49 | |
| 50 | func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() } |
| 51 | func (f *fakeControl) Start(_ context.Context, host string, role instancectl.Role, max int64) (string, error) { |
| 52 | f.rec("start " + host + " " + string(role)) |
| 53 | if f.onStart != nil { |
| 54 | f.onStart() |
| 55 | } |
| 56 | return f.inv, f.startErr |
| 57 | } |
| 58 | func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancectl.SeedResponse, error) { |
| 59 | f.rec("seed " + host) |
| 60 | if f.onSeed != nil { |
| 61 | f.onSeed() |
| 62 | } |
| 63 | return f.seed, nil |
| 64 | } |
| 65 | func (f *fakeControl) Cmd(_ context.Context, host string, max int64) (string, error) { |
| 66 | f.rec("cmd " + host) |
| 67 | return f.cmdInv, f.cmdErr |
| 68 | } |
| 69 | func (f *fakeControl) CmdStatus(_ context.Context, host, inv string) (instancectl.CmdStatus, error) { |
| 70 | f.rec("cmdstatus " + host + " " + inv) |
| 71 | return f.cmdStatus, nil |
| 72 | } |
| 73 | func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error { |
| 74 | f.rec("stop " + host + " " + string(role)) |
| 75 | if f.stopHook != nil { |
| 76 | f.stopHook(host, role) |
| 77 | } |
| 78 | return nil |
| 79 | } |
| 80 | func (f *fakeControl) Reset(_ context.Context, host string, role instancectl.Role) error { |
| 81 | f.rec("reset") |
| 82 | return nil |
| 83 | } |
| 84 | func (f *fakeControl) Sweep(context.Context) (instancectl.SweepResponse, error) { |
| 85 | f.rec("sweep") |
| 86 | return instancectl.SweepResponse{}, nil |
| 87 | } |
| 88 | func (f *fakeControl) Ready(context.Context) error { f.rec("ready"); return f.readyErr } |
| 89 | func (f *fakeControl) joined() string { |
| 90 | f.mu.Lock() |
| 91 | defer f.mu.Unlock() |
| 92 | return strings.Join(f.calls, "\n") |
| 93 | } |
| 94 | |
| 95 | // fakeUnit is the unit's observable state: what `systemctl show` and |
| 96 | // cgroup.events would say. |
| 97 | type fakeUnit struct { |
| 98 | mu sync.Mutex |
| 99 | active string |
| 100 | inv string |
| 101 | exists bool |
| 102 | populated bool |
| 103 | } |
| 104 | |
| 105 | func (u *fakeUnit) set(active, inv string, exists, populated bool) { |
| 106 | u.mu.Lock() |
| 107 | u.active, u.inv, u.exists, u.populated = active, inv, exists, populated |
| 108 | u.mu.Unlock() |
| 109 | } |
| 110 | func (u *fakeUnit) state(context.Context, string) (string, string, error) { |
| 111 | u.mu.Lock() |
| 112 | defer u.mu.Unlock() |
| 113 | return u.active, u.inv, nil |
| 114 | } |
| 115 | func (u *fakeUnit) events(string, string) (bool, bool, error) { |
| 116 | u.mu.Lock() |
| 117 | defer u.mu.Unlock() |
| 118 | return u.exists, u.populated, nil |
| 119 | } |
| 120 | |
| 121 | // nestedBackend returns an ExecBackend wired for a nested host with fakes, |
| 122 | // and the route. The data and run dirs are temp dirs standing in for |
| 123 | // /data/instances/<iid> and /run/carlos/instances/<iid>; provision's |
| 124 | // helpers are not used for the temp paths, so the route is built to |
| 125 | // classify as nested by construction through the Layout seam. |
| 126 | func nestedBackend(t *testing.T, ctl *fakeControl, unit *fakeUnit) (*ExecBackend, registry.Route) { |
| 127 | t.Helper() |
| 128 | root, err := os.MkdirTemp("", "nb") |
| 129 | if err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | t.Cleanup(func() { os.RemoveAll(root) }) |
| 133 | host := "h.example.com" |
| 134 | b := testExecBackend(t) |
| 135 | b.DataDir = filepath.Join(root, "data") |
| 136 | b.Control = ctl |
| 137 | b.UnitState = unit.state |
| 138 | b.CgroupEvents = unit.events |
| 139 | b.ObservePoll = 20 * time.Millisecond |
| 140 | b.RunDirOverride = filepath.Join(root, "run") |
| 141 | stageLive(t, b, host, "v1") |
| 142 | rt := registry.Route{Host: host, Kind: "instance", Hibernate: true, |
| 143 | DBPath: provision.NestedDBPath(b.DataDir, host), Socket: filepath.Join(b.RunDirOverride, "instances", host, "app.sock")} |
| 144 | return b, rt |
| 145 | } |
| 146 | |
| 147 | func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) { |
| 148 | t.Parallel() |
| 149 | ctl := &fakeControl{inv: "inv1"} |
| 150 | unit := &fakeUnit{} |
| 151 | b, rt := nestedBackend(t, ctl, unit) |
| 152 | // The fake control does not start anything, so health must be faked: |
| 153 | // bind a listener at rt.Socket that answers 200, from inside Start |
| 154 | // itself, where the app's own bind would actually happen, so |
| 155 | // prepare's stale-socket removal (which runs before Start ever calls |
| 156 | // the fake control) cannot race it. |
| 157 | var stopServing func() |
| 158 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 159 | defer func() { |
| 160 | if stopServing != nil { |
| 161 | stopServing() |
| 162 | } |
| 163 | }() |
| 164 | unit.set("active", "inv1", true, true) |
| 165 | p, ver, err := b.Start(context.Background(), rt) |
| 166 | if err != nil { |
| 167 | t.Fatalf("Start: %v", err) |
| 168 | } |
| 169 | _ = ver |
| 170 | up, ok := p.(*unitProc) |
| 171 | if !ok || up.invocation != "inv1" { |
| 172 | t.Fatalf("proc = %T %+v", p, p) |
| 173 | } |
| 174 | raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.run.json")) |
| 175 | if err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | var l instancectl.Launch |
| 179 | if err := json.Unmarshal(raw, &l); err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | if l.Schema != 1 || l.Host != rt.Host || len(l.Argv) != 5 || l.Argv[0] != filepath.Join(b.OptDir, "live", rt.Host) || l.Argv[2] != rt.Socket || l.Argv[4] != rt.DBPath { |
| 183 | t.Errorf("launch = %+v", l) |
| 184 | } |
| 185 | seenCtl := false |
| 186 | for _, kv := range l.Env { |
| 187 | if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") { |
| 188 | seenCtl = true |
| 189 | } |
| 190 | if strings.HasPrefix(kv, "CARLOS_ADMIN_TOKEN=") && len(kv) < 30 { |
| 191 | t.Errorf("token missing from env: %s", kv) |
| 192 | } |
| 193 | } |
| 194 | if !seenCtl { |
| 195 | t.Error("CARLOS_CONTROL_SOCKET not in the launch env") |
| 196 | } |
| 197 | if !strings.Contains(ctl.joined(), "start h.example.com run") { |
| 198 | t.Errorf("control calls: %s", ctl.joined()) |
| 199 | } |
| 200 | if p.Exited() { |
| 201 | t.Error("exited before the unit went inactive") |
| 202 | } |
| 203 | // Termination: terminal state plus empty cgroup. |
| 204 | unit.set("inactive", "inv1", false, false) |
| 205 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 206 | ctx, cancel := context.WithTimeout(context.Background(), time.Second) |
| 207 | defer cancel() |
| 208 | if err := p.Stop(ctx); err != nil { |
| 209 | t.Errorf("Stop after exit: %v", err) |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | func TestUnitProcDoesNotExitOnTerminalStateWhileTheCgroupIsPopulated(t *testing.T) { |
| 214 | t.Parallel() |
| 215 | ctl := &fakeControl{inv: "inv1"} |
| 216 | unit := &fakeUnit{} |
| 217 | b, rt := nestedBackend(t, ctl, unit) |
| 218 | var stopServing func() |
| 219 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 220 | defer func() { |
| 221 | if stopServing != nil { |
| 222 | stopServing() |
| 223 | } |
| 224 | }() |
| 225 | unit.set("active", "inv1", true, true) |
| 226 | p, _, err := b.Start(context.Background(), rt) |
| 227 | if err != nil { |
| 228 | t.Fatal(err) |
| 229 | } |
| 230 | unit.set("failed", "inv1", true, true) // a process survived SIGKILL |
| 231 | time.Sleep(150 * time.Millisecond) |
| 232 | if p.Exited() { |
| 233 | t.Fatal("exited while the cgroup is still populated (codex R2-5)") |
| 234 | } |
| 235 | ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) |
| 236 | defer cancel() |
| 237 | if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) { |
| 238 | t.Fatalf("Stop = %v, want ErrStillRunning", err) |
| 239 | } |
| 240 | unit.set("failed", "inv1", false, false) |
| 241 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 242 | } |
| 243 | |
| 244 | func TestUnitProcStopsAForeignInvocation(t *testing.T) { |
| 245 | t.Parallel() |
| 246 | ctl := &fakeControl{inv: "inv1"} |
| 247 | unit := &fakeUnit{} |
| 248 | b, rt := nestedBackend(t, ctl, unit) |
| 249 | var stopServing func() |
| 250 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 251 | defer func() { |
| 252 | if stopServing != nil { |
| 253 | stopServing() |
| 254 | } |
| 255 | }() |
| 256 | unit.set("active", "inv1", true, true) |
| 257 | p, _, err := b.Start(context.Background(), rt) |
| 258 | if err != nil { |
| 259 | t.Fatal(err) |
| 260 | } |
| 261 | ctl.stopHook = func(string, instancectl.Role) { unit.set("inactive", "", false, false) } |
| 262 | unit.set("active", "inv2", true, true) // someone restarted it outside the activator |
| 263 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 264 | if !strings.Contains(ctl.joined(), "stop h.example.com run") { |
| 265 | t.Errorf("foreign invocation was not stopped: %s", ctl.joined()) |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | func TestStartUnitRefusesWhenTheBoxIsNotReadyOrFenced(t *testing.T) { |
| 270 | ctl := &fakeControl{inv: "inv1"} |
| 271 | unit := &fakeUnit{} |
| 272 | b, rt := nestedBackend(t, ctl, unit) |
| 273 | b.NestedOK = func() error { return errors.New("probe: pid1 fail") } |
| 274 | if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "pid1") { |
| 275 | t.Fatalf("not-ready wake = %v", err) |
| 276 | } |
| 277 | if strings.Contains(ctl.joined(), "start") { |
| 278 | t.Error("a start was issued on a box that is not ready") |
| 279 | } |
| 280 | b.NestedOK = nil |
| 281 | b.LayoutFacts.Journal = func(string) bool { return true } |
| 282 | if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "migration in flight") { |
| 283 | t.Fatalf("fenced wake = %v", err) |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | func TestSeedUnitCleansUpOnlyItsOwnDBOnFailure(t *testing.T) { |
| 288 | ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}} |
| 289 | unit := &fakeUnit{} |
| 290 | b, rt := nestedBackend(t, ctl, unit) |
| 291 | rt.Seed = `["seed","--fast"]` |
| 292 | unit.set("inactive", "", false, false) |
| 293 | err := b.Seed(context.Background(), rt) |
| 294 | if err == nil || !strings.Contains(err.Error(), "exit 3") { |
| 295 | t.Fatalf("Seed = %v", err) |
| 296 | } |
| 297 | raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.seed.json")) |
| 298 | if err != nil { |
| 299 | t.Fatal(err) |
| 300 | } |
| 301 | var l instancectl.Launch |
| 302 | _ = json.Unmarshal(raw, &l) |
| 303 | if len(l.Argv) != 5 || l.Argv[1] != "seed" || l.Argv[2] != "--fast" || l.Argv[3] != "--db" || l.Argv[4] != rt.DBPath { |
| 304 | t.Errorf("seed argv = %v", l.Argv) |
| 305 | } |
| 306 | gotHost := false |
| 307 | for _, kv := range l.Env { |
| 308 | if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") || strings.HasPrefix(kv, "CARLOS_PURGE=") { |
| 309 | t.Errorf("seed env carries %s", kv) |
| 310 | } |
| 311 | gotHost = gotHost || kv == "CARLOS_HOST="+rt.Host |
| 312 | } |
| 313 | if !gotHost { |
| 314 | t.Errorf("seed env lacks CARLOS_HOST=%s (#256): %v", rt.Host, l.Env) |
| 315 | } |
| 316 | if _, err := os.Stat(rt.DBPath); !os.IsNotExist(err) { |
| 317 | t.Errorf("failed seed left its database: %v", err) |
| 318 | } |
| 319 | } |
| 320 | |
| 321 | // capturedEvent is one call eventCapture recorded. |
| 322 | type capturedEvent struct{ level, msg string } |
| 323 | |
| 324 | // eventCapture is a minimal EventSink that records Event calls for |
| 325 | // assertions; App is unused by these tests but required by the interface. |
| 326 | type eventCapture struct { |
| 327 | mu sync.Mutex |
| 328 | events []capturedEvent |
| 329 | } |
| 330 | |
| 331 | func (e *eventCapture) App(acct, app, host, line string) {} |
| 332 | func (e *eventCapture) Event(acct, app, host, level, msg string) { |
| 333 | e.mu.Lock() |
| 334 | e.events = append(e.events, capturedEvent{level, msg}) |
| 335 | e.mu.Unlock() |
| 336 | } |
| 337 | func (e *eventCapture) last() (capturedEvent, bool) { |
| 338 | e.mu.Lock() |
| 339 | defer e.mu.Unlock() |
| 340 | if len(e.events) == 0 { |
| 341 | return capturedEvent{}, false |
| 342 | } |
| 343 | return e.events[len(e.events)-1], true |
| 344 | } |
| 345 | |
| 346 | // TestUnitProcExitLineBoundsAForgedExitRecord covers the exit record's |
| 347 | // ruling: it is tenant-writable text, so a record with fields no real |
| 348 | // trampoline would ever write must still produce a bounded, single-line, |
| 349 | // member-safe event, and the record must be consumed (removed) exactly |
| 350 | // once. |
| 351 | func TestUnitProcExitLineBoundsAForgedExitRecord(t *testing.T) { |
| 352 | t.Parallel() |
| 353 | ctl := &fakeControl{inv: "inv1"} |
| 354 | unit := &fakeUnit{} |
| 355 | b, rt := nestedBackend(t, ctl, unit) |
| 356 | sink := &eventCapture{} |
| 357 | b.Logs = sink |
| 358 | var stopServing func() |
| 359 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 360 | defer func() { |
| 361 | if stopServing != nil { |
| 362 | stopServing() |
| 363 | } |
| 364 | }() |
| 365 | unit.set("active", "inv1", true, true) |
| 366 | p, _, err := b.Start(context.Background(), rt) |
| 367 | if err != nil { |
| 368 | t.Fatal(err) |
| 369 | } |
| 370 | up, ok := p.(*unitProc) |
| 371 | if !ok { |
| 372 | t.Fatalf("proc = %T", p) |
| 373 | } |
| 374 | rec := instancectl.ExitRecord{ |
| 375 | Invocation: up.invocation, |
| 376 | Result: "boom\nboom", // a newline: must never reach the event line raw |
| 377 | Status: strings.Repeat("9", 200), // far past recordToken's 32-byte bound |
| 378 | Code: "exi\x00ted", // a control character |
| 379 | } |
| 380 | raw, err := json.Marshal(rec) |
| 381 | if err != nil { |
| 382 | t.Fatal(err) |
| 383 | } |
| 384 | recordPath := filepath.Join(up.runDir, instancectl.ExitRecordName(instancectl.RoleRun, up.invocation)) |
| 385 | if err := os.WriteFile(recordPath, raw, 0o600); err != nil { |
| 386 | t.Fatal(err) |
| 387 | } |
| 388 | unit.set("inactive", "inv1", false, false) |
| 389 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 390 | ev, ok := sink.last() |
| 391 | if !ok { |
| 392 | t.Fatal("no event emitted") |
| 393 | } |
| 394 | if ev.level != "warn" { |
| 395 | t.Errorf("level = %q, want warn", ev.level) |
| 396 | } |
| 397 | if ev.msg != "instance exited: exit ? (?)" { |
| 398 | t.Errorf("msg = %q, want %q", ev.msg, "instance exited: exit ? (?)") |
| 399 | } |
| 400 | if strings.Contains(ev.msg, "\n") { |
| 401 | t.Errorf("event line contains a newline: %q", ev.msg) |
| 402 | } |
| 403 | if _, err := os.Stat(recordPath); !os.IsNotExist(err) { |
| 404 | t.Errorf("exit record was not removed after being consumed: %v", err) |
| 405 | } |
| 406 | } |
| 407 | |
| 408 | // TestStartUnitWithEmptyInvocationFailsCleanly covers the case the control |
| 409 | // service reports no invocation at all: the unit already exited before its |
| 410 | // start was observed (plan 2 final review). The wake must fail with a |
| 411 | // diagnosis naming that, and, because nothing is running, no unitProc |
| 412 | // is ever created to observe, so Control must never see a stop. |
| 413 | func TestStartUnitWithEmptyInvocationFailsCleanly(t *testing.T) { |
| 414 | ctl := &fakeControl{inv: ""} |
| 415 | unit := &fakeUnit{} |
| 416 | b, rt := nestedBackend(t, ctl, unit) |
| 417 | if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "exited before it was observed") { |
| 418 | t.Fatalf("Start = %v", err) |
| 419 | } |
| 420 | if strings.Contains(ctl.joined(), "stop") { |
| 421 | t.Errorf("nothing was ever running; nothing should have been stopped: %s", ctl.joined()) |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | // TestStopKeepsListenersOpenUntilTerminationIsProven covers Stop's |
| 426 | // ordering contract: an unproven stop (ErrStillRunning) must leave the |
| 427 | // control and log listeners open, since a writer that may still exist must |
| 428 | // still be able to reach them, and only a stop that observes done may |
| 429 | // close them. |
| 430 | func TestStopKeepsListenersOpenUntilTerminationIsProven(t *testing.T) { |
| 431 | ctl := &fakeControl{inv: "inv1"} |
| 432 | unit := &fakeUnit{} |
| 433 | b, rt := nestedBackend(t, ctl, unit) |
| 434 | var stopServing func() |
| 435 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 436 | defer func() { |
| 437 | if stopServing != nil { |
| 438 | stopServing() |
| 439 | } |
| 440 | }() |
| 441 | unit.set("active", "inv1", true, true) |
| 442 | p, _, err := b.Start(context.Background(), rt) |
| 443 | if err != nil { |
| 444 | t.Fatal(err) |
| 445 | } |
| 446 | logSocket := b.logSocket(rt.Host, instancectl.RoleRun) |
| 447 | |
| 448 | ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) |
| 449 | defer cancel() |
| 450 | if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) { |
| 451 | t.Fatalf("Stop = %v, want ErrStillRunning", err) |
| 452 | } |
| 453 | c, err := net.Dial("unix", logSocket) |
| 454 | if err != nil { |
| 455 | t.Fatalf("log socket refused a dial after an unproven stop: %v", err) |
| 456 | } |
| 457 | c.Close() |
| 458 | |
| 459 | unit.set("inactive", "inv1", false, false) |
| 460 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 461 | ctx2, cancel2 := context.WithTimeout(context.Background(), time.Second) |
| 462 | defer cancel2() |
| 463 | if err := p.Stop(ctx2); err != nil { |
| 464 | t.Errorf("second Stop after exit: %v", err) |
| 465 | } |
| 466 | if c2, err := net.Dial("unix", logSocket); err == nil { |
| 467 | c2.Close() |
| 468 | t.Error("log socket still accepts a dial after Stop closed the listeners") |
| 469 | } |
| 470 | } |
| 471 | |
| 472 | // TestStopClosesListenersAfterDoneWithNoSecondStopCall pins the fix for the |
| 473 | // leak Stop's ctx.Done branch used to have: several real callers (the |
| 474 | // sidecar co-stop, a discarded sidecar, a failed health wait) never call |
| 475 | // Stop a second time once it returns ErrStillRunning. The listeners must |
| 476 | // still close once termination is proven, with nobody asking again. |
| 477 | func TestStopClosesListenersAfterDoneWithNoSecondStopCall(t *testing.T) { |
| 478 | t.Parallel() |
| 479 | ctl := &fakeControl{inv: "inv1"} |
| 480 | unit := &fakeUnit{} |
| 481 | b, rt := nestedBackend(t, ctl, unit) |
| 482 | var stopServing func() |
| 483 | ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| 484 | defer func() { |
| 485 | if stopServing != nil { |
| 486 | stopServing() |
| 487 | } |
| 488 | }() |
| 489 | unit.set("active", "inv1", true, true) |
| 490 | p, _, err := b.Start(context.Background(), rt) |
| 491 | if err != nil { |
| 492 | t.Fatal(err) |
| 493 | } |
| 494 | logSocket := b.logSocket(rt.Host, instancectl.RoleRun) |
| 495 | |
| 496 | ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) |
| 497 | defer cancel() |
| 498 | if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) { |
| 499 | t.Fatalf("Stop = %v, want ErrStillRunning", err) |
| 500 | } |
| 501 | |
| 502 | unit.set("inactive", "inv1", false, false) |
| 503 | waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| 504 | |
| 505 | deadline := time.Now().Add(2 * time.Second) |
| 506 | for { |
| 507 | c, err := net.Dial("unix", logSocket) |
| 508 | if err != nil { |
| 509 | break // the listener closed on its own; no second Stop call was made |
| 510 | } |
| 511 | c.Close() |
| 512 | if time.Now().After(deadline) { |
| 513 | t.Fatal("log socket still accepts a dial 2s after the unit exited, with no second Stop call") |
| 514 | } |
| 515 | time.Sleep(10 * time.Millisecond) |
| 516 | } |
| 517 | } |
| 518 | |
| 519 | // TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain |
| 520 | // covers the ordering rule: Control.Stop for the seed unit runs, and is |
| 521 | // proven by an empty cgroup, BEFORE any database cleanup, and when the |
| 522 | // cgroup never reports empty, nothing is removed at all. The wait is |
| 523 | // bounded by roleEmptyBound (actStopTimeout in production), which this test |
| 524 | // genuinely waits out, shrunk to a fraction of a second. |
| 525 | func TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain(t *testing.T) { |
| 526 | if roleEmptyBound != actStopTimeout { |
| 527 | t.Fatalf("roleEmptyBound = %v, want actStopTimeout (%v) outside a test", roleEmptyBound, actStopTimeout) |
| 528 | } |
| 529 | old := roleEmptyBound |
| 530 | roleEmptyBound = 300 * time.Millisecond |
| 531 | t.Cleanup(func() { roleEmptyBound = old }) |
| 532 | ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}} |
| 533 | unit := &fakeUnit{} |
| 534 | b, rt := nestedBackend(t, ctl, unit) |
| 535 | rt.Seed = `["seed","--fast"]` |
| 536 | // Stand in for a real seed process that wrote its half-built database |
| 537 | // before dying, created only once Seed actually runs, so the PRE-seed |
| 538 | // stat that decides seedOwnsDB still sees it absent (seedOwnsDB=true). |
| 539 | ctl.onSeed = func() { |
| 540 | if err := os.WriteFile(rt.DBPath, []byte("half-built"), 0o600); err != nil { |
| 541 | t.Fatal(err) |
| 542 | } |
| 543 | } |
| 544 | // The cgroup stays populated for the whole wait: a process from the |
| 545 | // seed survived SIGKILL and is still there when the bound elapses. |
| 546 | unit.set("active", "s1", true, true) |
| 547 | err := b.Seed(context.Background(), rt) |
| 548 | if err == nil || !strings.Contains(err.Error(), "processes remain") { |
| 549 | t.Fatalf("Seed = %v, want a processes-remain error", err) |
| 550 | } |
| 551 | if !strings.Contains(ctl.joined(), "stop h.example.com seed") { |
| 552 | t.Errorf("seed unit was never stopped before cleanup was attempted: %s", ctl.joined()) |
| 553 | } |
| 554 | if _, err := os.Stat(rt.DBPath); err != nil { |
| 555 | t.Errorf("the database was removed despite its processes remaining: %v", err) |
| 556 | } |
| 557 | } |
| 558 | |
| 559 | // switchableEvents is a CgroupEvents seam a test can turn from "cannot |
| 560 | // read" (an unproven cgroup, answered at once rather than after |
| 561 | // actStopTimeout) to "gone". |
| 562 | type switchableEvents struct { |
| 563 | mu sync.Mutex |
| 564 | err error |
| 565 | reads []string |
| 566 | } |
| 567 | |
| 568 | func (s *switchableEvents) set(err error) { s.mu.Lock(); s.err = err; s.mu.Unlock() } |
| 569 | func (s *switchableEvents) events(slice, unit string) (bool, bool, error) { |
| 570 | s.mu.Lock() |
| 571 | defer s.mu.Unlock() |
| 572 | s.reads = append(s.reads, unit) |
| 573 | if s.err != nil { |
| 574 | return false, false, s.err |
| 575 | } |
| 576 | return false, false, nil |
| 577 | } |
| 578 | func (s *switchableEvents) read() []string { |
| 579 | s.mu.Lock() |
| 580 | defer s.mu.Unlock() |
| 581 | return append([]string(nil), s.reads...) |
| 582 | } |
| 583 | |
| 584 | // Task 8 fix round 1, finding 2: a nested seed writes its obligation ahead |
| 585 | // as a flat one does, before root is asked to run anything, and a seed |
| 586 | // whose unit is proven empty removes it. |
| 587 | func TestANestedSeedRecordsItsObligationBeforeItRuns(t *testing.T) { |
| 588 | ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "success"}} |
| 589 | unit := &fakeUnit{} |
| 590 | b, rt := nestedBackend(t, ctl, unit) |
| 591 | rt.Seed = `["seed"]` |
| 592 | ob := rt.DBPath + seedUnprovenSuffix |
| 593 | seen := false |
| 594 | ctl.onSeed = func() { |
| 595 | _, err := os.Stat(ob) |
| 596 | seen = err == nil |
| 597 | } |
| 598 | unit.set("inactive", "", false, false) |
| 599 | if err := b.Seed(context.Background(), rt); err != nil { |
| 600 | t.Fatalf("Seed: %v", err) |
| 601 | } |
| 602 | if !seen { |
| 603 | t.Fatal("root ran the seed with no obligation on disk") |
| 604 | } |
| 605 | if _, err := os.Stat(ob); !os.IsNotExist(err) { |
| 606 | t.Fatalf("obligation still there (%v) after a proven seed", err) |
| 607 | } |
| 608 | } |
| 609 | |
| 610 | // Finding 2: a nested seed whose unit is not proven empty (failed or |
| 611 | // successful) keeps its database AND leaves the obligation. The next wake |
| 612 | // proves the nested seed's own unit empty (stopped, then its cgroup read), |
| 613 | // never carlos-exec@<host>, which a nested seed never ran in: while that |
| 614 | // proof fails the wake is refused and the database is untouched; once it |
| 615 | // holds, the database and the obligation go and the wake is fresh. |
| 616 | func TestAnUnprovenNestedSeedLeavesItsObligationAndTheNextWakeProvesItsUnit(t *testing.T) { |
| 617 | for _, result := range []string{"exit-code", "success"} { |
| 618 | t.Run(result, func(t *testing.T) { |
| 619 | ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: result, Status: "3"}} |
| 620 | unit := &fakeUnit{} |
| 621 | b, rt := nestedBackend(t, ctl, unit) |
| 622 | rt.Seed = `["seed"]` |
| 623 | ev := &switchableEvents{err: errors.New("cgroup.events: unreadable")} |
| 624 | b.CgroupEvents = ev.events |
| 625 | var flat []string |
| 626 | b.cgroupKill = func(_ context.Context, _ registry.Route, child string) error { |
| 627 | flat = append(flat, child) |
| 628 | return errors.New("a nested seed never ran in carlos-exec@<host>") |
| 629 | } |
| 630 | swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) { |
| 631 | return false, nil |
| 632 | }) |
| 633 | ctl.onSeed = func() { |
| 634 | if err := os.WriteFile(rt.DBPath, []byte("half"), 0o600); err != nil { |
| 635 | t.Fatal(err) |
| 636 | } |
| 637 | } |
| 638 | ob := rt.DBPath + seedUnprovenSuffix |
| 639 | |
| 640 | if err := b.Seed(context.Background(), rt); !errors.Is(err, ErrSeedUnproven) { |
| 641 | t.Fatalf("Seed = %v, want ErrSeedUnproven", err) |
| 642 | } |
| 643 | if _, err := os.Stat(ob); err != nil { |
| 644 | t.Fatalf("no obligation beside the nested seed's database: %v", err) |
| 645 | } |
| 646 | |
| 647 | if fresh, err := b.Restore(context.Background(), rt); !errors.Is(err, ErrSeedUnproven) || fresh { |
| 648 | t.Fatalf("Restore = %v, %v; want a refusal while the seed unit is unproven", fresh, err) |
| 649 | } |
| 650 | if got, _ := os.ReadFile(rt.DBPath); string(got) != "half" { |
| 651 | t.Fatalf("database = %q after a refused wake, want it untouched", got) |
| 652 | } |
| 653 | |
| 654 | ev.set(nil) |
| 655 | before := strings.Count(ctl.joined(), "stop h.example.com seed") |
| 656 | if fresh, err := b.Restore(context.Background(), rt); err != nil || !fresh { |
| 657 | t.Fatalf("Restore = %v, %v; want fresh once the seed unit is proven empty", fresh, err) |
| 658 | } |
| 659 | if strings.Count(ctl.joined(), "stop h.example.com seed") <= before { |
| 660 | t.Fatalf("the wake did not stop the seed unit before its proof: %s", ctl.joined()) |
| 661 | } |
| 662 | if len(flat) != 0 { |
| 663 | t.Fatalf("the nested seed's proof asked carlos-exec@<host> (%q)", flat) |
| 664 | } |
| 665 | if reads := ev.read(); len(reads) == 0 || reads[len(reads)-1] != instancectl.UnitName(instancectl.RoleSeed, rt.Host) { |
| 666 | t.Fatalf("cgroup reads %q, want the seed unit's", reads) |
| 667 | } |
| 668 | for _, p := range []string{rt.DBPath, ob} { |
| 669 | if _, err := os.Stat(p); !os.IsNotExist(err) { |
| 670 | t.Fatalf("%s still there (%v) after the proof held", filepath.Base(p), err) |
| 671 | } |
| 672 | } |
| 673 | }) |
| 674 | } |
| 675 | } |
| 676 | |
| 677 | // The obligation sits beside the database, on ground the instance can |
| 678 | // write. A FIFO planted at its name must not hang the wake (the read never |
| 679 | // waits for a writer), and it must not hold the row down either: it is an |
| 680 | // obligation that does not say what the seed owned, so it is proven, the |
| 681 | // database is kept, and the planted name is removed. |
| 682 | func TestAPlantedFIFOAtTheObligationIsAnsweredNotWaitedOn(t *testing.T) { |
| 683 | const host = "fifo-obligation.example.com" |
| 684 | b, _ := delegatedBackend(t, host) |
| 685 | b.cgroupKill = func(context.Context, registry.Route, string) error { return nil } |
| 686 | rt := registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`} |
| 687 | db := b.pathsFor(rt).dbPath |
| 688 | if err := os.MkdirAll(filepath.Dir(db), 0o755); err != nil { |
| 689 | t.Fatal(err) |
| 690 | } |
| 691 | if err := os.WriteFile(db, []byte("data"), 0o600); err != nil { |
| 692 | t.Fatal(err) |
| 693 | } |
| 694 | if err := unix.Mkfifo(db+seedUnprovenSuffix, 0o600); err != nil { |
| 695 | t.Fatalf("Mkfifo: %v", err) |
| 696 | } |
| 697 | type result struct { |
| 698 | fresh bool |
| 699 | err error |
| 700 | } |
| 701 | done := make(chan result, 1) |
| 702 | go func() { |
| 703 | fresh, err := b.Restore(context.Background(), rt) |
| 704 | done <- result{fresh, err} |
| 705 | }() |
| 706 | select { |
| 707 | case r := <-done: |
| 708 | if r.err != nil || r.fresh { |
| 709 | t.Fatalf("Restore = %v, %v; want the warm database once the proof holds", r.fresh, r.err) |
| 710 | } |
| 711 | case <-time.After(5 * time.Second): |
| 712 | t.Fatal("Restore hung on a FIFO planted at the obligation's name") |
| 713 | } |
| 714 | if got, err := os.ReadFile(db); err != nil || string(got) != "data" { |
| 715 | t.Fatalf("database = %q, %v; want it kept: the planted name says nothing the seed owned", got, err) |
| 716 | } |
| 717 | if _, err := os.Lstat(db + seedUnprovenSuffix); !os.IsNotExist(err) { |
| 718 | t.Fatalf("the planted name is still there (%v)", err) |
| 719 | } |
| 720 | } |
| 721 | |
| 722 | // TestPrepareRemovesAStaleSocketFile covers prepare's stale-socket removal: |
| 723 | // a file left behind at rt.Socket by an earlier, SIGKILLed invocation must |
| 724 | // not survive into the next one, or every later wake's health check would |
| 725 | // fail with nothing but a dial timeout. |
| 726 | func TestPrepareRemovesAStaleSocketFile(t *testing.T) { |
| 727 | ctl := &fakeControl{inv: "inv1"} |
| 728 | unit := &fakeUnit{} |
| 729 | b, rt := nestedBackend(t, ctl, unit) |
| 730 | if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil { |
| 731 | t.Fatal(err) |
| 732 | } |
| 733 | if err := os.WriteFile(rt.Socket, []byte("stale"), 0o600); err != nil { |
| 734 | t.Fatal(err) |
| 735 | } |
| 736 | _, ctl2, logs, _, err := b.prepare(rt, instancectl.RoleRun) |
| 737 | if err != nil { |
| 738 | t.Fatalf("prepare: %v", err) |
| 739 | } |
| 740 | defer logs.Close() |
| 741 | if ctl2 != nil { |
| 742 | defer ctl2.Close() |
| 743 | } |
| 744 | if _, err := os.Stat(rt.Socket); !os.IsNotExist(err) { |
| 745 | t.Errorf("stale socket file was not removed by prepare: %v", err) |
| 746 | } |
| 747 | } |
| 748 | |
| 749 | // TestPrepareRefusesAFIFOAtTheAdminTokenPath pins EnsureAdminTokenBeneath's |
| 750 | // use from the nested prepare: the token file lives on the same |
| 751 | // tenant-writable ground as everything else prepare touches, so a FIFO |
| 752 | // planted at that name must fail prepare fast rather than hang it forever |
| 753 | // behind the token mint's own read. |
| 754 | func TestPrepareRefusesAFIFOAtTheAdminTokenPath(t *testing.T) { |
| 755 | ctl := &fakeControl{inv: "inv1"} |
| 756 | unit := &fakeUnit{} |
| 757 | b, rt := nestedBackend(t, ctl, unit) |
| 758 | instDir := provision.InstanceDir(b.DataDir, rt.Host) |
| 759 | if err := os.MkdirAll(instDir, 0o700); err != nil { |
| 760 | t.Fatal(err) |
| 761 | } |
| 762 | if err := unix.Mkfifo(filepath.Join(instDir, rt.Host+".env"), 0o600); err != nil { |
| 763 | t.Fatalf("Mkfifo: %v", err) |
| 764 | } |
| 765 | done := make(chan error, 1) |
| 766 | go func() { |
| 767 | _, _, _, _, err := b.prepare(rt, instancectl.RoleRun) |
| 768 | done <- err |
| 769 | }() |
| 770 | select { |
| 771 | case err := <-done: |
| 772 | if err == nil { |
| 773 | t.Fatal("prepare against a FIFO'd admin token file succeeded, want an error") |
| 774 | } |
| 775 | case <-time.After(2 * time.Second): |
| 776 | t.Fatal("prepare blocked on a FIFO'd admin token file instead of refusing it") |
| 777 | } |
| 778 | } |
| 779 | |
| 780 | // serveOK binds a unix socket at path and answers 200 to every request, so |
| 781 | // waitHealthyCtx passes without a real unit. |
| 782 | func serveOK(t *testing.T, path string) func() { |
| 783 | t.Helper() |
| 784 | if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { |
| 785 | t.Fatal(err) |
| 786 | } |
| 787 | ln, err := net.Listen("unix", path) |
| 788 | if err != nil { |
| 789 | t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err) |
| 790 | } |
| 791 | srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })} |
| 792 | go srv.Serve(ln) //nolint:errcheck |
| 793 | return func() { srv.Close(); ln.Close() } |
| 794 | } |
| 795 | |
amadan