amadan

carlos / platform Public

zdd plan 3 task 8: a retiring generation's schedule writes get 409 after the switch

schedule.MountGated gates PUT and DELETE on a retired func; Service.RetiredControl
names the retiring generation's control socket; ExecBackend.Retired is wired in
agent.go. Reads and the bearer check are unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 3ff569b88e95facc34b1797a2aec4339614ded73 parent dee0cc8
8 files changed, +313 −4
  • internal/activator/backend_exec.go +7 −1
  • internal/activator/genbackend.go +9 −0
  • internal/activator/opset.go +25 −0
  • internal/activator/prepare.go +3 −1
  • internal/activator/swap_schedule_test.go +156 −0
  • internal/host/agent.go +5 −0
  • internal/schedule/control.go +27 −2
  • internal/schedule/control_gate_test.go +81 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 688c29db0..de6d5644d 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -180,6 +180,12 @@ type ExecBackend struct {
// nested run dir and log socket beneath it, so a test can exercise the
// nested layout without writing to /run/carlos.
RunDirOverride string
+ // Retired answers whether the control socket at control belongs to a
+ // generation of host that the activator has switched away from
+ // (zero-downtime spec §7.10). Its schedule writes are then refused.
+ // Wired to Service.RetiredControl by the agent. Nil (tests, a box
+ // before this wiring) refuses nothing.
+ Retired func(host, control string) bool
// Strict, when non-nil and true for a host, makes every spawn of that
// host strict (moves design §5.3): the child starts only inside a group
// prepareStrict prepared, which is drained (KillAndEmpty) and closed
@@ -1372,7 +1378,7 @@ func (b *ExecBackend) spawnChildFrom(ctx context.Context, rt registry.Route, soc
var mount func(*http.ServeMux)
if b.Timers != nil {
mount = func(mux *http.ServeMux) {
- schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now)
+ schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.retiredGate(rt.Host, controlSocket))
}
}
ctl, err := edge.ListenControl(controlSocket, rt.Host, b.Cache, mount)
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
index 99a017f45..df1e2657c 100644
--- a/internal/activator/genbackend.go
+++ b/internal/activator/genbackend.go
@@ -296,3 +296,12 @@ func (b *SplitBackend) KillSidecar(ctx context.Context, rt registry.Route) error
}
return nil
}
+
+// retiredGate is MountGated's gate for the control socket at control. It
+// is nil when the backend has no Retired seam.
+func (b *ExecBackend) retiredGate(host, control string) func() bool {
+ if b.Retired == nil {
+ return nil
+ }
+ return func() bool { return b.Retired(host, control) }
+}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 75519b7b1..096d69f94 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -493,3 +493,28 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) {
u.busy = false
a.mu.Unlock()
}
+
+// RetiredControl reports whether control is the control socket of a
+// generation of host that has been switched away from and is still
+// draining (zero-downtime spec §7.10). Its schedule writes are refused
+// from the switch on. The current generation, a successor and a
+// hibernating generation are never retired. Once a retiring generation
+// has left the set its listener is closed, so the answer stops mattering.
+// With swaps off nothing is ever retired.
+func (a *Service) RetiredControl(host, control string) bool {
+ if !a.gens {
+ return false
+ }
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil {
+ return false
+ }
+ for _, m := range e.ops.members {
+ if m.role == roleRetiring && m.gen.Control == control {
+ return true
+ }
+ }
+ return false
+}
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go
index 929aea921..4c4acff25 100644
--- a/internal/activator/prepare.go
+++ b/internal/activator/prepare.go
@@ -138,7 +138,9 @@ func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string)
}
var mount func(*http.ServeMux)
if b.Timers != nil {
- mount = func(mux *http.ServeMux) { schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now) }
+ mount = func(mux *http.ServeMux) {
+ schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.retiredGate(rt.Host, control))
+ }
}
ctl, err := edge.ListenControl(control, rt.Host, b.Cache, mount)
if err != nil {
diff --git a/internal/activator/swap_schedule_test.go b/internal/activator/swap_schedule_test.go
new file mode 100644
index 000000000..ffe49d860
--- /dev/null
+++ b/internal/activator/swap_schedule_test.go
@@ -0,0 +1,156 @@
+package activator
+
+import (
+ "context"
+ "net/http"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// RetiredControl names the retiring generation's control socket and nothing
+// else (spec §7.10). The successor's writes apply while it is checked. A
+// hibernating stop after the swap leaves the current generation's writes
+// open.
+func TestRetiredControlNamesOnlyTheRetiringGeneration(t *testing.T) {
+ f := newSwapFixture(t)
+ duringHealth := map[string]bool{}
+ f.a.genHealth = func(_ context.Context, g Gen) error {
+ duringHealth["successor"] = f.a.RetiredControl(f.host, g.Control)
+ duringHealth["incumbent"] = f.a.RetiredControl(f.host, f.incumbent.Control)
+ return nil
+ }
+ gate := make(chan struct{})
+ f.b.lastProc().setStopGate(gate) // hold the retiring generation's drain open
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
+ waitFor(t, "the switch", func() bool { return indexPrefix(f.b.events(), "switch:") >= 0 })
+ if duringHealth["successor"] || duringHealth["incumbent"] {
+ t.Fatalf("before the switch: %v, want both open", duringHealth)
+ }
+ succ, _ := f.b.lastGen()
+ if !f.a.RetiredControl(f.host, f.incumbent.Control) {
+ t.Fatal("the retiring generation's control socket is still open for writes after the switch")
+ }
+ if f.a.RetiredControl(f.host, succ.Control) {
+ t.Fatal("the current generation's writes are refused")
+ }
+ close(gate)
+ if res := <-results; res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if f.a.RetiredControl(f.host, f.incumbent.Control) {
+ t.Fatal("a generation that has left the set is still reported retired")
+ }
+ // A hibernating stop is unchanged: the stopping generation is current.
+ stopGate := make(chan struct{})
+ _, cur := f.b.lastGen()
+ cur.setStopGate(stopGate)
+ f.clk.advance(2 * time.Hour)
+ f.a.sweepOnce()
+ waitFor(t, "the hibernating stop", func() bool {
+ f.a.mu.Lock()
+ defer f.a.mu.Unlock()
+ e := f.a.entries[f.host]
+ return e != nil && e.state == actStopping
+ })
+ if f.a.RetiredControl(f.host, succ.Control) {
+ t.Fatal("a hibernating generation's writes are refused")
+ }
+ close(stopGate)
+}
+
+// Flag off: nothing is ever retired. The entry carries a retiring member on
+// the very control socket that would be gated with swaps on, so a pass here
+// is the flag's doing and not an empty set's.
+func TestRetiredControlIsFalseWithSwapOff(t *testing.T) {
+ f := newSwapFixture(t)
+ retiring := &genMember{gen: Gen{N: 77, Build: "v1", Control: "/run/carlos/g/old.sock.control"}, role: roleRetiring, done: make(chan struct{})}
+ f.a.mu.Lock()
+ f.a.addMemberLocked(f.a.entries[f.host], retiring)
+ f.a.mu.Unlock()
+ t.Cleanup(func() {
+ f.a.mu.Lock()
+ delete(f.a.entries[f.host].ops.members, retiring.gen.N)
+ f.a.mu.Unlock()
+ })
+ if !f.a.RetiredControl(f.host, retiring.gen.Control) {
+ t.Fatal("control: with swaps on, a retiring member's socket is not reported retired")
+ }
+ f.a.gens = false
+ if f.a.RetiredControl(f.host, retiring.gen.Control) {
+ t.Fatal("RetiredControl reported true with swaps off")
+ }
+}
+
+func TestExecBackendRetiredGate(t *testing.T) {
+ b := testExecBackend(t)
+ if b.retiredGate("h.example.com", "/c") != nil {
+ t.Fatal("a backend with no Retired seam gated the schedule verbs")
+ }
+ var gotHost, gotControl string
+ b.Retired = func(h, c string) bool { gotHost, gotControl = h, c; return true }
+ if !b.retiredGate("h.example.com", "/run/carlos/g/abc.7.sock.control")() || gotHost != "h.example.com" || gotControl != "/run/carlos/g/abc.7.sock.control" {
+ t.Fatalf("gate asked %q %q", gotHost, gotControl)
+ }
+}
+
+// The seam reaches the real control socket of a spawned child: the verbs
+// are mounted through the gate, the gate is asked about this host and this
+// control path, and once it says retired a write is a 409 while a read
+// still answers.
+func TestExecBackendStartGatesTheControlSocketWrites(t *testing.T) {
+ b := testExecBackend(t)
+ b.Timers = &fakeTimerStore{}
+ var retired atomic.Bool
+ var askedHost, askedControl atomic.Value
+ b.Retired = func(host, control string) bool {
+ askedHost.Store(host)
+ askedControl.Store(control)
+ return retired.Load()
+ }
+ ctl, token, stop := startAndReachControlSocket(t, b, "gated.example.com")
+ defer stop()
+
+ send := func(method, path, body string) (int, string) {
+ req, err := http.NewRequest(method, "http://carlos"+path, strings.NewReader(body))
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+token)
+ resp, err := socketClient(ctl).Do(req)
+ if err != nil {
+ t.Fatalf("dial control socket %s: %v", ctl, err)
+ }
+ defer resp.Body.Close()
+ var sb strings.Builder
+ buf := make([]byte, 512)
+ for {
+ n, err := resp.Body.Read(buf)
+ sb.Write(buf[:n])
+ if err != nil {
+ break
+ }
+ }
+ return resp.StatusCode, sb.String()
+ }
+ if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent {
+ t.Fatalf("open DELETE = %d, want 204", code)
+ }
+ retired.Store(true)
+ if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) {
+ t.Fatalf("retired DELETE = %d %q, want 409", code, msg)
+ }
+ if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) {
+ t.Fatalf("retired PUT = %d %q, want 409", code, msg)
+ }
+ if code, _ := send("GET", "/schedule", ""); code != http.StatusOK {
+ t.Fatalf("retired GET = %d, want 200", code)
+ }
+ if askedHost.Load() != "gated.example.com" || askedControl.Load() != ctl {
+ t.Fatalf("gate asked %v %v, want gated.example.com %s", askedHost.Load(), askedControl.Load(), ctl)
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 738c89d71..b390b1234 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -399,6 +399,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// where the pass reads release.GenerationsPath from. Set before
// Start and before any wake.
svc.GenerationsFile = release.GenerationsPath(dd)
+ // A retiring generation's schedule writes are refused once a swap
+ // has switched away from it (zero-downtime spec §7.10). Wired before
+ // Start, like Routes. With swaps off nothing is ever retired, so
+ // every write applies as before.
+ eb.Retired = svc.RetiredControl
if fleetStore != nil {
// MUST be assigned before Start() below spawns the sweeper
// goroutine, which reads svc.OnWake on every successful wake from
diff --git a/internal/schedule/control.go b/internal/schedule/control.go
index acd30bd2f..5738e952f 100644
--- a/internal/schedule/control.go
+++ b/internal/schedule/control.go
@@ -25,7 +25,17 @@ type TimerStore interface {
CountTimers(host, kind string) (int, error)
}
-// Mount adds the once-timer verbs to an instance's control socket (spec
+// RetiredWriteMessage is the 409 body a retiring generation's schedule
+// writes get once the switch has happened (zero-downtime spec §7.10). App
+// developers read it in their logs; it is copy-reviewed.
+const RetiredWriteMessage = "this process has been replaced; the new process owns the schedule"
+
+// Mount is MountGated with no gate.
+func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time) {
+ MountGated(mux, host, store, token, now, nil)
+}
+
+// MountGated adds the once-timer verbs to an instance's control socket (spec
// §7.2). host is fixed here and never read from a request — the
// confused-deputy guarantee edge.ListenControl gives — and every verb needs
// the host's own CARLOS_ADMIN_TOKEN as a bearer. Be precise about what that
@@ -38,7 +48,14 @@ type TimerStore interface {
// with host B's token could already dial B's own socket with it; these
// verbs add reach (a platform wake plus a POST at a chosen time and path),
// not privilege — and per-instance uids are the real fix.
-func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time) {
+//
+// retired gates the two writing verbs: while it reports true, PUT and
+// DELETE answer 409 and change nothing. A generation the activator has
+// switched away from must not rewrite timers the successor now owns, which
+// is what its reassert-at-shutdown would do (zero-downtime spec §7.10).
+// Reads stay open, and the bearer is checked first. A nil retired is Mount
+// exactly.
+func MountGated(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time, retired func() bool) {
authed := func(fn http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
want, err := token()
@@ -64,6 +81,10 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}))
mux.HandleFunc("PUT /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ if retired != nil && retired() {
+ http.Error(w, RetiredWriteMessage, http.StatusConflict)
+ return
+ }
name := r.PathValue("name")
if !ValidName(name) {
http.Error(w, "name: want [a-z0-9][a-z0-9-]{0,31}", http.StatusBadRequest)
@@ -137,6 +158,10 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}))
mux.HandleFunc("DELETE /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
+ if retired != nil && retired() {
+ http.Error(w, RetiredWriteMessage, http.StatusConflict)
+ return
+ }
name := r.PathValue("name")
if !ValidName(name) {
http.Error(w, "bad name", http.StatusBadRequest)
diff --git a/internal/schedule/control_gate_test.go b/internal/schedule/control_gate_test.go
new file mode 100644
index 000000000..9a180a3b7
--- /dev/null
+++ b/internal/schedule/control_gate_test.go
@@ -0,0 +1,81 @@
+package schedule
+
+import (
+ "net/http"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// A retiring generation's writes are refused with 409 once the switch has
+// happened (zero-downtime spec §7.10). The successor's writes apply. Reads
+// stay open, and the bearer is still checked first.
+func TestMountGatedRefusesARetiredGenerationsWrites(t *testing.T) {
+ reg, _ := controlFixture(t)
+ var retired atomic.Bool
+ mux := http.NewServeMux()
+ MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
+ func() time.Time { return at("2026-08-23T13:07:00Z") }, retired.Load)
+ body := `{"at":"2026-09-01T08:00:00Z","path":"/jobs/remind"}`
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body); rec.Code != http.StatusOK {
+ t.Fatalf("current generation's write = %d %s, want 200", rec.Code, rec.Body)
+ }
+ retired.Store(true)
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body); rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), RetiredWriteMessage) {
+ t.Fatalf("retired PUT = %d %q, want 409 with RetiredWriteMessage", rec.Code, rec.Body)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusConflict {
+ t.Fatalf("retired DELETE = %d, want 409", rec.Code)
+ }
+ if row, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err != nil || row == nil {
+ t.Fatalf("a refused DELETE removed the timer: %v", err)
+ }
+ if rec := do(mux, "GET", "/schedule", "Bearer tok", ""); rec.Code != http.StatusOK {
+ t.Fatalf("retired GET = %d, want 200", rec.Code)
+ }
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer wrong", body); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("a retired generation's bad bearer = %d, want 401", rec.Code)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer wrong", ""); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("a retired generation's bad bearer on DELETE = %d, want 401", rec.Code)
+ }
+}
+
+// A refused PUT changes nothing: the row the successor wrote survives a
+// retiring generation's attempt to overwrite it.
+func TestMountGatedRefusedPutLeavesTheRowAlone(t *testing.T) {
+ reg, _ := controlFixture(t)
+ var retired atomic.Bool
+ mux := http.NewServeMux()
+ MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
+ func() time.Time { return at("2026-08-23T13:07:00Z") }, retired.Load)
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/new"}`); rec.Code != http.StatusOK {
+ t.Fatalf("PUT = %d %s", rec.Code, rec.Body)
+ }
+ retired.Store(true)
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/jobs/old"}`); rec.Code != http.StatusConflict {
+ t.Fatalf("retired PUT = %d, want 409", rec.Code)
+ }
+ row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil || row.Path != "/jobs/new" {
+ t.Fatalf("a refused PUT changed the row: %+v, %v", row, err)
+ }
+}
+
+// A nil gate is Mount exactly: both writing verbs apply.
+func TestMountGatedNilIsMount(t *testing.T) {
+ reg, _ := controlFixture(t)
+ mux := http.NewServeMux()
+ MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
+ func() time.Time { return at("2026-08-23T13:07:00Z") }, nil)
+ if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); rec.Code != http.StatusOK {
+ t.Fatalf("PUT = %d", rec.Code)
+ }
+ if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("DELETE = %d, want 204", rec.Code)
+ }
+ if _, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err == nil {
+ t.Fatal("a nil gate's DELETE left the timer in place")
+ }
+}