amadan

carlos / platform Public

instancectl: fix round 2 — peer-check race, seed/sweep partial failure, refusal log rate limit (plan 2, task 5)

Six items from the codex review of fix round 1:

1. Serve captured s.PeerOK into the vettedListener at start, so a caller
   setting PeerOK after Serve began was silently ignored, and the -race
   detector caught the read/write as a genuine data race. Added
   Server.peerOK, which calls through s.PeerOK per connection; Serve wires
   ok: s.peerOK. newServer in the tests now takes optional configure
   hooks applied before Serve starts, and
   TestPeerCheckClosesTheConnectionWithoutAResponse sets PeerOK through
   one instead of mutating the running server.

2. /seed now issues stop on a WaitCgroupEmpty timeout before returning the
   502 — stop is the action that actually kills survivors, not just a
   status check.

3. /sweep no longer aborts on the first unit whose cgroup won't empty: it
   stops and waits every target, collects failures, and returns 200 when
   all succeeded or 502 with {"units": [...], "failed": {unit: reason}}
   otherwise. SweepResponse gains a Failed map[string]string
   `json:"failed,omitempty"` field (the one names.go edit this round
   allows).

4. /sweep treats a stop error as that unit's failure and, after a
   successful stop, polls `show -p ActiveState --value` (50ms, same
   per-unit budget) until inactive or failed before the cgroup wait, so an
   still-activating unit with no cgroup yet is never reported swept.

5. peer_linux.go's edgePID now negative-caches a failed MainPID lookup for
   one second (previously only a success was cached), and the
   vettedListener rate-limits its refusal log to one line per second,
   folding the suppressed count into the next line.

6. WaitCgroupEmpty returns a read error other than ErrNotExist immediately
   (fmt.Errorf("%s: %w", unit, err)) instead of spinning to the budget.

New test TestSweepStopsOthersAndReportsAStuckUnitAsFailed covers items 3
and 4 together; TestSweepStopsEveryInstanceUnitAndJob now also asserts
Failed is empty on the all-succeeded path.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YZ1ZvWj7naf73GjTX8MBv5
Paul Campbell pushed by paul@keymail.dev 08a0ed8246b1ca3f65ff65b4e572349e699f4fc6 parent c15a7b8
4 files changed, +166 −17
  • internal/instancectl/names.go +2 −1
  • internal/instancectl/peer_linux.go +18 −4
  • internal/instancectl/server.go +88 −9
  • internal/instancectl/server_test.go +58 −3

CI passed — run details

diff --git a/internal/instancectl/names.go b/internal/instancectl/names.go
index 73810a815..7de0f994c 100644
--- a/internal/instancectl/names.go
+++ b/internal/instancectl/names.go
@@ -181,5 +181,6 @@ type SeedResponse struct {
}
type SweepResponse struct {
- Units []string `json:"units"`
+ Units []string `json:"units"`
+ Failed map[string]string `json:"failed,omitempty"`
}
diff --git a/internal/instancectl/peer_linux.go b/internal/instancectl/peer_linux.go
index c36f15059..1c3190f18 100644
--- a/internal/instancectl/peer_linux.go
+++ b/internal/instancectl/peer_linux.go
@@ -52,24 +52,38 @@ func productionPeerOK() func(net.Conn) error {
carlosUID, _ := strconv.Atoi(u.Uid)
var mu sync.Mutex
var mainPID int
+ var lastErr error
var at time.Time
edgePID := func() (int, error) {
mu.Lock()
defer mu.Unlock()
- if time.Since(at) < time.Second && mainPID > 0 {
- return mainPID, nil
+ // Cache both a success and a failure for one second: while the
+ // edge has no MainPID yet (a restart window), every refused
+ // connect would otherwise re-exec systemctl under this mutex —
+ // the negative case needs the same cache the positive one always
+ // had (codex review fix round 2 item 5).
+ if time.Since(at) < time.Second {
+ if mainPID > 0 {
+ return mainPID, nil
+ }
+ if lastErr != nil {
+ return 0, lastErr
+ }
}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "MainPID", "--value", edgeUnit).Output()
if err != nil {
+ mainPID, lastErr, at = 0, err, time.Now()
return 0, err
}
pid, err := strconv.Atoi(strings.TrimSpace(string(out)))
if err != nil || pid <= 0 {
- return 0, fmt.Errorf("carlos-edge.service has no MainPID")
+ err = fmt.Errorf("carlos-edge.service has no MainPID")
+ mainPID, lastErr, at = 0, err, time.Now()
+ return 0, err
}
- mainPID, at = pid, time.Now()
+ mainPID, lastErr, at = pid, nil, time.Now()
return pid, nil
}
return func(conn net.Conn) error {
diff --git a/internal/instancectl/server.go b/internal/instancectl/server.go
index 18425ba9a..65a35a256 100644
--- a/internal/instancectl/server.go
+++ b/internal/instancectl/server.go
@@ -108,7 +108,7 @@ func Run(args []string) {
// ones to the HTTP server one at a time through a listener that only ever
// yields vetted connections.
func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
- vetted := &vettedListener{Listener: ln, ok: s.PeerOK, logf: s.logf}
+ vetted := &vettedListener{Listener: ln, ok: s.peerOK, logf: s.logf}
srv := &http.Server{Handler: s.Handler(), ReadHeaderTimeout: 5 * time.Second}
go func() {
<-ctx.Done()
@@ -127,6 +127,15 @@ type vettedListener struct {
net.Listener
ok func(net.Conn) error
logf func(string, ...any)
+
+ // lastLog and suppressed rate-limit the refusal log to at most one
+ // line per second: a refusing peer that reconnects in a tight loop
+ // (the common shape while carlos-edge.service has no MainPID yet)
+ // would otherwise flood the log at the same rate. Accept is only ever
+ // called from the one goroutine net/http.Server.Serve runs its accept
+ // loop on, so these need no lock of their own.
+ lastLog time.Time
+ suppressed int
}
func (v *vettedListener) Accept() (net.Conn, error) {
@@ -136,7 +145,7 @@ func (v *vettedListener) Accept() (net.Conn, error) {
return nil, err
}
if err := v.ok(c); err != nil {
- v.logf("instance-control: refused connection: %v", err)
+ v.logRefusal(err)
c.Close()
continue
}
@@ -144,12 +153,34 @@ func (v *vettedListener) Accept() (net.Conn, error) {
}
}
+func (v *vettedListener) logRefusal(err error) {
+ now := time.Now()
+ if !v.lastLog.IsZero() && now.Sub(v.lastLog) < time.Second {
+ v.suppressed++
+ return
+ }
+ if v.suppressed > 0 {
+ v.logf("instance-control: refused connection: %v (%d more refusals suppressed)", err, v.suppressed)
+ } else {
+ v.logf("instance-control: refused connection: %v", err)
+ }
+ v.lastLog = now
+ v.suppressed = 0
+}
+
func (s *Server) logf(format string, args ...any) {
if s.Logf != nil {
s.Logf(format, args...)
}
}
+// peerOK calls through s.PeerOK per connection (never captured into a
+// closure at Serve start) so a caller that sets PeerOK after Serve has
+// begun — a test, or a production reconfiguration — takes effect on the
+// very next Accept instead of being silently ignored (codex review,
+// fix round 2 item 1).
+func (s *Server) peerOK(c net.Conn) error { return s.PeerOK(c) }
+
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("POST /start", s.handleStart)
@@ -280,6 +311,12 @@ func (s *Server) handleSeed(w http.ResponseWriter, r *http.Request) {
wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
defer cancel()
if err := WaitCgroupEmpty(wctx, s.CgroupRoot, SliceName(req.host), unit, 50*time.Millisecond); err != nil {
+ // The cgroup never emptied: stop is the action that actually kills
+ // the survivors, so issue it before reporting the failure rather
+ // than leaving them running behind a 502.
+ if _, stopErr := s.Systemctl(ctx, "stop", unit); stopErr != nil {
+ s.logf("instance-control: %s: stop after seed timeout: %v", req.host, stopErr)
+ }
http.Error(w, "seed: "+err.Error(), http.StatusBadGateway)
return
}
@@ -358,24 +395,64 @@ func (s *Server) handleSweep(w http.ResponseWriter, r *http.Request) {
targets[f[1]] = true
}
}
+ // Every target gets stopped and waited on; a stuck unit is recorded and
+ // the sweep continues to the rest rather than aborting on the first one
+ // (design §6.7, codex review fix round 2 item 3): a sweep exists to
+ // clear the box for the next thing, and giving up after unit one would
+ // leave every unit after it running.
var stopped []string
+ failed := map[string]string{}
for unit := range targets {
+ wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
if _, err := s.Systemctl(ctx, "stop", unit); err != nil {
- s.logf("instance-control: sweep: stop %s: %v", unit, err)
+ failed[unit] = "stop: " + err.Error()
+ cancel()
+ continue
+ }
+ // A unit that is still activating has no cgroup yet; wait for
+ // systemd to settle it to inactive or failed before trusting an
+ // absent cgroup as "swept" (codex review fix round 2 item 4).
+ if err := s.waitInactiveOrFailed(wctx, unit, 50*time.Millisecond); err != nil {
+ failed[unit] = err.Error()
+ cancel()
+ continue
}
_, esc, _ := ParseUnitName(unit)
- wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
err := WaitCgroupEmpty(wctx, s.CgroupRoot, "carlos-instances-"+esc+".slice", unit, 50*time.Millisecond)
cancel()
if err != nil {
- http.Error(w, "sweep: "+unit+": "+err.Error(), http.StatusBadGateway)
- return
+ failed[unit] = err.Error()
+ continue
}
stopped = append(stopped, unit)
}
+ if len(failed) > 0 {
+ writeJSON(w, http.StatusBadGateway, SweepResponse{Units: stopped, Failed: failed})
+ return
+ }
writeJSON(w, http.StatusOK, SweepResponse{Units: stopped})
}
+// waitInactiveOrFailed polls a unit's ActiveState until systemd reports it
+// settled (inactive after a clean stop, failed after one that wasn't),
+// bounded by ctx.
+func (s *Server) waitInactiveOrFailed(ctx context.Context, unit string, every time.Duration) error {
+ var last string
+ for {
+ if out, err := s.Systemctl(ctx, "show", "-p", "ActiveState", "--value", unit); err == nil {
+ last = strings.TrimSpace(out)
+ if last == "inactive" || last == "failed" {
+ return nil
+ }
+ }
+ select {
+ case <-ctx.Done():
+ return fmt.Errorf("%s: ActiveState still %q", unit, last)
+ case <-time.After(every):
+ }
+ }
+}
+
func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
rep, ready := s.Ready()
if ready {
@@ -407,10 +484,12 @@ func WaitCgroupEmpty(ctx context.Context, root, slice, unit string, every time.D
events := filepath.Join(root, parentSlices, slice, unit, "cgroup.events")
for {
b, err := os.ReadFile(events)
- if errors.Is(err, os.ErrNotExist) {
+ switch {
+ case errors.Is(err, os.ErrNotExist):
return nil
- }
- if err == nil && strings.Contains(string(b), "populated 0") {
+ case err != nil:
+ return fmt.Errorf("%s: %w", unit, err)
+ case strings.Contains(string(b), "populated 0"):
return nil
}
select {
diff --git a/internal/instancectl/server_test.go b/internal/instancectl/server_test.go
index 309c0d18b..f8683d879 100644
--- a/internal/instancectl/server_test.go
+++ b/internal/instancectl/server_test.go
@@ -46,7 +46,11 @@ func (f *fakeCtl) joined() string {
return strings.Join(f.calls, "\n")
}
-func newServer(t *testing.T, ctl *fakeCtl) (*Server, *http.Client) {
+// newServer starts a Server behind a real unix socket. Any configure hooks
+// run before the accept loop starts, so a test that wants a non-default
+// PeerOK (or anything else) sets it up front rather than racing Serve's
+// first Accept (codex review fix round 2 item 1).
+func newServer(t *testing.T, ctl *fakeCtl, configure ...func(*Server)) (*Server, *http.Client) {
t.Helper()
root := t.TempDir()
s := &Server{
@@ -58,6 +62,9 @@ func newServer(t *testing.T, ctl *fakeCtl) (*Server, *http.Client) {
ProbeEvery: time.Hour,
Logf: t.Logf,
}
+ for _, configure := range configure {
+ configure(s)
+ }
dir, err := os.MkdirTemp("", "ic")
if err != nil {
t.Fatal(err)
@@ -231,12 +238,59 @@ func TestSweepStopsEveryInstanceUnitAndJob(t *testing.T) {
if strings.Contains(got, "sidecar@a") {
t.Errorf("sweep stopped an inactive unit with no job: %v", out.Units)
}
+ if len(out.Failed) != 0 {
+ t.Errorf("nothing should have failed: %v", out.Failed)
+ }
calls := ctl.joined()
if !strings.Contains(calls, "stop carlos-instance@a.example.com.service") || !strings.Contains(calls, "stop carlos-instance@b.example.com.service") {
t.Errorf("stops not issued:\n%s", calls)
}
}
+// TestSweepStopsOthersAndReportsAStuckUnitAsFailed proves a sweep never
+// aborts on the first unit whose cgroup won't empty: the other targets are
+// still stopped and reported swept, and the stuck one is reported failed
+// rather than losing the whole response (codex review fix round 2 item 3).
+func TestSweepStopsOthersAndReportsAStuckUnitAsFailed(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "list-units": "carlos-instance@a.example.com.service loaded active running carlos exec-backed instance a.example.com\n" +
+ "carlos-instance@stuck.example.com.service loaded active running carlos exec-backed instance stuck.example.com\n",
+ "list-jobs": "",
+ "show -p ActiveState --value": "inactive\n",
+ }}
+ s, c := newServer(t, ctl)
+ s.cgroupWaitBudget = 200 * time.Millisecond
+ ev := filepath.Join(s.CgroupRoot, "carlos.slice", "carlos-instances.slice", "carlos-instances-stuck.example.com.slice", "carlos-instance@stuck.example.com.service", "cgroup.events")
+ if err := os.MkdirAll(filepath.Dir(ev), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(ev, []byte("populated 1\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ resp := post(t, c, "/sweep")
+ if resp.StatusCode != 502 {
+ t.Fatalf("status %d, want 502 while a unit is stuck", resp.StatusCode)
+ }
+ var out SweepResponse
+ if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+ t.Fatal(err)
+ }
+ got := strings.Join(out.Units, " ")
+ if !strings.Contains(got, "carlos-instance@a.example.com.service") {
+ t.Errorf("the other unit was not stopped and reported: %v", out.Units)
+ }
+ if strings.Contains(got, "stuck") {
+ t.Errorf("the stuck unit must not be reported as swept: %v", out.Units)
+ }
+ if _, ok := out.Failed["carlos-instance@stuck.example.com.service"]; !ok {
+ t.Errorf("the stuck unit is missing from failed: %v", out.Failed)
+ }
+ calls := ctl.joined()
+ if !strings.Contains(calls, "stop carlos-instance@a.example.com.service") || !strings.Contains(calls, "stop carlos-instance@stuck.example.com.service") {
+ t.Errorf("both units were not stopped:\n%s", calls)
+ }
+}
+
func TestReadyReflectsTheProbeVerdict(t *testing.T) {
ctl := &fakeCtl{}
s, c := newServer(t, ctl)
@@ -279,8 +333,9 @@ func TestReadyReflectsTheProbeVerdict(t *testing.T) {
func TestPeerCheckClosesTheConnectionWithoutAResponse(t *testing.T) {
ctl := &fakeCtl{}
- s, c := newServer(t, ctl)
- s.PeerOK = func(net.Conn) error { return errors.New("not the agent") }
+ _, c := newServer(t, ctl, func(s *Server) {
+ s.PeerOK = func(net.Conn) error { return errors.New("not the agent") }
+ })
_, err := c.Get("http://unix/ready")
if err == nil {
t.Fatal("a refused peer got an HTTP response")