| 1 | package activator |
| 2 | |
| 3 | // Unit-backed tenant tests: the same state machine (activator.go) driving a |
| 4 | // carlos-app@ unit through the Systemctl seam instead of an exec'd child — |
| 5 | // no systemd, no AWS. The end-to-end test at the bottom proves the edge |
| 6 | // shape: a request for a sleeping unit tenant is HELD while the "unit" |
| 7 | // starts and turns healthy, then forwarded. |
| 8 | |
| 9 | import ( |
| 10 | "context" |
| 11 | "errors" |
| 12 | "fmt" |
| 13 | "io" |
| 14 | "net" |
| 15 | "net/http" |
| 16 | "net/http/httptest" |
| 17 | "os" |
| 18 | "path/filepath" |
| 19 | "sync" |
| 20 | "testing" |
| 21 | "time" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/registry" |
| 24 | ) |
| 25 | |
| 26 | // fakeSystemctl records verbs and optionally runs hooks — the Proc-style |
| 27 | // seam's test double. Hooks let the end-to-end test bind/close a real |
| 28 | // listener exactly when the "unit" would. |
| 29 | type fakeSystemctl struct { |
| 30 | mu sync.Mutex |
| 31 | calls []string |
| 32 | startErr error |
| 33 | onStart func() // runs on "start", outside the lock |
| 34 | onStop func() // runs on "stop", outside the lock |
| 35 | } |
| 36 | |
| 37 | func (f *fakeSystemctl) run(_ context.Context, verb, unit string) error { |
| 38 | f.mu.Lock() |
| 39 | f.calls = append(f.calls, verb+" "+unit) |
| 40 | err := f.startErr |
| 41 | onStart, onStop := f.onStart, f.onStop |
| 42 | f.mu.Unlock() |
| 43 | if verb == "start" { |
| 44 | if err != nil { |
| 45 | return err |
| 46 | } |
| 47 | if onStart != nil { |
| 48 | onStart() |
| 49 | } |
| 50 | } |
| 51 | if verb == "stop" && onStop != nil { |
| 52 | onStop() |
| 53 | } |
| 54 | return nil |
| 55 | } |
| 56 | |
| 57 | func (f *fakeSystemctl) recorded() []string { |
| 58 | f.mu.Lock() |
| 59 | defer f.mu.Unlock() |
| 60 | return append([]string(nil), f.calls...) |
| 61 | } |
| 62 | |
| 63 | // testUnitRoute is testRoute's unit-backed sibling: a whole-app tenant |
| 64 | // (carlos-app@<host>) flagged for hibernation. |
| 65 | func testUnitRoute(host string) registry.Route { |
| 66 | return registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/apps/" + host + ".sock", |
| 67 | Hibernate: true, Backing: registry.BackingUnit} |
| 68 | } |
| 69 | |
| 70 | // testUnitActivator wires a Service over a SplitBackend: the usual |
| 71 | // fakeBackend on the exec side, a UnitBackend with the fake systemctl and an |
| 72 | // instant health gate on the unit side. |
| 73 | func testUnitActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeSystemctl, *fakeClock, *registry.Registry) { |
| 74 | t.Helper() |
| 75 | eb := &fakeBackend{} |
| 76 | sc := &fakeSystemctl{} |
| 77 | ub := &UnitBackend{ |
| 78 | Systemctl: sc.run, |
| 79 | OptDir: t.TempDir(), |
| 80 | WaitHealthy: func(context.Context, registry.Route) error { return nil }, |
| 81 | } |
| 82 | clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| 83 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 84 | if err != nil { |
| 85 | t.Fatalf("open registry: %v", err) |
| 86 | } |
| 87 | a := New(Config{Bucket: "test-bucket", Idle: idle}, &SplitBackend{Exec: eb, Unit: ub}, reg) |
| 88 | a.now = clk.now |
| 89 | return a, eb, sc, clk, reg |
| 90 | } |
| 91 | |
| 92 | // The whole-app hibernation cycle: a request starts the unit (systemctl |
| 93 | // start, health-gated); the idle window stops it (systemctl stop); the next |
| 94 | // request starts it again. The exec side is never touched. |
| 95 | func TestUnitTenantWakeIdleSleepCycle(t *testing.T) { |
| 96 | t.Parallel() |
| 97 | a, eb, sc, clk, _ := testUnitActivator(t, 10*time.Second) |
| 98 | route := testUnitRoute("kass.example.com") |
| 99 | unit := UnitName(route.Host) |
| 100 | |
| 101 | release, err := a.beginRequest(context.Background(), route) |
| 102 | if err != nil { |
| 103 | t.Fatalf("beginRequest: %v", err) |
| 104 | } |
| 105 | release() |
| 106 | if got := sc.recorded(); len(got) != 1 || got[0] != "start "+unit { |
| 107 | t.Fatalf("wake systemctl calls = %v, want [start %s]", got, unit) |
| 108 | } |
| 109 | |
| 110 | // Not yet idle: the unit stays up. |
| 111 | clk.advance(9 * time.Second) |
| 112 | a.sweepOnce() |
| 113 | if a.entryCount() != 1 { |
| 114 | t.Fatal("unit tenant hibernated before its idle window elapsed") |
| 115 | } |
| 116 | |
| 117 | // Idle: stopped via systemctl, entry forgotten. |
| 118 | clk.advance(2 * time.Second) |
| 119 | a.sweepOnce() |
| 120 | waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 }) |
| 121 | if got := sc.recorded(); len(got) != 2 || got[1] != "stop "+unit { |
| 122 | t.Fatalf("sleep systemctl calls = %v, want [... stop %s]", got, unit) |
| 123 | } |
| 124 | |
| 125 | // Re-wake: a fresh start. |
| 126 | release, err = a.beginRequest(context.Background(), route) |
| 127 | if err != nil { |
| 128 | t.Fatalf("beginRequest after sleep: %v", err) |
| 129 | } |
| 130 | release() |
| 131 | if got := sc.recorded(); len(got) != 3 || got[2] != "start "+unit { |
| 132 | t.Fatalf("re-wake systemctl calls = %v, want a second start", got) |
| 133 | } |
| 134 | |
| 135 | // The exec backend was never involved: no lease claims, no spawns. |
| 136 | if acquires, restores, starts := eb.counts(); acquires != 0 || restores != 0 || starts != 0 { |
| 137 | t.Fatalf("unit tenant leaked into the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts) |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | // SplitBackend dispatch, the other direction: an exec-backed route wakes |
| 142 | // through the fake exec backend and never touches systemctl. |
| 143 | func TestSplitBackendExecRouteBypassesUnit(t *testing.T) { |
| 144 | t.Parallel() |
| 145 | a, eb, sc, _, _ := testUnitActivator(t, time.Minute) |
| 146 | |
| 147 | release, err := a.beginRequest(context.Background(), testRoute("plain.example.com")) |
| 148 | if err != nil { |
| 149 | t.Fatalf("beginRequest: %v", err) |
| 150 | } |
| 151 | release() |
| 152 | if acquires, restores, starts := eb.counts(); acquires != 1 || restores != 1 || starts != 1 { |
| 153 | t.Fatalf("exec route skipped the exec backend: acquires=%d restores=%d starts=%d", acquires, restores, starts) |
| 154 | } |
| 155 | if got := sc.recorded(); len(got) != 0 { |
| 156 | t.Fatalf("exec route touched systemctl: %v", got) |
| 157 | } |
| 158 | } |
| 159 | |
| 160 | // Restore and Seed dispatch by Backing like every other SplitBackend method — |
| 161 | // and a unit tenant, whose Restore is never fresh, never reaches a seed even |
| 162 | // with one declared on its route. |
| 163 | func TestSplitBackendDispatchesFreshAndSeedByBacking(t *testing.T) { |
| 164 | eb := &fakeBackend{restoreFresh: true} |
| 165 | ub := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()} |
| 166 | sb := &SplitBackend{Exec: eb, Unit: ub} |
| 167 | ctx := context.Background() |
| 168 | |
| 169 | if fresh, err := sb.Restore(ctx, testRoute("plain.example.com")); err != nil || !fresh { |
| 170 | t.Fatalf("exec route Restore = %v, %v; want true, nil", fresh, err) |
| 171 | } |
| 172 | unitRoute := testUnitRoute("kass.example.com") |
| 173 | unitRoute.Seed = `["seed"]` |
| 174 | if fresh, err := sb.Restore(ctx, unitRoute); err != nil || fresh { |
| 175 | t.Fatalf("unit route Restore = %v, %v; want false, nil", fresh, err) |
| 176 | } |
| 177 | |
| 178 | if err := sb.Seed(ctx, testRoute("plain.example.com")); err != nil { |
| 179 | t.Fatalf("exec route Seed: %v", err) |
| 180 | } |
| 181 | if got := eb.seedCount(); got != 1 { |
| 182 | t.Fatalf("exec-side seeds = %d, want 1", got) |
| 183 | } |
| 184 | if err := sb.Seed(ctx, unitRoute); err != nil { |
| 185 | t.Fatalf("unit route Seed: %v", err) |
| 186 | } |
| 187 | if got := eb.seedCount(); got != 1 { |
| 188 | t.Fatalf("a unit tenant's seed leaked into the exec backend (seeds=%d)", got) |
| 189 | } |
| 190 | } |
| 191 | |
| 192 | // A unit that starts but never turns healthy fails the wake cleanly: the |
| 193 | // unit is put back down (no half-up tenant lingering behind a 503) and no |
| 194 | // entry is left, so the next request retries from scratch. |
| 195 | func TestUnitTenantFailedHealthStopsUnit(t *testing.T) { |
| 196 | route := testUnitRoute("broken.example.com") |
| 197 | unit := UnitName(route.Host) |
| 198 | |
| 199 | sc := &fakeSystemctl{} |
| 200 | ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir(), |
| 201 | WaitHealthy: func(context.Context, registry.Route) error { return errors.New("never healthy") }} |
| 202 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 203 | if err != nil { |
| 204 | t.Fatalf("open registry: %v", err) |
| 205 | } |
| 206 | a := New(Config{Bucket: "test-bucket", Idle: time.Minute}, &SplitBackend{Exec: &fakeBackend{}, Unit: ub}, reg) |
| 207 | |
| 208 | if _, err := a.beginRequest(context.Background(), route); err == nil { |
| 209 | t.Fatal("wake succeeded despite the health gate failing") |
| 210 | } |
| 211 | if got := sc.recorded(); len(got) != 2 || got[0] != "start "+unit || got[1] != "stop "+unit { |
| 212 | t.Fatalf("systemctl calls = %v, want [start %s, stop %s]", got, unit, unit) |
| 213 | } |
| 214 | if a.entryCount() != 0 { |
| 215 | t.Fatal("failed unit wake left an entry behind") |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // A start that fails outright (unit file missing, systemd unreachable) |
| 220 | // fails the wake with no stop attempt and no entry. |
| 221 | func TestUnitTenantStartFailureFailsWake(t *testing.T) { |
| 222 | a, _, sc, _, _ := testUnitActivator(t, time.Minute) |
| 223 | sc.startErr = errors.New("Unit carlos-app@gone.example.com.service not found") |
| 224 | |
| 225 | if _, err := a.beginRequest(context.Background(), testUnitRoute("gone.example.com")); err == nil { |
| 226 | t.Fatal("wake succeeded despite systemctl start failing") |
| 227 | } |
| 228 | if got := sc.recorded(); len(got) != 1 { |
| 229 | t.Fatalf("systemctl calls = %v, want just the failed start", got) |
| 230 | } |
| 231 | if a.entryCount() != 0 { |
| 232 | t.Fatal("failed start left an entry behind") |
| 233 | } |
| 234 | } |
| 235 | |
| 236 | // Shutdown puts unit tenants down like exec children — deliberately: with |
| 237 | // the tenant's carlos-app@.socket unit enabled, systemd's wake-on-connect |
| 238 | // brings it back on the next connection even while the agent is away, and |
| 239 | // the next agent's health-gated wake re-adopts it (start on a running unit |
| 240 | // is a no-op). |
| 241 | func TestUnitTenantShutdownStopsUnit(t *testing.T) { |
| 242 | a, _, sc, _, _ := testUnitActivator(t, time.Hour) |
| 243 | route := testUnitRoute("kass.example.com") |
| 244 | |
| 245 | release, err := a.beginRequest(context.Background(), route) |
| 246 | if err != nil { |
| 247 | t.Fatalf("beginRequest: %v", err) |
| 248 | } |
| 249 | release() |
| 250 | |
| 251 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 252 | defer cancel() |
| 253 | a.Shutdown(ctx) |
| 254 | if a.entryCount() != 0 { |
| 255 | t.Fatal("shutdown left the unit entry behind") |
| 256 | } |
| 257 | got := sc.recorded() |
| 258 | if len(got) != 2 || got[1] != "stop "+UnitName(route.Host) { |
| 259 | t.Fatalf("systemctl calls = %v, want the shutdown stop", got) |
| 260 | } |
| 261 | } |
| 262 | |
| 263 | // The inert lease/replication seams: everything the state machine calls on |
| 264 | // them must succeed and report healthy, so the machine treats a unit tenant |
| 265 | // exactly like a replicating exec child. (Replication ownership really |
| 266 | // lives with the root carlos-litestream unit — see backend_unit.go.) |
| 267 | func TestUnitBackendInertSeams(t *testing.T) { |
| 268 | b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: t.TempDir()} |
| 269 | ctx := context.Background() |
| 270 | rt := testUnitRoute("inert.example.com") |
| 271 | |
| 272 | lease, err := b.Acquire(ctx, rt) |
| 273 | if err != nil { |
| 274 | t.Fatalf("Acquire: %v", err) |
| 275 | } |
| 276 | rep, err := lease.Replicate(ctx, rt.DBPath, 0) |
| 277 | if err != nil { |
| 278 | t.Fatalf("Replicate: %v", err) |
| 279 | } |
| 280 | if err := rep.Err(); err != nil { |
| 281 | t.Fatalf("Rep.Err: %v", err) |
| 282 | } |
| 283 | if err := rep.Close(ctx); err != nil { |
| 284 | t.Fatalf("Rep.Close: %v", err) |
| 285 | } |
| 286 | if err := lease.Release(ctx); err != nil { |
| 287 | t.Fatalf("Release: %v", err) |
| 288 | } |
| 289 | // Never fresh: the agent cannot see inside a DynamicUser's state dir, so |
| 290 | // it must never claim a unit tenant is on its first boot — that claim is |
| 291 | // what would run an app's seed over live data. |
| 292 | if fresh, err := b.Restore(ctx, rt); err != nil || fresh { |
| 293 | t.Fatalf("Restore = %v, %v; want false, nil", fresh, err) |
| 294 | } |
| 295 | if err := b.Seed(ctx, rt); err != nil { |
| 296 | t.Fatalf("Seed: %v", err) |
| 297 | } |
| 298 | due, err := b.NextDue(ctx, rt) |
| 299 | if err != nil || due != 0 { |
| 300 | t.Fatalf("NextDue = %d, %v; want 0, nil (unit tenants sleep until a request)", due, err) |
| 301 | } |
| 302 | if b.HasSidecar(rt) { |
| 303 | t.Fatal("unit tenants must not report agent-managed sidecars") |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // Worker spawn is exec-child machinery, exactly like sidecar co-spawn: a |
| 308 | // unit tenant that wants a worker runs it inside its own unit. |
| 309 | func TestUnitBackendRefusesWorkers(t *testing.T) { |
| 310 | var b UnitBackend |
| 311 | if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "x"}, nil); err == nil { |
| 312 | t.Fatal("unit-backed tenants have no agent-managed workers, exactly as they have no agent-managed sidecars") |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // LiveVersion reads the same live/<host> symlink layout as ExecBackend, so |
| 317 | // build convergence cycles awake unit tenants onto shipped builds. |
| 318 | func TestUnitBackendLiveVersion(t *testing.T) { |
| 319 | opt := t.TempDir() |
| 320 | b := &UnitBackend{Systemctl: (&fakeSystemctl{}).run, OptDir: opt} |
| 321 | |
| 322 | if v, err := b.LiveVersion("new.example.com"); err != nil || v != "" { |
| 323 | t.Fatalf("LiveVersion(no symlink) = %q, %v; want empty, nil", v, err) |
| 324 | } |
| 325 | verDir := filepath.Join(opt, "versions", "abc1234") |
| 326 | if err := os.MkdirAll(verDir, 0o755); err != nil { |
| 327 | t.Fatal(err) |
| 328 | } |
| 329 | if err := os.MkdirAll(filepath.Join(opt, "live"), 0o755); err != nil { |
| 330 | t.Fatal(err) |
| 331 | } |
| 332 | if err := os.Symlink(filepath.Join(verDir, "app"), filepath.Join(opt, "live", "kass.example.com")); err != nil { |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | if v, err := b.LiveVersion("kass.example.com"); err != nil || v != "abc1234" { |
| 336 | t.Fatalf("LiveVersion = %q, %v; want abc1234", v, err) |
| 337 | } |
| 338 | |
| 339 | // A migrated tenant's link has the account sqid between the version and |
| 340 | // the binary; reading "the parent directory" would return the sqid and |
| 341 | // convince ConvergeDecision this tenant is converged forever. |
| 342 | stageLiveUnderAccount(t, opt, "moved.example.com", "kqmvexample", "def5678", "kass") |
| 343 | if v, err := b.LiveVersion("moved.example.com"); err != nil || v != "def5678" { |
| 344 | t.Fatalf("LiveVersion (account-scoped) = %q, %v; want def5678", v, err) |
| 345 | } |
| 346 | } |
| 347 | |
| 348 | // The end-to-end shape the feature promises: a request for a SLEEPING unit |
| 349 | // tenant is held at the edge while the unit starts and its upstream turns |
| 350 | // healthy (the real TCP prober — no injected health seam here), then |
| 351 | // forwarded; after the idle window the unit is stopped for real (listener |
| 352 | // closed); the next request repeats the cycle. TCP Addr keeps the test |
| 353 | // sandbox-safe; the unix-socket path differs only in the dialer. |
| 354 | func TestUnitTenantWakeServesHeldRequest(t *testing.T) { |
| 355 | // Pre-pick the tenant's port so the route can exist before the "unit" runs. |
| 356 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 357 | if err != nil { |
| 358 | t.Fatalf("pick port: %v", err) |
| 359 | } |
| 360 | addr := ln.Addr().String() |
| 361 | ln.Close() |
| 362 | |
| 363 | // The "unit": the fake systemctl binds/closes a real HTTP listener on |
| 364 | // start/stop, with a deliberate boot delay so the request provably waits |
| 365 | // on the health gate. |
| 366 | var srvMu sync.Mutex |
| 367 | var srv *http.Server |
| 368 | sc := &fakeSystemctl{} |
| 369 | sc.onStart = func() { |
| 370 | time.Sleep(200 * time.Millisecond) // boot time — the edge must hold the request |
| 371 | l, err := net.Listen("tcp", addr) |
| 372 | if err != nil { |
| 373 | t.Errorf("unit listen: %v", err) |
| 374 | return |
| 375 | } |
| 376 | s := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 377 | io.WriteString(w, "kass up") |
| 378 | })} |
| 379 | srvMu.Lock() |
| 380 | srv = s |
| 381 | srvMu.Unlock() |
| 382 | go s.Serve(l) |
| 383 | } |
| 384 | sc.onStop = func() { |
| 385 | srvMu.Lock() |
| 386 | s := srv |
| 387 | srv = nil |
| 388 | srvMu.Unlock() |
| 389 | if s != nil { |
| 390 | s.Close() |
| 391 | } |
| 392 | } |
| 393 | |
| 394 | old := spawnHealthTimeout |
| 395 | spawnHealthTimeout = 5 * time.Second |
| 396 | t.Cleanup(func() { spawnHealthTimeout = old }) |
| 397 | |
| 398 | eb := &fakeBackend{} |
| 399 | ub := &UnitBackend{Systemctl: sc.run, OptDir: t.TempDir()} // nil WaitHealthy: the real prober |
| 400 | clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| 401 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 402 | if err != nil { |
| 403 | t.Fatalf("open registry: %v", err) |
| 404 | } |
| 405 | a := New(Config{Bucket: "test-bucket", Idle: 10 * time.Second}, &SplitBackend{Exec: eb, Unit: ub}, reg) |
| 406 | a.now = clk.now |
| 407 | |
| 408 | route := registry.Route{Host: "kass.example.com", Kind: "instance", Addr: addr, |
| 409 | Hibernate: true, Backing: registry.BackingUnit} |
| 410 | if err := reg.Upsert(route); err != nil { |
| 411 | t.Fatal(err) |
| 412 | } |
| 413 | rt, err := newTestRouter(reg, a) |
| 414 | if err != nil { |
| 415 | t.Fatalf("new router: %v", err) |
| 416 | } |
| 417 | front := httptest.NewServer(rt) |
| 418 | t.Cleanup(front.Close) |
| 419 | t.Cleanup(func() { sc.onStop() }) // in case the test dies mid-cycle |
| 420 | |
| 421 | // Asleep → the request wakes the unit and is answered by it. |
| 422 | if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" { |
| 423 | t.Fatalf("held request = %d %q, want 200 kass up", code, body) |
| 424 | } |
| 425 | |
| 426 | // Idle out → the unit is stopped for real: its listener is gone. |
| 427 | clk.advance(11 * time.Second) |
| 428 | a.sweepOnce() |
| 429 | waitFor(t, "unit hibernation", func() bool { return a.entryCount() == 0 }) |
| 430 | waitFor(t, "listener closed", func() bool { |
| 431 | conn, err := net.DialTimeout("tcp", addr, 100*time.Millisecond) |
| 432 | if err == nil { |
| 433 | conn.Close() |
| 434 | return false |
| 435 | } |
| 436 | return true |
| 437 | }) |
| 438 | |
| 439 | // The next request wakes it again. |
| 440 | if body, code := proxyGet(t, front.URL, "kass.example.com"); code != 200 || body != "kass up" { |
| 441 | t.Fatalf("post-sleep request = %d %q, want 200 kass up", code, body) |
| 442 | } |
| 443 | want := []string{ |
| 444 | "start " + UnitName(route.Host), |
| 445 | "stop " + UnitName(route.Host), |
| 446 | "start " + UnitName(route.Host), |
| 447 | } |
| 448 | if got := sc.recorded(); fmt.Sprint(got) != fmt.Sprint(want) { |
| 449 | t.Fatalf("systemctl calls = %v, want %v", got, want) |
| 450 | } |
| 451 | } |
| 452 | |
amadan