carlos / platform Public

activator: start a purge listener alongside each spawned instance

ExecBackend gets a Cache field; when set, Start passes each spawned
instance a --control-socket flag and starts a purge listener bound to
it, so the instance can invalidate its own cached responses right
after a write. childProc closes the purge listener on Stop. nil Cache
disables this entirely (no purge listener, no error).

RunAgent now constructs one edge.Cache and shares it between the
activator's ExecBackend and edge.RunWithActivator, replacing the
inline cache Task 2's implementer had constructed only for the edge
side — without this fix, purges would land in a cache the edge never
reads from.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Paul Campbell 2 months ago bbbe47c6897bd7c35eab7db96fccae42daac433e parent 17c5afe

3 files changed, +117 −5

internal/activator/activator_test.go +84 −0
internal/activator/backend_exec.go +25 −4
internal/host/agent.go +8 −1

diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go

index 5aee62fd7..2a5db49b0 100644

--- a/internal/activator/activator_test.go

+++ b/internal/activator/activator_test.go

@@ -26,6 +26,7 @@ import (

"testing"

"time"

+ "github.com/carlosframework/platform/internal/edge"

"github.com/carlosframework/platform/internal/registry"

)

@@ -1909,3 +1910,86 @@ func TestExecBackendHasSidecar(t *testing.T) {

t.Fatal("HasSidecar false with the sidecar env file staged")

}

+// Start passes a --control-socket flag and starts a purge listener bound to

+// it whenever ExecBackend.Cache is set — the wiring that lets a spawned

+// instance invalidate its own cached responses right after a write.

+//

+// Deviation from the task brief's literal test: the brief asserted the

+// control socket file still exists AFTER Start returns, even on the

+// health-wait error stageLive's fake binary (`#!/bin/sh\n`, never binds

+// rt.Socket) is expected to produce here. That's not actually true of the

+// implementation the same brief specifies — Start's pre-existing

+// health-wait-failure path calls p.Stop() before returning, and Step 3's

+// own childProc.Stop change closes the purge listener there, which (per

+// Go's net.Listener.Close on a unix socket the listener itself created)

+// unlinks the socket file. So by the time Start returns with an error, the

+// file is already gone — confirmed empirically, not a guess. This test

+// instead polls for the socket WHILE Start's health-wait is still running

+// (i.e. before its internal Stop cleanup fires), which is the invariant

+// that's actually guaranteed and is what "Start wires up the purge

+// listener" means in practice.

+func TestExecBackendStartPassesControlSocketFlagAndStartsPurgeListener(t *testing.T) {

+ b := testExecBackend(t)

+ b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)

+ stageLive(t, b, "control-socket.example.com", "v1")

+ rt := registry.Route{

+ Host: "control-socket.example.com",

+ Kind: "instance",

+ Socket: filepath.Join(t.TempDir(), "instance.sock"),

+ }

+ wantControlSocket := rt.Socket + ".control"

+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)

+ defer cancel()

+ done := make(chan struct{})

+ var proc Proc

+ go func() {

+ proc, _, _ = b.Start(ctx, rt) // health-wait error from the fake binary is expected and irrelevant here

+ close(done)

+ }()

+ found := false

+ deadline := time.Now().Add(1500 * time.Millisecond)

+ for time.Now().Before(deadline) {

+ if _, err := os.Stat(wantControlSocket); err == nil {

+ found = true

+ break

+ }

+ time.Sleep(10 * time.Millisecond)

+ }

+ <-done

+ if proc != nil {

+ defer proc.Stop(context.Background())

+ }

+ if !found {

+ t.Fatalf("expected control socket %s to exist while Start's health-wait ran", wantControlSocket)

+ }

+// A nil ExecBackend.Cache disables the purge listener entirely — no control

+// socket, no error.

+func TestExecBackendStartWithNilCacheSkipsPurgeListener(t *testing.T) {

+ b := testExecBackend(t) // b.Cache left nil

+ stageLive(t, b, "no-cache.example.com", "v1")

+ rt := registry.Route{

+ Host: "no-cache.example.com",

+ Kind: "instance",

+ Socket: filepath.Join(t.TempDir(), "instance.sock"),

+ }

+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)

+ defer cancel()

+ proc, _, _ := b.Start(ctx, rt)

+ if proc != nil {

+ defer proc.Stop(context.Background())

+ }

+ if _, err := os.Stat(rt.Socket + ".control"); err == nil {

+ t.Fatal("expected no control socket when ExecBackend.Cache is nil")

+ }

diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go

index ffc0b128b..e6a52237f 100644

--- a/internal/activator/backend_exec.go

+++ b/internal/activator/backend_exec.go

@@ -19,6 +19,7 @@ import (

"syscall"

"time"

+ "github.com/carlosframework/platform/internal/edge"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/registry"

"github.com/carlosframework/platform/internal/release"

@@ -50,6 +51,10 @@ type ExecBackend struct {

// but a carlos app is a separate binary from the release store — there is

// nothing sensible to invent here, and the adopt pass owns the symlinks.

Materialize func(ctx context.Context, host string) error

+ // Cache, when non-nil, gets a purge listener started for each spawned

+ // instance so it can invalidate its own cached responses right after a

+ // write. Nil disables this — no purge listener, no error.

+ Cache *edge.Cache

}

var _ Backend = (*ExecBackend)(nil)

@@ -231,7 +236,8 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin

return nil, "", fmt.Errorf("read env file: %w", err)

}

- cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath)

+ controlSocket := rt.Socket + ".control"

+ cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath, "--control-socket", controlSocket)

cmd.Env = childEnv(fileEnv)

// The child logs through the agent's own stderr/stdout (journald under

// systemd); there's no per-instance unit to own a log stream anymore.

@@ -252,6 +258,17 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin

}

p := &childProc{cmd: cmd, done: make(chan struct{})}

+ if b.Cache != nil {

+ pl, err := edge.ListenPurge(controlSocket, rt.Host, b.Cache)

+ if err != nil {

+ // Not fatal to the wake — the instance still serves requests, it

+ // just can't purge its own cache until the next restart. A

+ // stuck-stale cache self-heals via the safety-net TTL anyway.

+ log.Printf("activator: purge listener for %s: %v (continuing without it)", rt.Host, err)

+ } else {

+ p.purge = pl

+ }

go func() {

p.err = cmd.Wait()

close(p.done)

@@ -402,9 +419,10 @@ func readEnvFile(path string) (map[string]string, error) {

// childProc is a spawned child: SIGTERM starts the instance's own graceful

// drain; SIGKILL only if it overstays.

type childProc struct {

- cmd *exec.Cmd

- done chan struct{}

- err error

+ cmd *exec.Cmd

+ done chan struct{}

+ err error

+ purge *edge.PurgeListener // nil if b.Cache was nil or ListenPurge failed

}

func (p *childProc) Exited() bool {

@@ -417,6 +435,9 @@ func (p *childProc) Exited() bool {

}

func (p *childProc) Stop(ctx context.Context) error {

+ if p.purge != nil {

+ p.purge.Close()

+ }

if p.Exited() {

return nil

}

diff --git a/internal/host/agent.go b/internal/host/agent.go

index a47d0fc4a..f4e09d086 100644

--- a/internal/host/agent.go

+++ b/internal/host/agent.go

@@ -110,6 +110,12 @@ func RunAgent(args []string) {

log.Fatalf("agent: open registry %s: %v", regPath, err)

}

+ // Constructed once here so the activator's ExecBackend (purges) and the

+ // edge (reads + serves cached hits) share the very same store — see the

+ // package doc comment above about caching applying before the

+ // hibernation branch.

+ cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)

cfg, err := activator.ConfigFromEnv()

if err != nil {

log.Fatalf("agent: %v", err)

@@ -126,6 +132,7 @@ func RunAgent(args []string) {

Cfg: *cfg,

OptDir: release.OptDir(),

DataDir: dd,

+ Cache: cache,

Unit: &activator.UnitBackend{OptDir: release.OptDir()},

}, reg) // the registry satisfies activator.DueStore directly

@@ -142,7 +149,7 @@ func RunAgent(args []string) {

// The same serving loop as `carlos edge`, with the activator injected.

// Caching applies here too: cached hits still avoid the activator, which

// is the entire point of checking the cache before the hibernation branch.

- edge.RunWithActivator(args, act, edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes))

+ edge.RunWithActivator(args, act, cache)

// The edge has drained; take the children down too (they're this process's

// children — orphaning them would strand their leases until TTL expiry).