amadan

carlos / platform Public

host: one transport per socket, bounded converge passes, and the wake story proven

Three deferred review findings on the delivery leg, plus the test the branch
review asked for before calling the story proven.

- The runner built a fresh http.Transport per delivery (edge.UnixProxy makes
  one), so every tick dialled from scratch and threw an unread connection
  pool away. One proxy per SOCKET PATH now, cached: dropped when a delivery
  errors (a socket that would not answer is exactly the one whose pool is
  worthless), and the whole map reset if it ever passes 512 entries, so the
  cache cannot grow without a ceiling.

- The agent's three converge call sites all hand Converger.Pass an unbounded
  context.Background(). A pass is store-bound, and a bucket that accepts a
  connection and never answers would park the pass — and every caller queued
  behind it on the converger's own mutex — for the life of the process. The
  timeout goes on scheduleDeps.converge, the choke point all three already go
  through, at a generous 5 minutes.

- The arming switch moves out of agent.go into armSchedule, so the decision
  itself is testable rather than only its parts.

Tests: the wake story end to end in one process (asleep → due → the tick
wakes it → the handler runs and sees the bearer and the occurrence header →
the in-flight hold carries a handler past the idle window → it hibernates
again); the I6 arm fires a once-timer on a box with NO deploy store; three
deliveries to one socket open ONE connection (3 without the cache); and the
cache's bound, stability and error-eviction. The wake test's limits are
documented on it — the fake is activator.Backend, so no real child, lease or
SIGTERM drain is exercised.
Paul Campbell pushed by paul@keymail.dev b028b56dca6d0ba9c5f25e6475841917c25d9bd8 parent 53b76e9
4 files changed, +569 −25
  • internal/host/agent.go +8 −19
  • internal/host/schedule_wake_test.go +452 −0
  • internal/host/schedule_wiring.go +108 −5
  • internal/host/schedule_wiring_test.go +1 −1
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 23b14482e..fab5d5121 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -322,14 +322,13 @@ func RunAgent(args []string) {
// a console bump fires the fast one); Converger.Pass serializes itself on
// its own mutex, so no call site here needs one.
//
- // The two halves arm on different conditions. Declared schedules live in
- // the bucket, so the converger needs the deploy store. The RUNNER needs
- // only the registry: once-timers are registered by instances on their
- // control socket (mounted above, Timers: reg, whenever the activator is
- // armed) straight into this box's rows, and a box that mounts the verb
- // must also fire what it accepts — so the runner starts whenever the
- // activator is armed, store or no store. One line each way, like every
- // other optional loop.
+ // The two halves arm on different conditions — armSchedule
+ // (schedule_wiring.go) is that switch, and carries the why. In short:
+ // declared schedules live in the bucket, so the converger needs the
+ // deploy store; the RUNNER needs only the registry, because once-timers
+ // are registered by instances on their control socket (mounted above,
+ // Timers: reg, whenever the activator is armed) straight into this box's
+ // rows, and a box that mounts the verb must also fire what it accepts.
//
// Guarded assignment, not a direct pass: logSink is a *applog.Sink, and
// a nil one handed to an interface parameter is NOT a nil interface —
@@ -338,18 +337,8 @@ func RunAgent(args []string) {
if logSink != nil {
sink = logSink
}
- var sd *scheduleDeps
schedStore, schedErr := openDeployStore(context.Background())
- switch {
- case schedErr == nil:
- sd = newScheduleDeps(reg, schedStore, svc, sink, dd)
- log.Printf("agent: scheduled work armed (converge on serial + hourly; runner every 15s)")
- case svc != nil:
- sd = newScheduleDeps(reg, nil, svc, sink, dd)
- log.Printf("agent: schedules from the bucket are off (%v); once-timers registered by instances still run (runner every 15s)", schedErr)
- default:
- log.Printf("agent: scheduled work disabled: %v", schedErr)
- }
+ sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
if sd != nil {
sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
}
diff --git a/internal/host/schedule_wake_test.go b/internal/host/schedule_wake_test.go
new file mode 100644
index 000000000..b0ca445f8
--- /dev/null
+++ b/internal/host/schedule_wake_test.go
@@ -0,0 +1,452 @@
+package host
+
+// The wake story, end to end, in one process: an instance asleep → its timer
+// comes due → the tick wakes it through the activator → the app's handler
+// runs and sees the contract headers → the instance hibernates again once it
+// is idle (spec §6.3, §7.1).
+//
+// The seam is activator.Backend — the same seam internal/activator's own
+// tests use, with a fake that models an exec child faithfully enough for this
+// story: the instance's unix socket exists ONLY while it is "running", so a
+// delivery that did not wake it would find nothing to dial, and the handler
+// running at all is proof the wake happened.
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/http"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// --- the fake instance and its backend ---------------------------------------
+
+// fakeInstanceProc is the spawned child: stopping it takes the socket away,
+// which is what "hibernated" means to anything downstream.
+type fakeInstanceProc struct {
+ inst *fakeInstance
+}
+
+func (p *fakeInstanceProc) Stop(ctx context.Context) error { return p.inst.stop(ctx) }
+func (p *fakeInstanceProc) Exited() bool { return !p.inst.running() }
+
+// fakeInstance is an app listening on a unix socket, started and stopped by
+// the backend below. It records every request it served.
+type fakeInstance struct {
+ socket string
+ handler http.Handler
+
+ mu sync.Mutex
+ srv *http.Server
+ starts int
+ conns int // accepted connections — the transport-reuse assertion
+ requests []*http.Request
+}
+
+func (f *fakeInstance) running() bool {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.srv != nil
+}
+
+func (f *fakeInstance) start() error {
+ ln, err := net.Listen("unix", f.socket)
+ if err != nil {
+ return err
+ }
+ srv := &http.Server{
+ Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ f.mu.Lock()
+ f.requests = append(f.requests, r.Clone(context.Background()))
+ f.mu.Unlock()
+ f.handler.ServeHTTP(w, r)
+ }),
+ ConnState: func(_ net.Conn, st http.ConnState) {
+ if st == http.StateNew {
+ f.mu.Lock()
+ f.conns++
+ f.mu.Unlock()
+ }
+ },
+ }
+ f.mu.Lock()
+ f.srv = srv
+ f.starts++
+ f.mu.Unlock()
+ go srv.Serve(ln)
+ return nil
+}
+
+func (f *fakeInstance) stop(ctx context.Context) error {
+ f.mu.Lock()
+ srv := f.srv
+ f.srv = nil
+ f.mu.Unlock()
+ if srv == nil {
+ return nil
+ }
+ return srv.Close()
+}
+
+func (f *fakeInstance) served() []*http.Request {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]*http.Request(nil), f.requests...)
+}
+
+func (f *fakeInstance) startCount() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.starts
+}
+
+func (f *fakeInstance) connCount() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.conns
+}
+
+// fakeWakeBackend is activator.Backend over one fakeInstance: no S3, no
+// processes, no replication — just the spawn/stop the wake path drives.
+type fakeWakeBackend struct{ inst *fakeInstance }
+
+type nopLease struct{}
+
+func (nopLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
+ return nopRep{}, nil
+}
+func (nopLease) Release(context.Context) error { return nil }
+
+type nopRep struct{}
+
+func (nopRep) Err() error { return nil }
+func (nopRep) Close(context.Context) error { return nil }
+
+func (b *fakeWakeBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
+ return nopLease{}, nil
+}
+func (b *fakeWakeBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+func (b *fakeWakeBackend) Seed(context.Context, registry.Route) error { return nil }
+func (b *fakeWakeBackend) Start(_ context.Context, _ registry.Route) (activator.Proc, string, error) {
+ if err := b.inst.start(); err != nil {
+ return nil, "", err
+ }
+ return &fakeInstanceProc{inst: b.inst}, "", nil
+}
+func (b *fakeWakeBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+func (b *fakeWakeBackend) HasSidecar(registry.Route) bool { return false }
+func (b *fakeWakeBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
+ return nil, nil
+}
+
+// wakeFixture builds a sleeping instance: a registry with one hibernate-
+// flagged exec route, a fakeInstance that listens only while it is "running",
+// and an activator over it with its sweeper started. dd is the agent's data
+// dir — where scheduleToken mints CARLOS_ADMIN_TOKEN.
+func wakeFixture(t *testing.T, host string, idle time.Duration, h http.HandlerFunc) (*registry.Registry, *fakeInstance, *activator.Service, string) {
+ t.Helper()
+ dd := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dd, "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: h}
+ if err := reg.Upsert(registry.Route{
+ Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready",
+ Hibernate: true, Socket: inst.socket, DBPath: filepath.Join(dd, "app.db"),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: idle}, &fakeWakeBackend{inst: inst}, reg)
+ svc.Start()
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(ctx)
+ })
+ if svc.Owns(host) || inst.running() {
+ t.Fatal("fixture: the instance is supposed to start asleep")
+ }
+ return reg, inst, svc, dd
+}
+
+// --- the story ---------------------------------------------------------------
+
+// TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain is the branch
+// review's "prove the wake story" test. It asserts, in order:
+//
+// 1. the instance is ASLEEP — nothing is listening on its socket;
+// 2. a due timer's tick wakes it (the backend spawns exactly once);
+// 3. the app's handler runs and sees the contract: the bearer token the
+// agent minted for THIS instance, and X-Carlos-Schedule-At carrying the
+// occurrence (spec §6.3's header table);
+// 4. the in-flight hold is real — the handler runs LONGER than the idle
+// window and is not cut short;
+// 5. the delivery is recorded as a 2xx on the row; and
+// 6. the instance hibernates once the tick has been idle for the window:
+// the activator drops the entry and the socket goes away again.
+//
+// What it does NOT prove: anything about the real ExecBackend — no process is
+// forked, no database is restored, no lease is taken, and the "hibernate" of
+// step 6 is the activator's stop path driving this fake's Stop, not a SIGTERM
+// drain of a real child. Those belong to internal/activator's own tests; what
+// this pins is that the runner's delivery leg drives that machinery correctly
+// from a sleeping start.
+func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
+ const host = "sleepy.bab.oncarlos.com"
+ // The idle window is short so the test can watch a real hibernation; the
+ // handler deliberately runs past it (see the in-flight assertion below).
+ const idle = 400 * time.Millisecond
+ const handlerRuns = 3 * idle
+
+ // 1. Asleep: no entry, and nothing listening (asserted in wakeFixture).
+ seen := make(chan *http.Request, 4)
+ reg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) {
+ seen <- r.Clone(context.Background())
+ time.Sleep(handlerRuns)
+ w.WriteHeader(http.StatusNoContent)
+ })
+
+ now := time.Date(2026, 8, 23, 13, 7, 0, 0, time.UTC)
+ occurrence := now.Unix() - 5
+ r := &schedule.Runner{
+ Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd),
+ Now: func() time.Time { return now }, Timeout: 30 * time.Second,
+ Box: "box1", Build: "test",
+ }
+ if err := reg.UpsertTimer(registry.Timer{
+ Host: host, Name: "sync", Kind: schedule.KindSchedule, Spec: "every 6h0m0s",
+ Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence,
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ // 2 + 3. The tick wakes it and the handler sees the contract.
+ r.Tick(context.Background())
+ var req *http.Request
+ select {
+ case req = <-seen:
+ case <-time.After(30 * time.Second):
+ t.Fatal("the handler never ran — the tick did not wake the sleeping instance")
+ }
+ tok, err := scheduleToken(dd)(host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := req.Header.Get("Authorization"); got != "Bearer "+tok {
+ t.Fatalf("Authorization = %q, want the bearer of this instance's own minted token", got)
+ }
+ if got := req.Header.Get("X-Carlos-Schedule-At"); got != strconv.FormatInt(occurrence, 10) {
+ t.Fatalf("X-Carlos-Schedule-At = %q, want the occurrence %d", got, occurrence)
+ }
+ if got := req.Header.Get("X-Carlos-Schedule"); got != "sync" {
+ t.Fatalf("X-Carlos-Schedule = %q", got)
+ }
+ if req.Method != http.MethodPost || req.URL.Path != "/jobs/sync" {
+ t.Fatalf("request = %s %s", req.Method, req.URL.Path)
+ }
+ if n := inst.startCount(); n != 1 {
+ t.Fatalf("the instance was started %d times, want exactly 1 wake", n)
+ }
+
+ // 4 + 5. The handler outlives the idle window and is not cut short: the
+ // activator's in-flight count is what holds the instance open (spec §6.3),
+ // and the row records the 204 the handler eventually wrote.
+ deadline := time.Now().Add(30 * time.Second)
+ for {
+ tm, _ := reg.GetTimer(host, "sync")
+ if tm != nil && tm.LastStatus != "" {
+ if tm.LastStatus != "204" {
+ t.Fatalf("recorded %q, want 204 — the handler ran %s, past the %s idle window, and must not be cut short",
+ tm.LastStatus, handlerRuns, idle)
+ }
+ if tm.NextAt <= now.Unix() {
+ t.Fatalf("row did not advance: %+v", tm)
+ }
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatal("the delivery was never recorded on the row")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ if !svc.Owns(host) {
+ t.Fatal("the instance was already hibernated when its tick finished — the in-flight hold did not hold")
+ }
+
+ // 6. Idle again: the sweep hibernates it, and the socket goes with it.
+ for {
+ if !svc.Owns(host) && !inst.running() {
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("the instance never hibernated after the tick (owns=%v running=%v)", svc.Owns(host), inst.running())
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ // And it woke exactly once for the whole story — the tick did not
+ // ping-pong the instance awake behind its own back.
+ if n := inst.startCount(); n != 1 {
+ t.Fatalf("the instance was started %d times over the story, want 1", n)
+ }
+ if reqs := inst.served(); len(reqs) != 1 {
+ t.Fatalf("the instance served %d requests, want the one tick", len(reqs))
+ }
+}
+
+// The delivery leg caches one proxy — and therefore one connection pool — per
+// socket path, so a schedule that fires every minute stops dialling from
+// scratch every time. Asserted where it is visible: the instance's accepted
+// connection count.
+func TestDeliveryReusesTheConnectionForASocket(t *testing.T) {
+ const host = "warm.bab.oncarlos.com"
+ dd := t.TempDir()
+ inst := &fakeInstance{
+ socket: filepath.Join(dd, "app.sock"),
+ handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }),
+ }
+ if err := inst.start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { inst.stop(context.Background()) })
+
+ // svc nil: an instance that is not activator-managed is proxied straight,
+ // which is the same delivery path minus the wake.
+ d := &activatorDeliverer{}
+ rt := registry.Route{Host: host, Kind: "instance", Socket: inst.socket}
+ for i := 0; i < 3; i++ {
+ req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ status, err := d.Deliver(context.Background(), rt, req)
+ if status != http.StatusNoContent || err != nil {
+ t.Fatalf("delivery %d = %d %v", i, status, err)
+ }
+ }
+ if n := inst.connCount(); n != 1 {
+ t.Fatalf("three deliveries opened %d connections, want 1 — the transport is not being reused", n)
+ }
+ if len(inst.served()) != 3 {
+ t.Fatalf("the instance served %d ticks, want 3", len(inst.served()))
+ }
+}
+
+// The arming switch (I6): a box with the activator armed and NO deploy store
+// still gets a RUNNING runner. Its once-timers arrive on the control socket
+// (spec §7.2) and nothing else on the box would ever fire them — the bug this
+// pins is gating the runner on the store, which left those rows inert.
+//
+// Asserted by firing one: a once-row on a storeless box's registry is
+// delivered to its instance and finishes.
+func TestArmScheduleFiresOnceTimersWithNoDeployStore(t *testing.T) {
+ const host = "storeless.bab.oncarlos.com"
+ served := make(chan struct{}, 1)
+ reg, _, svc, dd := wakeFixture(t, host, 30*time.Second, func(w http.ResponseWriter, r *http.Request) {
+ served <- struct{}{}
+ w.WriteHeader(http.StatusNoContent)
+ })
+
+ sd := armSchedule(reg, nil, errors.New("no deploy store on this box"), svc, nil, dd)
+ if sd == nil || sd.runner == nil {
+ t.Fatal("a box with the activator armed and no store got no runner — its once-timers would never fire")
+ }
+ if sd.conv != nil || scheduleLegFor(sd) != nil {
+ t.Fatal("no store, but a converger was built — it has nothing to read declarations from")
+ }
+ if sd.runner.Report != nil {
+ t.Fatal("no store, but the runner reports — there is nowhere to write a status object")
+ }
+
+ now := time.Now()
+ sd.runner.Now = func() time.Time { return now }
+ if err := reg.UpsertTimer(registry.Timer{
+ Host: host, Name: "remind", Kind: schedule.KindOnce, Spec: "at " + strconv.FormatInt(now.Unix()-1, 10),
+ Path: "/jobs/remind", NextAt: now.Unix() - 1, DueAt: now.Unix() - 1,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ sd.runner.Tick(context.Background())
+ select {
+ case <-served:
+ case <-time.After(30 * time.Second):
+ t.Fatal("the once-timer never reached the instance on a storeless box")
+ }
+ deadline := time.Now().Add(30 * time.Second)
+ for {
+ tm, _ := reg.GetTimer(host, "remind")
+ if tm != nil && tm.LastStatus == "204" && tm.NextAt == 0 { // a finished once-row
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("the once-row never finished: %+v", tm)
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+
+ // The other two arms of the same switch.
+ sd = armSchedule(reg, release.NewDirStore(t.TempDir()), nil, svc, nil, dd)
+ if sd == nil || sd.conv == nil || sd.runner == nil || sd.runner.Report == nil {
+ t.Fatalf("with a deploy store, want both halves and a reporter: %+v", sd)
+ }
+ if sd := armSchedule(reg, nil, errors.New("no deploy store"), nil, nil, dd); sd != nil {
+ t.Fatal("no store and no activator: nothing can write a row or fire one, so nothing should arm")
+ }
+}
+
+// The proxy cache is bounded: one entry per socket path, the same handler
+// back for the same path, dropped when its delivery errors, and the whole map
+// reset rather than grown past maxProxyCache.
+func TestProxyCacheIsStableAndBounded(t *testing.T) {
+ d := &activatorDeliverer{}
+ first := d.proxyFor("/run/carlos/a.sock")
+ if again := d.proxyFor("/run/carlos/a.sock"); again != first {
+ t.Fatal("the same socket got a second proxy — its connection pool is being thrown away")
+ }
+ if other := d.proxyFor("/run/carlos/b.sock"); other == first {
+ t.Fatal("two sockets share one proxy — the transport would dial the wrong file")
+ }
+ d.dropProxy("/run/carlos/a.sock")
+ if again := d.proxyFor("/run/carlos/a.sock"); again == first {
+ t.Fatal("dropProxy kept the entry")
+ }
+ for i := 0; i < maxProxyCache+10; i++ {
+ d.proxyFor("/run/carlos/" + strconv.Itoa(i) + ".sock")
+ }
+ d.mu.Lock()
+ n := len(d.proxies)
+ d.mu.Unlock()
+ if n > maxProxyCache {
+ t.Fatalf("cache holds %d entries, want at most maxProxyCache (%d)", n, maxProxyCache)
+ }
+
+ // A delivery that errors retires its socket: the next tick dials fresh.
+ rt := registry.Route{Host: "gone.bab.oncarlos.com", Kind: "instance", Socket: filepath.Join(t.TempDir(), "missing.sock")}
+ req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Short: the unix dialer retries a missing socket for the whole context
+ // (an instance mid-spawn), and this one is never coming.
+ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+ defer cancel()
+ if status, err := d.Deliver(ctx, rt, req.WithContext(ctx)); err == nil {
+ t.Fatalf("delivery to a socket that does not exist = %d, want an error", status)
+ }
+ d.mu.Lock()
+ _, kept := d.proxies[rt.Socket]
+ d.mu.Unlock()
+ if kept {
+ t.Fatal("a failed delivery left its proxy (and its dead pool) in the cache")
+ }
+}
diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go
index d6c26a17e..ce773bc03 100644
--- a/internal/host/schedule_wiring.go
+++ b/internal/host/schedule_wiring.go
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/activator"
@@ -74,7 +75,7 @@ func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator
sd := &scheduleDeps{
runner: &schedule.Runner{
Reg: reg,
- Deliver: activatorDeliverer{svc: svc},
+ Deliver: &activatorDeliverer{svc: svc},
Token: scheduleToken(dd),
Event: event,
Box: box,
@@ -100,6 +101,38 @@ func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator
return sd
}
+// armSchedule is the agent's arming switch for scheduled work, lifted out of
+// agent.go so the decision itself can be tested rather than only its parts.
+//
+// storeErr is openDeployStore's error, and the three arms are not
+// interchangeable:
+//
+// - a deploy store: both halves. Declarations converge into rows and every
+// fire reports back to the bucket.
+// - NO store but an armed activator: the RUNNER ALONE, no converger, no
+// reports. This is the arm a review found missing (I6). A box that mounts
+// the control socket's schedule verbs accepts once-timers from its own
+// instances (spec §7.2) straight into the registry — and a box that
+// accepts a timer must fire it. Gating the runner on the store would have
+// left those rows sitting in the table forever.
+// - neither: nothing at all, since nothing could write a row or fire one.
+//
+// It returns nil for that last arm, which every caller (and scheduleLegFor,
+// and converge) already tolerates.
+func armSchedule(reg *registry.Registry, store release.Store, storeErr error, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps {
+ switch {
+ case storeErr == nil:
+ log.Printf("agent: scheduled work armed (converge on serial + hourly; runner every 15s)")
+ return newScheduleDeps(reg, store, svc, sink, dd)
+ case svc != nil:
+ log.Printf("agent: schedules from the bucket are off (%v); once-timers registered by instances still run (runner every 15s)", storeErr)
+ return newScheduleDeps(reg, nil, svc, sink, dd)
+ default:
+ log.Printf("agent: scheduled work disabled: %v", storeErr)
+ return nil
+ }
+}
+
// scheduleToken is the runner's Token: ensure-then-read on the activator's
// own <dataDir>/<host>.env (spec §6.4).
//
@@ -128,13 +161,27 @@ func scheduleToken(dd string) func(host string) (string, error) {
}
}
+// convergePassTimeout bounds ONE converge pass. The agent's three call sites
+// (the seed and the serial leg in fastConvergeLoop, the hourly floor in
+// adoptLoop) all hand this context.Background() — an unbounded context — and
+// a pass is store-bound: a bucket that accepts a connection and then never
+// answers would park the pass, and with it every caller queued on
+// Converger's own mutex, for as long as the process lives. Generous on
+// purpose: a pass reads one object per app plus the run signal, so minutes
+// is far past any healthy bucket and still finite. The floors underneath
+// (hourly, and the next serial move) are what recover a pass cut short.
+const convergePassTimeout = 5 * time.Minute
+
// converge runs one converger pass, or nothing on a box with no deploy
// store. The agent's three call sites go through this so none of them can
-// reach into a nil conv.
+// reach into a nil conv — and so that all three get the timeout above
+// rather than each remembering to apply it.
func (sd *scheduleDeps) converge(ctx context.Context) error {
if sd == nil || sd.conv == nil {
return nil
}
+ ctx, cancel := context.WithTimeout(ctx, convergePassTimeout)
+ defer cancel()
return sd.conv.Pass(ctx)
}
@@ -163,9 +210,58 @@ func scheduleLegFor(sd *scheduleDeps) func(ctx context.Context) {
// in-flight count holds it open. No separate keep-awake to get wrong.
// - the tick never leaves the box: no DNS, no TLS, no edge auth, nothing an
// outsider could aim at the same path.
-type activatorDeliverer struct{ svc *activator.Service }
+type activatorDeliverer struct {
+ svc *activator.Service
+
+ // proxies caches one unix proxy — and therefore one http.Transport, and
+ // therefore one connection pool — per socket path. A fresh Transport per
+ // delivery meant a fresh TCP-over-unix connection for every tick and a
+ // pool that was thrown away unread; a schedule that fires every minute
+ // paid a dial (and the runtime a goroutine pair) each time for nothing.
+ // Keyed by socket path rather than host because the path is what the
+ // Transport actually dials: a route re-pointed at a new socket gets a new
+ // entry rather than a pool aimed at the old file.
+ //
+ // Bounded two ways: an entry is dropped when its delivery errors (a
+ // socket that would not answer is exactly the one whose pooled
+ // connections are worthless), and the whole map is dropped if it ever
+ // passes maxProxyCache — this box has one socket per exec instance, so
+ // the ceiling is only ever reached by a fleet far larger than one box
+ // holds, and a rebuild costs one dial per live schedule.
+ mu sync.Mutex
+ proxies map[string]http.Handler
+}
+
+// maxProxyCache is that ceiling. Deliberately well above the number of
+// exec instances a box runs (the 1000-per-host timer cap is per host, not
+// per box) so the reset is a safety net, never a working mechanism.
+const maxProxyCache = 512
-func (d activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
+// proxyFor is the cached unix proxy for socket.
+func (d *activatorDeliverer) proxyFor(socket string) http.Handler {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ if p, ok := d.proxies[socket]; ok {
+ return p
+ }
+ if d.proxies == nil {
+ d.proxies = map[string]http.Handler{}
+ } else if len(d.proxies) >= maxProxyCache {
+ clear(d.proxies)
+ }
+ p := edge.UnixProxy(socket)
+ d.proxies[socket] = p
+ return p
+}
+
+// dropProxy forgets socket's proxy, and with it its connection pool.
+func (d *activatorDeliverer) dropProxy(socket string) {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ delete(d.proxies, socket)
+}
+
+func (d *activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
if rt.Socket == "" {
// An addr-backed (or tunnel-backed) route has no socket to dial from
// here. Converge only ever writes rows for exec-backed instances, so
@@ -173,7 +269,7 @@ func (d activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req
// to an empty path (which would be a connect-refused every 15s).
return 0, errors.New("no socket for route " + rt.Host)
}
- proxy := edge.UnixProxy(rt.Socket)
+ proxy := d.proxyFor(rt.Socket)
h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// ServeHTTP reports false for a route it does not manage (not an
// instance, or not hibernate-flagged); that route is proxied straight,
@@ -183,6 +279,13 @@ func (d activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req
}
})
status, err := deliverVia(ctx, h, rt, req)
+ if err != nil {
+ // Anything that did not come back as an HTTP status — a refused
+ // wake, a socket that would not dial, a deadline — retires this
+ // socket's pool. The next tick dials fresh; a route that has gone
+ // away leaves nothing behind.
+ d.dropProxy(rt.Socket)
+ }
if err != nil && ctx.Err() == nil {
// Redacted for the same reason as the token error: this string is
// the row's last_status. The context's own error is left alone —
diff --git a/internal/host/schedule_wiring_test.go b/internal/host/schedule_wiring_test.go
index 2cd397fc8..8e5940f75 100644
--- a/internal/host/schedule_wiring_test.go
+++ b/internal/host/schedule_wiring_test.go
@@ -81,7 +81,7 @@ func TestDeliverViaReportsContextError(t *testing.T) {
// a route with no socket is an error, not a dial to an empty path.
func TestActivatorDelivererRefusesSocketlessRoute(t *testing.T) {
req := httptest.NewRequest("POST", "http://a.bab.oncarlos.com/jobs/x", nil)
- got, err := activatorDeliverer{}.Deliver(context.Background(), registry.Route{Host: "a.bab.oncarlos.com"}, req)
+ got, err := (&activatorDeliverer{}).Deliver(context.Background(), registry.Route{Host: "a.bab.oncarlos.com"}, req)
if got != 0 || err == nil || !strings.Contains(err.Error(), "no socket") {
t.Fatalf("socketless route → %d %v; want 0 and a no-socket error", got, err)
}