amadan
zdd plan 3 final fix: retiring schedule writes go through a write fence the switch takes (F8, T8 nested gate)
schedule.MountGated takes a WriteFence instead of a retired bool: a PUT or DELETE validates its request (body parsed and checked) first, then runs every store check and its mutation inside the fence, which answers false (409, RetiredWriteMessage) once the generation is retired. The activator's fence (schedfence.go) is per host: writes run in it concurrently, and an overlap's switch takes it exclusively before a.mu, waiting for the writes in flight, keeping new ones out, and latching the retiring generation in the same step. A generation once retired stays refused, so a handler that began before the switch, or outlives the member's place in the set, never writes after it. Flag off, ScheduleWrite runs the write at once, as before. ExecBackend.Retired becomes ScheduleFence(host, gen, write), bound per listener with the generation it serves: the flat spawn's (n) and the nested prepareAt's (gen). The agent wires it to Service.ScheduleWrite. RetiredControl (by control path, cleared when the member left) is replaced by RetiredGen (latched). Tests: the stalled-body PUT interleaving (schedule), every store access of PUT and DELETE inside the fence (schedule), the switch waiting on a write in flight and the latch after drain (activator), flag off, the flat listener asked for its generation and the nested prepareAt listener fenced end to end.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
11 files changed,
+594
−154
internal/activator/activator.go+4 −0internal/activator/backend_exec.go+9 −7internal/activator/genbackend.go+6 −5internal/activator/opset.go+0 −25internal/activator/prepare.go+7 −3internal/activator/schedfence.go+121 −0internal/activator/swap.go+8 −0internal/activator/swap_schedule_test.go+198 −47internal/host/agent.go+5 −4internal/schedule/control.go+97 −57internal/schedule/control_gate_test.go+139 −6
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 9b1d362a2..dfae1aab2 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -927,6 +927,10 @@ type Service struct { |
| // before any wake, like OnWake. |
| GenerationsFile string |
| + // schedMu guards schedFences, each host's schedule write fence |
| + // (schedfence.go). It is never held with a.mu or a fence's own lock. |
| + schedMu sync.Mutex |
| + schedFences map[string]*schedFence |
| // genMu serialises the generations file's writers (persistGenerations). |
| // It is taken BEFORE a.mu, never while holding it. genWritten (a write |
| // has succeeded) and genDirty (the last write failed; the sweep retries) |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 601262f98..0f25b2d66 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -180,12 +180,14 @@ 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 |
| + // ScheduleFence runs a schedule write made on the control socket of |
| + // host's generation gen (the number the listener was bound for, 0 for |
| + // a spawn that is not a generation's), unless the activator has |
| + // switched away from that generation, and reports whether it ran |
| + // (zero-downtime spec §7.10, final review F8). Wired to |
| + // Service.ScheduleWrite by the agent. Nil (tests, a box before this |
| + // wiring) refuses nothing. |
| + ScheduleFence func(host string, gen uint64, write func()) 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 |
| @@ -1382,7 +1384,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.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.retiredGate(rt.Host, controlSocket)) |
| + schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.writeFence(rt.Host, n)) |
| } |
| } |
| ctl, err := edge.ListenControl(controlSocket, rt.Host, b.Cache, mount) |
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go| index 4c2c6c237..a883bc340 100644 |
| --- a/internal/activator/genbackend.go |
| +++ b/internal/activator/genbackend.go |
| @@ -9,6 +9,7 @@ import ( |
| "github.com/carlosframework/platform/internal/cgroup" |
| "github.com/carlosframework/platform/internal/instancectl" |
| "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/schedule" |
| ) |
| var ( |
| @@ -316,13 +317,13 @@ 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 { |
| +// writeFence is MountGated's fence for the control socket of host's |
| +// generation gen. It is nil when the backend has no ScheduleFence seam. |
| +func (b *ExecBackend) writeFence(host string, gen uint64) schedule.WriteFence { |
| + if b.ScheduleFence == nil { |
| return nil |
| } |
| - return func() bool { return b.Retired(host, control) } |
| + return func(write func()) bool { return b.ScheduleFence(host, gen, write) } |
| } |
| // genPather is a backend that can name a generation's sockets. |
diff --git a/internal/activator/opset.go b/internal/activator/opset.go| index 6dfe5d00b..268c53f4e 100644 |
| --- a/internal/activator/opset.go |
| +++ b/internal/activator/opset.go |
| @@ -600,28 +600,3 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| _ = a.persistGenerations() |
| } |
| } |
| - |
| -// 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 4c4acff25..16e3d45a8 100644 |
| --- a/internal/activator/prepare.go |
| +++ b/internal/activator/prepare.go |
| @@ -100,7 +100,7 @@ func (b *ExecBackend) prepareAt(rt registry.Route, role instancectl.Role, sock s |
| // handling, which at least turns the bind failure into a named |
| // error (memberSafeWakeReason). |
| _ = os.Remove(sock) |
| - ctl = b.listenControl(rt, sock+".control", p.envFile) |
| + ctl = b.listenControl(rt, sock+".control", p.envFile, gen) |
| } |
| logs, err := listenLogs(b.logSocketFor(rt.Host, role, gen), b.logEmitter(rt)) |
| if err != nil { |
| @@ -128,7 +128,11 @@ func exitPrefix(role instancectl.Role, gen uint64) string { |
| // reading beneath the instance directory rather than following a plain |
| // path, design §3.2); a bind failure is logged and the instance runs |
| // without its control verbs, as today. |
| -func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string) *edge.ControlListener { |
| +// |
| +// gen is the generation the listener serves (0 for the original |
| +// templates): its schedule writes go through that generation's write fence |
| +// (writeFence, final review F8). |
| +func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string, gen uint64) *edge.ControlListener { |
| tokenReader := func() (string, error) { |
| env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(envPath)) |
| if err != nil { |
| @@ -139,7 +143,7 @@ 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.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.retiredGate(rt.Host, control)) |
| + schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.writeFence(rt.Host, gen)) |
| } |
| } |
| ctl, err := edge.ListenControl(control, rt.Host, b.Cache, mount) |
diff --git a/internal/activator/schedfence.go b/internal/activator/schedfence.go| new file mode 100644 |
| index 000000000..0ef9498b0 |
| --- /dev/null |
| +++ b/internal/activator/schedfence.go |
| @@ -0,0 +1,121 @@ |
| +package activator |
| + |
| +import "sync" |
| + |
| +// schedFence is one host's schedule write fence (zero-downtime spec §7.10, |
| +// final review F8). Every schedule write a control listener of the host |
| +// takes (a PUT's or DELETE's store checks and its mutation) runs inside it, |
| +// and an overlap's switch holds it exclusively: it waits for the writes in |
| +// flight, keeps new ones out, and latches the retiring generation in the |
| +// same step as the switch. A generation once retired stays refused, so a |
| +// handler that began before the switch, or that outlives the retiring |
| +// member's place in the operation set, can never write after it. |
| +// |
| +// Lock order: the switch takes the fence, then a.mu; a write holds no |
| +// lock while it runs, and never takes a.mu inside the fence, so the two |
| +// never invert. mu guards the fields; cond is on mu. |
| +type schedFence struct { |
| + mu sync.Mutex |
| + cond *sync.Cond |
| + writes int // schedule writes in flight |
| + switching bool // a switch holds the fence (or is waiting for its writes) |
| + retired map[uint64]bool // generations switched away from: latched |
| +} |
| + |
| +// schedFenceFor is host's fence, made on first use. |
| +func (a *Service) schedFenceFor(host string) *schedFence { |
| + a.schedMu.Lock() |
| + defer a.schedMu.Unlock() |
| + if a.schedFences == nil { |
| + a.schedFences = map[string]*schedFence{} |
| + } |
| + f := a.schedFences[host] |
| + if f == nil { |
| + f = &schedFence{retired: map[uint64]bool{}} |
| + f.cond = sync.NewCond(&f.mu) |
| + a.schedFences[host] = f |
| + } |
| + return f |
| +} |
| + |
| +// write runs w for generation gen unless gen is retired, and reports |
| +// whether it ran. It waits while a switch holds the fence. |
| +func (f *schedFence) write(gen uint64, w func()) bool { |
| + f.mu.Lock() |
| + for f.switching { |
| + f.cond.Wait() |
| + } |
| + if f.retired[gen] { |
| + f.mu.Unlock() |
| + return false |
| + } |
| + f.writes++ |
| + f.mu.Unlock() |
| + defer func() { |
| + f.mu.Lock() |
| + f.writes-- |
| + f.cond.Broadcast() |
| + f.mu.Unlock() |
| + }() |
| + w() |
| + return true |
| +} |
| + |
| +// lock takes the fence for a switch: one switch at a time, and none while |
| +// a write runs. Never call it with a.mu held. |
| +func (f *schedFence) lock() { |
| + f.mu.Lock() |
| + for f.switching { |
| + f.cond.Wait() |
| + } |
| + f.switching = true |
| + for f.writes > 0 { |
| + f.cond.Wait() |
| + } |
| + f.mu.Unlock() |
| +} |
| + |
| +// retire latches gen as switched away from. The caller holds the fence |
| +// (lock). |
| +func (f *schedFence) retire(gen uint64) { |
| + f.mu.Lock() |
| + f.retired[gen] = true |
| + f.mu.Unlock() |
| +} |
| + |
| +func (f *schedFence) unlock() { |
| + f.mu.Lock() |
| + f.switching = false |
| + f.cond.Broadcast() |
| + f.mu.Unlock() |
| +} |
| + |
| +// ScheduleWrite is the control listeners' write fence (schedule.WriteFence, |
| +// bound per listener through ExecBackend.ScheduleFence): it runs write for |
| +// a schedule write made on the control socket of host's generation gen, |
| +// unless an overlap's switch has retired that generation, and reports |
| +// whether it ran. The switch waits for a write in flight, so the retirement |
| +// decision and the mutation are one step against it. |
| +// |
| +// With swaps off nothing is ever retired and nothing is serialized: write |
| +// runs at once, as before. |
| +func (a *Service) ScheduleWrite(host string, gen uint64, write func()) bool { |
| + if !a.gens { |
| + write() |
| + return true |
| + } |
| + return a.schedFenceFor(host).write(gen, write) |
| +} |
| + |
| +// RetiredGen reports whether host's generation gen has been switched |
| +// away from (its schedule writes are refused for good). False with swaps |
| +// off. |
| +func (a *Service) RetiredGen(host string, gen uint64) bool { |
| + if !a.gens { |
| + return false |
| + } |
| + f := a.schedFenceFor(host) |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return f.retired[gen] |
| +} |
diff --git a/internal/activator/swap.go b/internal/activator/swap.go| index 66c1de21a..5c01e0ae5 100644 |
| --- a/internal/activator/swap.go |
| +++ b/internal/activator/swap.go |
| @@ -495,7 +495,14 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| retiring *genMember |
| switched bool |
| ) |
| + // The host's schedule write fence (final review F8): the switch waits |
| + // for the schedule writes in flight, keeps new ones out, and latches the |
| + // retiring generation in the same step, so no write of it lands after |
| + // the switch. Taken before a.mu (the fence's lock order). |
| + fence := a.schedFenceFor(at.Host) |
| err = a.admit(ctx, work, func() error { |
| + fence.lock() |
| + defer fence.unlock() |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| if err := a.memberCurrentLocked(e, m, mctx, at, newest); err != nil { |
| @@ -512,6 +519,7 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| return err |
| } |
| retiring = &genMember{gen: e.gen, role: roleRetiring, proc: e.proc, done: make(chan struct{})} |
| + fence.retire(retiring.gen.N) // its schedule writes are refused from here, for good |
| delete(e.ops.members, m.gen.N) |
| a.addMemberLocked(e, retiring) // swapsim:M2 |
| e.gen, e.proc = m.gen, m.proc |
diff --git a/internal/activator/swap_schedule_test.go b/internal/activator/swap_schedule_test.go| index ffe49d860..4ade5e8d4 100644 |
| --- a/internal/activator/swap_schedule_test.go |
| +++ b/internal/activator/swap_schedule_test.go |
| @@ -2,25 +2,31 @@ package activator |
| import ( |
| "context" |
| + "io" |
| "net/http" |
| + "os" |
| "strings" |
| + "sync" |
| "sync/atomic" |
| "testing" |
| "time" |
| + "github.com/carlosframework/platform/internal/instancectl" |
| "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) { |
| +// The switch retires exactly the generation it switched away from (spec |
| +// §7.10), and the retirement is latched (final review F8): the retiring |
| +// generation's writes stay refused after it has drained and left the |
| +// operation set, since a handler that began before the switch may still be |
| +// running. The successor's writes apply throughout, and a hibernating stop |
| +// after the swap retires nothing. |
| +func TestRetiredGenNamesOnlyTheRetiringGeneration(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) |
| + duringHealth["successor"] = f.a.RetiredGen(f.host, g.N) |
| + duringHealth["incumbent"] = f.a.RetiredGen(f.host, f.incumbent.N) |
| return nil |
| } |
| gate := make(chan struct{}) |
| @@ -32,18 +38,28 @@ func TestRetiredControlNamesOnlyTheRetiringGeneration(t *testing.T) { |
| 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.RetiredGen(f.host, f.incumbent.N) { |
| + t.Fatal("the retiring generation's schedule writes are still open after the switch") |
| } |
| - if f.a.RetiredControl(f.host, succ.Control) { |
| + if f.a.RetiredGen(f.host, succ.N) { |
| t.Fatal("the current generation's writes are refused") |
| } |
| + if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) { |
| + t.Fatal("ScheduleWrite ran the retired generation's write") |
| + } |
| 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") |
| + if _, members := f.a.genStateForTest(f.host); len(members) != 0 { |
| + t.Fatalf("members %v: the retiring generation should have left the set", members) |
| + } |
| + if !f.a.RetiredGen(f.host, f.incumbent.N) { |
| + t.Fatal("the retirement was not latched: the drained generation's writes reopened") |
| + } |
| + ran := false |
| + if !f.a.ScheduleWrite(f.host, succ.N, func() { ran = true }) || !ran { |
| + t.Fatal("the successor's write was refused") |
| } |
| // A hibernating stop is unchanged: the stopping generation is current. |
| stopGate := make(chan struct{}) |
| @@ -57,60 +73,112 @@ func TestRetiredControlNamesOnlyTheRetiringGeneration(t *testing.T) { |
| e := f.a.entries[f.host] |
| return e != nil && e.state == actStopping |
| }) |
| - if f.a.RetiredControl(f.host, succ.Control) { |
| + if f.a.RetiredGen(f.host, succ.N) { |
| 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) { |
| +// schedSwitchWaiting reports whether a switch holds host's write fence and |
| +// is waiting for a write in flight. |
| +func (a *Service) schedSwitchWaiting(host string) bool { |
| + f := a.schedFenceFor(host) |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return f.switching && f.writes > 0 |
| +} |
| + |
| +// Final review F8: the switch is serialized against the schedule writes. |
| +// A write of the incumbent that is under way when the switch comes holds |
| +// the switch until it finishes (so it lands before the switch, while its |
| +// generation is current), and once the switch has happened every write of |
| +// the retired generation is refused. |
| +func TestScheduleWriteIsSerializedWithTheSwitch(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") |
| + entered, release := make(chan struct{}), make(chan struct{}) |
| + wrote := make(chan bool, 1) |
| + go func() { |
| + wrote <- f.a.ScheduleWrite(f.host, f.incumbent.N, func() { |
| + close(entered) |
| + <-release |
| + f.b.record("schedule-write") |
| + }) |
| + }() |
| + <-entered |
| + results := make(chan SwapResult, 1) |
| + go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| + waitFor(t, "the switch waiting on the write", func() bool { return f.a.schedSwitchWaiting(f.host) }) |
| + if ev := f.b.events(); indexPrefix(ev, "switch:") >= 0 { |
| + t.Fatalf("events %v: the switch ran while a schedule write was under way", ev) |
| + } |
| + close(release) |
| + if !<-wrote { |
| + t.Fatal("the incumbent's write, begun before the switch, was refused") |
| + } |
| + if res := <-results; res.Outcome != SwapDone { |
| + t.Fatalf("Swap = %+v", res) |
| + } |
| + ev := f.b.events() |
| + if w, sw := indexPrefix(ev, "schedule-write"), indexPrefix(ev, "switch:"); w < 0 || sw < 0 || w > sw { |
| + t.Fatalf("events %v: want the write before the switch", ev) |
| + } |
| + if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) { |
| + t.Fatal("the retired generation's write after the switch was not refused") |
| + } |
| +} |
| + |
| +// Flag off: nothing is retired and nothing is serialized. A generation the |
| +// fence has latched (here by hand, as a switch would with swaps on) still |
| +// writes once the switch is off. |
| +func TestScheduleWriteIsUnfencedWithSwapOff(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.schedFenceFor(f.host).retire(77) |
| + if f.a.ScheduleWrite(f.host, 77, func() {}) || !f.a.RetiredGen(f.host, 77) { |
| + t.Fatal("with swaps on, a latched generation's write was not refused") |
| } |
| f.a.gens = false |
| - if f.a.RetiredControl(f.host, retiring.gen.Control) { |
| - t.Fatal("RetiredControl reported true with swaps off") |
| + ran := false |
| + if !f.a.ScheduleWrite(f.host, 77, func() { ran = true }) || !ran || f.a.RetiredGen(f.host, 77) { |
| + t.Fatal("with swaps off, a write was refused or a generation reported retired") |
| } |
| } |
| -func TestExecBackendRetiredGate(t *testing.T) { |
| +func TestExecBackendWriteFence(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") |
| + if b.writeFence("h.example.com", 7) != nil { |
| + t.Fatal("a backend with no ScheduleFence seam fenced 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) |
| + var gotHost string |
| + var gotGen uint64 |
| + b.ScheduleFence = func(h string, g uint64, write func()) bool { gotHost, gotGen = h, g; write(); return true } |
| + ran := false |
| + if !b.writeFence("h.example.com", 7)(func() { ran = true }) || !ran || gotHost != "h.example.com" || gotGen != 7 { |
| + t.Fatalf("fence asked %q %d, ran %v", gotHost, gotGen, ran) |
| } |
| } |
| -// 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. |
| +// The seam reaches the real control socket of a spawned child (the flat |
| +// listener): the verbs are mounted through the fence, the fence is asked |
| +// about this host and the generation the listener was bound for (0 for a |
| +// plain Start), and once it refuses 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 { |
| + var askedHost atomic.Value |
| + askedGen := uint64(99) |
| + var mu sync.Mutex |
| + b.ScheduleFence = func(host string, gen uint64, write func()) bool { |
| askedHost.Store(host) |
| - askedControl.Store(control) |
| - return retired.Load() |
| + mu.Lock() |
| + askedGen = gen |
| + mu.Unlock() |
| + if retired.Load() { |
| + return false |
| + } |
| + write() |
| + return true |
| } |
| ctl, token, stop := startAndReachControlSocket(t, b, "gated.example.com") |
| defer stop() |
| @@ -150,7 +218,90 @@ func TestExecBackendStartGatesTheControlSocketWrites(t *testing.T) { |
| 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) |
| + mu.Lock() |
| + gen := askedGen |
| + mu.Unlock() |
| + if askedHost.Load() != "gated.example.com" || gen != 0 { |
| + t.Fatalf("fence asked %v generation %d, want gated.example.com generation 0", askedHost.Load(), gen) |
| + } |
| +} |
| + |
| +// Final review F8, T8: the nested layout's control listener (prepareAt, |
| +// which the nested generation start uses) is mounted through the same |
| +// fence, asked for the generation it was prepared for. |
| +func TestPreparedControlListenerIsFenced(t *testing.T) { |
| + b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{}) |
| + b.GenContainment = true |
| + b.Timers = &fakeTimerStore{} |
| + var retired atomic.Bool |
| + var mu sync.Mutex |
| + asked := map[uint64]int{} |
| + b.ScheduleFence = func(host string, gen uint64, write func()) bool { |
| + mu.Lock() |
| + if host == rt.Host { |
| + asked[gen]++ |
| + } |
| + mu.Unlock() |
| + if retired.Load() { |
| + return false |
| + } |
| + write() |
| + return true |
| + } |
| + if l, _ := b.Layout(rt); l != LayoutNested { |
| + t.Fatalf("layout %v, want nested", l) |
| + } |
| + p0 := b.pathsFor(rt) |
| + if err := os.MkdirAll(p0.runDir, 0o750); err != nil { |
| + t.Fatal(err) |
| + } |
| + sock := nestedGenSocketIn(p0.runDir, 7) |
| + p, ctl, logs, _, err := b.prepareAt(rt, instancectl.RoleRun, sock, 7) |
| + if err != nil { |
| + t.Fatalf("prepareAt: %v", err) |
| + } |
| + if logs != nil { |
| + defer logs.Close() |
| + } |
| + if ctl == nil { |
| + t.Fatal("prepareAt bound no control listener") |
| + } |
| + defer ctl.Close() |
| + env, err := readEnvFileBeneath(p.dataDir, rt.Host+".env") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + token := env["CARLOS_ADMIN_TOKEN"] |
| + 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(sock + ".control").Do(req) |
| + if err != nil { |
| + t.Fatalf("dial control socket: %v", err) |
| + } |
| + defer resp.Body.Close() |
| + buf, _ := io.ReadAll(resp.Body) |
| + return resp.StatusCode, string(buf) |
| + } |
| + if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent { |
| + t.Fatalf("open DELETE = %d %q, want 204", code, msg) |
| + } |
| + retired.Store(true) |
| + if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-02T08:00:00Z","path":"/r2"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) { |
| + t.Fatalf("retired PUT = %d %q, want 409", code, msg) |
| + } |
| + if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict { |
| + t.Fatalf("retired DELETE = %d, want 409", code) |
| + } |
| + if code, _ := send("GET", "/schedule", ""); code != http.StatusOK { |
| + t.Fatalf("retired GET = %d, want 200", code) |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if asked[7] != 3 || len(asked) != 1 { |
| + t.Fatalf("fence asked %v, want generation 7 for each of the three writes", asked) |
| } |
| } |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index b390b1234..d014e8323 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -400,10 +400,11 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| // 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 |
| + // has switched away from it, through each host's write fence, which |
| + // the switch takes (zero-downtime spec §7.10). Wired before Start, |
| + // like Routes. With swaps off nothing is ever retired or serialized, |
| + // so every write applies as before. |
| + eb.ScheduleFence = svc.ScheduleWrite |
| 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 5738e952f..7479d6621 100644 |
| --- a/internal/schedule/control.go |
| +++ b/internal/schedule/control.go |
| @@ -30,6 +30,17 @@ type TimerStore interface { |
| // developers read it in their logs; it is copy-reviewed. |
| const RetiredWriteMessage = "this process has been replaced; the new process owns the schedule" |
| +// WriteFence serializes one generation's schedule writes against the |
| +// switch that retires it (zero-downtime spec §7.10, final review F8). It |
| +// runs write, the final retirement decision and the mutation together, |
| +// unless the generation has been retired, and reports whether it ran. |
| +// While write runs, the switch that would retire this generation waits; |
| +// once it has retired it, every later call answers false, including one |
| +// for a request that began before the switch. A request's body is parsed |
| +// and validated before the fence is asked, so a slow client never holds |
| +// the switch. |
| +type WriteFence func(write func()) bool |
| + |
| // 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) |
| @@ -49,13 +60,23 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri |
| // 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. |
| // |
| -// 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) { |
| +// fence gates the two writing verbs: every store access a PUT or DELETE |
| +// makes runs inside it, and once it refuses, the verb answers 409 and |
| +// changes 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, the bearer is checked first, and the request is validated before |
| +// the fence is asked. A nil fence is Mount exactly. |
| +func MountGated(mux *http.ServeMux, host string, store TimerStore, token func() (string, error), now func() time.Time, fence WriteFence) { |
| + // fenced runs write inside the fence, and reports false (the caller |
| + // answers 409) when the generation is retired. |
| + fenced := func(write func()) bool { |
| + if fence == nil { |
| + write() |
| + return true |
| + } |
| + return fence(write) |
| + } |
| authed := func(fn http.HandlerFunc) http.HandlerFunc { |
| return func(w http.ResponseWriter, r *http.Request) { |
| want, err := token() |
| @@ -81,10 +102,6 @@ func MountGated(mux *http.ServeMux, host string, store TimerStore, token func() |
| })) |
| 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) |
| @@ -120,66 +137,89 @@ func MountGated(mux *http.ServeMux, host string, store TimerStore, token func() |
| http.Error(w, err.Error(), http.StatusBadRequest) |
| return |
| } |
| - existing, err := store.GetTimer(host, name) |
| - switch { |
| - case err == nil && existing.Kind != KindOnce: |
| - http.Error(w, "name is a declared schedule", http.StatusConflict) |
| - return |
| - case err != nil && !errors.Is(err, os.ErrNotExist): |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| - return |
| - case err != nil: |
| - n, cerr := store.CountTimers(host, KindOnce) |
| - if cerr != nil { |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| + // The store's checks and the write run inside the fence, so the |
| + // retirement decision and the mutation are one step against the |
| + // switch. The response is written after it. |
| + var ( |
| + fail string |
| + code int |
| + row *registry.Timer |
| + ) |
| + ran := fenced(func() { |
| + existing, err := store.GetTimer(host, name) |
| + switch { |
| + case err == nil && existing.Kind != KindOnce: |
| + fail, code = "name is a declared schedule", http.StatusConflict |
| + return |
| + case err != nil && !errors.Is(err, os.ErrNotExist): |
| + fail, code = "registry error", http.StatusInternalServerError |
| return |
| + case err != nil: |
| + n, cerr := store.CountTimers(host, KindOnce) |
| + if cerr != nil { |
| + fail, code = "registry error", http.StatusInternalServerError |
| + return |
| + } |
| + if n >= MaxOncePerHost { |
| + fail, code = "too many one-shot timers (limit 1000) — delete finished ones or use a schedule", http.StatusTooManyRequests |
| + return |
| + } |
| } |
| - if n >= MaxOncePerHost { |
| - http.Error(w, "too many one-shot timers (limit 1000) — delete finished ones or use a schedule", http.StatusTooManyRequests) |
| + sp := Spec{Kind: KindOnce, At: atT.UTC()} |
| + next := atT.Unix() |
| + if next <= 0 { |
| + next = max(int64(1), now().Unix()) |
| + } |
| + t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: next, DueAt: atT.Unix(), CreatedAt: now().Unix()} |
| + if err := store.UpsertTimer(t); err != nil { |
| + fail, code = "registry error", http.StatusInternalServerError |
| return |
| } |
| + if row, err = store.GetTimer(host, name); err != nil { |
| + fail, code = "registry error", http.StatusInternalServerError |
| + } |
| + }) |
| + switch { |
| + case !ran: |
| + http.Error(w, RetiredWriteMessage, http.StatusConflict) |
| + case fail != "": |
| + http.Error(w, fail, code) |
| + default: |
| + writeTimer(w, *row, now()) |
| } |
| - sp := Spec{Kind: KindOnce, At: atT.UTC()} |
| - next := atT.Unix() |
| - if next <= 0 { |
| - next = max(int64(1), now().Unix()) |
| - } |
| - t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: next, DueAt: atT.Unix(), CreatedAt: now().Unix()} |
| - if err := store.UpsertTimer(t); err != nil { |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| - return |
| - } |
| - row, err := store.GetTimer(host, name) |
| - if err != nil { |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| - return |
| - } |
| - writeTimer(w, *row, now()) |
| })) |
| 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) |
| return |
| } |
| - switch row, err := store.GetTimer(host, name); { |
| - case err == nil && row.Kind != KindOnce: |
| - http.Error(w, "name is a declared schedule — remove it with carlos schedule rm", http.StatusConflict) |
| - return |
| - case err != nil && !errors.Is(err, os.ErrNotExist): |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| - return |
| - } |
| - if err := store.DeleteTimer(host, name); err != nil { |
| - http.Error(w, "registry error", http.StatusInternalServerError) |
| - return |
| + var ( |
| + fail string |
| + code int |
| + ) |
| + ran := fenced(func() { |
| + switch row, err := store.GetTimer(host, name); { |
| + case err == nil && row.Kind != KindOnce: |
| + fail, code = "name is a declared schedule — remove it with carlos schedule rm", http.StatusConflict |
| + return |
| + case err != nil && !errors.Is(err, os.ErrNotExist): |
| + fail, code = "registry error", http.StatusInternalServerError |
| + return |
| + } |
| + if err := store.DeleteTimer(host, name); err != nil { |
| + fail, code = "registry error", http.StatusInternalServerError |
| + } |
| + }) |
| + switch { |
| + case !ran: |
| + http.Error(w, RetiredWriteMessage, http.StatusConflict) |
| + case fail != "": |
| + http.Error(w, fail, code) |
| + default: |
| + w.WriteHeader(http.StatusNoContent) |
| } |
| - w.WriteHeader(http.StatusNoContent) |
| })) |
| mux.HandleFunc("GET /schedule", authed(func(w http.ResponseWriter, r *http.Request) { |
diff --git a/internal/schedule/control_gate_test.go b/internal/schedule/control_gate_test.go| index 9a180a3b7..724e07b71 100644 |
| --- a/internal/schedule/control_gate_test.go |
| +++ b/internal/schedule/control_gate_test.go |
| @@ -2,26 +2,57 @@ package schedule |
| import ( |
| "net/http" |
| + "net/http/httptest" |
| "strings" |
| + "sync" |
| "sync/atomic" |
| "testing" |
| "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| +// gateFence is a WriteFence over a retired flag, serialized against |
| +// retire() the way the activator's is against its switch. inside is true |
| +// while a write runs in it. |
| +type gateFence struct { |
| + mu sync.RWMutex |
| + retired bool |
| + inside atomic.Bool |
| +} |
| + |
| +func (f *gateFence) fence(write func()) bool { |
| + f.mu.RLock() |
| + defer f.mu.RUnlock() |
| + if f.retired { |
| + return false |
| + } |
| + f.inside.Store(true) |
| + defer f.inside.Store(false) |
| + write() |
| + return true |
| +} |
| + |
| +func (f *gateFence) retire() { |
| + f.mu.Lock() |
| + f.retired = true |
| + f.mu.Unlock() |
| +} |
| + |
| // 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 |
| + f := &gateFence{} |
| 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) |
| + func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence) |
| 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) |
| + f.retire() |
| 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) |
| } |
| @@ -46,14 +77,14 @@ func TestMountGatedRefusesARetiredGenerationsWrites(t *testing.T) { |
| // retiring generation's attempt to overwrite it. |
| func TestMountGatedRefusedPutLeavesTheRowAlone(t *testing.T) { |
| reg, _ := controlFixture(t) |
| - var retired atomic.Bool |
| + f := &gateFence{} |
| 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) |
| + func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence) |
| 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) |
| + f.retire() |
| 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) |
| } |
| @@ -79,3 +110,105 @@ func TestMountGatedNilIsMount(t *testing.T) { |
| t.Fatal("a nil gate's DELETE left the timer in place") |
| } |
| } |
| + |
| +// stallBody is a request body whose first Read signals entered and then |
| +// blocks until release closes: a client stalled mid-body. |
| +type stallBody struct { |
| + r *strings.Reader |
| + entered chan struct{} |
| + release chan struct{} |
| + first atomic.Bool |
| +} |
| + |
| +func (b *stallBody) Read(p []byte) (int, error) { |
| + if b.first.CompareAndSwap(false, true) { |
| + close(b.entered) |
| + <-b.release |
| + } |
| + return b.r.Read(p) |
| +} |
| +func (b *stallBody) Close() error { return nil } |
| + |
| +// Final review F8: a retiring generation's PUT that began before the |
| +// switch and stalled in its body cannot overwrite the successor's write. |
| +// The retirement decision is made inside the fence, after the body, and |
| +// the switch is serialized against it. |
| +func TestRetiredPutStalledInItsBodyCannotOverwriteTheSuccessor(t *testing.T) { |
| + reg, _ := controlFixture(t) |
| + now := func() time.Time { return at("2026-08-23T13:07:00Z") } |
| + tok := func() (string, error) { return "tok", nil } |
| + f := &gateFence{} |
| + oldMux, newMux := http.NewServeMux(), http.NewServeMux() |
| + MountGated(oldMux, "a.bab.oncarlos.com", reg, tok, now, f.fence) |
| + MountGated(newMux, "a.bab.oncarlos.com", reg, tok, now, (&gateFence{}).fence) |
| + body := &stallBody{r: strings.NewReader(`{"at":"2026-09-02T08:00:00Z","path":"/jobs/old"}`), entered: make(chan struct{}), release: make(chan struct{})} |
| + req := httptest.NewRequest("PUT", "/schedule/remind", nil) |
| + req.Body = body |
| + req.Header.Set("Authorization", "Bearer tok") |
| + rec := httptest.NewRecorder() |
| + done := make(chan struct{}) |
| + go func() { oldMux.ServeHTTP(rec, req); close(done) }() |
| + <-body.entered |
| + f.retire() // the switch: it does not wait on a request still reading its body |
| + if r := do(newMux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/new"}`); r.Code != http.StatusOK { |
| + t.Fatalf("successor PUT = %d %s", r.Code, r.Body) |
| + } |
| + close(body.release) |
| + <-done |
| + if rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), RetiredWriteMessage) { |
| + t.Errorf("the retired generation's stalled PUT = %d %q, want 409", rec.Code, rec.Body) |
| + } |
| + if row, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err != nil || row.Path != "/jobs/new" { |
| + t.Fatalf("the successor's timer was overwritten: %+v, %v", row, err) |
| + } |
| +} |
| + |
| +// fenceCheckStore fails the test when a writing verb touches the store |
| +// outside its generation's fence. |
| +type fenceCheckStore struct { |
| + TimerStore |
| + t *testing.T |
| + f *gateFence |
| +} |
| + |
| +func (s fenceCheckStore) in(op string) { |
| + if !s.f.inside.Load() { |
| + s.t.Errorf("%s ran outside the write fence", op) |
| + } |
| +} |
| +func (s fenceCheckStore) UpsertTimer(t registry.Timer) error { |
| + s.in("UpsertTimer") |
| + return s.TimerStore.UpsertTimer(t) |
| +} |
| +func (s fenceCheckStore) DeleteTimer(h, n string) error { |
| + s.in("DeleteTimer") |
| + return s.TimerStore.DeleteTimer(h, n) |
| +} |
| +func (s fenceCheckStore) GetTimer(h, n string) (*registry.Timer, error) { |
| + s.in("GetTimer") |
| + return s.TimerStore.GetTimer(h, n) |
| +} |
| +func (s fenceCheckStore) CountTimers(h, k string) (int, error) { |
| + s.in("CountTimers") |
| + return s.TimerStore.CountTimers(h, k) |
| +} |
| + |
| +// Final review F8: PUT's and DELETE's store checks and mutations all run |
| +// inside the fence, so the final retirement decision and the write are one |
| +// step against the switch (DELETE's check/use race included). |
| +func TestMountGatedWritesOnlyInsideTheFence(t *testing.T) { |
| + reg, _ := controlFixture(t) |
| + f := &gateFence{} |
| + mux := http.NewServeMux() |
| + MountGated(mux, "a.bab.oncarlos.com", fenceCheckStore{TimerStore: reg, t: t, f: f}, func() (string, error) { return "tok", nil }, |
| + func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence) |
| + 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 %s", rec.Code, rec.Body) |
| + } |
| + if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/r2"}`); rec.Code != http.StatusOK { |
| + t.Fatalf("second PUT = %d %s", rec.Code, rec.Body) |
| + } |
| + if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent { |
| + t.Fatalf("DELETE = %d %s", rec.Code, rec.Body) |
| + } |
| +} |