fix(host): release every hold, keep a cold burst under the ceiling, decode lists in one pass; retract v1.37.31

.gitignore

@@ -8,3 +8,4 @@ rust/examples/*/target/
 CLAUDE.md
 AGENTS.md
 GEMINI.md
+/testplugin

LLM.md

@@ -54,8 +54,8 @@ Patch tags, one above the highest `v*` on GitHub. Push the commit to main, read
 - `hasBody(method)` is the one rule for which methods carry a JSON body. GET, HEAD and DELETE do not.
 - A refusal is written by the error handler from the route the request matched: an RFC 9457 problem document (`application/problem+json`), or `{error, error_description}` at addresses declared with `OAuth`. `composeOAuth` builds that address set; `buildOpenAPI` reads the same set to pick each operation's `default` response (`problem-details` or `oauth-error`). Schemas in the document are objects, never booleans: `CommandsFromSpec` and other readers parse them as objects.
 - The call-plane body is ZAP, laid out from struct field order (`internal/zapenc`). Append fields at the end. Reordering, inserting or retyping a field changes the wire.
-- A pointer to a scalar, string or `[]byte` crosses as `list<T>` of at most one element: nil is the empty list, a pointer to zero is one zero element. A pointer to a struct is its message in a `bytes` slot, present when the slot is not null. `zapenc`, the `Layouts` emitter and the Rust projector (`TypeRef.Opt`) all follow this.
-- A plugin request holds its instance (`hold`, `instance.busy`) until the reply is written, a streamed body is closed, or an upgrade's tunnel ends. `evict` raises `closing` before it reads `busy`; `claim` raises `busy` before it reads `closing`. Eviction never stops a busy instance, ranks idle ones by when their last request ended, and `makeRoom` waits up to the starter's `Plugin.Start` for an idle one, then answers 503.
+- A pointer to a scalar, string or `[]byte` crosses as `list<T>` of at most one element: nil is the empty list, a pointer to zero is one zero element. A pointer to a struct is its message in a `bytes` slot, present when the slot is not null. `zapenc`, the `Layouts` emitter and the Rust projector (`TypeRef.Opt`) all follow this. A `*[N]byte` stays inline, so a pointer to an all-zero array reads back nil. Lists decode in one pass (`List.EachBytes`) and grow from the entries present, never the claimed count.
+- A plugin request holds its instance (`hold`, `instance.busy`) until the reply is written, a streamed body is closed (zap-proto/http closes it when the connection ends, v0.3.12+), or an upgrade's tunnel ends; an upgrade over ZAP is refused 501 because ZAP's server never runs a hijack, and a declined upgrade's reply is read as a reply, bounded by `declineWait`. `evict` raises `closing` before it reads `busy`; `claim` raises `busy` before it reads `closing`. Eviction never stops a busy instance and takes idle ones coldest first by when their last request ended, skipping one whose lock is held. A starter waits for room until its own deadline (`Plugin.Start` from when it arrived), then 503; the ceiling is exceeded only when no process may be stopped and no start is in flight. `lazy`/`idle`/`start` are copied off the spec at Load because the request path reads them unlocked.
 - Doc prose is filed under `DocKey(pkg, method, path)`. zipdoc writes the package's import path. `pkgOf` reads the handler's package from its runtime name and maps `main` to the build info path, because a built binary names package main's functions `main.` while `go test` names them by import path.
 - MCP over HTTP: POST only (GET and DELETE answer 405), one message per request, no sessions, no batches, notifications 202. `negotiate` answers `initialize` in the requested revision when it is 2025-06-18, 2025-11-25 or 2026-07-28, else in 2026-07-28. The official TypeScript (1.30.0) and Python (2.2.0) clients refuse an answer in a revision they did not request and do not know.
 - `AdaptNetHTTP` carries an upgrade: a request with `Connection: Upgrade` runs the handler inside fasthttp's hijack, on a writer that is an `http.Hijacker`, so a net/http WebSocket server (coder/websocket, gorilla) writes its own 101 and owns the socket; a handler that declines answers an ordinary HTTP/1.1 reply. `wsx` stays the native way for a zip handler.

go.mod

@@ -14,6 +14,10 @@ go 1.27.1
 // it is told rather than silently moved. v1.36.22 is the re-cut.
 retract v1.36.21
 
+// Carries a 14.7MB test binary at the module root, and a plugin request over
+// ZAP or a declined upgrade could hold its process for good.
+retract v1.37.31
+
 require (
 	github.com/dop251/goja v0.0.0-20260607120635-348e6bea910d
 	github.com/evanw/esbuild v0.28.1
@@ -24,7 +28,7 @@ require (
 	github.com/valyala/fasthttp v1.70.0
 	github.com/zap-proto/fiber/v3 v3.2.1
 	github.com/zap-proto/go v1.8.3
-	github.com/zap-proto/http v0.3.11
+	github.com/zap-proto/http v0.3.12
 	github.com/zap-proto/mcp v1.0.5
 	golang.org/x/sys v0.45.0
 	golang.org/x/tools v0.45.0

go.sum

@@ -87,8 +87,8 @@ github.com/zap-proto/fiber/v3 v3.2.1 h1:k45oKyTwySPtGt8sPz2Ao8OUHc7pDEhai8Np2Ym6
 github.com/zap-proto/fiber/v3 v3.2.1/go.mod h1:eDm2z+ufJrkuE4MeX0Mea4oc/p7/HpXiZjVB+BXCKOA=
 github.com/zap-proto/go v1.8.3 h1:GWhb8y7ylgVcaVOR8XN/It56Xovm5Mkv+nQqiblnnoA=
 github.com/zap-proto/go v1.8.3/go.mod h1:fbNeWcssRRbTkQEDCrmUmzqJA1gKcAlgi1z4QrtEFZg=
-github.com/zap-proto/http v0.3.11 h1:LdaTSQtxcHRooQMwz0bZGjw+7IuONEwCG3QtLH/fHr0=
-github.com/zap-proto/http v0.3.11/go.mod h1:Gw4Kn/PpF1P+CHJyBQQ3aoFDI3P4p0wsjURFwjPnHzE=
+github.com/zap-proto/http v0.3.12 h1:74EvTeTZxikxctOLKno3UvGWVv3hiaE35UleAU5Nrc0=
+github.com/zap-proto/http v0.3.12/go.mod h1:Gw4Kn/PpF1P+CHJyBQQ3aoFDI3P4p0wsjURFwjPnHzE=
 github.com/zap-proto/mcp v1.0.5 h1:1Hi0KNi4YlWW2zkQUMDK9n3jP3wZaRJ1rseMW9p9F0c=
 github.com/zap-proto/mcp v1.0.5/go.mod h1:cYOsq+33NyW4X8nUVtIs/qtezVovYazw1s0V7mrQBYA=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

hold_internal_test.go

@@ -0,0 +1,353 @@
+package zip
+
+import (
+	enc "encoding/binary"
+	"io"
+	"net"
+	nethttp "net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/valyala/fasthttp"
+	zaphttp "github.com/zap-proto/http"
+)
+
+// A hold is released on every way a request can end, and the ceiling admits
+// over itself only when nothing could ever make room. Each test here is a path
+// that left a plugin held for good, or let a burst past the ceiling.
+
+var (
+	holdBinOnce sync.Once
+	holdBin     []byte
+	holdBinErr  error
+)
+
+func holdPlugin(t *testing.T) []byte {
+	t.Helper()
+	holdBinOnce.Do(func() {
+		d, err := os.MkdirTemp("", "hold")
+		if err != nil {
+			holdBinErr = err
+			return
+		}
+		defer os.RemoveAll(d)
+		out := filepath.Join(d, "tp")
+		cmd := exec.Command("go", "build", "-o", out, "./internal/testplugin")
+		cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
+		if b, err := cmd.CombinedOutput(); err != nil {
+			holdBinErr = err
+			t.Logf("%s", b)
+			return
+		}
+		holdBin, holdBinErr = os.ReadFile(out)
+	})
+	if holdBinErr != nil {
+		t.Skipf("cannot build test plugin: %v", holdBinErr)
+	}
+	return holdBin
+}
+
+func holdHost(t *testing.T, warm int, specs ...Plugin) *App {
+	t.Helper()
+	app := New(Config{AppName: "host", DisableStartupMessage: true, Warm: warm})
+	for _, p := range specs {
+		l, err := Load(p, "/v1/"+p.Name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		app.Use(l)
+	}
+	t.Cleanup(func() { _ = app.Shutdown() })
+	return app
+}
+
+func lazy(t *testing.T, name string, idle time.Duration) Plugin {
+	return Plugin{Name: name, Bin: holdPlugin(t), Dir: sockDir(t), Lazy: true, IdleAfter: idle}
+}
+
+func busyOf(a *App, name string) int64 {
+	_, p := a.pluginNamed(name)
+	if in := p.cur.Load(); in != nil {
+		return in.busy.Load()
+	}
+	return -1
+}
+
+func running(a *App) int {
+	n := 0
+	for _, p := range a.pluginSet() {
+		if p.cur.Load() != nil {
+			n++
+		}
+	}
+	return n
+}
+
+func get(t *testing.T, a *App, target string, hdr ...string) (int, string) {
+	t.Helper()
+	req, _ := nethttp.NewRequest("GET", target, nil)
+	for i := 0; i+1 < len(hdr); i += 2 {
+		req.Header.Set(hdr[i], hdr[i+1])
+	}
+	resp, err := a.Test(req, TestConfig{Timeout: 30 * time.Second, FailOnTimeout: true})
+	if err != nil {
+		t.Errorf("GET %s: %v", target, err) // Errorf: callers run it off the test goroutine
+		return 0, ""
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(resp.Body)
+	return resp.StatusCode, string(b)
+}
+
+func serveZAP(t *testing.T, a *App) string {
+	t.Helper()
+	sock := filepath.Join(sockDir(t), "h.sock")
+	go func() { _ = a.Listen(sock) }()
+	for range 300 {
+		if c, err := net.Dial("unix", sock); err == nil {
+			_ = c.Close()
+			return sock
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("host never listened")
+	return ""
+}
+
+func settle(limit time.Duration, ok func() bool) bool {
+	for end := time.Now().Add(limit); time.Now().Before(end); time.Sleep(10 * time.Millisecond) {
+		if ok() {
+			return true
+		}
+	}
+	return ok()
+}
+
+// A ZAP caller that leaves before a streamed head is written releases its hold:
+// the connection ending closes the body, and the body carries the hold.
+func TestHold_ACallerWhoLeavesOverZAPReleases(t *testing.T) {
+	app := holdHost(t, 1, lazy(t, "demo0", 100*time.Millisecond))
+	sock := serveZAP(t, app)
+
+	var req fasthttp.Request
+	req.Header.SetMethod("GET")
+	req.SetRequestURI("/v1/demo0/feed?drip=200ms")
+	buf, err := zaphttp.AppendRequest([]byte{0, 0, 0, 0}, &req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	enc.BigEndian.PutUint32(buf, uint32(len(buf)-4))
+	c, err := net.Dial("unix", sock)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.Write(buf); err != nil {
+		t.Fatal(err)
+	}
+	_ = c.Close() // gone before the plugin even starts
+
+	if !settle(5*time.Second, func() bool { return busyOf(app, "demo0") >= 0 }) {
+		t.Fatal("demo0 never started for the request")
+	}
+	if !settle(5*time.Second, func() bool { return busyOf(app, "demo0") == 0 }) {
+		t.Fatalf("demo0 still holds %d requests after its only caller left", busyOf(app, "demo0"))
+	}
+}
+
+// An upgrade cannot cross ZAP, whose server never runs a hijack. It is refused,
+// and the hold it took is released.
+func TestHold_AnUpgradeOverZAPIsRefused(t *testing.T) {
+	app := holdHost(t, 1, lazy(t, "demo0", time.Hour))
+	sock := serveZAP(t, app)
+
+	tr := zaphttp.Dial("unix", sock)
+	defer tr.CloseIdleConnections()
+	var req fasthttp.Request
+	var resp fasthttp.Response
+	req.SetRequestURI("/v1/demo0/socket")
+	req.Header.SetHost("box")
+	req.Header.Set("Connection", "Upgrade")
+	req.Header.Set("Upgrade", "websocket")
+	if err := tr.Do(&req, &resp); err != nil {
+		t.Fatalf("do: %v", err)
+	}
+	if resp.StatusCode() != 501 {
+		t.Fatalf("status %d, want 501 for an upgrade over ZAP", resp.StatusCode())
+	}
+	if b := busyOf(app, "demo0"); b > 0 {
+		t.Fatalf("demo0 holds %d requests after the refusal", b)
+	}
+}
+
+// A plugin that declines an upgrade with a chunked reply is answered once the
+// reply ends, and the hold is released. It used to read "until the connection
+// closes", which a keep-alive plugin never does.
+func TestHold_ADeclinedUpgradeWithAChunkedReplyEnds(t *testing.T) {
+	app := holdHost(t, 1, lazy(t, "demo0", time.Hour))
+
+	done := make(chan struct{})
+	var status int
+	var body string
+	go func() {
+		defer close(done)
+		status, body = get(t, app, "/v1/demo0/feed?drip=300ms", "Connection", "Upgrade", "Upgrade", "websocket")
+	}()
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatalf("the declined upgrade never answered; demo0 holds %d", busyOf(app, "demo0"))
+	}
+	if status != 200 || !strings.HasSuffix(body, "done\n") {
+		t.Fatalf("status %d body %q, want the plugin's whole reply", status, body)
+	}
+	if b := busyOf(app, "demo0"); b != 0 {
+		t.Fatalf("demo0 holds %d after the reply ended", b)
+	}
+}
+
+// A cold burst stays under the ceiling. Starts in flight are neither idle nor
+// busy, and they used to read as "nothing may be stopped", which admitted every
+// one of them over the ceiling.
+func TestHold_AColdBurstStaysUnderTheCeiling(t *testing.T) {
+	const warm, n = 2, 8
+	var specs []Plugin
+	for i := range n {
+		specs = append(specs, lazy(t, "demo"+string(rune('0'+i)), time.Hour))
+	}
+	app := holdHost(t, warm, specs...)
+
+	stop := make(chan struct{})
+	peak := 0
+	var mu sync.Mutex
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			if r := running(app); r > peak {
+				mu.Lock()
+				peak = r
+				mu.Unlock()
+			}
+			time.Sleep(time.Millisecond)
+		}
+	}()
+	var wg sync.WaitGroup
+	codes := make([]int, n)
+	for i := range n {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			codes[i], _ = get(t, app, "/v1/demo"+string(rune('0'+i))+"/version")
+		}(i)
+	}
+	wg.Wait()
+	close(stop)
+	for i, c := range codes {
+		if c != 200 {
+			t.Errorf("demo%d answered %d", i, c)
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if peak > warm {
+		t.Fatalf("peak %d processes against a ceiling of %d", peak, warm)
+	}
+}
+
+// When the coldest candidate cannot be stopped — its lock is held mid-Reload —
+// room is taken from the next one rather than refused.
+func TestHold_ALockedCandidateDoesNotBlockRoom(t *testing.T) {
+	app := holdHost(t, 2,
+		lazy(t, "demo0", time.Hour), lazy(t, "demo1", time.Hour), lazy(t, "demo2", time.Hour))
+	for _, n := range []string{"demo0", "demo1"} {
+		if c, _ := get(t, app, "/v1/"+n+"/version"); c != 200 {
+			t.Fatalf("%s: %d", n, c)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	_, cold := app.pluginNamed("demo0")
+	cold.mu.Lock() // demo0, the coldest, is mid-transition
+	defer cold.mu.Unlock()
+
+	if c, _ := get(t, app, "/v1/demo2/version"); c != 200 {
+		t.Fatalf("demo2 answered %d with demo1 idle and stoppable", c)
+	}
+	_, warmer := app.pluginNamed("demo1")
+	if warmer.cur.Load() != nil {
+		t.Fatal("demo1 is still running; room came from nowhere")
+	}
+}
+
+// Callers queued on one starter each wait their own bound, not one after
+// another's: the wait is counted from when the caller arrived.
+func TestHold_QueuedStartersWaitTogether(t *testing.T) {
+	slow := lazy(t, "demo1", time.Hour)
+	slow.Start = 400 * time.Millisecond
+	app := holdHost(t, 1, lazy(t, "demo0", time.Hour), slow)
+
+	go func() { _, _ = get(t, app, "/v1/demo0/version?sleep=4s") }()
+	if !settle(5*time.Second, func() bool { return busyOf(app, "demo0") == 1 }) {
+		t.Fatal("demo0 never became busy")
+	}
+	const callers = 5
+	var wg sync.WaitGroup
+	took := make([]time.Duration, callers)
+	for i := range callers {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			start := time.Now()
+			if c, _ := get(t, app, "/v1/demo1/version"); c != 503 {
+				t.Errorf("caller %d: %d, want 503 while the only process is busy", i, c)
+			}
+			took[i] = time.Since(start)
+		}(i)
+	}
+	wg.Wait()
+	for i, d := range took {
+		if d > 1500*time.Millisecond {
+			t.Errorf("caller %d waited %s against its own 400ms bound", i, d)
+		}
+	}
+}
+
+// Reload rewrites spec under p.mu while requests read the plugin unlocked. Run
+// with -race: the detector is the assertion.
+func TestHold_ReloadRacesNoRequest(t *testing.T) {
+	app := holdHost(t, 1, lazy(t, "demo0", time.Hour))
+	if c, _ := get(t, app, "/v1/demo0/version"); c != 200 {
+		t.Fatalf("demo0: %d", c)
+	}
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	for range 4 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				_, _ = get(t, app, "/v1/demo0/version")
+				app.Evict()
+			}
+		}()
+	}
+	for range 3 {
+		if err := app.Reload("demo0", Plugin{}); err != nil {
+			t.Fatalf("reload: %v", err)
+		}
+	}
+	close(stop)
+	wg.Wait()
+}

idle.go

@@ -30,14 +30,6 @@ import (
 // that CAS always fails and no restart storm follows. It is the same contract
 // Unload and Reload use, minus the disabled flag.
 
-// IdleAfter is how long a lazy plugin may go unused before its process is
-// stopped. Zero means never — the historical behaviour, and the right setting
-// for a plugin whose start is expensive or whose first request must not pay a
-// cold start (an identity or config service every other call goes through).
-//
-// It applies only to Lazy plugins: an eager one was started deliberately at
-// Load and stopping it would contradict that.
-func (p Plugin) idleAfter() time.Duration { return p.IdleAfter }
 
 // Evict reclaims plugin processes under two bounds and reports how many it
 // stopped: first every lazy plugin that has not served for its IdleAfter, then,
@@ -103,31 +95,15 @@ func (p *plugin) host() *App {
 // answers 503 and nothing in flight is killed.
 //
 // On success the starter holds one unit of room until it calls started.
-func (a *App) makeRoom(starter *plugin) error {
+func (a *App) makeRoom(starter *plugin, deadline time.Time) error {
 	if a.warm <= 0 {
 		return nil
 	}
-	deadline := time.Now().Add(starter.spec.startLimit())
 	for {
-		a.roomMu.Lock()
-		live, idle, busy := a.census(starter)
-		if live+a.starting < a.warm || idle == nil && busy == 0 {
-			// Under the ceiling, or over it on processes nothing may stop: a host
-			// is better over its budget than deprived of an identity or config
-			// service every other call goes through.
-			a.starting++
-			a.roomMu.Unlock()
+		room, live, busy := a.room(starter)
+		if room {
 			return nil
 		}
-		if idle != nil {
-			seen := idle.lastUse.Load()
-			if idle.evict("room", time.Since(time.Unix(0, seen)), seen) {
-				a.starting++
-				a.roomMu.Unlock()
-				return nil
-			}
-		}
-		a.roomMu.Unlock()
 		if time.Now().After(deadline) {
 			return fmt.Errorf("%d of %d processes running, %d busy and none idle", live, a.warm, busy)
 		}
@@ -135,6 +111,33 @@ func (a *App) makeRoom(starter *plugin) error {
 	}
 }
 
+// room is one look for room under roomMu, and reports whether it took one.
+//
+// Over the ceiling is admitted only when nothing could ever make room: no
+// process this host may stop, busy or idle, and no start in flight, which
+// becomes such a process once it runs. Admitting while starts are in flight let
+// a cold burst of eight first requests start eight processes against a ceiling
+// of two.
+func (a *App) room(starter *plugin) (ok bool, live, busy int) {
+	a.roomMu.Lock()
+	defer a.roomMu.Unlock()
+	live, idle, busy := a.census(starter)
+	if live+a.starting < a.warm || len(idle) == 0 && busy == 0 && a.starting == 0 {
+		a.starting++
+		return true, live, busy
+	}
+	// Coldest first, and on to the next when one cannot be stopped: a plugin
+	// mid-Reload holds its lock, and the room behind it is still room.
+	for _, p := range idle {
+		seen := p.lastUse.Load()
+		if p.evict("room", time.Since(time.Unix(0, seen)), seen) {
+			a.starting++
+			return true, live, busy
+		}
+	}
+	return false, live, busy
+}
+
 // roomPoll is how often a starter waiting for room looks again. A request on the
 // process it is waiting for takes far longer than this to finish.
 const roomPoll = 10 * time.Millisecond
@@ -151,9 +154,9 @@ func (a *App) started() {
 }
 
 // census counts the running processes, and among the ones this host may stop
-// returns the coldest idle one and how many are busy. The starter is never a
-// candidate.
-func (a *App) census(starter *plugin) (live int, coldest *plugin, busy int) {
+// returns the idle ones coldest first and how many are busy. The starter is
+// never a candidate.
+func (a *App) census(starter *plugin) (live int, idle []*plugin, busy int) {
 	for _, p := range a.pluginSet() {
 		in := p.cur.Load()
 		if in == nil {
@@ -167,16 +170,17 @@ func (a *App) census(starter *plugin) (live int, coldest *plugin, busy int) {
 			busy++
 			continue
 		}
-		if coldest == nil || p.lastUse.Load() < coldest.lastUse.Load() {
-			coldest = p
-		}
+		idle = append(idle, p)
 	}
-	return live, coldest, busy
+	slices.SortFunc(idle, func(x, y *plugin) int {
+		return cmp.Compare(x.lastUse.Load(), y.lastUse.Load())
+	})
+	return live, idle, busy
 }
 
 // evictable is whether the ceiling and the age bound may stop p at all: only a
 // lazy plugin with an idle bound, because the next request has to bring it back.
-func (p *plugin) evictable() bool { return p.spec.Lazy && p.spec.idleAfter() > 0 }
+func (p *plugin) evictable() bool { return p.lazy && p.idle > 0 }
 
 // pluginSet is every plugin this host and its composed hosts hold, read once
 // under each host's lock. Both passes need the same set and neither may hold a
@@ -256,8 +260,8 @@ func (a *App) evictOver(warm int, now time.Time) int {
 // evictIfIdle stops p's current instance if it is lazy, running, and has been
 // unused for at least its IdleAfter. It reports whether it stopped one.
 func (p *plugin) evictIfIdle(now time.Time) bool {
-	after := p.spec.idleAfter()
-	if after <= 0 || !p.spec.Lazy {
+	after := p.idle
+	if after <= 0 || !p.lazy {
 		return false
 	}
 	// No lock: the swap that takes the child down happens under p.mu inside
@@ -329,6 +333,7 @@ func (p *plugin) evict(reason string, idle time.Duration, seen int64) bool {
 		p.mu.Unlock()
 		return false // it exited on its own; the supervisor has it
 	}
+	drain := p.spec.Drain // read under the lock Reload writes spec under
 	p.mu.Unlock()
 
 	p.evictions.Add(1)
@@ -337,7 +342,7 @@ func (p *plugin) evict(reason string, idle time.Duration, seen int64) bool {
 			"name", p.name, "pid", in.cmd.Process.Pid, "reason", reason,
 			"idle", idle.Round(time.Second).String())
 	}
-	p.retire(in, p.spec.Drain)
+	p.retire(in, drain)
 	return true
 }
 

internal/layout/zap_gen.go

@@ -72,10 +72,13 @@ func (x *Ided) UnmarshalZAP(data []byte) error {
 	copy(x.ID[:], o.BytesFixed(idedIDAt, 32))
 	x.Name = string(strings.Clone(o.Text(idedNameAt)))
 	if l := o.List(idedIDsAt); l.Len() > 0 {
-		rows := make([][32]uint8, l.Len())
-		for i := range rows {
-			copy(rows[i][:], l.BytesAt(i))
-		}
+		var rows [][32]uint8
+		l.EachBytes(func(_ int, raw []byte) bool {
+			var e [32]uint8
+			copy(e[:], raw)
+			rows = append(rows, e)
+			return true
+		})
 		x.IDs = rows
 	}
 	if raw := o.Bytes(idedLeafAt); len(raw) > 0 {
@@ -624,95 +627,131 @@ func (x *Trunk) UnmarshalZAP(data []byte) error {
 		x.PtrL = &v
 	}
 	if l := o.List(trunkNumsAt); l.Len() > 0 {
-		rows := make([]uint32, l.Len())
-		for i := range rows {
+		var rows []uint32
+		l.EachBytes(func(_ int, raw []byte) bool {
 			var full [8]byte
-			copy(full[:], l.BytesAt(i))
-			rows[i] = uint32(binary.LittleEndian.Uint64(full[:]))
-		}
+			copy(full[:], raw)
+			e := uint32(binary.LittleEndian.Uint64(full[:]))
+			rows = append(rows, e)
+			return true
+		})
 		x.Nums = rows
 	}
 	if l := o.List(trunkSigsAt); l.Len() > 0 {
-		rows := make([]int32, l.Len())
-		for i := range rows {
+		var rows []int32
+		l.EachBytes(func(_ int, raw []byte) bool {
 			var full [8]byte
-			copy(full[:], l.BytesAt(i))
-			rows[i] = int32(int64(binary.LittleEndian.Uint64(full[:])<<32) >> 32)
-		}
+			copy(full[:], raw)
+			e := int32(int64(binary.LittleEndian.Uint64(full[:])<<32) >> 32)
+			rows = append(rows, e)
+			return true
+		})
 		x.Sigs = rows
 	}
 	if l := o.List(trunkBigsAt); l.Len() > 0 {
-		rows := make([]float64, l.Len())
-		for i := range rows {
+		var rows []float64
+		l.EachBytes(func(_ int, raw []byte) bool {
 			var full [8]byte
-			copy(full[:], l.BytesAt(i))
-			rows[i] = float64(math.Float64frombits(binary.LittleEndian.Uint64(full[:])))
-		}
+			copy(full[:], raw)
+			e := float64(math.Float64frombits(binary.LittleEndian.Uint64(full[:])))
+			rows = append(rows, e)
+			return true
+		})
 		x.Bigs = rows
 	}
 	if l := o.List(trunkStrsAt); l.Len() > 0 {
-		rows := make([]string, l.Len())
-		for i := range rows {
-			rows[i] = string(l.BytesAt(i))
-		}
+		var rows []string
+		l.EachBytes(func(_ int, raw []byte) bool {
+			e := string(raw)
+			rows = append(rows, e)
+			return true
+		})
 		x.Strs = rows
 	}
 	if l := o.List(trunkBufsAt); l.Len() > 0 {
-		rows := make([][]uint8, l.Len())
-		for i := range rows {
-			rows[i] = []uint8(append([]byte(nil), l.BytesAt(i)...))
-		}
+		var rows [][]uint8
+		l.EachBytes(func(_ int, raw []byte) bool {
+			e := []uint8(append([]byte(nil), raw...))
+			rows = append(rows, e)
+			return true
+		})
 		x.Bufs = rows
 	}
 	if l := o.List(trunkBitsAt); l.Len() > 0 {
-		rows := make([]bool, l.Len())
-		for i := range rows {
-			raw := l.BytesAt(i)
-			rows[i] = bool(len(raw) > 0 && raw[0] != 0)
-		}
+		var rows []bool
+		l.EachBytes(func(_ int, raw []byte) bool {
+			e := bool(len(raw) > 0 && raw[0] != 0)
+			rows = append(rows, e)
+			return true
+		})
 		x.Bits = rows
 	}
 	if l := o.List(trunkKidsAt); l.Len() > 0 {
-		rows := make([]Leaf, l.Len())
-		for i := range rows {
-			if err := rows[i].UnmarshalZAP(l.BytesAt(i)); err != nil {
-				return err
+		var rows []Leaf
+		var ferr error
+		l.EachBytes(func(_ int, raw []byte) bool {
+			var e Leaf
+			if ferr = e.UnmarshalZAP(raw); ferr != nil {
+				return false
 			}
+			rows = append(rows, e)
+			return true
+		})
+		if ferr != nil {
+			return ferr
 		}
 		x.Kids = rows
 	}
 	if l := o.List(trunkPtrsAt); l.Len() > 0 {
-		rows := make([]*Leaf, l.Len())
-		for i := range rows {
-			rows[i] = new(Leaf)
-			if err := rows[i].UnmarshalZAP(l.BytesAt(i)); err != nil {
-				return err
+		var rows []*Leaf
+		var ferr error
+		l.EachBytes(func(_ int, raw []byte) bool {
+			e := new(Leaf)
+			if ferr = e.UnmarshalZAP(raw); ferr != nil {
+				return false
 			}
+			rows = append(rows, e)
+			return true
+		})
+		if ferr != nil {
+			return ferr
 		}
 		x.Ptrs = rows
 	}
 	if l := o.List(trunkSealsAt); l.Len() > 0 {
-		rows := make([]Sealed, l.Len())
-		for range rows {
-		}
+		var rows []Sealed
+		l.EachBytes(func(_ int, raw []byte) bool {
+			var e Sealed
+			rows = append(rows, e)
+			return true
+		})
 		x.Seals = rows
 	}
 	if l := o.List(trunkPKidsAt); l.Len() > 0 {
-		rows := make([]Leaf, l.Len())
-		for i := range rows {
-			if err := rows[i].UnmarshalZAP(l.BytesAt(i)); err != nil {
-				return err
+		var rows []Leaf
+		var ferr error
+		l.EachBytes(func(_ int, raw []byte) bool {
+			var e Leaf
+			if ferr = e.UnmarshalZAP(raw); ferr != nil {
+				return false
 			}
+			rows = append(rows, e)
+			return true
+		})
+		if ferr != nil {
+			return ferr
 		}
 		x.PKids = &rows
 	}
 	if l := o.List(trunkPNumsAt); l.Len() > 0 {
-		rows := make([]uint32, l.Len())
-		for i := range rows {
+		var rows []uint32
+		l.EachBytes(func(_ int, raw []byte) bool {
 			var full [8]byte
-			copy(full[:], l.BytesAt(i))
-			rows[i] = uint32(binary.LittleEndian.Uint64(full[:]))
-		}
+			copy(full[:], raw)
+			e := uint32(binary.LittleEndian.Uint64(full[:]))
+			rows = append(rows, e)
+			return true
+		})
 		x.PNums = &rows
 	}
 	if l := o.List(trunkPtrI8At); l.Len() > 0 {

internal/zapenc/list_test.go

@@ -0,0 +1,86 @@
+package zapenc_test
+
+import (
+	"encoding/binary"
+	"testing"
+	"time"
+
+	zap "github.com/zap-proto/go"
+	"github.com/zap-proto/zip/internal/zapenc"
+)
+
+type rows struct {
+	Rows []inner
+}
+
+type names struct {
+	Names []string
+}
+
+// list writes an object whose one slot is a list holding blob, claiming count.
+func list(blob []byte, count int) []byte {
+	b := zap.NewBuilder(len(blob) + 64)
+	off := b.WriteBytes(blob)
+	ob := b.StartObject(8)
+	ob.SetList(0, off, count)
+	ob.FinishAsRoot()
+	return b.Finish()
+}
+
+func entries(n int, payload []byte) []byte {
+	var blob []byte
+	for range n {
+		blob = binary.LittleEndian.AppendUint32(blob, uint32(len(payload)))
+		blob = append(blob, payload...)
+	}
+	return blob
+}
+
+// A list element that is not a message is an error. It used to be read as a
+// null object, and the first field read through it dereferenced nil: one
+// malformed call took the whole process down.
+func TestAMalformedElementIsAnError(t *testing.T) {
+	msg := list(entries(1, []byte("not a message")), 1)
+	var got rows
+	if err := zapenc.Unmarshal(msg, &got); err == nil {
+		t.Fatal("a garbage element decoded without error")
+	}
+}
+
+// A list is read in one pass. Reading element i by index re-walks the list from
+// its start, so a long list took quadratic time — minutes of CPU for a body the
+// transport admits.
+func TestAListDecodesInOnePass(t *testing.T) {
+	const n = 200_000
+	msg := list(entries(n, []byte("a")), n)
+	start := time.Now()
+	var got names
+	if err := zapenc.Unmarshal(msg, &got); err != nil {
+		t.Fatal(err)
+	}
+	if len(got.Names) != n {
+		t.Fatalf("decoded %d of %d", len(got.Names), n)
+	}
+	if took := time.Since(start); took > 2*time.Second {
+		t.Fatalf("%d elements took %s", n, took)
+	}
+}
+
+// The elements are what the message holds, not what it claims to hold: an entry
+// is at least its 4-byte length, so a list in B bytes has at most B/4 of them
+// whatever its count says. (A count larger than the message itself is refused by
+// zap's own bound and reads as no list.)
+func TestAForgedCountYieldsOnlyRealElements(t *testing.T) {
+	const claim = 32
+	msg := list(entries(2, []byte("ab")), claim)
+	if len(msg) <= claim {
+		t.Fatalf("message of %d bytes cannot carry a claim of %d past zap's bound", len(msg), claim)
+	}
+	var got names
+	if err := zapenc.Unmarshal(msg, &got); err != nil {
+		t.Fatal(err)
+	}
+	if len(got.Names) > len(msg)/4 {
+		t.Fatalf("decoded %d elements from %d bytes, claimed %d", len(got.Names), len(msg), claim)
+	}
+}

internal/zapenc/zapenc.go

@@ -574,6 +574,9 @@ func leInt(v uint64, n int) []byte {
 // ---- decode ---------------------------------------------------------------
 
 func readStruct(o zap.Object, lay *layout, rv reflect.Value) error {
+	if o.IsNull() {
+		return nil // nothing to read; every accessor would dereference a nil message
+	}
 	for _, f := range lay.fields {
 		fv := rv.Field(f.index)
 		if f.ptr {
@@ -600,7 +603,7 @@ func readPointer(o zap.Object, f field, fv reflect.Value) error {
 		if l.Len() == 0 {
 			return nil
 		}
-		if err := readElem(l, 0, f.slice, nv.Elem()); err != nil {
+		if err := readElem(l.BytesAt(0), f.slice, nv.Elem()); err != nil {
 			return err
 		}
 	case kStruct:
@@ -687,36 +690,50 @@ func fixedRefusal(t reflect.Type) error {
 		"declare the type's wire (MarshalZAP/UnmarshalZAP) and it crosses inline", t, t.Len())
 }
 
+// readList walks the list once. Element i by index re-walks the stream from its
+// start, so decoding a list that way is quadratic in its length — a 160KB body
+// was over a second of CPU. The slice grows from the elements that are actually
+// there rather than being allocated from the count the message claims.
 func readList(o zap.Object, f field, target reflect.Value) error {
 	l := o.List(f.offset)
-	n := l.Len()
-	if n == 0 {
+	if l.Len() == 0 {
 		return nil
 	}
-	out := reflect.MakeSlice(target.Type(), n, n)
-	for i := 0; i < n; i++ {
-		ev := out.Index(i)
+	out := reflect.MakeSlice(target.Type(), 0, 0)
+	var err error
+	l.EachBytes(func(_ int, raw []byte) bool {
+		ev := reflect.New(f.slice.typ)
+		if err = readElem(raw, f.slice, ev.Elem()); err != nil {
+			return false
+		}
 		if f.slice.ptr {
-			nv := reflect.New(f.slice.typ)
-			if err := readElem(l, i, f.slice, nv.Elem()); err != nil {
-				return err
-			}
-			ev.Set(nv)
-			continue
-		}
-		if err := readElem(l, i, f.slice, ev); err != nil {
-			return err
+			out = reflect.Append(out, ev)
+		} else {
+			out = reflect.Append(out, ev.Elem())
 		}
+		return true
+	})
+	if err != nil {
+		return err
 	}
 	target.Set(out)
 	return nil
 }
 
-func readElem(l zap.List, i int, se *sliceElem, target reflect.Value) error {
+// readElem decodes one element from its own bytes. A struct element is a
+// complete message; one that does not parse is an error, never a nil object read
+// as if it were one — a single malformed element used to take the process down.
+func readElem(raw []byte, se *sliceElem, target reflect.Value) error {
 	if se.kind == kStruct {
-		return readStruct(l.ObjectAt(i), se.elem, target)
+		if len(raw) == 0 {
+			return nil // an empty element is the zero value, as the generated codec reads it
+		}
+		msg, err := zap.Parse(raw)
+		if err != nil {
+			return fmt.Errorf("zapenc: list element: %w", err)
+		}
+		return readStruct(msg.Root(), se.elem, target)
 	}
-	raw := l.BytesAt(i)
 	switch se.kind {
 	case kText:
 		target.SetString(string(raw))

layout.go

@@ -586,8 +586,10 @@ func readField(w *bytes.Buffer, lo, fn string, s Slot, f reflect.StructField, p
 		//
 		// An inline slot has no null: the writer leaves an absent pointer's
 		// bytes zeroed (see [writeField]), so all-zero is what absence looks
-		// like coming back. That is the rule the pointer SCALARS already read
-		// by, and reading it here is what keeps the two directions symmetric.
+		// like coming back, and a pointer to an all-zero array reads back nil.
+		// This is the one pointer that still cannot say present-and-zero: a
+		// scalar crosses as a list of at most one element, and bytes_fixed has no
+		// list form inline. Giving it one moves the wire of every optional id.
 		name := p.spell(ft)
 		fmt.Fprintf(w, "\tif raw := o.BytesFixed(%s, %d); len(raw) > 0 {\n", at, s.N)
 		fmt.Fprintf(w, "\t\tvar v %s\n\t\tcopy(v[:], raw)\n", name)
@@ -771,49 +773,68 @@ func bits(zaptype, v string) string {
 	return fmt.Sprintf("uint64(%s)", v)
 }
 
+// readList emits a single pass over the list. Element i by index re-walks the
+// stream from its start, so reading a list that way is quadratic in its length;
+// the rows grow from the elements actually there rather than from the count the
+// message claims.
 func readList(w *bytes.Buffer, at, fn string, s Slot, f reflect.StructField, p *pkg) error {
 	ft := under(f.Type)
 	et := ft.Elem()
 	ptr := et.Kind() == reflect.Pointer
 	el := under(et)
+	fallible := el.Kind() == reflect.Struct && !empty(el)
 
 	var elem bytes.Buffer
 	w, outer := &elem, w
+	// A scalar element arrives as its value and is pointed at after; a struct
+	// element is decoded in place, so a pointer one is allocated first.
+	value, point := "e", ""
+	if ptr && el.Kind() != reflect.Struct {
+		value, point = "v", "\t\t\te := &v\n"
+	}
 	switch {
-	case el.Kind() == reflect.Struct && empty(el):
-		if ptr {
-			fmt.Fprintf(w, "\t\t\trows[i] = new(%s)\n", p.spell(el))
-		}
 	case el.Kind() == reflect.Struct:
 		if ptr {
-			fmt.Fprintf(w, "\t\t\trows[i] = new(%s)\n", p.spell(el))
+			fmt.Fprintf(w, "\t\t\te := new(%s)\n", p.spell(el))
+		} else {
+			fmt.Fprintf(w, "\t\t\tvar e %s\n", p.spell(el))
+		}
+		if fallible {
+			fmt.Fprint(w, "\t\t\tif ferr = e.UnmarshalZAP(raw); ferr != nil {\n\t\t\t\treturn false\n\t\t\t}\n")
 		}
-		fmt.Fprint(w, "\t\t\tif err := rows[i].UnmarshalZAP(l.BytesAt(i)); err != nil {\n\t\t\t\treturn err\n\t\t\t}\n")
 	case el.Kind() == reflect.Array:
-		fmt.Fprint(w, "\t\t\tcopy(rows[i][:], l.BytesAt(i))\n")
+		fmt.Fprintf(w, "\t\t\tvar %s %s\n\t\t\tcopy(%s[:], raw)\n", value, p.spell(el), value)
 	case s.Elem == "text":
-		fmt.Fprintf(w, "\t\t\trows[i] = %s(l.BytesAt(i))\n", p.spell(et))
+		fmt.Fprintf(w, "\t\t\t%s := %s(raw)\n", value, p.spell(el))
 	case s.Elem == "bytes":
-		fmt.Fprintf(w, "\t\t\trows[i] = %s(append([]byte(nil), l.BytesAt(i)...))\n", p.spell(et))
+		fmt.Fprintf(w, "\t\t\t%s := %s(append([]byte(nil), raw...))\n", value, p.spell(el))
 	case s.Elem == "bool":
-		fmt.Fprintf(w, "\t\t\traw := l.BytesAt(i)\n\t\t\trows[i] = %s(len(raw) > 0 && raw[0] != 0)\n", p.spell(et))
+		fmt.Fprintf(w, "\t\t\t%s := %s(len(raw) > 0 && raw[0] != 0)\n", value, p.spell(el))
 	default:
-		fmt.Fprint(w, "\t\t\tvar full [8]byte\n\t\t\tcopy(full[:], l.BytesAt(i))\n")
-		fmt.Fprintf(w, "\t\t\trows[i] = %s(%s)\n", p.spell(et), elementRead(s))
+		fmt.Fprint(w, "\t\t\tvar full [8]byte\n\t\t\tcopy(full[:], raw)\n")
+		fmt.Fprintf(w, "\t\t\t%s := %s(%s)\n", value, p.spell(el), elementRead(s))
 		p.std("encoding/binary")
 		if s.Elem == "f32" || s.Elem == "f64" {
 			p.std("math")
 		}
 	}
+	fmt.Fprint(w, point)
 	w = outer
-	fmt.Fprintf(w, "\tif l := o.List(%s); l.Len() > 0 {\n", at)
-	fmt.Fprintf(w, "\t\trows := make(%s, l.Len())\n\t\tfor %s range rows {\n", p.spell(ft), index(&elem))
+	fmt.Fprintf(w, "\tif l := o.List(%s); l.Len() > 0 {\n\t\tvar rows %s\n", at, p.spell(ft))
+	if fallible {
+		fmt.Fprint(w, "\t\tvar ferr error\n")
+	}
+	fmt.Fprint(w, "\t\tl.EachBytes(func(_ int, raw []byte) bool {\n")
 	w.Write(elem.Bytes())
+	fmt.Fprint(w, "\t\t\trows = append(rows, e)\n\t\t\treturn true\n\t\t})\n")
+	if fallible {
+		fmt.Fprint(w, "\t\tif ferr != nil {\n\t\t\treturn ferr\n\t\t}\n")
+	}
 	assign := "x." + fn + " = rows"
 	if s.Ptr {
 		assign = "x." + fn + " = &rows"
 	}
-	fmt.Fprintf(w, "\t\t}\n\t\t%s\n\t}\n", assign)
+	fmt.Fprintf(w, "\t\t%s\n\t}\n", assign)
 	return nil
 }
 

load.go

@@ -209,6 +209,13 @@ type plugin struct {
 	spec     Plugin
 
 	app *App // for logging and supervision when started on demand
+
+	// lazy, idle and start are spec.Lazy, spec.IdleAfter and spec.startLimit(),
+	// copied at Load because the request path reads them with no lock while
+	// Reload rewrites spec under p.mu. Reload replaces the source, never these.
+	lazy  bool
+	idle  time.Duration
+	start time.Duration
 	// owner is the ROOT app of the composition this plugin ended up in, stamped
 	// when a generation is built. app is the definition that Load'ed it, which for
 	// a composed service is a sub-app holding one plugin and no ceiling — so it is
@@ -307,7 +314,7 @@ func (h hold) release() {
 	if h.in == nil {
 		return
 	}
-	if h.p.spec.IdleAfter > 0 {
+	if h.p.idle > 0 {
 		h.p.lastUse.Store(time.Now().UnixNano())
 	}
 	h.in.busy.Add(-1)
@@ -361,10 +368,12 @@ func (p *plugin) target() (Client, string, hold) {
 		// Closing under us. startOnDemand waits on the lock evict holds, then
 		// finds it either kept or gone.
 	}
-	if p.app == nil || !p.spec.Lazy || p.disabled.Load() {
+	if p.app == nil || !p.lazy || p.disabled.Load() {
 		return nil, "", hold{} // eager and down, disabled, or unloaded — the route answers 503
 	}
-	return p.startOnDemand()
+	// The wait for room is this caller's own, counted from now: a caller queued
+	// on p.mu behind another starter has already spent part of it.
+	return p.startOnDemand(time.Now().Add(p.start))
 }
 
 // Start brings the plugin named name up if it is not already, and reports the
@@ -403,7 +412,7 @@ func (a *App) Start(name string) (string, error) {
 // so a burst of concurrent first requests produces ONE child rather than one
 // per request. A start failure is not cached: the next request tries again,
 // because the usual cause is a dependency that has not come up yet.
-func (p *plugin) startOnDemand() (Client, string, hold) {
+func (p *plugin) startOnDemand(deadline time.Time) (Client, string, hold) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	// Under p.mu no evict is deciding — evict holds this lock from raising closing
@@ -420,7 +429,7 @@ func (p *plugin) startOnDemand() (Client, string, hold) {
 	// The ceiling is applied HERE, before another process exists, so warm bounds
 	// what this host holds rather than what a sweep restores a minute later.
 	host := p.host()
-	if err := host.makeRoom(p); err != nil {
+	if err := host.makeRoom(p, deadline); err != nil {
 		p.app.logger.Warn("zip lazy plugin found no room", "name", p.name, "err", err)
 		return nil, "", hold{}
 	}
@@ -548,7 +557,8 @@ func (a *App) load(prefixes []string, spec Plugin) error {
 		return fmt.Errorf("zip: Load(%s): plugin %q has URL but no Sum — refusing to run an unverified download", prefix, spec.Name)
 	}
 
-	p := &plugin{name: spec.Name, prefix: prefix, prefixes: prefixes, spec: spec, app: a}
+	p := &plugin{name: spec.Name, prefix: prefix, prefixes: prefixes, spec: spec, app: a,
+		lazy: spec.Lazy, idle: spec.IdleAfter, start: spec.startLimit()}
 	var in *instance
 	if !spec.Lazy {
 		var err error

transport.go

@@ -100,7 +100,7 @@ var (
 	transports   = map[string]Transport{
 		"zap": {
 			Serve: func(addr string, h fasthttp.RequestHandler) Server {
-				return &zapServer{addr: addr, srv: &http.Server{Network: NetworkOf(addr), Addr: addr, Handler: h}}
+				return &zapServer{addr: addr, srv: &http.Server{Network: NetworkOf(addr), Addr: addr, Handler: overZAP(h)}}
 			},
 			Dial: func(addr string) Client { return http.Dial(NetworkOf(addr), addr) },
 		},
@@ -133,6 +133,24 @@ var (
 	}
 )
 
+// viaZAP marks a request the ZAP transport carried. ZAP frames a request and its
+// reply and has no connection to hand over, so its server never runs a hijack:
+// an upgrade relayed over it would answer an empty 200 and hold the plugin for
+// good.
+type viaZAP struct{}
+
+func overZAP(h fasthttp.RequestHandler) fasthttp.RequestHandler {
+	return func(rc *fasthttp.RequestCtx) {
+		rc.SetUserValue(viaZAP{}, true)
+		h(rc)
+	}
+}
+
+func carriedByZAP(rc *fasthttp.RequestCtx) bool {
+	v, _ := rc.UserValue(viaZAP{}).(bool)
+	return v
+}
+
 // withPort defaults the port of a tcp address, because a URL usually omits it
 // and a dialer never can. A unix path is returned untouched.
 func withPort(addr, port string) string {
@@ -363,6 +381,10 @@ func (a *App) mountVia(prefix string, to func() (Client, string, hold)) {
 	h := func(c *Ctx) error {
 		client, host, held := to()
 		if upgrading(c.fc.Request()) {
+			if carriedByZAP(c.fc.RequestCtx()) {
+				held.release()
+				return Errorf(501, "mount %s: a protocol upgrade does not cross ZAP", prefix)
+			}
 			return relay(c.fc.RequestCtx(), host, "mount "+prefix, held)
 		}
 		return forward(c.Context(), c.fc.Request(), c.fc.Response(), client, host, "", "mount "+prefix, held)
@@ -471,6 +493,11 @@ func (b *heldBody) Close() error {
 	return err
 }
 
+// declineWait bounds reading the reply a plugin gave instead of switching
+// protocols. It is an ordinary answer and arrives at once; one still streaming
+// after this is not an answer to an upgrade.
+const declineWait = 10 * time.Second
+
 // upgradeWait bounds reaching a plugin for an upgrade. It is a dial on a socket
 // in this pod, so a second is already generous; what it exists to stop is a
 // caller hanging on a plugin that is not listening yet.
@@ -587,11 +614,16 @@ func relay(rc *fasthttp.RequestCtx, addr, what string, h hold) error {
 	if head.StatusCode() != fasthttp.StatusSwitchingProtocols {
 		defer func() { _ = up.Close() }()
 		head.CopyTo(&rc.Response.Header)
-		body, err := body(br, head.ContentLength())
-		if err != nil {
-			return Errorf(502, "%s: upgrade: %v", what, err)
+		// A declined upgrade is an ordinary reply, and it may be chunked or never
+		// end: read it as a reply, for as long as a reply takes and no longer, or
+		// the request holding the plugin waits on a body that is still streaming.
+		_ = up.SetReadDeadline(time.Now().Add(declineWait))
+		var got fasthttp.Response
+		head.CopyTo(&got.Header)
+		if err := got.ReadBody(br, 0); err != nil {
+			return Errorf(502, "%s: upgrade declined: %v", what, err)
 		}
-		rc.Response.SetBody(body)
+		rc.Response.SetBody(got.Body())
 		return nil
 	}
 
@@ -618,19 +650,6 @@ func relay(rc *fasthttp.RequestCtx, addr, what string, h hold) error {
 	return nil
 }
 
-// body reads a reply's body given the length its header declared. A negative
-// length is "until the connection ends", which is what identity encoding means.
-func body(br *bufio.Reader, n int) ([]byte, error) {
-	if n < 0 {
-		return io.ReadAll(br)
-	}
-	b := make([]byte, n)
-	if _, err := io.ReadFull(br, b); err != nil {
-		return nil, err
-	}
-	return b, nil
-}
-
 // makeSocketDir creates the directory a unix socket will be bound in, so
 // serving at the canonical zip.SocketPath("name") works on a host where the
 // runtime dir does not exist yet — which is every host, the first time.