amadan

carlos / platform Public

host: wire scheduled work — converge leg, runner, activator-backed delivery, status reports

The converge runs once at agent start (the seed), on every serial move (a
third fast leg, after the release pass set and the restart leg), and hourly
underneath both. The runner ticks every 15s and delivers through the
activator, so a hibernating instance wakes for its tick and is held awake by
the in-flight count. Wake refusals map to errors rather than app 503s, so the
member-facing status says which. Status reports go to the app's own key.

Also sets ExecBackend.Timers, mounting the schedule verbs on each instance's
control socket against the same registry this agent converges.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev bfe3c732360109e8adc71976a1811ac663f876e8 parent a2e4b50
3 files changed, +404 −4
  • internal/host/agent.go +79 −4
  • internal/host/schedule_wiring.go +218 −0
  • internal/host/schedule_wiring_test.go +107 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index f8bd49872..f3d00da79 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -266,6 +266,13 @@ func RunAgent(args []string) {
Cache: cache,
Materialize: materializeHook(reg, dd),
Cgroups: cgm,
+ // The scheduled-work verbs on every spawned instance's control
+ // socket write to the SAME registry this agent's converger and
+ // runner read (schedule spec §7.2): an app that asks for a
+ // once-timer from inside itself is asking this box's timer table,
+ // so the row is due-indexed by the next tick with nothing in
+ // between. reg satisfies schedule.TimerStore directly.
+ Timers: reg,
}
if logSink != nil { // a typed-nil in the interface would NOT be nil
eb.Logs = logSink
@@ -305,8 +312,34 @@ func RunAgent(args []string) {
go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
}
- go adoptLoop(reg, svc)
- go fastConvergeLoop(reg, svc)
+ // Scheduled work (schedule spec §6.2/§6.3), built ONCE here and shared by
+ // all three legs below: the seed and the fast leg in fastConvergeLoop, the
+ // hourly leg in adoptLoop, and the runner started here. One converger is a
+ // correctness requirement, not tidiness — its etag memory is what makes a
+ // declaration change cheap and a run signal exactly-once, and a second
+ // instance elsewhere would replay or swallow one.
+ //
+ // A box with no deploy store simply has no scheduled work: schedules live
+ // in the bucket, so there is nothing to converge and nowhere to report.
+ // One line, like every other optional loop.
+ var sd *scheduleDeps
+ if schedStore, err := openDeployStore(context.Background()); err != nil {
+ log.Printf("agent: scheduled work disabled: %v", err)
+ } else {
+ // 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 —
+ // the same trap the activator's Logs assignment above avoids.
+ var sink activator.EventSink
+ if logSink != nil {
+ sink = logSink
+ }
+ sd = newScheduleDeps(reg, schedStore, svc, sink, dd)
+ sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
+ log.Printf("agent: scheduled work armed (converge on serial + hourly; runner every 15s)")
+ }
+
+ go adoptLoop(reg, svc, sd)
+ go fastConvergeLoop(reg, svc, sd)
if logSink != nil {
go logshipLoop(logSink, status.Label(readMachineID()))
}
@@ -459,7 +492,11 @@ func materializeHook(reg *registry.Registry, dd string) func(context.Context, st
// The activator is taken so the pass can cycle exec-backed instances the way
// only this process can (see makeRestartRoute); nil is fine — a box with no
// hibernation armed simply has no children to bounce.
-func adoptLoop(reg *registry.Registry, svc *activator.Service) {
+//
+// sd carries the scheduled-work converger; this loop is its HOURLY floor (the
+// fast one rides the serial in fastConvergeLoop). Nil when the box has no
+// deploy store — the agent logged that once at arming.
+func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -513,6 +550,15 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service) {
release.RestartHibernating(restart)); err != nil {
log.Printf("agent: adopt pass: %v", err)
}
+ // The schedule converge's hourly floor (schedule spec §6.2), AFTER the
+ // release pass: an app whose schedules.json arrived with a new version
+ // gets its rows against the version that is actually staged. Die-free
+ // per app inside Pass — the joined error is one log line, never fatal.
+ if sd != nil {
+ if err := sd.conv.Pass(ctx); err != nil {
+ log.Printf("agent: schedule converge (hourly): %v", err)
+ }
+ }
}
}
@@ -971,6 +1017,7 @@ type fastDeps struct {
passOpts []release.PassOption // fastPassOpts — every sweep pass carries all of them
fullPass func(ctx context.Context) // reconcile → adopt → config sync, or a test hook
restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
+ scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
lastSerial int64
seeded bool
lastFull time.Time
@@ -1114,6 +1161,8 @@ func (d *fastDeps) backoffFor(host string) time.Duration { return d.step[host] }
// fullPass because it is not release convergence — it restarts what is
// already correct — and because it must still run on a box with no
// CARLOS_SECRETS_KEY, which fullPass's config leg silently skips.
+// The schedule converge (schedule spec §6.2) is a third leg on that same
+// move, last, for both of the same reasons — see the call site.
// 2. Fast sweep: fastMembers, and only when non-empty one PassFilter over
// exactly those hosts — an empty sweep costs zero store reads. After
// the sweep, members still unstaged back off; staged members clear
@@ -1144,6 +1193,16 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
if d.restartLeg != nil {
d.restartLeg(ctx)
}
+ // The schedule converge rides the same move, after the pass set for
+ // the same reason the restart leg does: a bump can carry a promote
+ // and a `carlos schedule set` at once, and converging schedules
+ // before the version they belong to has staged would write rows
+ // against the outgoing build. Also a separate leg rather than a
+ // member of fullPass — it is not release convergence, and it must
+ // still run on a box with no CARLOS_SECRETS_KEY.
+ if d.scheduleLeg != nil {
+ d.scheduleLeg(ctx)
+ }
d.lastSerial = doc.Serial
d.lastFull = now
}
@@ -1190,7 +1249,7 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in fastTick/fastMembers, which are tested.
-func fastConvergeLoop(reg *registry.Registry, svc *activator.Service) {
+func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -1219,6 +1278,18 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service) {
restartCache := map[[2]string]string{}
restartSeed(ctx, reg, store, dd, restartCache)
restartBounce := bounceFn(svc)
+ // The schedule converge's own seed, on restartSeed's precedent and for the
+ // same shape of reason: the leg below runs only on a serial MOVE, so
+ // without one pass here a box that booted after the bump would carry no
+ // timer rows (and report no status) until the hourly floor in adoptLoop —
+ // up to an hour of a schedule silently not running. It also seeds the
+ // converger's etag memory, which is what makes a run signal older than this
+ // process read as history rather than as a fresh request to run.
+ if sd != nil {
+ if err := sd.conv.Pass(ctx); err != nil {
+ log.Printf("agent: schedule converge (start): %v", err)
+ }
+ }
d := &fastDeps{
reg: reg, store: store, pub: pub, optDir: optDir,
restart: restart, passOpts: opts,
@@ -1253,6 +1324,10 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service) {
}
restartRequestLeg(ctx, routes, store, dd, restartBounce, restartCache)
},
+ // The schedule leg: converge this box's timer rows from the apps'
+ // declarations. Nil when the box has no scheduled work armed, which
+ // fastTick tolerates.
+ scheduleLeg: scheduleLegFor(sd),
}
log.Printf("agent: fast converge loop armed (2s + jitter: serial poll + canary/* + unstaged routes)")
for {
diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go
new file mode 100644
index 000000000..683f3484b
--- /dev/null
+++ b/internal/host/schedule_wiring.go
@@ -0,0 +1,218 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "log"
+ "net/http"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
+ "github.com/carlosframework/platform/internal/version"
+)
+
+// Scheduled work, host side (spec §6.2, §6.3, §4.3).
+//
+// Two halves, one set of dependencies:
+//
+// - the CONVERGER turns apps/<acct>/<app>/schedules.json into timer rows in
+// this box's registry. It runs on every serial move (so a `carlos schedule
+// set` lands in seconds, like every other console-mediated write), hourly
+// underneath that (the floor a missed bump falls back to), and once at
+// agent start (a box that booted after the bump must not wait an hour to
+// learn its schedules).
+// - the RUNNER fires the rows that are due, every 15s, and reports what
+// happened back to the bucket so `carlos schedule status` can answer
+// without a box command.
+//
+// Both are built ONCE, in the agent's startup, and shared: the converger's
+// etag memory is what keeps an idle box at one GET per app per pass and makes
+// a run signal exactly-once, so a second instance in another loop would
+// re-seed that memory and either replay a run or swallow one.
+type scheduleDeps struct {
+ conv *schedule.Converger
+ runner *schedule.Runner
+}
+
+// newScheduleDeps assembles the converger and the runner around this agent's
+// registry, deploy store and activator.
+//
+// svc may be nil (no hibernation armed on this box): delivery then proxies
+// straight to the instance's socket, which is exactly right — an instance that
+// is not activator-managed is either already running or not this leg's problem.
+// sink may be nil (no log store): the runner simply emits no member events.
+// Callers must pass a genuinely nil interface for sink, not a nil *applog.Sink
+// — see the trap agent.go documents where logSink is threaded.
+func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps {
+ box, err := os.Hostname()
+ if err != nil || box == "" {
+ box = "unknown"
+ }
+ // The reporter is shared by both halves on purpose: a converge that adds a
+ // row and a tick that fires it write the SAME status object, so whichever
+ // ran last is what a member reads. Its own timeout, not the caller's — a
+ // slow bucket must not hold a converge pass or a tick goroutine open.
+ report := func(rt registry.Route, st schedule.Status) {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := schedule.WriteStatus(ctx, store, release.RouteAccount(rt), rt.App, st); err != nil {
+ log.Printf("schedule: report %s: %v", rt.Host, err)
+ }
+ }
+ var event func(rt registry.Route, level, msg string)
+ if sink != nil {
+ event = func(rt registry.Route, level, msg string) {
+ sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg)
+ }
+ }
+ return &scheduleDeps{
+ conv: schedule.NewConverger(reg, store, box, report),
+ runner: &schedule.Runner{
+ Reg: reg,
+ Deliver: activatorDeliverer{svc: svc},
+ Token: schedule.TokenFromEnvFile(dd), // the activator's own <host>.env
+ Event: event,
+ Report: report,
+ Box: box,
+ Build: version.Build,
+ },
+ }
+}
+
+// scheduleLegFor is fastDeps.scheduleLeg for sd — a genuinely nil func when
+// scheduled work is not armed, so the tick skips the leg rather than calling a
+// closure over a nil converger.
+func scheduleLegFor(sd *scheduleDeps) func(ctx context.Context) {
+ if sd == nil {
+ return nil
+ }
+ return func(ctx context.Context) {
+ if err := sd.conv.Pass(ctx); err != nil {
+ log.Printf("agent: schedule converge: %v", err)
+ }
+ }
+}
+
+// activatorDeliverer delivers one tick the way a member request reaches an
+// instance from inside this process — through the activator, then the
+// instance's unix socket — and never through the public edge (spec §6.3).
+// Two properties come from that and only that:
+//
+// - a hibernating instance WAKES for its tick, exactly as it would for a
+// visitor, and stays awake for the tick's duration because the activator's
+// 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 }
+
+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
+ // this is a row that outlived its route's shape — an error, not a dial
+ // 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)
+ 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,
+ // the same call shape the edge itself uses.
+ if d.svc == nil || !d.svc.ServeHTTP(w, r, rt, proxy) {
+ proxy.ServeHTTP(w, r)
+ }
+ })
+ return deliverVia(ctx, h, rt, req)
+}
+
+// deliverVia is the pure core: run the handler against a recorder and turn
+// what it wrote into the runner's (status, error) pair.
+//
+// The mapping is the whole point. The activator answers its own refusals with
+// a 503 whose body names the instance ("instance is waking", "instance failed
+// to wake", "instance is waking elsewhere"), which is indistinguishable at the
+// status line from an app that answered 503 itself. Both retry — but the
+// member-facing status must say WHICH, so a wake refusal comes back as an
+// error (classified "503: instance is waking…") and an app's own 503 comes
+// back as the status 503. Anything else is returned as the app said it.
+func deliverVia(ctx context.Context, h http.Handler, rt registry.Route, req *http.Request) (int, error) {
+ rec := &statusRecorder{}
+ h.ServeHTTP(rec, req)
+ // Checked AFTER the handler returns, and before the status: a tick that
+ // ran past its deadline gets a 502 written under it by the proxy on the
+ // way out, and reporting that 502 would retry a job that may well have
+ // completed. The runner maps DeadlineExceeded to ErrTimeout, the one
+ // outcome that does not retry.
+ if err := ctx.Err(); err != nil {
+ return 0, err
+ }
+ line := rec.firstLine()
+ if rec.status == http.StatusServiceUnavailable && strings.HasPrefix(line, "instance") {
+ return 0, errors.New(line)
+ }
+ if rec.status == 0 {
+ // The handler returned without writing anything and the context is
+ // fine — nothing an instance can legitimately do (even a failed dial
+ // is a 502). Reporting status 0 would classify as "app refused" and
+ // never retry; this says what actually happened and retries.
+ return 0, errors.New("no response from " + rt.Host)
+ }
+ return rec.status, nil
+}
+
+// firstLineMax bounds what the recorder keeps of a body. A tick's response
+// body is never delivered anywhere — only the first line of a wake refusal is
+// ever read — so a chatty app cannot make this leg hold its output in memory.
+const firstLineMax = 256
+
+// statusRecorder is the ResponseWriter the activator and the proxy write
+// through: it keeps the status and the first line of the body and discards
+// everything else. Not httptest.ResponseRecorder — that one buffers the whole
+// body, and this runs in the agent on every tick.
+type statusRecorder struct {
+ header http.Header
+ status int
+ line []byte
+ lineEnd bool // newline seen or cap reached — later writes are discarded
+}
+
+func (r *statusRecorder) Header() http.Header {
+ if r.header == nil {
+ r.header = http.Header{}
+ }
+ return r.header
+}
+
+func (r *statusRecorder) WriteHeader(code int) {
+ if r.status == 0 {
+ r.status = code
+ }
+}
+
+func (r *statusRecorder) Write(b []byte) (int, error) {
+ r.WriteHeader(http.StatusOK) // implicit 200, as on a real writer
+ if !r.lineEnd {
+ if i := bytes.IndexByte(b, '\n'); i >= 0 {
+ r.line = append(r.line, b[:i]...)
+ r.lineEnd = true
+ } else {
+ r.line = append(r.line, b...)
+ }
+ if len(r.line) >= firstLineMax {
+ r.line = r.line[:firstLineMax]
+ r.lineEnd = true
+ }
+ }
+ return len(b), nil
+}
+
+// firstLine is the body's first line, trimmed — what the wake-refusal test
+// above matches on.
+func (r *statusRecorder) firstLine() string { return strings.TrimSpace(string(r.line)) }
diff --git a/internal/host/schedule_wiring_test.go b/internal/host/schedule_wiring_test.go
new file mode 100644
index 000000000..b25138a3e
--- /dev/null
+++ b/internal/host/schedule_wiring_test.go
@@ -0,0 +1,107 @@
+package host
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// A tick's outcome is read from the handler's own answer, so the mapping from
+// "what the activator wrote" to "what the policy sees" is the whole contract
+// of this leg: an app's 503 must retry as an app 503, while the activator's
+// own wake refusal must retry as a wake failure — same retry, different
+// member-facing status.
+func TestDeliverViaHandlerMapsWakeRefusals(t *testing.T) {
+ cases := []struct {
+ status int
+ body string
+ want int
+ wantErr string
+ }{
+ {204, "", 204, ""},
+ {503, "instance failed to wake\n", 0, "instance failed to wake"},
+ {503, "instance is waking; retry shortly\n", 0, "instance is waking; retry shortly"},
+ {503, "instance is waking elsewhere; retry shortly\n", 0, "instance is waking elsewhere"},
+ {503, "app says no", 503, ""},
+ {500, "instance blew up", 500, ""}, // only a 503 is a wake refusal
+ }
+ for _, c := range cases {
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(c.status)
+ w.Write([]byte(c.body))
+ })
+ req := httptest.NewRequest("POST", "http://a.bab.oncarlos.com/jobs/x", nil)
+ got, err := deliverVia(context.Background(), h, registry.Route{Host: "a.bab.oncarlos.com"}, req)
+ if got != c.want || (err == nil) != (c.wantErr == "") || (err != nil && !strings.Contains(err.Error(), c.wantErr)) {
+ t.Errorf("%d %q → %d %v; want %d %q", c.status, c.body, got, err, c.want, c.wantErr)
+ }
+ }
+}
+
+// A handler that writes a body without WriteHeader has answered 200 — the
+// recorder must say so, or every well-behaved app that just writes its
+// response would be recorded as "0 (app refused)".
+func TestDeliverViaImplicit200(t *testing.T) {
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "text/plain")
+ w.Write([]byte("ok"))
+ })
+ req := httptest.NewRequest("POST", "http://a.bab.oncarlos.com/jobs/x", nil)
+ got, err := deliverVia(context.Background(), h, registry.Route{Host: "a.bab.oncarlos.com"}, req)
+ if got != 200 || err != nil {
+ t.Fatalf("implicit 200 → %d %v", got, err)
+ }
+}
+
+// A tick that ran past its deadline is the context's error, not the 502 the
+// proxy writes on the way out: the runner maps DeadlineExceeded to ErrTimeout,
+// which is the one outcome that does NOT retry (the job may still be running).
+func TestDeliverViaReportsContextError(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ cancel() // the handler runs long; the deadline fires under it
+ w.WriteHeader(http.StatusBadGateway)
+ })
+ req := httptest.NewRequest("POST", "http://a.bab.oncarlos.com/jobs/x", nil)
+ got, err := deliverVia(ctx, h, registry.Route{Host: "a.bab.oncarlos.com"}, req)
+ if got != 0 || err != context.Canceled {
+ t.Fatalf("cancelled ctx → %d %v; want 0 context.Canceled", got, err)
+ }
+}
+
+// A tick has nowhere to go on an addr-backed route: the delivery leg dials the
+// instance's own unix socket on purpose (spec §6.3 — never the public edge), so
+// 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)
+ 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)
+ }
+}
+
+// The recorder is a ResponseWriter the activator writes through, so it must
+// keep the headers the activator sets (Retry-After on a wake refusal) without
+// panicking on a nil map, and must keep only the FIRST line of a body of any
+// size — a tick's response body is never read, only classified.
+func TestStatusRecorderHeadersAndFirstLine(t *testing.T) {
+ rec := &statusRecorder{}
+ rec.Header().Set("Retry-After", "5")
+ rec.WriteHeader(503)
+ rec.Write([]byte("instance is waking; retry shortly\n"))
+ rec.Write([]byte(strings.Repeat("x", 1<<20)))
+ rec.WriteHeader(200) // a second WriteHeader is a no-op, as on a real writer
+ if rec.status != 503 {
+ t.Fatalf("status = %d, want 503", rec.status)
+ }
+ if got := rec.firstLine(); got != "instance is waking; retry shortly" {
+ t.Fatalf("firstLine = %q", got)
+ }
+ if got := rec.Header().Get("Retry-After"); got != "5" {
+ t.Fatalf("Retry-After = %q", got)
+ }
+}