amadan

carlos / platform Public

zdd plan 1 task 8: scheduled delivery and the status probe follow AppSocket

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c8dbe3deb018f18e9b92f4fc85c12545bc9e55ec parent 5ee6243
4 files changed, +246 −6
  • internal/host/agent.go +14 −3
  • internal/host/generations_test.go +168 −0
  • internal/host/probes.go +24 −0
  • internal/host/schedule_wiring.go +40 −3
diff --git a/internal/host/agent.go b/internal/host/agent.go
index c713732d2..2049b8b69 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -558,7 +558,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
if errSink != nil {
go errshipLoop(errSink, status.Label(readMachineID()))
}
- go statusLoop(reg)
+ go statusLoop(reg, svc)
// The memory early-warning (memory-pressure spec §2.4): per box, gated
// by CARLOS_HOST_METRICS — one log line and nothing else where unset.
go memwatchLoop()
@@ -695,6 +695,17 @@ func runAgent(args []string, buildClaims ClaimFactory) {
stopHousekeeper()
}
+// statusSocketProbe is the kernel-table probe the status loop hands
+// wrapListeningProbe: listeningProbe itself with generations off (or no
+// activator), and pointed at each managed route's current generation with
+// them on (currentSocketProbe).
+func statusSocketProbe(routes []registry.Route, svc *activator.Service) func(string) *bool {
+ if svc == nil || !svc.GenerationsOn() {
+ return listeningProbe
+ }
+ return currentSocketProbe(routes, svc.AppSocket, listeningProbe)
+}
+
func configureWakeClaims(w *ClaimWiring, cfg *activator.Config) error {
if w == nil {
return hostwrite.ErrUnavailable
@@ -3075,7 +3086,7 @@ func upstreamStates() map[string][]status.UpstreamState {
// logging that every 5 minutes forever would bury the log in noise. Any
// OTHER acquire error (a real S3/credential problem) is logged, same as any
// other tick failure.
-func statusLoop(reg *registry.Registry) {
+func statusLoop(reg *registry.Registry, svc *activator.Service) {
ctx := context.Background()
// bucket is read for the lease key alone (lease.Acquire below), which a
// fleet box never reaches: shouldSkipPublisherElection short-circuits the
@@ -3188,7 +3199,7 @@ func statusLoop(reg *registry.Registry) {
Region: boxRegion,
MemFrac: readMemFrac(),
DiskFrac: readDiskFrac(),
- Listening: wrapListeningProbe(unitBackedSockets(routes), listeningProbe),
+ Listening: wrapListeningProbe(unitBackedSockets(routes), statusSocketProbe(routes, svc)),
// One systemctl call per 5-minute tick, asking the only question
// that catches an enabled-but-never-armed adoption timer — the
// flagship's silent no-adopt state, which nothing on any page
diff --git a/internal/host/generations_test.go b/internal/host/generations_test.go
new file mode 100644
index 000000000..0a564f909
--- /dev/null
+++ b/internal/host/generations_test.go
@@ -0,0 +1,168 @@
+package host
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// genWakeBackend is fakeWakeBackend with the generation capability: with
+// perGen it starts the fake instance on <dir>/g<n>.sock, never on rt.Socket,
+// so a tick that reached the canonical path would find nothing.
+type genWakeBackend struct {
+ fakeWakeBackend
+ dir string
+}
+
+func (b *genWakeBackend) StartGen(_ context.Context, rt registry.Route, n uint64, perGen bool) (activator.Proc, activator.Gen, error) {
+ sock := rt.Socket
+ if perGen {
+ sock = filepath.Join(b.dir, fmt.Sprintf("g%d.sock", n))
+ }
+ b.inst.socket = sock
+ if err := b.inst.start(); err != nil {
+ return nil, activator.Gen{}, err
+ }
+ return &fakeInstanceProc{inst: b.inst}, activator.Gen{N: n, Socket: sock, Control: sock + ".control"}, nil
+}
+
+func (b *genWakeBackend) NextDueVia(context.Context, registry.Route, string) (int64, error) {
+ return 0, nil
+}
+
+// genDeliveryFixture is wakeFixture over genWakeBackend, with the
+// generations switch set before the activator reads it.
+func genDeliveryFixture(t *testing.T, on bool) (*activator.Service, *fakeInstance, registry.Route) {
+ t.Helper()
+ prev := activator.GenerationsEnabled()
+ activator.SetGenerations(on)
+ t.Cleanup(func() { activator.SetGenerations(prev) })
+ dd, err := os.MkdirTemp("", "h")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dd) })
+ reg, err := registry.Open(filepath.Join(dd, "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ inst := &fakeInstance{handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })}
+ rt := registry.Route{Host: "gen.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready",
+ Hibernate: true, Socket: filepath.Join(dd, "canonical.sock"), DBPath: filepath.Join(dd, "app.db")}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, &genWakeBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, dir: dd}, reg)
+ svc.Start()
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(ctx)
+ })
+ return svc, inst, rt
+}
+
+func deliverTick(t *testing.T, d *activatorDeliverer, rt registry.Route) int {
+ t.Helper()
+ 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 err != nil {
+ t.Fatalf("Deliver: %v", err)
+ }
+ return status
+}
+
+// Scheduled delivery dials AppSocket (spec §7.1): the tick wakes the
+// instance and reaches its generation socket, though nothing listens on the
+// route's canonical path.
+func TestScheduledDeliveryReachesTheCurrentGeneration(t *testing.T) {
+ requireUnixSockets(t)
+ svc, inst, rt := genDeliveryFixture(t, true)
+ d := &activatorDeliverer{svc: svc}
+ if got := deliverTick(t, d, rt); got != http.StatusNoContent {
+ t.Fatalf("tick status = %d, want 204 from the instance", got)
+ }
+ cur, ok := svc.AppSocket(rt.Host)
+ if !ok || inst.socket != cur || cur == rt.Socket {
+ t.Fatalf("instance bound %q, AppSocket = %q (%v); want a generation socket, not %q", inst.socket, cur, ok, rt.Socket)
+ }
+ if len(inst.served()) != 1 {
+ t.Fatalf("instance served %d ticks, want 1", len(inst.served()))
+ }
+ if d.genPools == nil || d.genPools.Len() != 1 {
+ t.Fatal("the tick did not go through the leg's generation pools")
+ }
+}
+
+// Flag off: the canonical socket and the cached per-socket proxy, as before.
+func TestScheduledDeliveryWithGenerationsOffUsesTheCanonicalSocket(t *testing.T) {
+ requireUnixSockets(t)
+ svc, inst, rt := genDeliveryFixture(t, false)
+ d := &activatorDeliverer{svc: svc}
+ if got := deliverTick(t, d, rt); got != http.StatusNoContent {
+ t.Fatalf("tick status = %d, want 204", got)
+ }
+ if inst.socket != rt.Socket {
+ t.Fatalf("instance bound %q, want the canonical %q", inst.socket, rt.Socket)
+ }
+ if d.genPools != nil {
+ t.Fatal("generation pools were made with the switch off")
+ }
+ if _, ok := d.proxies[rt.Socket]; !ok {
+ t.Fatal("the canonical per-socket proxy was not used")
+ }
+}
+
+// The status probe asks about each managed route's current generation, and
+// about everything else exactly as before.
+func TestCurrentSocketProbePointsAtTheCurrentGeneration(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "awake.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/awake.example.com.sock"},
+ {Host: "asleep.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/asleep.example.com.sock"},
+ {Host: "svc.example.com", Kind: "service", Socket: "/run/carlos/svc.example.com.sock"},
+ }
+ appSocket := func(host string) (string, bool) {
+ if host == "awake.example.com" {
+ return "/run/carlos/g/0123456789ab.4.sock", true
+ }
+ return "", false
+ }
+ var asked []string
+ probe := currentSocketProbe(routes, appSocket, func(sock string) *bool {
+ asked = append(asked, sock)
+ v := true
+ return &v
+ })
+ for _, r := range routes {
+ probe(r.Socket)
+ }
+ want := []string{"/run/carlos/g/0123456789ab.4.sock", "/run/carlos/asleep.example.com.sock", "/run/carlos/svc.example.com.sock"}
+ if fmt.Sprint(asked) != fmt.Sprint(want) {
+ t.Fatalf("probed %v, want %v", asked, want)
+ }
+}
+
+// Flag off (or no activator): the status loop's probe is listeningProbe.
+func TestStatusSocketProbeIsUnchangedWithGenerationsOff(t *testing.T) {
+ got := statusSocketProbe(nil, nil)
+ if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", listeningProbe) {
+ t.Fatal("with no activator the status probe is not listeningProbe")
+ }
+ svc, _, _ := genDeliveryFixture(t, false)
+ got = statusSocketProbe(nil, svc)
+ if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", listeningProbe) {
+ t.Fatal("with generations off the status probe is not listeningProbe")
+ }
+}
diff --git a/internal/host/probes.go b/internal/host/probes.go
index d9e889387..dc42fd9ac 100644
--- a/internal/host/probes.go
+++ b/internal/host/probes.go
@@ -297,6 +297,30 @@ func wrapListeningProbe(unitBacked map[string]bool, probe func(string) *bool) fu
}
}
+// currentSocketProbe points probe at each activator-managed route's CURRENT
+// generation socket (zero-downtime spec §7.1): with generations on, nothing
+// binds a managed route's canonical path, so the kernel table would call
+// every awake instance asleep. A route the activator reports no current
+// generation for (asleep) is probed at its canonical path, where nothing
+// listens: asleep, which is the truth. Any other socket passes straight
+// through. appSocket is the activator's AppSocket.
+func currentSocketProbe(routes []registry.Route, appSocket func(host string) (string, bool), probe func(string) *bool) func(string) *bool {
+ hostOf := map[string]string{}
+ for _, r := range routes {
+ if r.Kind == "instance" && r.Hibernate && r.Socket != "" && r.Backing != registry.BackingUnit {
+ hostOf[r.Socket] = r.Host
+ }
+ }
+ return func(sock string) *bool {
+ if host, ok := hostOf[sock]; ok {
+ if cur, ok := appSocket(host); ok {
+ return probe(cur)
+ }
+ }
+ return probe(sock)
+ }
+}
+
// parseMeminfoBytes reads /proc/meminfo's MemTotal and MemAvailable (both in
// kB) and returns them converted to bytes. Either field missing or
// unparsable -> nil for both.
diff --git a/internal/host/schedule_wiring.go b/internal/host/schedule_wiring.go
index 03eb5be31..b89e0dcf7 100644
--- a/internal/host/schedule_wiring.go
+++ b/internal/host/schedule_wiring.go
@@ -300,6 +300,31 @@ type activatorDeliverer struct {
// rather than waiting out edge.UnixProxy's 90s IdleConnTimeout.
mu sync.Mutex
proxies map[string]http.Handler
+
+ // genPools is the delivery leg's own per-generation-socket pools, used
+ // instead of proxies for an activator-managed instance once generations
+ // are on (zero-downtime spec §7.1): a tick reaches the CURRENT
+ // generation, asked per connection. Made on first use and retired
+ // through the activator's hook, so a retired generation's pool closes
+ // with it rather than waiting for the daily prune. Nil with generations
+ // off, and nothing below reads it then.
+ genOnce sync.Once
+ genPools *edge.GenPools
+}
+
+// managedByGenerations reports whether rt's ticks go through the current
+// generation: generations on, and an activator-managed instance.
+func (d *activatorDeliverer) managedByGenerations(rt registry.Route) bool {
+ return d.svc != nil && d.svc.GenerationsOn() && rt.Kind == "instance" && rt.Hibernate
+}
+
+// genProxyFor is the AppSocket proxy for host over the leg's own pools.
+func (d *activatorDeliverer) genProxyFor(host string) http.Handler {
+ d.genOnce.Do(func() {
+ d.genPools = edge.NewGenPools()
+ d.svc.OnGenRetired(d.genPools.Retire)
+ })
+ return edge.AppSocketProxy(host, d.svc, d.genPools)
}
// maxProxyCache is that ceiling. Deliberately well above the number of
@@ -381,7 +406,16 @@ func (d *activatorDeliverer) deliverSelected(ctx context.Context, rt registry.Ro
// to an empty path (which would be a connect-refused every 15s).
return 0, errors.New("no socket for route " + rt.Host)
}
- proxy := d.proxyFor(rt.Socket)
+ // Generations on: the proxy asks the activator for the current
+ // generation's socket per connection (spec §7.1). Off: the cached proxy
+ // on rt.Socket, exactly as before.
+ gen := d.managedByGenerations(rt)
+ var proxy http.Handler
+ if gen {
+ proxy = d.genProxyFor(rt.Host)
+ } else {
+ proxy = d.proxyFor(rt.Socket)
+ }
var admissionErr error
if admit != nil {
next := proxy
@@ -416,8 +450,11 @@ func (d *activatorDeliverer) deliverSelected(ctx context.Context, rt registry.Ro
// this socket's pool; the next tick dials fresh. Not the case a
// reader expects here: a socket that will not dial is ReverseProxy's
// own 502 with no error, and it has nothing pooled to retire anyway.
- // Retired ROUTES are reclaimed by RetainSockets, not from here.
- d.dropProxy(rt.Socket)
+ // Retired ROUTES are reclaimed by RetainSockets, not from here. A
+ // generation's pool is retired by the activator's hook instead.
+ if !gen {
+ d.dropProxy(rt.Socket)
+ }
}
if err != nil && ctx.Err() == nil {
// Redacted for the same reason as the token error: this string is