| 1 | package host |
| 2 | |
| 3 | // The wake story, end to end, in one process: an instance asleep → its timer |
| 4 | // comes due → the tick wakes it through the activator → the app's handler |
| 5 | // runs and sees the contract headers → the instance hibernates again once it |
| 6 | // is idle (spec §6.3, §7.1). |
| 7 | // |
| 8 | // The seam is activator.Backend — the same seam internal/activator's own |
| 9 | // tests use, with a fake that models an exec child faithfully enough for this |
| 10 | // story: the instance's unix socket exists ONLY while it is "running", so a |
| 11 | // delivery that did not wake it would find nothing to dial, and the handler |
| 12 | // running at all is proof the wake happened. |
| 13 | |
| 14 | import ( |
| 15 | "context" |
| 16 | "errors" |
| 17 | "net" |
| 18 | "net/http" |
| 19 | "net/http/httptest" |
| 20 | "net/http/httputil" |
| 21 | "path/filepath" |
| 22 | "strconv" |
| 23 | "sync" |
| 24 | "testing" |
| 25 | "time" |
| 26 | |
| 27 | "github.com/carlosframework/platform/internal/activator" |
| 28 | "github.com/carlosframework/platform/internal/edge" |
| 29 | "github.com/carlosframework/platform/internal/registry" |
| 30 | "github.com/carlosframework/platform/internal/release" |
| 31 | "github.com/carlosframework/platform/internal/schedule" |
| 32 | ) |
| 33 | |
| 34 | // --- the fake instance and its backend --------------------------------------- |
| 35 | |
| 36 | // fakeInstanceProc is the spawned child: stopping it takes the socket away, |
| 37 | // which is what "hibernated" means to anything downstream. |
| 38 | type fakeInstanceProc struct { |
| 39 | inst *fakeInstance |
| 40 | } |
| 41 | |
| 42 | func (p *fakeInstanceProc) Stop(ctx context.Context) error { return p.inst.stop(ctx) } |
| 43 | func (p *fakeInstanceProc) Exited() bool { return !p.inst.running() } |
| 44 | |
| 45 | // fakeInstance is an app listening on a unix socket, started and stopped by |
| 46 | // the backend below. It records every request it served. |
| 47 | type fakeInstance struct { |
| 48 | socket string |
| 49 | handler http.Handler |
| 50 | |
| 51 | mu sync.Mutex |
| 52 | srv *http.Server |
| 53 | starts int |
| 54 | conns int // accepted connections — the transport-reuse assertion |
| 55 | requests []*http.Request |
| 56 | } |
| 57 | |
| 58 | func (f *fakeInstance) running() bool { |
| 59 | f.mu.Lock() |
| 60 | defer f.mu.Unlock() |
| 61 | return f.srv != nil |
| 62 | } |
| 63 | |
| 64 | func (f *fakeInstance) start() error { |
| 65 | ln, err := net.Listen("unix", f.socket) |
| 66 | if err != nil { |
| 67 | return err |
| 68 | } |
| 69 | srv := &http.Server{ |
| 70 | Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 71 | f.mu.Lock() |
| 72 | f.requests = append(f.requests, r.Clone(context.Background())) |
| 73 | f.mu.Unlock() |
| 74 | f.handler.ServeHTTP(w, r) |
| 75 | }), |
| 76 | ConnState: func(_ net.Conn, st http.ConnState) { |
| 77 | if st == http.StateNew { |
| 78 | f.mu.Lock() |
| 79 | f.conns++ |
| 80 | f.mu.Unlock() |
| 81 | } |
| 82 | }, |
| 83 | } |
| 84 | f.mu.Lock() |
| 85 | f.srv = srv |
| 86 | f.starts++ |
| 87 | f.mu.Unlock() |
| 88 | go srv.Serve(ln) |
| 89 | return nil |
| 90 | } |
| 91 | |
| 92 | func (f *fakeInstance) stop(ctx context.Context) error { |
| 93 | f.mu.Lock() |
| 94 | srv := f.srv |
| 95 | f.srv = nil |
| 96 | f.mu.Unlock() |
| 97 | if srv == nil { |
| 98 | return nil |
| 99 | } |
| 100 | return srv.Close() |
| 101 | } |
| 102 | |
| 103 | func (f *fakeInstance) served() []*http.Request { |
| 104 | f.mu.Lock() |
| 105 | defer f.mu.Unlock() |
| 106 | return append([]*http.Request(nil), f.requests...) |
| 107 | } |
| 108 | |
| 109 | func (f *fakeInstance) startCount() int { |
| 110 | f.mu.Lock() |
| 111 | defer f.mu.Unlock() |
| 112 | return f.starts |
| 113 | } |
| 114 | |
| 115 | func (f *fakeInstance) connCount() int { |
| 116 | f.mu.Lock() |
| 117 | defer f.mu.Unlock() |
| 118 | return f.conns |
| 119 | } |
| 120 | |
| 121 | // fakeWakeBackend is activator.Backend over one fakeInstance: no S3, no |
| 122 | // processes, no replication — just the spawn/stop the wake path drives. |
| 123 | type fakeWakeBackend struct{ inst *fakeInstance } |
| 124 | |
| 125 | type nopLease struct{} |
| 126 | |
| 127 | func (nopLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) { |
| 128 | return nopRep{}, nil |
| 129 | } |
| 130 | func (nopLease) Release(context.Context) error { return nil } |
| 131 | |
| 132 | type nopRep struct{} |
| 133 | |
| 134 | func (nopRep) Err() error { return nil } |
| 135 | func (nopRep) Close(context.Context) error { return nil } |
| 136 | |
| 137 | func (b *fakeWakeBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) { |
| 138 | return nopLease{}, nil |
| 139 | } |
| 140 | func (b *fakeWakeBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil } |
| 141 | func (b *fakeWakeBackend) Seed(context.Context, registry.Route) error { return nil } |
| 142 | func (b *fakeWakeBackend) Start(_ context.Context, _ registry.Route) (activator.Proc, string, error) { |
| 143 | if err := b.inst.start(); err != nil { |
| 144 | return nil, "", err |
| 145 | } |
| 146 | return &fakeInstanceProc{inst: b.inst}, "", nil |
| 147 | } |
| 148 | func (b *fakeWakeBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| 149 | func (b *fakeWakeBackend) HasSidecar(registry.Route) bool { return false } |
| 150 | func (b *fakeWakeBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) { |
| 151 | return nil, nil |
| 152 | } |
| 153 | |
| 154 | // StartWorker: this fixture drives the INSTANCE wake path, so nothing here |
| 155 | // ever asks for a worker. Refusing rather than returning a nil Proc is the |
| 156 | // safer stub — a nil Proc would be dereferenced by the caller and read as a |
| 157 | // crash inside the wake, which is a confusing way to learn a test wandered |
| 158 | // into the wrong door. |
| 159 | func (b *fakeWakeBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) { |
| 160 | return nil, "", errors.New("fakeWakeBackend: no worker path in this fixture") |
| 161 | } |
| 162 | |
| 163 | // wakeFixture builds a sleeping instance: a registry with one hibernate- |
| 164 | // flagged exec route, a fakeInstance that listens only while it is "running", |
| 165 | // and an activator over it with its sweeper started. dd is the agent's data |
| 166 | // dir — where scheduleToken mints CARLOS_ADMIN_TOKEN. |
| 167 | func wakeFixture(t *testing.T, host string, idle time.Duration, h http.HandlerFunc) (*registry.Registry, *fakeInstance, *activator.Service, string) { |
| 168 | t.Helper() |
| 169 | dd := t.TempDir() |
| 170 | reg, err := registry.Open(filepath.Join(dd, "r.db")) |
| 171 | if err != nil { |
| 172 | t.Fatal(err) |
| 173 | } |
| 174 | t.Cleanup(func() { reg.Close() }) |
| 175 | inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: h} |
| 176 | if err := reg.Upsert(registry.Route{ |
| 177 | Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready", |
| 178 | Hibernate: true, Socket: inst.socket, DBPath: filepath.Join(dd, "app.db"), |
| 179 | }); err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: idle}, &fakeWakeBackend{inst: inst}, reg) |
| 183 | svc.Start() |
| 184 | t.Cleanup(func() { |
| 185 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 186 | defer cancel() |
| 187 | svc.Shutdown(ctx) |
| 188 | }) |
| 189 | if svc.Owns(host) || inst.running() { |
| 190 | t.Fatal("fixture: the instance is supposed to start asleep") |
| 191 | } |
| 192 | return reg, inst, svc, dd |
| 193 | } |
| 194 | |
| 195 | // --- the story --------------------------------------------------------------- |
| 196 | |
| 197 | // TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain is the branch |
| 198 | // review's "prove the wake story" test. It asserts, in order: |
| 199 | // |
| 200 | // 1. the instance is ASLEEP — nothing is listening on its socket; |
| 201 | // 2. a due timer's tick wakes it (the backend spawns exactly once); |
| 202 | // 3. the app's handler runs and sees the contract: the bearer token the |
| 203 | // agent minted for THIS instance, and X-Carlos-Schedule-At carrying the |
| 204 | // occurrence (spec §6.3's header table); |
| 205 | // 4. the in-flight hold is real — the handler is held open past the idle |
| 206 | // window, over several sweeps, and the instance stays awake; |
| 207 | // 5. the delivery is recorded as a 2xx on the row; and |
| 208 | // 6. the instance hibernates once the tick has been idle for the window: |
| 209 | // the activator drops the entry and the socket goes away again. |
| 210 | // |
| 211 | // What it does NOT prove: anything about the real ExecBackend — no process is |
| 212 | // forked, no database is restored, no lease is taken, and the "hibernate" of |
| 213 | // step 6 is the activator's stop path driving this fake's Stop, not a SIGTERM |
| 214 | // drain of a real child. Those belong to internal/activator's own tests; what |
| 215 | // this pins is that the runner's delivery leg drives that machinery correctly |
| 216 | // from a sleeping start. |
| 217 | func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) { |
| 218 | t.Parallel() |
| 219 | const host = "sleepy.bab.oncarlos.com" |
| 220 | // The idle window is short so the test can watch a real hibernation. |
| 221 | const idle = 400 * time.Millisecond |
| 222 | |
| 223 | // The handler PARKS until the test releases it, rather than sleeping a |
| 224 | // fixed span. That is what takes the clock out of step 4: the instance |
| 225 | // must stay awake for as long as the handler runs, so the test can look |
| 226 | // at "still awake" whenever it likes, and a stalled machine only makes |
| 227 | // the handler run longer — it cannot turn the assertion red. |
| 228 | seen := make(chan *http.Request, 4) |
| 229 | release := make(chan struct{}) |
| 230 | // If an assertion t.Fatalfs before the release below, the parked |
| 231 | // handler goroutine must not outlive the test holding this channel: |
| 232 | // close it on cleanup unless the success path already has. Reading a |
| 233 | // closed channel succeeds immediately, so the select stays correct |
| 234 | // after the normal close. |
| 235 | t.Cleanup(func() { |
| 236 | select { |
| 237 | case <-release: |
| 238 | default: |
| 239 | close(release) |
| 240 | } |
| 241 | }) |
| 242 | reg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) { |
| 243 | seen <- r.Clone(context.Background()) |
| 244 | <-release |
| 245 | w.WriteHeader(http.StatusNoContent) |
| 246 | }) |
| 247 | |
| 248 | now := time.Now().UTC() |
| 249 | occurrence := now.Unix() - 5 |
| 250 | r := &schedule.Runner{ |
| 251 | Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg), |
| 252 | Now: func() time.Time { return now }, Timeout: 30 * time.Second, |
| 253 | Box: "box1", Build: "test", |
| 254 | } |
| 255 | if err := reg.UpsertTimer(registry.Timer{ |
| 256 | Host: host, Name: "sync", Kind: schedule.KindSchedule, Spec: "every 6h0m0s", |
| 257 | Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence, |
| 258 | }); err != nil { |
| 259 | t.Fatal(err) |
| 260 | } |
| 261 | |
| 262 | // 2 + 3. The tick wakes it and the handler sees the contract. |
| 263 | r.Tick(context.Background()) |
| 264 | var req *http.Request |
| 265 | select { |
| 266 | case req = <-seen: |
| 267 | case <-time.After(30 * time.Second): |
| 268 | t.Fatal("the handler never ran — the tick did not wake the sleeping instance") |
| 269 | } |
| 270 | tok, err := scheduleToken(dd, svc, reg)(host) |
| 271 | if err != nil { |
| 272 | t.Fatal(err) |
| 273 | } |
| 274 | if got := req.Header.Get("Authorization"); got != "Bearer "+tok { |
| 275 | t.Fatalf("Authorization = %q, want the bearer of this instance's own minted token", got) |
| 276 | } |
| 277 | if got := req.Header.Get("X-Carlos-Schedule-At"); got != strconv.FormatInt(occurrence, 10) { |
| 278 | t.Fatalf("X-Carlos-Schedule-At = %q, want the occurrence %d", got, occurrence) |
| 279 | } |
| 280 | if got := req.Header.Get("X-Carlos-Schedule"); got != "sync" { |
| 281 | t.Fatalf("X-Carlos-Schedule = %q", got) |
| 282 | } |
| 283 | if got := req.Header.Get("X-Carlos-Schedule-Kind"); got != schedule.KindSchedule { |
| 284 | t.Fatalf("X-Carlos-Schedule-Kind = %q, want %q", got, schedule.KindSchedule) |
| 285 | } |
| 286 | if req.Method != http.MethodPost || req.URL.Path != "/jobs/sync" { |
| 287 | t.Fatalf("request = %s %s", req.Method, req.URL.Path) |
| 288 | } |
| 289 | if n := inst.startCount(); n != 1 { |
| 290 | t.Fatalf("the instance was started %d times, want exactly 1 wake", n) |
| 291 | } |
| 292 | |
| 293 | // 4. The in-flight hold. The handler is parked, so the instance has now |
| 294 | // been "idle" (no activity stamp — that lands when the request finishes) |
| 295 | // for longer than the whole idle window, and the sweeper has run several |
| 296 | // times over it. It must still be awake: the activator's in-flight count |
| 297 | // is what holds it open for the length of a handler (spec §6.3), and |
| 298 | // without that hold a 30-minute job would be hibernated out from under |
| 299 | // itself. No deadline pressure here — the handler waits for the test. |
| 300 | time.Sleep(3 * idle) |
| 301 | if !svc.Owns(host) || !inst.running() { |
| 302 | t.Fatalf("the instance was hibernated with its tick still running (owns=%v running=%v) — the in-flight hold did not hold", |
| 303 | svc.Owns(host), inst.running()) |
| 304 | } |
| 305 | close(release) |
| 306 | |
| 307 | // 5. The row records the 204 the handler wrote once released, and |
| 308 | // advances to the next occurrence. |
| 309 | deadline := time.Now().Add(30 * time.Second) |
| 310 | for { |
| 311 | tm, _ := reg.GetTimer(host, "sync") |
| 312 | if tm != nil && tm.LastStatus != "" { |
| 313 | if tm.LastStatus != "204" { |
| 314 | t.Fatalf("recorded %q, want 204 — the handler ran past the %s idle window and must not be cut short", tm.LastStatus, idle) |
| 315 | } |
| 316 | if tm.NextAt <= now.Unix() { |
| 317 | t.Fatalf("row did not advance: %+v", tm) |
| 318 | } |
| 319 | break |
| 320 | } |
| 321 | if time.Now().After(deadline) { |
| 322 | t.Fatal("the delivery was never recorded on the row") |
| 323 | } |
| 324 | time.Sleep(5 * time.Millisecond) |
| 325 | } |
| 326 | |
| 327 | // 6. Idle again: the sweep hibernates it, and the socket goes with it. |
| 328 | for { |
| 329 | if !svc.Owns(host) && !inst.running() { |
| 330 | break |
| 331 | } |
| 332 | if time.Now().After(deadline) { |
| 333 | t.Fatalf("the instance never hibernated after the tick (owns=%v running=%v)", svc.Owns(host), inst.running()) |
| 334 | } |
| 335 | time.Sleep(5 * time.Millisecond) |
| 336 | } |
| 337 | // And it woke exactly once for the whole story — the tick did not |
| 338 | // ping-pong the instance awake behind its own back. |
| 339 | if n := inst.startCount(); n != 1 { |
| 340 | t.Fatalf("the instance was started %d times over the story, want 1", n) |
| 341 | } |
| 342 | if reqs := inst.served(); len(reqs) != 1 { |
| 343 | t.Fatalf("the instance served %d requests, want the one tick", len(reqs)) |
| 344 | } |
| 345 | } |
| 346 | |
| 347 | // The delivery leg caches one proxy — and therefore one connection pool — per |
| 348 | // socket path, so a schedule that fires every minute stops dialling from |
| 349 | // scratch every time. Asserted where it is visible: the instance's accepted |
| 350 | // connection count. |
| 351 | func TestDeliveryReusesTheConnectionForASocket(t *testing.T) { |
| 352 | const host = "warm.bab.oncarlos.com" |
| 353 | dd := t.TempDir() |
| 354 | inst := &fakeInstance{ |
| 355 | socket: filepath.Join(dd, "app.sock"), |
| 356 | handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }), |
| 357 | } |
| 358 | if err := inst.start(); err != nil { |
| 359 | t.Fatal(err) |
| 360 | } |
| 361 | t.Cleanup(func() { inst.stop(context.Background()) }) |
| 362 | |
| 363 | // svc nil: an instance that is not activator-managed is proxied straight, |
| 364 | // which is the same delivery path minus the wake. |
| 365 | d := &activatorDeliverer{} |
| 366 | rt := registry.Route{Host: host, Kind: "instance", Socket: inst.socket} |
| 367 | for i := 0; i < 3; i++ { |
| 368 | req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test") |
| 369 | if err != nil { |
| 370 | t.Fatal(err) |
| 371 | } |
| 372 | status, err := d.Deliver(context.Background(), rt, req) |
| 373 | if status != http.StatusNoContent || err != nil { |
| 374 | t.Fatalf("delivery %d = %d %v", i, status, err) |
| 375 | } |
| 376 | } |
| 377 | if n := inst.connCount(); n != 1 { |
| 378 | t.Fatalf("three deliveries opened %d connections, want 1 — the transport is not being reused", n) |
| 379 | } |
| 380 | if len(inst.served()) != 3 { |
| 381 | t.Fatalf("the instance served %d ticks, want 3", len(inst.served())) |
| 382 | } |
| 383 | } |
| 384 | |
| 385 | // The arming switch (I6): a box with the activator armed and NO deploy store |
| 386 | // still gets a RUNNING runner. Its once-timers arrive on the control socket |
| 387 | // (spec §7.2) and nothing else on the box would ever fire them — the bug this |
| 388 | // pins is gating the runner on the store, which left those rows inert. |
| 389 | // |
| 390 | // Asserted by firing one: a once-row on a storeless box's registry is |
| 391 | // delivered to its instance and finishes. |
| 392 | func TestArmScheduleFiresOnceTimersWithNoDeployStore(t *testing.T) { |
| 393 | t.Parallel() |
| 394 | const host = "storeless.bab.oncarlos.com" |
| 395 | served := make(chan struct{}, 1) |
| 396 | reg, _, svc, dd := wakeFixture(t, host, 30*time.Second, func(w http.ResponseWriter, r *http.Request) { |
| 397 | served <- struct{}{} |
| 398 | w.WriteHeader(http.StatusNoContent) |
| 399 | }) |
| 400 | |
| 401 | sd := armSchedule(reg, nil, errors.New("no deploy store on this box"), svc, nil, dd) |
| 402 | if sd == nil || sd.runner == nil { |
| 403 | t.Fatal("a box with the activator armed and no store got no runner — its once-timers would never fire") |
| 404 | } |
| 405 | if sd.conv != nil || scheduleLegFor(sd) != nil { |
| 406 | t.Fatal("no store, but a converger was built — it has nothing to read declarations from") |
| 407 | } |
| 408 | if sd.runner.Report != nil { |
| 409 | t.Fatal("no store, but the runner reports — there is nowhere to write a status object") |
| 410 | } |
| 411 | |
| 412 | now := time.Now() |
| 413 | sd.runner.Now = func() time.Time { return now } |
| 414 | if err := reg.UpsertTimer(registry.Timer{ |
| 415 | Host: host, Name: "remind", Kind: schedule.KindOnce, Spec: "at " + strconv.FormatInt(now.Unix()-1, 10), |
| 416 | Path: "/jobs/remind", NextAt: now.Unix() - 1, DueAt: now.Unix() - 1, |
| 417 | }); err != nil { |
| 418 | t.Fatal(err) |
| 419 | } |
| 420 | sd.runner.Tick(context.Background()) |
| 421 | select { |
| 422 | case <-served: |
| 423 | case <-time.After(30 * time.Second): |
| 424 | t.Fatal("the once-timer never reached the instance on a storeless box") |
| 425 | } |
| 426 | deadline := time.Now().Add(30 * time.Second) |
| 427 | for { |
| 428 | tm, _ := reg.GetTimer(host, "remind") |
| 429 | if tm != nil && tm.LastStatus == "204" && tm.NextAt == 0 { // a finished once-row |
| 430 | break |
| 431 | } |
| 432 | if time.Now().After(deadline) { |
| 433 | t.Fatalf("the once-row never finished: %+v", tm) |
| 434 | } |
| 435 | time.Sleep(5 * time.Millisecond) |
| 436 | } |
| 437 | |
| 438 | // The other two arms of the same switch. |
| 439 | sd = armSchedule(reg, release.NewDirStore(t.TempDir()), nil, svc, nil, dd) |
| 440 | if sd == nil || sd.conv == nil || sd.runner == nil || sd.runner.Report == nil { |
| 441 | t.Fatalf("with a deploy store, want both halves and a reporter: %+v", sd) |
| 442 | } |
| 443 | if sd := armSchedule(reg, nil, errors.New("no deploy store"), nil, nil, dd); sd != nil { |
| 444 | t.Fatal("no store and no activator: nothing can write a row or fire one, so nothing should arm") |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | // The proxy cache is bounded: one entry per socket path, the same handler |
| 449 | // back for the same path, dropped when its delivery errors, and the whole map |
| 450 | // reset rather than grown past maxProxyCache. |
| 451 | func TestProxyCacheIsStableAndBounded(t *testing.T) { |
| 452 | t.Parallel() |
| 453 | d := &activatorDeliverer{} |
| 454 | first := d.proxyFor("/run/carlos/a.sock") |
| 455 | if again := d.proxyFor("/run/carlos/a.sock"); again != first { |
| 456 | t.Fatal("the same socket got a second proxy — its connection pool is being thrown away") |
| 457 | } |
| 458 | if other := d.proxyFor("/run/carlos/b.sock"); other == first { |
| 459 | t.Fatal("two sockets share one proxy — the transport would dial the wrong file") |
| 460 | } |
| 461 | d.dropProxy("/run/carlos/a.sock") |
| 462 | if again := d.proxyFor("/run/carlos/a.sock"); again == first { |
| 463 | t.Fatal("dropProxy kept the entry") |
| 464 | } |
| 465 | for i := 0; i < maxProxyCache+10; i++ { |
| 466 | d.proxyFor("/run/carlos/" + strconv.Itoa(i) + ".sock") |
| 467 | } |
| 468 | d.mu.Lock() |
| 469 | n := len(d.proxies) |
| 470 | d.mu.Unlock() |
| 471 | if n > maxProxyCache { |
| 472 | t.Fatalf("cache holds %d entries, want at most maxProxyCache (%d)", n, maxProxyCache) |
| 473 | } |
| 474 | |
| 475 | // A delivery that ERRORS retires its socket: the next tick dials fresh. |
| 476 | // Reached here through the delivery's own deadline — deliberately, and |
| 477 | // not through a missing socket, which is NOT an error at this seam: the |
| 478 | // unix dialer retries a missing socket for a fixed 13s window |
| 479 | // (edge.dialUnixWithRetry, for an instance mid-spawn) and ReverseProxy |
| 480 | // then answers 502 itself, with no error for Deliver to see. |
| 481 | rt := registry.Route{Host: "gone.bab.oncarlos.com", Kind: "instance", Socket: filepath.Join(t.TempDir(), "missing.sock")} |
| 482 | req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test") |
| 483 | if err != nil { |
| 484 | t.Fatal(err) |
| 485 | } |
| 486 | ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) |
| 487 | defer cancel() |
| 488 | if status, err := d.Deliver(ctx, rt, req.WithContext(ctx)); err == nil { |
| 489 | t.Fatalf("delivery that ran past its deadline = %d, want an error", status) |
| 490 | } |
| 491 | d.mu.Lock() |
| 492 | _, kept := d.proxies[rt.Socket] |
| 493 | d.mu.Unlock() |
| 494 | if kept { |
| 495 | t.Fatal("a failed delivery left its proxy (and its dead pool) in the cache") |
| 496 | } |
| 497 | } |
| 498 | |
| 499 | // The prune leg's eviction (schedule.SocketRetainer): a route that goes away |
| 500 | // stops receiving deliveries, so nothing it caches can error its own way out |
| 501 | // — the daily sweep's live-socket set is what reclaims it. |
| 502 | func TestRetainSocketsDropsRetiredRoutes(t *testing.T) { |
| 503 | d := &activatorDeliverer{} |
| 504 | keep := d.proxyFor("/run/carlos/keep.sock") |
| 505 | d.proxyFor("/run/carlos/retired.sock") |
| 506 | d.RetainSockets(map[string]bool{"/run/carlos/keep.sock": true}) |
| 507 | d.mu.Lock() |
| 508 | _, retired := d.proxies["/run/carlos/retired.sock"] |
| 509 | n := len(d.proxies) |
| 510 | d.mu.Unlock() |
| 511 | if retired || n != 1 { |
| 512 | t.Fatalf("after the sweep the cache holds %d entries (retired kept: %v), want just the live one", n, retired) |
| 513 | } |
| 514 | if again := d.proxyFor("/run/carlos/keep.sock"); again != keep { |
| 515 | t.Fatal("the sweep dropped a socket that is still routed here") |
| 516 | } |
| 517 | // An empty live set is a box with no exec routes at all: nothing to keep. |
| 518 | d.RetainSockets(map[string]bool{}) |
| 519 | d.mu.Lock() |
| 520 | n = len(d.proxies) |
| 521 | d.mu.Unlock() |
| 522 | if n != 0 { |
| 523 | t.Fatalf("cache holds %d entries after an empty sweep, want 0", n) |
| 524 | } |
| 525 | // The runner must be able to find this method through the interface, or |
| 526 | // the sweep silently never runs. |
| 527 | if _, ok := any(d).(schedule.SocketRetainer); !ok { |
| 528 | t.Fatal("activatorDeliverer no longer satisfies schedule.SocketRetainer — the prune leg's eviction is dead code") |
| 529 | } |
| 530 | } |
| 531 | |
| 532 | // closeIdle returns a dropped proxy's pooled connections immediately instead |
| 533 | // of waiting out the 90s IdleConnTimeout, and it can only do that through two |
| 534 | // type assertions on what edge.UnixProxy builds. Pin the shape: if edge ever |
| 535 | // returns something else, the eager close silently stops happening, and this |
| 536 | // test is what says so. |
| 537 | func TestUnixProxyKeepsTheShapeCloseIdleNeeds(t *testing.T) { |
| 538 | rp, ok := edge.UnixProxy("/run/carlos/x.sock").(*httputil.ReverseProxy) |
| 539 | if !ok { |
| 540 | t.Fatal("edge.UnixProxy no longer returns a *httputil.ReverseProxy — closeIdle is now a no-op") |
| 541 | } |
| 542 | if _, ok := rp.Transport.(*http.Transport); !ok { |
| 543 | t.Fatal("edge.UnixProxy's Transport is no longer an *http.Transport — closeIdle is now a no-op") |
| 544 | } |
| 545 | } |
| 546 | |
| 547 | // The bug, as a test: a schedule that fires more often than the idle window |
| 548 | // used to pin its instance awake forever, because each tick stamped activity |
| 549 | // and bought another full window. With the tick activity-invisible, the |
| 550 | // instance sleeps between ticks and wakes again for the next one. |
| 551 | // |
| 552 | // No timer rows are armed here, deliberately: this exercises the delivery |
| 553 | // leg alone. The sweep guard (which reads the timers table) would correctly |
| 554 | // hold a host awake for a tick due this soon, and that is Task 3's test. |
| 555 | func TestDeliveryDoesNotHoldTheInstanceAwake(t *testing.T) { |
| 556 | t.Parallel() |
| 557 | const host = "ticky.bab.oncarlos.com" |
| 558 | const idle = 300 * time.Millisecond |
| 559 | reg, inst, svc, _ := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) { |
| 560 | w.WriteHeader(http.StatusNoContent) |
| 561 | }) |
| 562 | rt, err := reg.Get(host) |
| 563 | if err != nil { |
| 564 | t.Fatalf("get route: %v", err) |
| 565 | } |
| 566 | d := &activatorDeliverer{svc: svc} |
| 567 | |
| 568 | // Tick twice per idle window for a few windows. Under the old accounting |
| 569 | // the instance could never reach the sweep's threshold; now it does, |
| 570 | // between ticks, and the next tick wakes it again. |
| 571 | deadline := time.Now().Add(20 * time.Second) |
| 572 | for inst.startCount() < 2 && time.Now().Before(deadline) { |
| 573 | req := httptest.NewRequest(http.MethodPost, "http://carlos/jobs/sweep", nil) |
| 574 | if _, err := d.Deliver(context.Background(), *rt, req); err != nil { |
| 575 | t.Fatalf("Deliver: %v", err) |
| 576 | } |
| 577 | time.Sleep(idle / 2) |
| 578 | } |
| 579 | if n := inst.startCount(); n < 2 { |
| 580 | t.Fatalf("the instance started %d time(s): it never hibernated between ticks, so the tick is still buying residency", n) |
| 581 | } |
| 582 | } |
| 583 | |
| 584 | // The app-facing half of the tick contract is untouched by any of this. |
| 585 | func TestDeliveryStillCarriesScheduleAtHeader(t *testing.T) { |
| 586 | const host = "hdr.bab.oncarlos.com" |
| 587 | seen := make(chan string, 1) |
| 588 | reg, _, svc, _ := wakeFixture(t, host, time.Hour, func(w http.ResponseWriter, r *http.Request) { |
| 589 | select { |
| 590 | case seen <- r.Header.Get("X-Carlos-Schedule-At"): |
| 591 | default: |
| 592 | } |
| 593 | w.WriteHeader(http.StatusNoContent) |
| 594 | }) |
| 595 | rt, err := reg.Get(host) |
| 596 | if err != nil { |
| 597 | t.Fatalf("get route: %v", err) |
| 598 | } |
| 599 | d := &activatorDeliverer{svc: svc} |
| 600 | |
| 601 | req := httptest.NewRequest(http.MethodPost, "http://carlos/jobs/sweep", nil) |
| 602 | req.Header.Set("X-Carlos-Schedule-At", "1788126300") |
| 603 | status, err := d.Deliver(context.Background(), *rt, req) |
| 604 | if err != nil { |
| 605 | t.Fatalf("Deliver: %v", err) |
| 606 | } |
| 607 | if status != http.StatusNoContent { |
| 608 | t.Fatalf("status = %d, want 204", status) |
| 609 | } |
| 610 | select { |
| 611 | case got := <-seen: |
| 612 | if got != "1788126300" { |
| 613 | t.Errorf("handler saw X-Carlos-Schedule-At = %q, want it passed through unchanged", got) |
| 614 | } |
| 615 | case <-time.After(10 * time.Second): |
| 616 | t.Fatal("the handler never ran") |
| 617 | } |
| 618 | } |
| 619 | |
amadan