carlos / platform Public

registry: sidecar kind, args column, and due-time widening

Paul Campbell pushed by paul@keymail.dev d80276410f027a97be1f198f6b3fd4d8b63e09ab parent ae1e462
2 files changed, +139 −10
  • internal/registry/registry.go +50 −10
  • internal/registry/registry_test.go +89 −0
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 14b7be821..b68b7089e 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -28,13 +28,15 @@ type Registry struct {
// Route is one host a deployment serves: an app instance the platform
// lifecycle-manages (kind "instance"), a sibling service it only proxies
-// (kind "service"), or a served file tree (kind "static").
+// (kind "service"), a served file tree (kind "static"), or a process that
+// serves no requests and that nothing can dial (kind "sidecar" — its
+// next_due wake, via DueInstances, is its only timer wake source).
type Route struct {
Host string
// Slug is the app-account short name usable on a root domain. Empty =
// subdomain-only.
Slug string
- Kind string // instance | service | static
+ Kind string // instance | service | static | sidecar
Socket string // unix socket upstream; wins over Addr when both set
Addr string // TCP upstream host:port — whole-app tenants (Kass v1)
// Addrs is a route's multi-upstream set (HA design, issue #152): a
@@ -191,6 +193,18 @@ type Route struct {
// it — which it must be able to do, for a manifest whose seed was
// removed.
Seed string
+ // Args is a sidecar's argv as a JSON array, materialized from the app's
+ // sidecars manifest on every reconcile pass (truth stays in the bucket).
+ // Empty for every other kind of route.
+ Args string
+ // IdleSeconds and MaxRunSeconds are a sidecar worker's ceilings, in
+ // seconds, materialized from the app's manifest exactly as Args and Seed
+ // are — 0 means "use the platform default". The sweeper reads them
+ // locally on every tick to decide whether a worker has outrun its
+ // ceiling; a bucket read per wake would put the control plane in the hot
+ // path of a timer.
+ IdleSeconds int
+ MaxRunSeconds int
// DNSStatus is the last DNS health verdict for this host: "pointing"
// (resolves to a deployment edge IP), "elsewhere", or "" — unknown, the
// honest answer for a resolver failure, never a guess.
@@ -362,6 +376,9 @@ CREATE TABLE IF NOT EXISTS routes (
`ALTER TABLE routes ADD COLUMN fleet_box TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN addrs TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN health TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN args TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN idle_seconds INTEGER NOT NULL DEFAULT 0`,
+ `ALTER TABLE routes ADD COLUMN max_run_seconds INTEGER NOT NULL DEFAULT 0`,
} {
if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
@@ -557,7 +574,7 @@ func (r *Registry) Accounts() ([]Account, error) {
}
// routeCols is the column list every route read and the one route write
-// share — 35 names, in this exact order. scanRoute's destinations, Upsert's
+// share — 38 names, in this exact order. scanRoute's destinations, Upsert's
// placeholder row and Upsert's argument list all count from here: adding a
// column means adding it in FOUR places, and a miscount is a silent column
// shift rather than an error (TestUpsertRoundTripsAllColumns is what catches
@@ -566,7 +583,8 @@ const routeCols = `host, slug, kind, socket, addr, status, last_error, db_path,
version, prev_version, app, pool, region, channel, hibernate, canary,
replication, next_due, backing, status_named, platform, dns_status,
dns_checked_at, cert_expires_at, account, environment, caps, tunnel,
-provisioned, seed, unit, fleet_box, addrs, health, created_at`
+provisioned, seed, unit, fleet_box, addrs, health, created_at,
+args, idle_seconds, max_run_seconds`
func scanRoute(scan func(...any) error) (Route, error) {
var t Route
@@ -576,7 +594,8 @@ func scanRoute(scan func(...any) error) (Route, error) {
&t.Region, &t.Channel, &hib, &can, &t.Replication, &t.NextDue, &t.Backing,
&sn, &plat, &t.DNSStatus, &t.DNSCheckedAt, &t.CertExpiresAt, &t.Account,
&t.Environment, &t.Caps, &t.Tunnel, &prov, &t.Seed, &t.Unit, &t.FleetBox,
- &t.Addrs, &t.Health, &t.CreatedAt)
+ &t.Addrs, &t.Health, &t.CreatedAt,
+ &t.Args, &t.IdleSeconds, &t.MaxRunSeconds)
t.Hibernate = hib != 0
t.Canary = can != 0
t.StatusNamed = sn != 0
@@ -712,7 +731,7 @@ func (r *Registry) Upsert(t Route) error {
}
_, err := r.sql.Exec(`
INSERT INTO routes (`+routeCols+`)
-VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(host) DO UPDATE SET
slug=excluded.slug, kind=excluded.kind, socket=excluded.socket,
addr=excluded.addr, status=excluded.status, last_error=excluded.last_error,
@@ -738,7 +757,8 @@ ON CONFLICT(host) DO UPDATE SET
b2i(t.Hibernate), b2i(t.Canary), t.Replication, t.NextDue, t.Backing,
b2i(t.StatusNamed), b2i(t.Platform), t.DNSStatus, t.DNSCheckedAt,
t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel,
- b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health, t.CreatedAt)
+ b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health, t.CreatedAt,
+ t.Args, t.IdleSeconds, t.MaxRunSeconds)
return err
}
@@ -1281,6 +1301,24 @@ func (r *Registry) SetSeed(host, seed string) error {
return nil
}
+// SetArgs repoints a sidecar's argv — the narrow UPDATE the sidecar
+// reconcile pass uses when the app's manifest changed.
+func (r *Registry) SetArgs(host, args string) error {
+ _, err := r.sql.Exec(`UPDATE routes SET args = ? WHERE host = ?`, args, host)
+ return err
+}
+
+// SetWorkerLimits repoints a sidecar's idle window and run ceiling, both in
+// seconds, 0 meaning "use the platform default". Same narrow-UPDATE shape
+// and same reason as SetArgs: the manifest is the truth, this is the copy
+// the sweeper reads on every tick.
+func (r *Registry) SetWorkerLimits(host string, idle, maxRun int) error {
+ _, err := r.sql.Exec(
+ `UPDATE routes SET idle_seconds = ?, max_run_seconds = ? WHERE host = ?`,
+ idle, maxRun, host)
+ return err
+}
+
// SetProvisioned stamps (or clears) the flag that marks a route as the
// instance-provisioning reconcile pass's own — a narrow UPDATE, same
// rationale as SetSeed. It errors when no row matched, the SetChannel
@@ -1336,11 +1374,13 @@ func (r *Registry) SetStatusNamed(host string, on bool) error {
return nil
}
-// DueInstances returns hibernatable instance routes whose next_due has
-// arrived (0 means no schedule and never matches).
+// DueInstances returns hibernatable instance AND sidecar routes whose
+// next_due has arrived (0 means no schedule and never matches). Sidecars
+// are included because the due-time sweep is their ONLY timer wake source:
+// nothing can dial a sidecar, so no request will ever wake one.
func (r *Registry) DueInstances(now int64) ([]Route, error) {
rows, err := r.sql.Query(`SELECT `+routeCols+` FROM routes
-WHERE kind = 'instance' AND next_due > 0 AND next_due <= ? ORDER BY host`, now)
+WHERE kind IN ('instance','sidecar') AND next_due > 0 AND next_due <= ? ORDER BY host`, now)
if err != nil {
return nil, err
}
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index fdef66c1a..0bdba818d 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -2378,3 +2378,92 @@ func TestWildcardDomains(t *testing.T) {
t.Fatalf("after delete: %+v", got)
}
}
+
+// TestDueInstancesIncludesSidecars pins the reason this task exists: a
+// sidecar cannot be dialed, so its due-time wake is its ONLY schedule wake
+// source. Left filtering kind='instance', DueInstances would store a
+// sidecar's next_due and never once act on it.
+func TestDueInstancesIncludesSidecars(t *testing.T) {
+ r := openTestRegistry(t)
+ if err := r.Upsert(Route{
+ Host: "bot.eleven.bdm.sidecar.internal", Kind: "sidecar",
+ DBPath: "/tmp/bot.db", Hibernate: true, Status: "ready",
+ }); err != nil {
+ t.Fatalf("upsert sidecar: %v", err)
+ }
+ if err := r.SetNextDue("bot.eleven.bdm.sidecar.internal", 100); err != nil {
+ t.Fatalf("SetNextDue: %v", err)
+ }
+
+ due, err := r.DueInstances(200)
+ if err != nil {
+ t.Fatalf("DueInstances: %v", err)
+ }
+ if len(due) != 1 || due[0].Host != "bot.eleven.bdm.sidecar.internal" {
+ t.Fatalf("DueInstances = %+v, want the due sidecar — a sidecar's timer wake is its ONLY schedule wake source", due)
+ }
+}
+
+// TestArgsRoundTrips checks Args survives Upsert and can be repointed
+// narrowly by SetArgs, the door the sidecar reconcile pass uses when the
+// app's manifest argv changes.
+func TestArgsRoundTrips(t *testing.T) {
+ r := openTestRegistry(t)
+ if err := r.Upsert(Route{Host: "bot.x.internal", Kind: "sidecar", Args: `["sidecar","run"]`}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ got, err := r.Get("bot.x.internal")
+ if err != nil {
+ t.Fatalf("get: %v", err)
+ }
+ if got.Args != `["sidecar","run"]` {
+ t.Fatalf("Args = %q, want the argv back", got.Args)
+ }
+
+ if err := r.SetArgs("bot.x.internal", `["worker","run"]`); err != nil {
+ t.Fatalf("SetArgs: %v", err)
+ }
+ got, _ = r.Get("bot.x.internal")
+ if got.Args != `["worker","run"]` {
+ t.Fatalf("Args after SetArgs = %q", got.Args)
+ }
+}
+
+// TestWorkerLimitsRoundTrip checks IdleSeconds/MaxRunSeconds survive Upsert
+// and can be repointed narrowly by SetWorkerLimits — the sweeper reads these
+// two columns locally on every tick, so a bucket read must never be in that
+// hot path.
+func TestWorkerLimitsRoundTrip(t *testing.T) {
+ r := openTestRegistry(t)
+ if err := r.Upsert(Route{Host: "bot.x.internal", Kind: "sidecar", IdleSeconds: 300, MaxRunSeconds: 3600}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ got, _ := r.Get("bot.x.internal")
+ if got.IdleSeconds != 300 || got.MaxRunSeconds != 3600 {
+ t.Fatalf("limits = %d/%d, want 300/3600 — the sweeper reads these locally on every tick", got.IdleSeconds, got.MaxRunSeconds)
+ }
+ if err := r.SetWorkerLimits("bot.x.internal", 60, 120); err != nil {
+ t.Fatalf("SetWorkerLimits: %v", err)
+ }
+ got, _ = r.Get("bot.x.internal")
+ if got.IdleSeconds != 60 || got.MaxRunSeconds != 120 {
+ t.Fatalf("limits after set = %d/%d", got.IdleSeconds, got.MaxRunSeconds)
+ }
+}
+
+// TestRouteWrittenBeforeArgsStillParses pins the additive rule: a row
+// written by code that predates this column has no args/idle/max-run value,
+// and must still read back as the zero value rather than fail to parse.
+func TestRouteWrittenBeforeArgsStillParses(t *testing.T) {
+ r := openTestRegistry(t)
+ if err := r.Upsert(Route{Host: "old.example.com", Kind: "instance"}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ got, err := r.Get("old.example.com")
+ if err != nil {
+ t.Fatalf("get: %v", err)
+ }
+ if got.Args != "" || got.IdleSeconds != 0 || got.MaxRunSeconds != 0 {
+ t.Fatalf("row = %+v, want zero values for a pre-existing row", got)
+ }
+}