carlos / platform Public

instance-control fences legacy starts at the accepted-connection watermark with one request per connection; proofs already running end with Shutdown (#292)

Astra's round 3 review (REJECT):

P1. A keep-alive connection let a delayed /start on a later connection
escape the fence: the /stop recorded its own (older) connection number.
instance-control now answers one request per connection
(SetKeepAlivesEnabled(false)), so a connection's accept order is its
request's, and a /stop fences at the accepted-connection watermark read
under legacyMu at the moment it takes the fence: every start accepted at
or before it is refused, however late it registers; any accepted after it
runs normally.

P2. Closed by the same change: every request gets a fresh stamp, so a
keep-alive client's start after a stop runs.

P2. A legacy proof already running when Shutdown began inherited
Background and could run 30 s past the budget. Every proof is now
registered, and Shutdown cancels them all when its context ends; a proof
cut short is unproven and the lease is kept.

Liveness pins for the census owner (platform-1e): the fence covers only
starts accepted before the stop and lives in memory only; the join is per
host; a stuck stop answers within its bound and the next stop does not
hang; sidecar starts and seed stops pass through during a legacy stop.
The join budget is a test-settable Server field.

README: the release carrying #292 restarts instance-control on roll.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev c560c60bc6faeb1d9a433e648ae805589af6ca30 parent b12360d
9 files changed, +324 −21
  • README.md +5 −0
  • internal/activator/activator.go +5 −0
  • internal/activator/activator_test.go +9 −0
  • internal/activator/containment.go +27 −5
  • internal/activator/shutdown_lease_test.go +32 −0
  • internal/instancectl/legacystart.go +22 −13
  • internal/instancectl/legacystart_live_test.go +215 −0
  • internal/instancectl/legacystart_test.go +2 −2
  • internal/instancectl/server.go +7 −1
diff --git a/README.md b/README.md
index 492ee8361..c84d1511e 100644
--- a/README.md
+++ b/README.md
@@ -352,6 +352,11 @@ The first release that understands app regions is v0.23.0.
6. Turn the switch on. Re-run step 5 after any period with the switch off,
and whenever a new region becomes eligible.
+**#292: instance-control changes.** The release carrying #292 changes
+`carlos-instance-control`: it orders a host's legacy run starts against
+its stops. So the service must restart on roll. `carlos system update` does
+that since v0.26.1, so no manual step is needed.
+
### Tenant email senders, plan 2 of 5 (2026-09-21)
This release turns the one implicit per-app email sender into an explicit
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index a3f7b1b31..236a41ab7 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1161,6 +1161,9 @@ type Service struct {
// shutdownCtx is Shutdown's budget, set with shuttingDown: a legacy
// proof runs under the sooner of its own budget and this (#292).
shutdownCtx context.Context
+ // proofCancels are the legacy proofs running now (legacyRunGone), so
+ // Shutdown can end them with its budget. Under a.mu.
+ proofCancels map[*byte]context.CancelFunc
// convergeBudget is how many build-convergence cycles this sweep may
// still start: reset to maxConvergeCyclesPerSweep at the top of every
@@ -4995,6 +4998,8 @@ func (a *Service) Shutdown(ctx context.Context) {
a.shutdownCtx = ctx
sweepStop, sweepDone := a.sweepStop, a.sweepDone
a.mu.Unlock()
+ // …and those already running end with it too (Astra, round 3).
+ context.AfterFunc(ctx, a.cancelProofs)
// …and cancel the service lifetime, which TEARS DOWN any worker spawn
// already past that check rather than letting the stop below wait it out.
// The wait is right for a Bounce or an idle sleep — no deadline, and
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index dbadf7e90..0000ed2e2 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -383,6 +383,9 @@ type fakeBackend struct {
// stop that never proves); legacyEnded counts those that returned.
legacyBlock bool
legacyEnded int
+ // startLegacyFail fails every Start as a legacy run unit's start root
+ // may still carry out, at once, outside any shutdown.
+ startLegacyFail bool
// leaseLapsed: the owed records have passed their expiry, so a retry
// that fails again answers lease.ErrReleaseLapsed, as lease.Held does.
leaseLapsed bool
@@ -574,6 +577,12 @@ func (b *fakeBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
if startErr != nil {
return nil, "", startErr
}
+ b.mu.Lock()
+ legacyFail := b.startLegacyFail
+ b.mu.Unlock()
+ if legacyFail {
+ return nil, "", markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: root did not answer", rt.Host)))
+ }
if gate != nil {
<-gate
}
diff --git a/internal/activator/containment.go b/internal/activator/containment.go
index da3a2a6e7..0426d36ab 100644
--- a/internal/activator/containment.go
+++ b/internal/activator/containment.go
@@ -77,9 +77,11 @@ var errNoLegacyProof = errors.New("activator: this backend cannot prove a run un
// context is often why the start was aborted).
const legacyStopBudget = instancectl.DeadlineControl
-// legacyRunGone runs the backend's proof for rt under legacyStopBudget or,
-// once Shutdown has begun, whichever ends first of that and Shutdown's own
-// budget (Astra, round 2): no proof outlives Shutdown, and one cut short is
+// legacyRunGone runs the backend's proof for rt under legacyStopBudget and
+// never past Shutdown's own budget (Astra, rounds 2 and 3): a proof started
+// once Shutdown has begun runs under its context, and every proof is
+// registered so that one already running when Shutdown began is cancelled
+// the moment that budget ends (cancelProofs). A proof cut short is
// unproven, so the lease is kept.
func (a *Service) legacyRunGone(rt registry.Route) error {
p, ok := a.backend.(legacyRunProver)
@@ -87,16 +89,36 @@ func (a *Service) legacyRunGone(rt registry.Route) error {
return errNoLegacyProof
}
parent := context.Background()
+ token := new(byte)
a.mu.Lock()
if a.shutdownCtx != nil {
parent = a.shutdownCtx
}
- a.mu.Unlock()
ctx, cancel := context.WithTimeout(parent, legacyStopBudget)
- defer cancel()
+ if a.proofCancels == nil {
+ a.proofCancels = make(map[*byte]context.CancelFunc)
+ }
+ a.proofCancels[token] = cancel
+ a.mu.Unlock()
+ defer func() {
+ a.mu.Lock()
+ delete(a.proofCancels, token)
+ a.mu.Unlock()
+ cancel()
+ }()
return p.LegacyRunGone(ctx, rt)
}
+// cancelProofs cancels every legacy proof still running: Shutdown arms it
+// on its own context, so none outlives the shutdown budget.
+func (a *Service) cancelProofs() {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ for _, cancel := range a.proofCancels {
+ cancel()
+ }
+}
+
// errStartedThenExited is the one uncertain start whose outcome is known to
// be a crash at the start: root started the unit and it had exited before
// root read its invocation (Astra R3 ruling 206, zdd-readiness Task 6). The
diff --git a/internal/activator/shutdown_lease_test.go b/internal/activator/shutdown_lease_test.go
index 6bb46e6fe..4b7c24297 100644
--- a/internal/activator/shutdown_lease_test.go
+++ b/internal/activator/shutdown_lease_test.go
@@ -436,3 +436,35 @@ func TestTheLegacyProofEndsWithShutdownsBudget(t *testing.T) {
t.Fatalf("lease releases = %d, want %d: an unproven start keeps its lease", got, acq-1)
}
}
+
+// Astra round 3 P2: a proof already running when Shutdown begins (here a
+// failed start's, from before the shutdown, under its own 30 s budget)
+// ends when Shutdown's budget does. Cut short, it is unproven, and the
+// lease is kept.
+func TestAProofRunningBeforeShutdownEndsWithItsBudget(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ b.mu.Lock()
+ b.startLegacyFail, b.legacyBlock = true, true
+ b.mu.Unlock()
+ go func() {
+ if release, err := a.beginRequest(context.Background(), testRoute("early.example.com")); err == nil {
+ release()
+ }
+ }()
+ waitFor(t, "the failed start's proof to begin", func() bool {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.legacyStops == 1
+ })
+ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+ defer cancel()
+ a.Shutdown(ctx)
+ waitFor(t, "the proof to end with Shutdown's budget", func() bool {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.legacyEnded > 0
+ })
+ if got := b.releaseCount(); got != 0 {
+ t.Fatalf("lease releases = %d, want 0: a proof cut short keeps the lease", got)
+ }
+}
diff --git a/internal/instancectl/legacystart.go b/internal/instancectl/legacystart.go
index abc64f542..35355f2b6 100644
--- a/internal/instancectl/legacystart.go
+++ b/internal/instancectl/legacystart.go
@@ -23,10 +23,11 @@ import (
// - a /stop fences the host: new starts are refused while it runs, every
// attempt in flight is cancelled and joined (bounded), and only then is
// the unit stopped, which replaces any job a joined attempt submitted;
-// - a start whose request was accepted before a stop of the host (the
-// connection's accept order, connSeq) is an earlier request, and is
-// refused once that stop has fenced the host, however late its handler
-// runs.
+// - a start whose request was accepted at or before the moment a stop
+// fences the host (the accepted-connection watermark, connSeq) is an
+// earlier request, and is refused, however late its handler registers;
+// one accepted after it runs normally. Serve answers one request per
+// connection, so a connection's accept order is its request's.
//
// Once /stop has answered, no start from an earlier request is in flight or
// can be submitted afterwards. Generation units, seeds, sidecars and the
@@ -98,14 +99,18 @@ func (s *Server) beginLegacyStart(parent context.Context, host string, seq uint6
return ctx, finish, nil
}
-// fenceLegacyStarts is a /stop's first step for host, its request of accept
-// order seq: refuse new starts and every earlier request's, cancel the
-// attempts in flight and join them, bounded by legacyJoinBudget and ctx.
-// release lifts the refusal of new starts once the stop is done; the
-// earlier requests stay refused. An error means an attempt did not return
-// in time and may still submit.
-func (s *Server) fenceLegacyStarts(ctx context.Context, host string, seq uint64) (func(), error) {
+// fenceLegacyStarts is a /stop's first step for host: refuse new starts
+// while the stop runs, refuse every start accepted at or before this moment
+// (the accepted-connection watermark, read under legacyMu, so no start can
+// register between the watermark and the snapshot), cancel the attempts in
+// flight and join them, bounded by the join budget and ctx. The fence is
+// per host and in memory only: release lifts the refusal of new starts when
+// the stop is done, and only connections accepted before the watermark
+// stay refused. An error means an attempt did not return in time and may
+// still submit.
+func (s *Server) fenceLegacyStarts(ctx context.Context, host string) (func(), error) {
s.legacyMu.Lock()
+ seq := s.connSeq.Load()
if s.legacyStopping == nil {
s.legacyStopping = make(map[string]int)
}
@@ -128,14 +133,18 @@ func (s *Server) fenceLegacyStarts(ctx context.Context, host string, seq uint64)
}
s.legacyMu.Unlock()
}
- timer := time.NewTimer(legacyJoinBudget)
+ budget := legacyJoinBudget
+ if s.legacyJoin > 0 {
+ budget = s.legacyJoin
+ }
+ timer := time.NewTimer(budget)
defer timer.Stop()
for _, a := range inFlight {
a.cancel()
select {
case <-a.done:
case <-timer.C:
- return release, fmt.Errorf("a start of %s is still in flight after %s", host, legacyJoinBudget)
+ return release, fmt.Errorf("a start of %s is still in flight after %s", host, budget)
case <-ctx.Done():
return release, fmt.Errorf("a start of %s is still in flight: %w", host, ctx.Err())
}
diff --git a/internal/instancectl/legacystart_live_test.go b/internal/instancectl/legacystart_live_test.go
new file mode 100644
index 000000000..617e9a0ff
--- /dev/null
+++ b/internal/instancectl/legacystart_live_test.go
@@ -0,0 +1,215 @@
+package instancectl
+
+import (
+ "context"
+ "io"
+ "net"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+)
+
+// Astra round 3 P1: connection 1 (keep-alive) is accepted first and later
+// carries the /stop; a delayed /start arrives on connection 2 and has not
+// registered when the stop fences the host. The fence is the accepted
+// connection watermark at that moment, not the stop's own number, so the
+// delayed start is refused when it finally registers.
+func TestADelayedStartAcceptedBeforeTheFenceIsRefusedWhenItRegisters(t *testing.T) {
+ host := "watermark.example.com"
+ s, _ := newServer(t, &fakeCtl{})
+ _ = s.nextConnSeq() // connection 1: will carry the /stop
+ startConn := s.nextConnSeq() // connection 2: the delayed /start
+ release, err := s.fenceLegacyStarts(context.Background(), host)
+ if err != nil {
+ t.Fatalf("fence: %v", err)
+ }
+ release()
+ if _, finish, err := s.beginLegacyStart(context.Background(), host, startConn); err == nil {
+ finish()
+ t.Fatal("a start accepted before the fence registered after it and was admitted")
+ }
+}
+
+// Astra round 3 P2: a client that reuses its connection gets a fresh stamp
+// for every request, because the server answers each on its own connection.
+// A start after a stop, from the same keep-alive client, runs.
+func TestAKeepAliveClientsStartAfterAStopRuns(t *testing.T) {
+ host := "reuse.example.com"
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ _, sock := serveOnSocket(t, ctl)
+ var dials int
+ c := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ dials++
+ return (&net.Dialer{}).DialContext(ctx, "unix", sock)
+ }}}
+ for i, path := range []string{"/start?host=" + host + "&role=run&memory_max=536870912", "/stop?host=" + host + "&role=run", "/start?host=" + host + "&role=run&memory_max=536870912"} {
+ resp, err := c.Post("http://unix"+path, "", nil)
+ if err != nil {
+ t.Fatalf("POST %s: %v", path, err)
+ }
+ body, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ if resp.StatusCode/100 != 2 {
+ t.Fatalf("request %d %s: status %d: %s", i, path, resp.StatusCode, body)
+ }
+ }
+ if dials != 3 {
+ t.Fatalf("dials = %d, want one connection per request", dials)
+ }
+}
+
+// Liveness 1: the fence covers only starts accepted before the stop. A
+// /start accepted after the /stop has returned runs normally, and nothing
+// of a fence outlives the server: a restarted instance-control has none.
+func TestTheFenceCoversOnlyStartsAcceptedBeforeTheStop(t *testing.T) {
+ host := "after.example.com"
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ s, c := newServer(t, ctl)
+ stop := post(t, c, "/stop?host="+host+"&role=run")
+ stop.Body.Close()
+ if stop.StatusCode != http.StatusNoContent {
+ t.Fatalf("/stop status %d", stop.StatusCode)
+ }
+ start := post(t, c, "/start?host="+host+"&role=run&memory_max=536870912")
+ start.Body.Close()
+ if start.StatusCode != http.StatusOK {
+ t.Fatalf("a start accepted after the stop returned: status %d, want 200", start.StatusCode)
+ }
+ s.legacyMu.Lock()
+ stopping := s.legacyStopping[host]
+ s.legacyMu.Unlock()
+ if stopping != 0 {
+ t.Fatalf("the stop's refusal of new starts outlived it (%d)", stopping)
+ }
+ // A restarted service: the fence lived in memory only.
+ _, c2 := newServer(t, &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv2\n"}})
+ again := post(t, c2, "/start?host="+host+"&role=run&memory_max=536870912")
+ again.Body.Close()
+ if again.StatusCode != http.StatusOK {
+ t.Fatalf("a restarted instance-control refused a start: status %d", again.StatusCode)
+ }
+}
+
+// heldStart parks host's legacy /start before its tier step until release
+// is closed, and returns once the handler is parked there. The handler does
+// not see its own cancellation (unseenCancel).
+func heldStart(t *testing.T, ctl *fakeCtl, c *http.Client, host string) (release chan struct{}) {
+ t.Helper()
+ entered, gate := make(chan struct{}), make(chan struct{})
+ ctl.mu.Lock()
+ if ctl.before == nil {
+ ctl.before = map[string]func(){}
+ }
+ ctl.before["set-property"] = func() {
+ close(entered)
+ <-gate
+ }
+ ctl.unseenCancel = true
+ ctl.mu.Unlock()
+ postAsync(c, "/start?host="+host+"&role=run&memory_max=536870912")
+ select {
+ case <-entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("the held start never reached its tier step")
+ }
+ return gate
+}
+
+// Liveness 2: the join is per host. Host A's attempt will not join; /start
+// and /stop of host B answer at once while A's /stop is still joining.
+func TestAStuckJoinDelaysNoOtherHost(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ _, c := newServer(t, ctl)
+ gate := heldStart(t, ctl, c, "stuck.example.com")
+ stopA := postAsync(c, "/stop?host=stuck.example.com&role=run")
+ defer func() { close(gate); <-stopA }()
+ time.Sleep(50 * time.Millisecond) // A's stop is joining
+ for _, path := range []string{"/start?host=free.example.com&role=run&memory_max=536870912", "/stop?host=free.example.com&role=run"} {
+ done := postAsync(c, path)
+ select {
+ case code := <-done:
+ if code/100 != 2 {
+ t.Fatalf("%s: status %d", path, code)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatalf("%s waited behind another host's stuck join", path)
+ }
+ }
+ select {
+ case <-stopA:
+ t.Fatal("A's stop answered while its attempt was still stuck")
+ default:
+ }
+}
+
+// Liveness 3: a stuck stop answers once, within the bound, and the next
+// /stop of the host does not hang behind the stuck attempt: it answers the
+// same error within the bound, and joins it once it returns.
+func TestAStuckStopAnswersAndTheNextStopDoesNotHang(t *testing.T) {
+ host := "wedged.example.com"
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ _, c := newServer(t, ctl, func(s *Server) { s.legacyJoin = 200 * time.Millisecond })
+ gate := heldStart(t, ctl, c, host)
+ stopOnce := func() (int, string, time.Duration) {
+ begun := time.Now()
+ resp := post(t, c, "/stop?host="+host+"&role=run")
+ body, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ return resp.StatusCode, string(body), time.Since(begun)
+ }
+ for i := 0; i < 2; i++ {
+ code, body, took := stopOnce()
+ if code != http.StatusBadGateway || !strings.Contains(body, "still in flight") {
+ t.Fatalf("stop %d with the attempt stuck: %d %q, want 502 naming the start in flight", i+1, code, body)
+ }
+ if took > 2*time.Second {
+ t.Fatalf("stop %d took %s, past its bound", i+1, took)
+ }
+ }
+ close(gate)
+ time.Sleep(50 * time.Millisecond)
+ if code, body, _ := stopOnce(); code != http.StatusNoContent {
+ t.Fatalf("stop once the attempt returned: %d %q, want 204", code, body)
+ }
+}
+
+// Liveness 4: the other verbs pass through unchanged during a legacy stop.
+// While host A's legacy /stop is joining a stuck attempt, A's sidecar
+// /start and seed /stop answer at once.
+func TestOtherVerbsPassThroughDuringALegacyStop(t *testing.T) {
+ host := "verbs.example.com"
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ _, c := newServer(t, ctl)
+ gate := heldStart(t, ctl, c, host)
+ stop := postAsync(c, "/stop?host="+host+"&role=run")
+ defer func() { close(gate); <-stop }()
+ time.Sleep(50 * time.Millisecond) // the legacy stop is joining
+ for _, path := range []string{"/start?host=" + host + "&role=sidecar&memory_max=536870912", "/stop?host=" + host + "&role=seed"} {
+ done := postAsync(c, path)
+ select {
+ case code := <-done:
+ if code/100 != 2 {
+ t.Fatalf("%s during a legacy stop: status %d", path, code)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatalf("%s waited behind a legacy stop", path)
+ }
+ }
+}
+
+// postAsync POSTs path from its own goroutine, never touching t, and
+// delivers the status (0 on a transport error).
+func postAsync(c *http.Client, path string) <-chan int {
+ done := make(chan int, 1)
+ go func() {
+ resp, err := c.Post("http://unix"+path, "", nil)
+ if err != nil {
+ done <- 0
+ return
+ }
+ resp.Body.Close()
+ done <- resp.StatusCode
+ }()
+ return done
+}
diff --git a/internal/instancectl/legacystart_test.go b/internal/instancectl/legacystart_test.go
index 785b57c5d..ad8b3acf9 100644
--- a/internal/instancectl/legacystart_test.go
+++ b/internal/instancectl/legacystart_test.go
@@ -97,8 +97,8 @@ func TestALegacyStartFromBeforeAStopIsRefused(t *testing.T) {
s, _ := newServer(t, ctl)
// Stamp two connections in accept order: the start's first.
startSeq := s.nextConnSeq()
- stopSeq := s.nextConnSeq()
- release, err := s.fenceLegacyStarts(context.Background(), host, stopSeq)
+ _ = s.nextConnSeq() // the stop's own connection
+ release, err := s.fenceLegacyStarts(context.Background(), host)
if err != nil {
t.Fatalf("fence: %v", err)
}
diff --git a/internal/instancectl/server.go b/internal/instancectl/server.go
index ed023fb29..110851e7b 100644
--- a/internal/instancectl/server.go
+++ b/internal/instancectl/server.go
@@ -126,6 +126,8 @@ type Server struct {
legacyStopping map[string]int
legacyBarrier map[string]uint64
connSeq atomic.Uint64
+ // legacyJoin overrides legacyJoinBudget; tests shorten it.
+ legacyJoin time.Duration
mu sync.Mutex
report ProbeReport
@@ -215,6 +217,10 @@ func runWithClaim(args []string, claim, heldRelease http.Handler, hostApp func(s
func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
vetted := &vettedListener{Listener: ln, ok: s.peerOK, logf: s.logf}
srv := &http.Server{Handler: s.Handler(), ReadHeaderTimeout: 5 * time.Second, ConnContext: s.connContext}
+ // One request per connection (#292): a connection's accept order is the
+ // request's (connContext), which the legacy start fence relies on, so a
+ // client may not reuse one. The agent's client never does.
+ srv.SetKeepAlivesEnabled(false)
go func() {
<-ctx.Done()
srv.Close()
@@ -651,7 +657,7 @@ func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) {
// on (legacystart.go, #292): once this answers, no start from an
// earlier request can still be submitted, and the stop below
// replaces any job a joined start had submitted.
- release, err := s.fenceLegacyStarts(r.Context(), req.host, requestSeq(r.Context()))
+ release, err := s.fenceLegacyStarts(r.Context(), req.host)
defer release()
if err != nil {
// Stopped anyway, but not proven final: a start may follow.