feat: App.Plugins() — a host reports what it is actually running

A fleet view needs one thing this package did not expose: the ability to ask a live process what it loaded, at which version, and whether it is still up. Deployment config answers what was INTENDED; only the process knows what is TRUE — which is exactly the difference that matters mid-rollout, when the two disagree by design. for _, p := range app.Plugins() { // Name, Prefix, Source, Version, Addr, PID, Running, Since, Reloads } Version is the artifact's SHA-256 for a URL-installed plugin — the only version identifier that cannot drift from the bits running, because it IS the bits. Since resets on Reload, so it reports the age of what is running rather than of the mount, and Reloads climbing on one host and not its peers is the signal that a rollout is uneven. An Unloaded plugin stays listed with Running=false, because "deployed but down" and "not deployed" are different states and its routes are still registered answering 503. Remote (Addr) mounts are recorded too. They took an early return and were invisible before; a fleet view that silently omits part of the surface is worse than none. Two fixes found by having the tests exercise this: Shutdown now waits for instances a Reload or Unload is still draining. Those retire asynchronously so the caller is not blocked for the drain window, which let a child outlive the host that spawned it — and since children inherit the host's stdout, the process appeared to hang after main returned. The suite showed it as "Test I/O incomplete 1m0s after exiting"; it now finishes in 22s instead of 74s. Tests no longer hardcode ports. A fixed 19653 meant an unrelated process on the box made TestListen_ZAP fail with 404 rather than a connection error, because the port answers — it is just answering for someone else. That reads as a routing bug and is not one. Co-authored-by: Hanzo Dev <[email protected]>
2026-10-19 15:38:57 +00:00 · 2026-07-27 12:51:44 -07:00
parent 2869190b45
commit 4403d199ce
8 changed files with 248 additions and 21 deletions
@@ -135,8 +135,28 @@ type plugin struct {
 	prefix string
 	spec   Plugin

-	cur atomic.Pointer[instance]
-	mu  sync.Mutex
+	cur     atomic.Pointer[instance]
+	mu      sync.Mutex
+	reloads atomic.Int64
+
+	// stopping tracks the goroutines draining replaced instances. Reload and
+	// Unload retire an instance asynchronously so the caller is not blocked for
+	// the drain window, which means a child can outlive the call that replaced
+	// it. Shutdown waits on this: a host that exits must not leave a child
+	// holding its stdout, or the process appears to hang after main returns.
+	stopping sync.WaitGroup
+}
+
+// retire stops in asynchronously, tracked so Shutdown can wait for it.
+func (p *plugin) retire(in *instance, grace time.Duration) {
+	if in == nil {
+		return
+	}
+	p.stopping.Add(1)
+	go func() {
+		defer p.stopping.Done()
+		stop(in, grace)
+	}()
 }

 // instance is one running child process and everything that must be released
@@ -145,8 +165,9 @@ type instance struct {
 	cmd    *exec.Cmd
 	dir    string
 	sock   string
-	client Client
-	exited chan error // cmd.Wait's single result
+	client  Client
+	exited  chan error // cmd.Wait's single result
+	started time.Time
 }

 // target is the hot path: what the mounted route should talk to right now.
@@ -177,7 +198,19 @@ func (a *App) load(prefix string, spec Plugin) error {
 		return fmt.Errorf("zip: Load(%s) needs a Plugin.Name", prefix)
 	}
 	if spec.Addr != "" {
-		return a.Mount(prefix, spec.Addr)
+		if err := a.Mount(prefix, spec.Addr); err != nil {
+			return err
+		}
+		// Recorded even though this host did not start it, so Plugins() reports
+		// the whole surface a request can reach rather than only the children.
+		// A fleet view that silently omits remote mounts is worse than none.
+		a.plugMu.Lock()
+		if a.plugins == nil {
+			a.plugins = map[string]*plugin{}
+		}
+		a.plugins[spec.Name] = &plugin{name: spec.Name, prefix: prefix, spec: spec}
+		a.plugMu.Unlock()
+		return nil
 	}
 	if len(spec.Bin) == 0 && spec.Path == "" && spec.URL == "" {
 		return fmt.Errorf("zip: Load(%s): plugin %q has no Addr, Bin, Path, or URL", prefix, spec.Name)
@@ -216,6 +249,9 @@ func (a *App) load(prefix string, spec Plugin) error {
 		if cur := p.cur.Swap(nil); cur != nil {
 			stop(cur, 0)
 		}
+		// Wait out any instance a Reload or Unload is still draining, so no
+		// child survives this host.
+		p.stopping.Wait()
 		return nil
 	})
 	return nil
@@ -254,6 +290,7 @@ func (a *App) Reload(name string, bin []byte) error {

 	prev := p.cur.Swap(next) // every request from here uses next
 	p.spec = spec
+	p.reloads.Add(1)
 	a.logger.Info("zip reloaded plugin", "name", name,
 		"pid", next.cmd.Process.Pid, "addr", next.sock)

@@ -262,7 +299,7 @@ func (a *App) Reload(name string, bin []byte) error {
 		if drain <= 0 {
 			drain = 5 * time.Second
 		}
-		go stop(prev, drain)
+		p.retire(prev, drain)
 	}
 	return nil
 }
@@ -279,9 +316,7 @@ func (a *App) Unload(name string) error {
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if prev := p.cur.Swap(nil); prev != nil {
-		go stop(prev, 0)
-	}
+	p.retire(p.cur.Swap(nil), 0)
 	return nil
 }

@@ -331,7 +366,7 @@ func start(spec Plugin) (*instance, error) {
 	exited := make(chan error, 1)
 	go func() { exited <- cmd.Wait() }()

-	in := &instance{cmd: cmd, dir: dir, sock: sock, exited: exited}
+	in := &instance{cmd: cmd, dir: dir, sock: sock, exited: exited, started: time.Now()}
 	if err := waitListening(sock, spec.Start, exited); err != nil {
 		stop(in, 0)
 		return nil, err
@@ -243,3 +243,74 @@ func TestLoad_ReleaseRequiresSum(t *testing.T) {
 		t.Fatalf("err = %v, want a refusal naming the unverified download", err)
 	}
 }
+
+// TestPlugins_Status proves a host can report what it is actually running —
+// the primitive a fleet view needs, since deployment config says what was
+// INTENDED and only the process knows what is TRUE.
+func TestPlugins_Status(t *testing.T) {
+	v1, v2 := buildPlugin(t, "v1"), buildPlugin(t, "v2")
+	sum := sha256.Sum256(v1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write(v1)
+	}))
+	defer srv.Close()
+
+	cache := t.TempDir()
+	app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
+	if err := app.Add(
+		zip.Load("/v1/embedded", zip.Plugin{Name: "embedded", Bin: v1, Dir: cache}),
+		zip.Load("/v1/installed", zip.Plugin{
+			Name: "installed", URL: srv.URL + "/x", Sum: hex.EncodeToString(sum[:]), Dir: cache,
+		}),
+	); err != nil {
+		t.Fatalf("Add: %v", err)
+	}
+	defer func() { _ = app.Shutdown() }()
+
+	got := app.Plugins()
+	if len(got) != 2 {
+		t.Fatalf("Plugins() returned %d, want 2", len(got))
+	}
+	// Sorted by name, so this order is stable and a diff between hosts is too.
+	if got[0].Name != "embedded" || got[1].Name != "installed" {
+		t.Fatalf("not sorted by name: %v", []string{got[0].Name, got[1].Name})
+	}
+	if got[0].Source != "embedded" || got[1].Source != "url" {
+		t.Fatalf("sources = %q/%q, want embedded/url", got[0].Source, got[1].Source)
+	}
+	// The digest IS the version — it cannot drift from the bits running.
+	if got[1].Version != hex.EncodeToString(sum[:]) {
+		t.Fatalf("installed Version = %q, want the artifact digest", got[1].Version)
+	}
+	for _, p := range got {
+		if !p.Running || p.PID == 0 || p.Since.IsZero() {
+			t.Fatalf("%s: running=%v pid=%d since=%v — want a live instance", p.Name, p.Running, p.PID, p.Since)
+		}
+		if p.Reloads != 0 {
+			t.Fatalf("%s: Reloads=%d before any reload", p.Name, p.Reloads)
+		}
+	}
+
+	// A reload must be visible: the counter climbs and Since resets.
+	before := got[0].Since
+	if err := app.Reload("embedded", v2); err != nil {
+		t.Fatalf("Reload: %v", err)
+	}
+	after := app.Plugins()[0]
+	if after.Reloads != 1 {
+		t.Fatalf("Reloads = %d after one reload, want 1", after.Reloads)
+	}
+	if !after.Since.After(before) {
+		t.Fatalf("Since did not advance on reload (%v -> %v)", before, after.Since)
+	}
+
+	// Unload must read as "deployed but down", not as absent — its routes are
+	// still registered and answering 503.
+	if err := app.Unload("embedded"); err != nil {
+		t.Fatalf("Unload: %v", err)
+	}
+	down := app.Plugins()[0]
+	if down.Running || down.Name != "embedded" {
+		t.Fatalf("after Unload: %+v — want the plugin still listed, Running=false", down)
+	}
+}
@@ -29,7 +29,7 @@ func TestMCP_FreeToolSurface(t *testing.T) {
 		return &greetOut{Message: "hello " + in.Name}, nil
 	}, zip.WithOperationID("greet"), zip.WithSummary("Greet someone by name"))

-	const httpAddr = "127.0.0.1:18099"
+	httpAddr := freeAddr(t)
 	go func() { _ = app.Listen("http://" + httpAddr) }()
 	defer func() { _ = app.Shutdown() }()
 	waitHTTP(t, "http://"+httpAddr+"/.well-known/openapi.json")
@@ -70,7 +70,7 @@ func TestMount_OverZAP(t *testing.T) {
 		return c.JSON(201, map[string]string{"echo": string(c.Body())})
 	})

-	const pluginAddr = "127.0.0.1:19661"
+	pluginAddr := freeAddr(t)
 	go func() { _ = plugin.Listen(pluginAddr) }() // bare addr = ZAP
 	defer func() { _ = plugin.Shutdown() }()
 	waitReachable(t, pluginAddr)
@@ -110,7 +110,7 @@ func TestMount_StaticBeatsRemoteMount(t *testing.T) {
 		return c.JSON(200, map[string]string{"servedBy": "plugin"})
 	})

-	const pluginAddr = "127.0.0.1:19662"
+	pluginAddr := freeAddr(t)
 	go func() { _ = plugin.Listen(pluginAddr) }()
 	defer func() { _ = plugin.Shutdown() }()
 	waitReachable(t, pluginAddr)
@@ -0,0 +1,26 @@
+package zip_test
+
+import (
+	"net"
+	"testing"
+)
+
+// freeAddr returns a loopback address nothing is listening on.
+//
+// These tests used fixed ports, which is fine in isolation and wrong on a
+// shared machine: an unrelated process holding 19653 makes TestListen_ZAP fail
+// with a 404 rather than a connection error, because the port answers — it is
+// just answering for someone else. That reads as a routing bug and is not one.
+//
+// Binding :0 and releasing has a race window, but it is bounded by the next
+// line of the test rather than by whatever else lives on the box.
+func freeAddr(t *testing.T) string {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("reserve a port: %v", err)
+	}
+	addr := l.Addr().String()
+	_ = l.Close()
+	return addr
+}
@@ -0,0 +1,95 @@
+package zip
+
+import (
+	"sort"
+	"time"
+)
+
+// What a host is running. A fleet view needs one thing this package did not
+// expose: the ability to ask a live process what it actually loaded, at which
+// version, and whether it is still up. Reading it from deployment config
+// answers what was INTENDED; only the process knows what is TRUE — which is the
+// difference that matters during a rolling upgrade, when the two disagree by
+// design.
+
+// PluginStatus is one loaded plugin as the host currently sees it.
+type PluginStatus struct {
+	Name   string `json:"name"`
+	Prefix string `json:"prefix"`
+
+	// Source is where the binary came from: "embedded", "path", "url", or
+	// "remote" for an instance this host did not start.
+	Source string `json:"source"`
+
+	// Version is the artifact's SHA-256 when it was installed from a URL —
+	// the only version identifier that cannot drift from the bits actually
+	// running, since it IS the bits. Empty for the other sources.
+	Version string `json:"version,omitempty"`
+
+	// Addr is the socket or address serving it.
+	Addr string `json:"addr,omitempty"`
+
+	// PID is the child process, or 0 when this host did not start it.
+	PID int `json:"pid,omitempty"`
+
+	// Running is false after Unload, or after a child exited and no Reload has
+	// replaced it. Its routes stay registered and answer 503, so a false here
+	// is the difference between "not deployed" and "deployed but down".
+	Running bool `json:"running"`
+
+	// Since is when the CURRENT instance started — it resets on Reload, so it
+	// reports the age of what is running, not of the mount.
+	Since time.Time `json:"since,omitzero"`
+
+	// 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"`
+}
+
+// Plugins reports every plugin this host has loaded, ordered by name so a diff
+// between two hosts is stable. Safe to call while requests are in flight.
+func (a *App) Plugins() []PluginStatus {
+	a.plugMu.Lock()
+	ps := make([]*plugin, 0, len(a.plugins))
+	for _, p := range a.plugins {
+		ps = append(ps, p)
+	}
+	a.plugMu.Unlock()
+
+	out := make([]PluginStatus, 0, len(ps))
+	for _, p := range ps {
+		s := PluginStatus{
+			Name:    p.name,
+			Prefix:  p.prefix,
+			Source:  p.source(),
+			Version: p.spec.Sum,
+			Reloads: int(p.reloads.Load()),
+		}
+		if in := p.cur.Load(); in != nil {
+			s.Running = true
+			s.Addr = in.sock
+			s.Since = in.started
+			if in.cmd != nil && in.cmd.Process != nil {
+				s.PID = in.cmd.Process.Pid
+			}
+		}
+		out = append(out, s)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
+	return out
+}
+
+// source names where this plugin's binary came from.
+func (p *plugin) source() string {
+	switch {
+	case p.spec.Addr != "":
+		return "remote"
+	case p.spec.URL != "":
+		return "url"
+	case len(p.spec.Bin) > 0:
+		return "embedded"
+	case p.spec.Path != "":
+		return "path"
+	}
+	return "unknown"
+}
@@ -32,7 +32,7 @@ func TestListenZAP_Streams(t *testing.T) {
 		})
 	})

-	const addr = "127.0.0.1:19655"
+	addr := freeAddr(t)
 	go func() { _ = app.Listen(addr) }() // bare addr = ZAP
 	defer func() { _ = app.Shutdown() }()

@@ -24,7 +24,7 @@ func TestListen_ZAP(t *testing.T) {
 		return c.JSON(200, map[string]string{"status": "ok", "transport": "zap"})
 	})

-	const addr = "127.0.0.1:19653"
+	addr := freeAddr(t)
 	go func() { _ = app.Listen(addr) }() // bare addr = ZAP (DefaultScheme)
 	defer func() { _ = app.Shutdown() }()

@@ -62,8 +62,8 @@ func TestListen_DualTransport(t *testing.T) {
 		return c.JSON(200, map[string]string{"status": "ok"})
 	})

-	const zapAddr = "127.0.0.1:19654"
-	const httpAddr = "127.0.0.1:18080"
+	zapAddr := freeAddr(t)
+	httpAddr := freeAddr(t)
 	go func() { _ = app.Listen(zapAddr, "http://"+httpAddr) }() // ONE call, both transports
 	defer func() { _ = app.Shutdown() }()
 	waitReachable(t, zapAddr)
@@ -110,7 +110,7 @@ func TestHTTPTransport_ReadBufferSize_Raises431Ceiling(t *testing.T) {
 	// Control: no ReadBufferSize -> fasthttp's 4 KiB default -> 431.
 	ctrl := zip.New(zip.Config{AppName: "ctrl", DisableStartupMessage: true})
 	ctrl.Get("/v1/health", func(c *zip.Ctx) error { return c.JSON(200, map[string]string{"ok": "1"}) })
-	const ctrlAddr = "127.0.0.1:19701"
+	ctrlAddr := freeAddr(t)
 	go func() { _ = ctrl.Listen("http://" + ctrlAddr) }()
 	defer func() { _ = ctrl.Shutdown() }()
 	waitDialable(t, ctrlAddr)
@@ -121,7 +121,7 @@ func TestHTTPTransport_ReadBufferSize_Raises431Ceiling(t *testing.T) {
 	// Fixed: ReadBufferSize 32 KiB -> the SAME 9 KiB header is accepted.
 	fixed := zip.New(zip.Config{AppName: "fixed", DisableStartupMessage: true, ReadBufferSize: 32768})
 	fixed.Get("/v1/health", func(c *zip.Ctx) error { return c.JSON(200, map[string]string{"ok": "1"}) })
-	const fixedAddr = "127.0.0.1:19702"
+	fixedAddr := freeAddr(t)
 	go func() { _ = fixed.Listen("http://" + fixedAddr) }()
 	defer func() { _ = fixed.Shutdown() }()
 	waitDialable(t, fixedAddr)
@@ -145,7 +145,7 @@ func TestHTTPTransport_ServerHeaderCoversPreRoutingErrors(t *testing.T) {
 	// Branded: the pre-routing 431 carries Server: <brand>, never the framework.
 	brand := zip.New(zip.Config{AppName: "brand", DisableStartupMessage: true, ServerHeader: "hanzo"})
 	brand.Get("/v1/health", func(c *zip.Ctx) error { return c.JSON(200, map[string]string{"ok": "1"}) })
-	const brandAddr = "127.0.0.1:19703"
+	brandAddr := freeAddr(t)
 	go func() { _ = brand.Listen("http://" + brandAddr) }()
 	defer func() { _ = brand.Shutdown() }()
 	waitDialable(t, brandAddr)
@@ -166,7 +166,7 @@ func TestHTTPTransport_ServerHeaderCoversPreRoutingErrors(t *testing.T) {
 	// Suppressed: ServerHeader "-" emits NO Server header on the pre-routing error.
 	quiet := zip.New(zip.Config{AppName: "quiet", DisableStartupMessage: true, ServerHeader: "-"})
 	quiet.Get("/v1/health", func(c *zip.Ctx) error { return c.JSON(200, map[string]string{"ok": "1"}) })
-	const quietAddr = "127.0.0.1:19704"
+	quietAddr := freeAddr(t)
 	go func() { _ = quiet.Listen("http://" + quietAddr) }()
 	defer func() { _ = quiet.Shutdown() }()
 	waitDialable(t, quietAddr)