feat: supervise plugins, and account for what each one costs

A plugin that panicked was invisible and permanent. Nothing watched the child, so when it died cur still pointed at the corpse: every request became a connection error dressed as a 502, Plugins() kept reporting Running, and only a manual Reload recovered. A plugin is a separate process precisely so a panic cannot take the host down — which is only true if the host notices. It notices now. The supervisor sees the exit, stops serving from the dead instance (503 "deployed but down", the truth, rather than 502 blaming an upstream that no longer exists), and restarts with capped backoff. An instance retired by Reload or Unload also exits; a compare-and-swap tells the two apart, so only a death that was still current counts as a crash. Restarts is reported separately from Reloads — deliberate swaps and crashes should never be one number, and a climbing Restarts is a crash loop. Proven, not argued: TestPlugins_SurvivesPanic panics the plugin on a goroutine, where no recover() can catch it, and asserts the host survives, the pid changes, Restarts increments, and the same routes serve again with no re-registration. Child exit is now a closed channel rather than a single-value one, because the supervisor, a drain and Shutdown all need to observe the same exit and the first reader would otherwise starve the rest. Also adds per-plugin resource accounting, read from /proc: Usage{CPU, RSS, Threads, FDs} // measured, not self-reported This is the observability a monolith cannot offer at any price. One process has one RSS and one CPU clock, and nothing inside it can be attributed to a subsystem — "which app is eating the memory" is an argument. Running a service as its own process makes it a question with an exact answer. Measured on a live plugin: rss=12.7MB threads=5 fds=7. And plugin output is tagged with the plugin name. An untagged line in a merged stream is worse than useless: you can see something is wrong and not what. Co-authored-by: Hanzo Dev <[email protected]>
2026-10-19 15:38:57 +00:00 · 2026-07-27 15:40:43 -07:00
parent 1d67323b50
commit 3d7fc4e43c
6 changed files with 342 additions and 23 deletions
@@ -22,6 +22,13 @@ func main() {
 	app.All("/*", func(c *zip.Ctx) error {
 		return c.JSON(200, map[string]string{"echo": c.Path(), "version": version})
 	})
+	// Crashes the process the way a real bug does — a panic on a goroutine,
+	// which no handler recover() can catch. Used to prove the host survives a
+	// plugin dying and brings it back.
+	app.Get("/v1/demo/crash", func(c *zip.Ctx) error {
+		go func() { panic("testplugin: deliberate crash") }()
+		return c.JSON(200, map[string]string{"crashing": "true"})
+	})
 	// Addr is the whole plugin side of the contract: serve where the host said.
 	_ = app.Listen(zip.Addr(":9999"))
 }
@@ -1,6 +1,7 @@
 package zip

 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
@@ -136,9 +137,11 @@ type plugin struct {
 	prefixes []string // every subtree this plugin answers
 	spec     Plugin

-	cur     atomic.Pointer[instance]
-	mu      sync.Mutex
-	reloads atomic.Int64
+	cur      atomic.Pointer[instance]
+	mu       sync.Mutex
+	reloads  atomic.Int64
+	restarts atomic.Int64
+	closed   bool // Shutdown ran; the supervisor must stop resurrecting it

 	// stopping tracks the goroutines draining replaced instances. Reload and
 	// Unload retire an instance asynchronously so the caller is not blocked for
@@ -163,12 +166,18 @@ func (p *plugin) retire(in *instance, grace time.Duration) {
 // instance is one running child process and everything that must be released
 // when it stops.
 type instance struct {
-	cmd    *exec.Cmd
-	dir    string
-	sock   string
+	cmd     *exec.Cmd
+	dir     string
+	sock    string
 	client  Client
-	exited  chan error // cmd.Wait's single result
 	started time.Time
+
+	// done is CLOSED when the child has exited, and exitErr holds why. A
+	// closed channel broadcasts: the supervisor, a drain, and Shutdown can all
+	// observe the same exit. A single-value channel would let whoever read
+	// first starve the others.
+	done    chan struct{}
+	exitErr error
 }

 // target is the hot path: what the mounted route should talk to right now.
@@ -256,8 +265,12 @@ func (a *App) load(prefixes []string, spec Plugin) error {
 	for _, pre := range prefixes {
 		a.mountVia(pre, p.target)
 	}
+	go a.supervise(p, in)

 	a.OnShutdown(func(context.Context) error {
+		p.mu.Lock()
+		p.closed = true
+		p.mu.Unlock()
 		if cur := p.cur.Swap(nil); cur != nil {
 			stop(cur, 0)
 		}
@@ -364,23 +377,28 @@ func start(spec Plugin) (*instance, error) {
 	sock := filepath.Join(dir, spec.Name+".sock")
 	cmd := exec.Command(bin, spec.Args...)
 	cmd.Env = append(append(os.Environ(), spec.Env...), AddrEnv+"="+sock)
-	// The child's output is the operator's only window into it, so it goes
-	// where the host's own does rather than being swallowed.
-	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
+	// The child's output is the operator's only window into it, so it goes where
+	// the host's own does rather than being swallowed — but tagged, because an
+	// untagged line in a merged stream is worse than useless: you can see that
+	// something is wrong and not which plugin is wrong. Tagging is the whole
+	// difference between one log stream and N attributable ones.
+	cmd.Stdout = &tagWriter{w: os.Stdout, tag: []byte("[" + spec.Name + "] ")}
+	cmd.Stderr = &tagWriter{w: os.Stderr, tag: []byte("[" + spec.Name + "] ")}
 	tieToHost(cmd)
 	if err := cmd.Start(); err != nil {
 		_ = os.RemoveAll(dir)
 		return nil, fmt.Errorf("start: %w", err)
 	}

-	// Exactly one Wait for this process's lifetime; everything else reads the
-	// result off this channel. Two Waits is an error, and a child nobody waits
-	// on becomes a zombie.
-	exited := make(chan error, 1)
-	go func() { exited <- cmd.Wait() }()
+	// Exactly one Wait for this process's lifetime; everyone else observes the
+	// close. Two Waits is an error, and a child nobody waits on is a zombie.
+	in := &instance{cmd: cmd, dir: dir, sock: sock, started: time.Now(), done: make(chan struct{})}
+	go func() {
+		in.exitErr = cmd.Wait()
+		close(in.done)
+	}()

-	in := &instance{cmd: cmd, dir: dir, sock: sock, exited: exited, started: time.Now()}
-	if err := waitListening(sock, spec.Start, exited); err != nil {
+	if err := waitListening(sock, spec.Start, in.done, in); err != nil {
 		stop(in, 0)
 		return nil, err
 	}
@@ -463,14 +481,14 @@ func stop(in *instance, grace time.Duration) {
 	}
 	if grace > 0 {
 		select {
-		case <-in.exited: // already gone; nothing to wait out
+		case <-in.done: // already gone; nothing to wait out
 		case <-time.After(grace):
 		}
 	}
 	if in.cmd != nil && in.cmd.Process != nil {
 		_ = in.cmd.Process.Kill()
 		select {
-		case <-in.exited:
+		case <-in.done:
 		case <-time.After(5 * time.Second):
 		}
 	}
@@ -487,7 +505,7 @@ func stop(in *instance, grace time.Duration) {
 // waitListening blocks until the socket accepts, the child exits, or the
 // deadline passes. Watching the child matters: a plugin that dies immediately
 // should say so, not time out.
-func waitListening(sock string, limit time.Duration, exited <-chan error) error {
+func waitListening(sock string, limit time.Duration, done <-chan struct{}, in *instance) error {
 	if limit <= 0 {
 		limit = 10 * time.Second
 	}
@@ -498,8 +516,8 @@ func waitListening(sock string, limit time.Duration, exited <-chan error) error
 			return nil
 		}
 		select {
-		case err := <-exited:
-			return fmt.Errorf("exited before listening: %v", err)
+		case <-done:
+			return fmt.Errorf("exited before listening: %v", in.exitErr)
 		default:
 		}
 		if time.Now().After(deadline) {
@@ -508,3 +526,95 @@ func waitListening(sock string, limit time.Duration, exited <-chan error) error
 		time.Sleep(5 * time.Millisecond)
 	}
 }
+
+// tagWriter prefixes each line a plugin writes with its name, so a host serving
+// several plugins produces one stream that is still attributable. It splits on
+// newlines rather than on Write calls, because a child may emit a line across
+// several writes or several lines in one.
+type tagWriter struct {
+	w       io.Writer
+	tag     []byte
+	mu      sync.Mutex
+	midline bool // a previous Write ended without a newline
+}
+
+func (t *tagWriter) Write(p []byte) (int, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	n := len(p)
+	for len(p) > 0 {
+		if !t.midline {
+			if _, err := t.w.Write(t.tag); err != nil {
+				return 0, err
+			}
+			t.midline = true
+		}
+		i := bytes.IndexByte(p, '\n')
+		if i < 0 {
+			_, err := t.w.Write(p)
+			return n, err
+		}
+		if _, err := t.w.Write(p[:i+1]); err != nil {
+			return 0, err
+		}
+		t.midline = false
+		p = p[i+1:]
+	}
+	return n, nil
+}
+
+// supervise watches one instance and brings the plugin back if it dies on its
+// own. Without this a crashed plugin is invisible and permanent: the child is
+// gone but cur still points at it, so every request gets a connection error
+// dressed up as a 502, Plugins() keeps reporting Running, and only a manual
+// Reload recovers. A plugin is a separate process precisely so a panic cannot
+// take the host down — that is only true if the host notices.
+//
+// An instance replaced by Reload or Unload also "exits", which is expected and
+// not a crash. The compare-and-swap distinguishes them: only the instance that
+// is still current when it dies was unexpected.
+func (a *App) supervise(p *plugin, in *instance) {
+	<-in.done
+	if !p.cur.CompareAndSwap(in, nil) {
+		return // replaced deliberately; its retirement is already someone's job
+	}
+	a.logger.Error("zip plugin exited unexpectedly",
+		"name", p.name, "pid", in.cmd.Process.Pid, "err", in.exitErr)
+	stop(in, 0)
+
+	// Restart with backoff. Requests answer 503 in the gap — "deployed but
+	// down", which is the truth — rather than 502, which would blame the
+	// upstream for a process that no longer exists.
+	//
+	// Backoff is capped and retried indefinitely rather than given up on: the
+	// common cause is a dependency that is itself restarting, and a plugin that
+	// recovers on its own is worth far more than one that stays dead because a
+	// retry budget ran out while nobody was watching.
+	delay := 100 * time.Millisecond
+	for {
+		p.mu.Lock()
+		shuttingDown := p.closed
+		spec := p.spec
+		p.mu.Unlock()
+		if shuttingDown {
+			return
+		}
+		time.Sleep(delay)
+		if delay *= 2; delay > 30*time.Second {
+			delay = 30 * time.Second
+		}
+		next, err := start(spec)
+		if err != nil {
+			a.logger.Error("zip plugin restart failed", "name", p.name, "err", err)
+			continue
+		}
+		if !p.cur.CompareAndSwap(nil, next) {
+			stop(next, 0) // a Reload beat us to it
+			return
+		}
+		p.restarts.Add(1)
+		a.logger.Info("zip plugin restarted", "name", p.name, "pid", next.cmd.Process.Pid)
+		go a.supervise(p, next)
+		return
+	}
+}
@@ -400,3 +400,104 @@ func TestLoad_NoPrefix(t *testing.T) {
 		t.Fatalf("err = %v, want a refusal naming the missing prefix", err)
 	}
 }
+
+// TestPlugins_Usage proves per-plugin resource accounting is real. This is the
+// observability a monolith cannot offer at any price: one process has one RSS
+// and one CPU clock, and nothing inside it can be attributed to a subsystem.
+// Running a service as its own process turns "which app is eating the memory"
+// into a question with an exact, kernel-measured answer.
+func TestPlugins_Usage(t *testing.T) {
+	if runtime.GOOS != "linux" {
+		t.Skip("resource accounting reads /proc")
+	}
+	bin := buildPlugin(t, "v1")
+
+	app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
+	if err := app.Add(zip.Load(
+		zip.Plugin{Name: "demo", Bin: bin, Dir: t.TempDir()}, "/v1/demo",
+	)); err != nil {
+		t.Fatalf("Add(Load): %v", err)
+	}
+	defer func() { _ = app.Shutdown() }()
+
+	// Drive it so there is CPU to account for.
+	for i := 0; i < 20; i++ {
+		if status, _ := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
+			t.Fatalf("request %d: status %d", i, status)
+		}
+	}
+
+	u := app.Plugins()[0].Usage
+	if u.RSS <= 0 {
+		t.Fatalf("RSS = %d, want a real resident set for a running child", u.RSS)
+	}
+	if u.Threads <= 0 {
+		t.Fatalf("Threads = %d, want at least one", u.Threads)
+	}
+	if u.FDs <= 0 {
+		t.Fatalf("FDs = %d, want at least the socket", u.FDs)
+	}
+	// A Go runtime always has several threads and holds more than a trivial
+	// resident set, so these also guard against parsing the wrong /proc fields.
+	if u.Threads > 10000 || u.RSS > 1<<40 {
+		t.Fatalf("implausible usage %+v — likely misparsed /proc fields", u)
+	}
+	t.Logf("plugin usage: cpu=%v rss=%.1fMB threads=%d fds=%d",
+		u.CPU, float64(u.RSS)/(1<<20), u.Threads, u.FDs)
+}
+
+// TestPlugins_SurvivesPanic proves the central safety claim of running a plugin
+// as its own process: a panic in the plugin cannot take the host down, the host
+// NOTICES rather than serving from a corpse, and the plugin comes back on its
+// own.
+//
+// Without a supervisor the failure is silent and permanent — the child is gone
+// but the mount still points at it, so every request gets a connection error
+// dressed as a 502 and the status keeps reporting Running.
+func TestPlugins_SurvivesPanic(t *testing.T) {
+	bin := buildPlugin(t, "v1")
+
+	app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
+	if err := app.Add(zip.Load(
+		zip.Plugin{Name: "demo", Bin: bin, Dir: t.TempDir()}, "/v1/demo",
+	)); err != nil {
+		t.Fatalf("Add(Load): %v", err)
+	}
+	defer func() { _ = app.Shutdown() }()
+
+	first := app.Plugins()[0].PID
+	if status, _ := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
+		t.Fatalf("before crash: status %d", status)
+	}
+
+	// Panic it on a goroutine — unrecoverable, so the process really dies.
+	if status, _ := call(t, app, "GET", "/v1/demo/crash", ""); status != 200 {
+		t.Fatalf("crash trigger: status %d", status)
+	}
+
+	// THE HOST MUST SURVIVE. If a plugin panic could kill the host, none of
+	// this architecture is worth anything.
+	deadline := time.Now().Add(20 * time.Second)
+	var restarted bool
+	for time.Now().Before(deadline) {
+		ps := app.Plugins()
+		if len(ps) != 1 {
+			t.Fatalf("plugin vanished from status: %+v", ps)
+		}
+		if ps[0].Restarts > 0 && ps[0].Running && ps[0].PID != first {
+			restarted = true
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	if !restarted {
+		t.Fatalf("plugin did not come back: %+v", app.Plugins()[0])
+	}
+
+	// And it serves again, on the same routes, with no re-registration.
+	if status, body := call(t, app, "GET", "/v1/demo/version", ""); status != 200 ||
+		!strings.Contains(body, `"version":"v1"`) {
+		t.Fatalf("after restart: status=%d body=%q", status, body)
+	}
+	t.Logf("survived: pid %d -> %d, restarts=%d", first, app.Plugins()[0].PID, app.Plugins()[0].Restarts)
+}
@@ -44,6 +44,36 @@ type PluginStatus struct {
 	// Reloads counts successful swaps since Load. A climbing number on one
 	// host and not its peers is the signal that a rollout is uneven.
 	Reloads int `json:"reloads"`
+
+	// Restarts counts times the supervisor brought this plugin back after it
+	// died on its own. Distinct from Reloads, which are deliberate: a nonzero
+	// Restarts is a plugin crashing, and a climbing one is a crash loop.
+	Restarts int `json:"restarts"`
+
+	// Usage is what this plugin costs right now, read from the kernel.
+	Usage Usage `json:"usage,omitzero"`
+}
+
+// Usage is a plugin's resource cost, measured rather than self-reported.
+//
+// This exists because a plugin is a process. A linked-in subsystem cannot be
+// measured this way at any price — one process has one RSS and one CPU clock,
+// and nothing inside it can be attributed to a subsystem. Running a service as
+// its own process makes "which app is eating the memory" a question with an
+// exact answer instead of an argument.
+//
+// Zero means not measured (no /proc, or nothing running), not measured-as-zero.
+type Usage struct {
+	// CPU is total user+system time this instance has consumed since it
+	// started. It only ever climbs, so a rate is the interesting derivative.
+	CPU time.Duration `json:"cpuNs,omitzero"`
+	// RSS is resident memory in bytes — the honest number for "what does this
+	// plugin cost", as opposed to virtual size.
+	RSS int64 `json:"rssBytes,omitzero"`
+	// Threads and FDs are the two limits a busy service hits first, and both
+	// are leaks when they climb without bound.
+	Threads int `json:"threads,omitzero"`
+	FDs     int `json:"fds,omitzero"`
 }

 // Plugins reports every plugin this host has loaded, ordered by name so a diff
@@ -63,7 +93,8 @@ func (a *App) Plugins() []PluginStatus {
 			Prefix:  p.prefix,
 			Source:  p.source(),
 			Version: p.spec.Sum,
-			Reloads: int(p.reloads.Load()),
+			Reloads:  int(p.reloads.Load()),
+			Restarts: int(p.restarts.Load()),
 		}
 		if in := p.cur.Load(); in != nil {
 			s.Running = true
@@ -71,6 +102,7 @@ func (a *App) Plugins() []PluginStatus {
 			s.Since = in.started
 			if in.cmd != nil && in.cmd.Process != nil {
 				s.PID = in.cmd.Process.Pid
+				s.Usage = readUsage(s.PID)
 			}
 		}
 		out = append(out, s)
@@ -0,0 +1,62 @@
+//go:build linux
+
+package zip
+
+import (
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// readUsage reports what the kernel already knows about a running plugin.
+//
+// This is the observability a monolith cannot give you. In one process you can
+// measure the process; you cannot say which subsystem is holding the RSS or
+// burning the CPU, because there is no boundary to measure at. A plugin IS a
+// process, so per-plugin CPU, memory, thread and descriptor counts come from
+// /proc for free — exact, continuous, and with no instrumentation in the
+// plugin at all. It cannot drift from reality because it is not self-reported.
+func readUsage(pid int) Usage {
+	var u Usage
+	if pid <= 0 {
+		return u
+	}
+	dir := "/proc/" + strconv.Itoa(pid)
+
+	// utime+stime are fields 14 and 15, and num_threads is 20 — but comm
+	// (field 2) is parenthesised and may itself contain spaces and parens, so
+	// fields are counted from the LAST ')' rather than by splitting the line.
+	if b, err := os.ReadFile(dir + "/stat"); err == nil {
+		if i := strings.LastIndexByte(string(b), ')'); i > 0 && i+2 < len(b) {
+			f := strings.Fields(string(b)[i+2:])
+			// After comm, field 3 is state, so utime/stime/num_threads land at
+			// offsets 11, 12 and 17 of this slice.
+			if len(f) > 17 {
+				ut, _ := strconv.ParseInt(f[11], 10, 64)
+				st, _ := strconv.ParseInt(f[12], 10, 64)
+				// USER_HZ is 100 on every Linux target we ship to.
+				u.CPU = time.Duration(ut+st) * 10 * time.Millisecond
+				u.Threads, _ = strconv.Atoi(f[17])
+			}
+		}
+	}
+
+	// VmRSS is the resident set — what this plugin actually costs in memory.
+	if b, err := os.ReadFile(dir + "/status"); err == nil {
+		for line := range strings.SplitSeq(string(b), "\n") {
+			if kb, ok := strings.CutPrefix(line, "VmRSS:"); ok {
+				if f := strings.Fields(kb); len(f) > 0 {
+					n, _ := strconv.ParseInt(f[0], 10, 64)
+					u.RSS = n * 1024
+				}
+				break
+			}
+		}
+	}
+
+	if ents, err := os.ReadDir(dir + "/fd"); err == nil {
+		u.FDs = len(ents)
+	}
+	return u
+}
@@ -0,0 +1,7 @@
+//go:build !linux
+
+package zip
+
+// readUsage reports nothing where /proc is unavailable. The zero Usage is
+// meaningful: it says "not measured here", not "measured as zero".
+func readUsage(int) Usage { return Usage{} }