fix(host): never evict a process with requests in flight, rank recency by request end

host_forward_test.go

@@ -52,7 +52,7 @@ func TestForwardKeepsTheCallersHost(t *testing.T) {
 	req.Header.SetHost("lux.id")
 	req.Header.SetRequestURI("/.well-known/openid-configuration")
 
-	if err := forward(context.Background(), req, resp, rec, "/var/run/iam.sock", "", "mount /v1/iam"); err != nil {
+	if err := forward(context.Background(), req, resp, rec, "/var/run/iam.sock", "", "mount /v1/iam", hold{}); err != nil {
 		t.Fatalf("forward: %v", err)
 	}
 	if rec.sawHost != "lux.id" {
@@ -71,7 +71,7 @@ func TestForwardFallsBackToTheDialAddress(t *testing.T) {
 
 	req.Header.SetRequestURI("/mcp")
 
-	if err := forward(context.Background(), req, resp, rec, "plugin.sock", "", "mcp iam"); err != nil {
+	if err := forward(context.Background(), req, resp, rec, "plugin.sock", "", "mcp iam", hold{}); err != nil {
 		t.Fatalf("forward: %v", err)
 	}
 	if rec.sawHost != "plugin.sock" {
@@ -89,7 +89,7 @@ func TestForwardStillRewritesThePath(t *testing.T) {
 	req.Header.SetHost("hanzo.ai")
 	req.Header.SetRequestURI("/v1/iam/mcp")
 
-	if err := forward(context.Background(), req, resp, rec, "iam.sock", "/mcp", "mcp iam"); err != nil {
+	if err := forward(context.Background(), req, resp, rec, "iam.sock", "/mcp", "mcp iam", hold{}); err != nil {
 		t.Fatalf("forward: %v", err)
 	}
 	if got := string(req.URI().Path()); got != "/mcp" {
@@ -105,7 +105,7 @@ func TestForwardWithNoClientIs503(t *testing.T) {
 	req, resp := fasthttp.AcquireRequest(), fasthttp.AcquireResponse()
 	defer fasthttp.ReleaseRequest(req)
 	defer fasthttp.ReleaseResponse(resp)
-	err := forward(context.Background(), req, resp, nil, "iam.sock", "", "mount /v1/iam")
+	err := forward(context.Background(), req, resp, nil, "iam.sock", "", "mount /v1/iam", hold{})
 	if err == nil || !strings.Contains(err.Error(), "no instance running") {
 		t.Fatalf("err = %v, want a 503 naming the mount", err)
 	}

idle.go

@@ -2,6 +2,7 @@ package zip
 
 import (
 	"cmp"
+	"fmt"
 	"slices"
 	"sync"
 	"time"
@@ -91,40 +92,92 @@ func (p *plugin) host() *App {
 // door asks every subsystem at once: ~100 children started inside 90 seconds, the
 // pod stopped answering its own liveness probe, and the kubelet killed it — the
 // "ceiling" observed only in the logs of a container that was already gone.
-func (a *App) makeRoom(starter *plugin) {
+//
+// A BUSY PROCESS IS NEVER THE ONE STOPPED. Ranking by when a request started, and
+// not counting the ones in flight, stopped a subsystem two to eleven seconds into
+// a request it was still answering: the caller read "http: read response: EOF" as
+// a 502, and the host evicted 162 times in eight minutes, each stop the next
+// start's cold miss. Candidates are idle instances, coldest by when their last
+// request ended. When every candidate is busy the starter waits for one to finish,
+// as long as its own start may take, and then gives up with an error — the caller
+// 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 {
+	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()
+			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)
+		}
+		time.Sleep(roomPoll)
+	}
+}
+
+// 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
+
+// started returns the unit of room makeRoom handed out: the process is now
+// counted as running, or it failed to start and holds nothing.
+func (a *App) started() {
 	if a.warm <= 0 {
 		return
 	}
-	now := time.Now()
-	for {
-		var (
-			live      int
-			evictable []*plugin
-		)
-		for _, p := range a.pluginSet() {
-			if p.cur.Load() == nil {
-				continue
-			}
-			live++
-			if p != starter && p.spec.Lazy && p.spec.idleAfter() > 0 {
-				evictable = append(evictable, p)
-			}
+	a.roomMu.Lock()
+	a.starting--
+	a.roomMu.Unlock()
+}
+
+// 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) {
+	for _, p := range a.pluginSet() {
+		in := p.cur.Load()
+		if in == nil {
+			continue
 		}
-		// The starter is not running yet, so room for it means strictly under.
-		if live < a.warm || len(evictable) == 0 {
-			return
+		live++
+		if p == starter || !p.evictable() {
+			continue
 		}
-		slices.SortFunc(evictable, func(x, y *plugin) int {
-			return cmp.Compare(x.lastUse.Load(), y.lastUse.Load())
-		})
-		cold := evictable[0]
-		coldSeen := cold.lastUse.Load()
-		if !cold.evict("room", now.Sub(time.Unix(0, coldSeen)), coldSeen) {
-			return // already down; re-reading would spin
+		if in.busy.Load() > 0 {
+			busy++
+			continue
+		}
+		if coldest == nil || p.lastUse.Load() < coldest.lastUse.Load() {
+			coldest = p
 		}
 	}
+	return live, coldest, 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 }
+
 // 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
 // lock while stopping a child, because retire() takes the plugin's own.
@@ -169,20 +222,21 @@ func (a *App) evictOver(warm int, now time.Time) int {
 		evictable []*plugin
 	)
 	for _, p := range a.pluginSet() {
-		if p.cur.Load() == nil {
+		in := p.cur.Load()
+		if in == nil {
 			continue
 		}
 		live++
-		if p.spec.Lazy && p.spec.idleAfter() > 0 {
+		if p.evictable() && in.busy.Load() == 0 {
 			evictable = append(evictable, p)
 		}
 	}
 	if live <= warm {
 		return 0
 	}
-	// Ascending by last use, so the front of the slice is the coldest. A
-	// plugin's lastUse is stamped on every resolve, and only stamped for the
-	// ones collected above, so the ordering is total.
+	// Ascending by last use, so the front of the slice is the coldest. lastUse
+	// is stamped when a request ENDS, so a plugin that has been answering for a
+	// minute is not mistaken for one asked a minute ago and since forgotten.
 	slices.SortFunc(evictable, func(x, y *plugin) int {
 		return cmp.Compare(x.lastUse.Load(), y.lastUse.Load())
 	})
@@ -224,12 +278,17 @@ func (p *plugin) evictIfIdle(now time.Time) bool {
 // the one place a plugin is reclaimed, so the two policies above cannot come to
 // disagree about how a child is taken down; reason says which asked.
 func (p *plugin) evict(reason string, idle time.Duration, seen int64) bool {
-	p.mu.Lock()
+	// A plugin whose lock is held is mid-transition — starting, reloading,
+	// unloading — and is nobody's candidate. Waiting for it would hold the room
+	// lock across a child's whole start.
+	if !p.mu.TryLock() {
+		return false
+	}
 	if p.closed || p.disabled.Load() {
 		p.mu.Unlock()
 		return false
 	}
-	// THE CLOCK IS RE-READ UNDER THE LOCK, and this is the whole of the fix.
+	// THE CLOCK IS RE-READ UNDER THE LOCK.
 	//
 	// The caller decided to evict from a lastUse it read outside the lock. A
 	// request arriving in that window does not just get served by the instance
@@ -248,15 +307,30 @@ func (p *plugin) evict(reason string, idle time.Duration, seen int64) bool {
 		p.mu.Unlock()
 		return false
 	}
+	in := p.cur.Load()
+	if in == nil {
+		p.mu.Unlock()
+		return false // already down; nothing to stop
+	}
+	// NOTHING IN FLIGHT, decided against every request that will ever reach in.
+	// closing is raised before busy is read, and claim raises busy before it
+	// reads closing, so a request either sees closing and goes to the lock this
+	// holds, or is counted here and keeps the instance.
+	in.closing.Store(true)
+	if in.busy.Load() > 0 {
+		in.closing.Store(false)
+		p.mu.Unlock()
+		return false
+	}
 	// Swap to nil BEFORE the child dies so supervise()'s CAS fails and it does
 	// not treat this as a crash to recover from. Not disabled: the next request
 	// through target() is meant to bring it back.
-	in := p.cur.Swap(nil)
+	if !p.cur.CompareAndSwap(in, nil) {
+		p.mu.Unlock()
+		return false // it exited on its own; the supervisor has it
+	}
 	p.mu.Unlock()
 
-	if in == nil {
-		return false // already down; nothing to stop
-	}
 	p.evictions.Add(1)
 	if p.app != nil {
 		p.app.logger.Info("zip idle plugin evicted",

inflight_test.go

@@ -0,0 +1,214 @@
+package zip_test
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/zap-proto/zip"
+)
+
+// A process answering a request is never the one stopped for room, and recency is
+// when a request ENDED. Each test here puts a request in flight on demo0, makes
+// the ceiling ask for room, and reads which process went — the arrangement that
+// stopped a subsystem mid-answer in production and read as a 502.
+
+// fetchAsync is call for a goroutine: it reports instead of failing the test.
+func fetchAsync(app *zip.App, target string) (int, string, error) {
+	req, err := http.NewRequest("GET", target, nil)
+	if err != nil {
+		return 0, "", err
+	}
+	resp, err := app.Test(req, zip.TestConfig{Timeout: deadline, FailOnTimeout: true})
+	if err != nil {
+		return 0, "", err
+	}
+	defer resp.Body.Close()
+	b, err := io.ReadAll(resp.Body)
+	return resp.StatusCode, string(b), err
+}
+
+type answer struct {
+	status int
+	body   string
+	err    error
+	took   time.Duration
+}
+
+// inflight starts a request in the background and returns once its plugin is
+// running, so the request is being served when the caller acts.
+func inflight(t *testing.T, app *zip.App, name, target string) <-chan answer {
+	t.Helper()
+	out := make(chan answer, 1)
+	go func() {
+		start := time.Now()
+		status, body, err := fetchAsync(app, target)
+		out <- answer{status, body, err, time.Since(start)}
+	}()
+	if !settles(5*time.Second, func() bool { return pidOf(app, name) != 0 }) {
+		t.Fatalf("%s never started", name)
+	}
+	time.Sleep(50 * time.Millisecond) // the plugin is up; let the request reach it
+	return out
+}
+
+func pidOf(app *zip.App, name string) int {
+	for _, p := range app.Plugins() {
+		if p.Name == name && p.Running {
+			return p.PID
+		}
+	}
+	return 0
+}
+
+func completed(t *testing.T, a answer, want string) {
+	t.Helper()
+	if a.err != nil || a.status != 200 || !strings.Contains(a.body, want) {
+		t.Fatalf("the request in flight did not complete: status=%d err=%v body=%q",
+			a.status, a.err, a.body)
+	}
+}
+
+// TestInflight_ALongRequestKeepsItsProcess is the production failure: demo0 is
+// asked first and is still answering when the ceiling needs room. By start time
+// demo0 is the coldest; it is also the only one busy. demo1 must go.
+func TestInflight_ALongRequestKeepsItsProcess(t *testing.T) {
+	app := warmHost(t, 3, 2, time.Hour)
+
+	slow := inflight(t, app, "demo0", "/v1/demo0/version?sleep=1500ms")
+	pid := pidOf(app, "demo0")
+	if status, _ := call(t, app, "GET", "/v1/demo1/version", ""); status != 200 {
+		t.Fatalf("demo1 did not serve: %d", status)
+	}
+	if status, _ := call(t, app, "GET", "/v1/demo2/version", ""); status != 200 {
+		t.Fatalf("demo2 did not serve: %d", status)
+	}
+
+	completed(t, <-slow, `"echo":"/v1/demo0/version"`)
+	if got := pidOf(app, "demo0"); got != pid {
+		t.Fatalf("demo0 was stopped mid-request: pid %d, now %d", pid, got)
+	}
+	if pidOf(app, "demo1") != 0 {
+		t.Fatal("demo1 — the idle one — is still running; something else made the room")
+	}
+}
+
+// TestInflight_AStreamKeepsItsProcess holds the same line for a reply that is
+// still streaming: the request is in flight until the body ends, not until the
+// head arrives.
+func TestInflight_AStreamKeepsItsProcess(t *testing.T) {
+	app := warmHost(t, 3, 2, time.Hour)
+
+	stream := inflight(t, app, "demo0", "/v1/demo0/feed?drip=1500ms")
+	pid := pidOf(app, "demo0")
+	for _, name := range []string{"demo1", "demo2"} {
+		if status, _ := call(t, app, "GET", "/v1/"+name+"/version", ""); status != 200 {
+			t.Fatalf("%s did not serve: %d", name, status)
+		}
+	}
+
+	a := <-stream
+	completed(t, a, "tick\n")
+	if !strings.HasSuffix(a.body, "done\n") {
+		t.Fatalf("the stream was cut before its end: %q", a.body)
+	}
+	if got := pidOf(app, "demo0"); got != pid {
+		t.Fatalf("demo0 was stopped mid-stream: pid %d, now %d", pid, got)
+	}
+}
+
+// TestInflight_EveryoneBusyWaits is the case with no idle candidate at all. The
+// starter waits for a request to finish and takes that room; neither request in
+// flight is killed to make it.
+func TestInflight_EveryoneBusyWaits(t *testing.T) {
+	app := warmHost(t, 3, 2, time.Hour)
+
+	first := inflight(t, app, "demo0", "/v1/demo0/version?sleep=1200ms")
+	second := inflight(t, app, "demo1", "/v1/demo1/version?sleep=1200ms")
+
+	start := time.Now()
+	status, body := call(t, app, "GET", "/v1/demo2/version", "")
+	waited := time.Since(start)
+	if status != 200 {
+		t.Fatalf("demo2 did not serve after waiting: %d %q", status, body)
+	}
+	completed(t, <-first, `"echo":"/v1/demo0/version"`)
+	completed(t, <-second, `"echo":"/v1/demo1/version"`)
+	if waited < 500*time.Millisecond {
+		t.Fatalf("demo2 started after %s, before any request in flight finished — "+
+			"someone busy was stopped for it", waited)
+	}
+	if live := len(runningPIDs(t, app)); live > 2 {
+		t.Fatalf("%d processes against a ceiling of 2", live)
+	}
+}
+
+// TestInflight_NoRoomIsA503 bounds the wait: a starter waits as long as its own
+// start may take, then is refused. The request in flight still completes.
+func TestInflight_NoRoomIsA503(t *testing.T) {
+	bin := buildPlugin(t, "v1")
+	app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true, Warm: 1})
+	for _, p := range []zip.Plugin{
+		{Name: "demo0", Bin: bin, Dir: sockDir(t), Lazy: true, IdleAfter: time.Hour},
+		{Name: "demo1", Bin: bin, Dir: sockDir(t), Lazy: true, IdleAfter: time.Hour, Start: 300 * time.Millisecond},
+	} {
+		app.Use(must(zip.Load(p, "/v1/"+p.Name)))
+	}
+	t.Cleanup(func() { _ = app.Shutdown() })
+
+	slow := inflight(t, app, "demo0", "/v1/demo0/version?sleep=2s")
+	start := time.Now()
+	status, _ := call(t, app, "GET", "/v1/demo1/version", "")
+	if status != 503 {
+		t.Fatalf("status %d, want 503 while the only process is busy", status)
+	}
+	if took := time.Since(start); took < 250*time.Millisecond {
+		t.Fatalf("refused after %s, without waiting for room", took)
+	}
+	completed(t, <-slow, `"echo":"/v1/demo0/version"`)
+}
+
+// TestInflight_RecencyIsWhenARequestEnded is the ranking. demo0 is asked first
+// and answers last; demo1 is asked after it and answers at once. By start time
+// demo0 is the coldest. By end time it is the warmest, and demo1 goes.
+func TestInflight_RecencyIsWhenARequestEnded(t *testing.T) {
+	app := warmHost(t, 3, 2, time.Hour)
+
+	slow := inflight(t, app, "demo0", "/v1/demo0/version?sleep=800ms")
+	if status, _ := call(t, app, "GET", "/v1/demo1/version", ""); status != 200 {
+		t.Fatal("demo1 did not serve")
+	}
+	completed(t, <-slow, `"echo":"/v1/demo0/version"`)
+	pid := pidOf(app, "demo0")
+
+	if status, _ := call(t, app, "GET", "/v1/demo2/version", ""); status != 200 {
+		t.Fatal("demo2 did not serve")
+	}
+	if got := pidOf(app, "demo0"); got != pid {
+		t.Fatal("demo0 was evicted, ranked by when its request STARTED")
+	}
+	if pidOf(app, "demo1") != 0 {
+		t.Fatal("demo1, the least recently active, is still running")
+	}
+}
+
+// TestInflight_TheSweepSparesTheBusy is the age bound's half: a process older than
+// its IdleAfter is not stopped while it is answering.
+func TestInflight_TheSweepSparesTheBusy(t *testing.T) {
+	app := warmHost(t, 1, 0, 200*time.Millisecond)
+
+	slow := inflight(t, app, "demo0", "/v1/demo0/version?sleep=1s")
+	pid := pidOf(app, "demo0")
+	for range 4 {
+		time.Sleep(250 * time.Millisecond)
+		if n := app.Evict(); n != 0 {
+			t.Fatalf("Evict stopped %d processes while demo0 was answering", n)
+		}
+	}
+	completed(t, <-slow, `"echo":"/v1/demo0/version"`)
+	if got := pidOf(app, "demo0"); got != pid {
+		t.Fatalf("demo0 changed under its request: %d → %d", pid, got)
+	}
+}

internal/testplugin/main.go

@@ -5,10 +5,12 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"log"
 	"os"
+	"time"
 
 	"github.com/zap-proto/zip"
 )
@@ -89,7 +91,27 @@ func main() {
 	// otherwise identical from the outside. A wildcard answers paths that do not
 	// exist, so there is no operation for it to be: an op is one named endpoint
 	// with one schema, and this is deliberately neither.
+	//
+	// ?sleep=<duration> answers after that long and ?drip=<duration> streams a
+	// line every 50ms for that long, then "done". They make a request that is
+	// still in flight when something else happens — what the eviction tests need,
+	// at any prefix the plugin is mounted under.
 	app.Raw(zip.MethodAll, "/*", func(c *zip.Ctx) error {
+		if d, err := time.ParseDuration(c.Query("drip")); err == nil {
+			return c.SendStreamWriter(func(w *bufio.Writer) {
+				for end := time.Now().Add(d); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
+					_, _ = w.WriteString("tick\n")
+					if w.Flush() != nil {
+						return
+					}
+				}
+				_, _ = w.WriteString("done\n")
+				_ = w.Flush()
+			})
+		}
+		if d, err := time.ParseDuration(c.Query("sleep")); err == nil {
+			time.Sleep(d)
+		}
 		return c.JSON(200, map[string]string{"echo": c.Path(), "version": version})
 	})
 

load.go

@@ -283,6 +283,53 @@ type instance struct {
 	// first starve the others.
 	done    chan struct{}
 	exitErr error
+
+	// busy counts the requests this instance is serving: a reply until it is
+	// written, a streamed reply until its body is closed, an upgrade until its
+	// tunnel ends. Eviction never stops an instance whose count is above zero.
+	busy atomic.Int64
+	// closing is raised by evict BEFORE it reads busy, and a request raises busy
+	// BEFORE it reads closing. Whichever comes second sees the first, so a
+	// request is either refused by a closing instance or counted by evict.
+	closing atomic.Bool
+}
+
+// hold is one request's claim on the instance serving it. release ends the
+// claim and stamps the plugin's last use, so recency is when a request ENDED:
+// a request that ran for a minute is a minute more recent than its start. The
+// zero hold, for a mount with no instance behind it, releases nothing.
+type hold struct {
+	p  *plugin
+	in *instance
+}
+
+func (h hold) release() {
+	if h.in == nil {
+		return
+	}
+	if h.p.spec.IdleAfter > 0 {
+		h.p.lastUse.Store(time.Now().UnixNano())
+	}
+	h.in.busy.Add(-1)
+}
+
+// claim counts one request against in, or reports that in is closing.
+func (p *plugin) claim(in *instance) (hold, bool) {
+	in.busy.Add(1)
+	if in.closing.Load() {
+		in.busy.Add(-1)
+		return hold{}, false
+	}
+	return hold{p: p, in: in}, true
+}
+
+// startLimit is how long a start may wait for the child to listen, and how long
+// it may wait for room before that.
+func (p Plugin) startLimit() time.Duration {
+	if p.Start <= 0 {
+		return 10 * time.Second
+	}
+	return p.Start
 }
 
 // startError reports why the last start attempt failed, or "" when none has. It is
@@ -302,20 +349,20 @@ func (p *plugin) startError() string {
 	return ""
 }
 
-// target is the hot path: what the mounted route should talk to right now.
-// For a lazy plugin the first caller through here starts it; the atomic load
-// keeps the steady-state cost to one load once it is running.
-func (p *plugin) target() (Client, string) {
+// target is the hot path: what the mounted route should talk to right now, held
+// for the request — the caller releases the hold when the request is over. For a
+// lazy plugin the first caller through here starts it; once it is running the
+// cost is a load and two atomic adds.
+func (p *plugin) target() (Client, string, hold) {
 	if in := p.cur.Load(); in != nil {
-		// One store on the hot path, and only when idle eviction is armed:
-		// a plugin that can never be evicted has no reason to be timed.
-		if p.spec.IdleAfter > 0 {
-			p.lastUse.Store(time.Now().UnixNano())
+		if h, ok := p.claim(in); ok {
+			return in.client, in.sock, h
 		}
-		return in.client, in.sock
+		// 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() {
-		return nil, "" // eager and down, disabled, or unloaded — the route answers 503
+		return nil, "", hold{} // eager and down, disabled, or unloaded — the route answers 503
 	}
 	return p.startOnDemand()
 }
@@ -341,7 +388,9 @@ func (a *App) Start(name string) (string, error) {
 	if p == nil {
 		return "", fmt.Errorf("zip: Start: no plugin named %q", name)
 	}
-	if client, addr := p.target(); client != nil {
+	client, addr, h := p.target()
+	h.release() // a start is not a request; nothing is in flight
+	if client != nil {
 		return addr, nil
 	}
 	if p.disabled.Load() {
@@ -354,20 +403,28 @@ 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) {
+func (p *plugin) startOnDemand() (Client, string, hold) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	// Under p.mu no evict is deciding — evict holds this lock from raising closing
+	// to the swap — so a current instance here is not closing and the claim holds.
 	if in := p.cur.Load(); in != nil {
-		return in.client, in.sock // won by another caller while we waited
+		h, _ := p.claim(in)
+		return in.client, in.sock, h // won by another caller while we waited
 	}
 	// Re-checked under the lock, where Unload set it, so a request racing an
 	// Unload cannot slip a child in behind it.
 	if p.closed || p.disabled.Load() {
-		return nil, ""
+		return nil, "", 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.
-	p.host().makeRoom(p)
+	host := p.host()
+	if err := host.makeRoom(p); err != nil {
+		p.app.logger.Warn("zip lazy plugin found no room", "name", p.name, "err", err)
+		return nil, "", hold{}
+	}
+	defer host.started()
 
 	in, err := start(p.spec)
 	if err != nil {
@@ -376,8 +433,11 @@ func (p *plugin) startOnDemand() (Client, string) {
 		// later request too, so the reason has to outlive the log line.
 		msg := err.Error()
 		p.lastErr.Store(&msg)
-		return nil, ""
+		return nil, "", hold{}
 	}
+	// The caller's claim is counted before anyone can see the instance, so the
+	// first request on a fresh child is never the one an eviction lands on.
+	in.busy.Store(1)
 	// PUBLISHED BY COMPARE-AND-SWAP, because p.mu does not exclude the supervisor.
 	//
 	// A restart swaps its own instance in without taking this lock (supervise holds
@@ -398,15 +458,16 @@ func (p *plugin) startOnDemand() (Client, string) {
 	if !p.cur.CompareAndSwap(nil, in) {
 		stop(in, 0)
 		if cur := p.cur.Load(); cur != nil {
-			return cur.client, cur.sock
+			h, _ := p.claim(cur)
+			return cur.client, cur.sock, h
 		}
-		return nil, ""
+		return nil, "", hold{}
 	}
 	p.lastUse.Store(time.Now().UnixNano()) // or the next sweep evicts what just started
 	p.app.logger.Info("zip lazy plugin started on first request",
 		"name", p.name, "pid", in.cmd.Process.Pid, "addr", in.sock)
 	go p.app.supervise(p, in)
-	return in.client, in.sock
+	return in.client, in.sock, hold{p: p, in: in}
 }
 
 // When p.Addr is set the remote is proxied directly. Otherwise the binary is
@@ -723,7 +784,7 @@ func start(spec Plugin) (*instance, error) {
 		close(in.done)
 	}()
 
-	if err := waitListening(sock, spec.Start, in.done, in); err != nil {
+	if err := waitListening(sock, spec.startLimit(), in.done, in); err != nil {
 		stop(in, 0)
 		return nil, err
 	}
@@ -835,9 +896,6 @@ func stop(in *instance, grace time.Duration) {
 // deadline passes. Watching the child matters: a plugin that dies immediately
 // should say so, not time out.
 func waitListening(sock string, limit time.Duration, done <-chan struct{}, in *instance) error {
-	if limit <= 0 {
-		limit = 10 * time.Second
-	}
 	deadline := time.Now().Add(limit)
 	for {
 		if c, err := net.Dial("unix", sock); err == nil {

mcp.go

@@ -803,9 +803,10 @@ func (a *App) tools() []mcpTool {
 // spec: the model sees "this tool is not available right now" and can react,
 // where a 503 body is a failure it cannot interpret.
 func (a *App) relay(ctx context.Context, f *zapmcp.Frame, p *plugin) *zapmcp.Frame {
-	client, host := p.target()
+	client, host, h := p.target()
 	body, err := json.Marshal(f)
 	if err != nil {
+		h.release()
 		return a.answer(f, nil, err)
 	}
 	req := fasthttp.AcquireRequest()
@@ -817,7 +818,7 @@ func (a *App) relay(ctx context.Context, f *zapmcp.Frame, p *plugin) *zapmcp.Fra
 	req.Header.SetContentType(mimeJSON)
 	req.SetBody(body)
 	forwardIdentity(ctx, req)
-	if err := forward(ctx, req, resp, client, host, p.spec.mcpPath(), "mcp "+p.name); err != nil {
+	if err := forward(ctx, req, resp, client, host, p.spec.mcpPath(), "mcp "+p.name, h); err != nil {
 		return a.answer(f, nil, err)
 	}
 	var ans zapmcp.Frame

remote.go

@@ -104,7 +104,7 @@ func remoteApp(parent *App, prefix, addr string, d Declaration) (*App, error) {
 	r := newApp(cfg)
 
 	proxy := func(c *Ctx) error {
-		return forward(c.Context(), c.fc.Request(), c.fc.Response(), client, hostport, "", "mount "+prefix)
+		return forward(c.Context(), c.fc.Request(), c.fc.Response(), client, hostport, "", "mount "+prefix, hold{})
 	}
 	if len(d.Routes) == 0 {
 		// Nothing declared: the prefix and everything under it, which is what a

transport.go

@@ -343,7 +343,7 @@ func (a *App) mount(prefix, addr string) error {
 	}
 	client := t.Dial(hostport)
 	a.logger.Info("zip mounting", "prefix", prefix, "transport", scheme, "addr", hostport)
-	a.mountVia(prefix, func() (Client, string) { return client, hostport })
+	a.mountVia(prefix, func() (Client, string, hold) { return client, hostport, hold{} })
 	return nil
 }
 
@@ -356,13 +356,16 @@ func (a *App) mount(prefix, addr string) error {
 // Registering the route is a separate concern from CLAIMING the prefix (see
 // [App.claim]): a reload re-resolves the target without re-registering, and an
 // ownership assertion happens once, at the door the composition came in.
-func (a *App) mountVia(prefix string, to func() (Client, string)) {
+//
+// The target comes back HELD: the instance behind it counts the request until it
+// is over, which is what keeps eviction off a process mid-answer.
+func (a *App) mountVia(prefix string, to func() (Client, string, hold)) {
 	h := func(c *Ctx) error {
-		client, host := to()
+		client, host, held := to()
 		if upgrading(c.fc.Request()) {
-			return relay(c.fc.RequestCtx(), host, "mount "+prefix)
+			return relay(c.fc.RequestCtx(), host, "mount "+prefix, held)
 		}
-		return forward(c.Context(), c.fc.Request(), c.fc.Response(), client, host, "", "mount "+prefix)
+		return forward(c.Context(), c.fc.Request(), c.fc.Response(), client, host, "", "mount "+prefix, held)
 	}
 	prefix = strings.TrimSuffix(normPath(prefix), "/")
 	site := here(1)
@@ -388,8 +391,13 @@ func (a *App) mountVia(prefix string, to func() (Client, string)) {
 //
 // what names the caller in the error, because "no instance running" is worth
 // nothing without knowing what did not run.
-func forward(ctx context.Context, req *fasthttp.Request, resp *fasthttp.Response, client Client, host, path, what string) error {
+//
+// h is released when the reply is over: here for a whole reply, when its body is
+// closed for a streamed one. A zero hold — a remote, which no host may stop —
+// takes the direct path.
+func forward(ctx context.Context, req *fasthttp.Request, resp *fasthttp.Response, client Client, host, path, what string, h hold) error {
 	if client == nil {
+		h.release()
 		return Errorf(503, "%s: no instance running", what)
 	}
 	// The caller's Host survives the hop. A client dials its own address and never
@@ -403,13 +411,66 @@ func forward(ctx context.Context, req *fasthttp.Request, resp *fasthttp.Response
 	if path != "" {
 		req.URI().SetPath(path)
 	}
-	if err := do(ctx, client, req, resp); err != nil {
-		// The upstream, not this hop, is what failed.
+	if h.in == nil {
+		if err := do(ctx, client, req, resp); err != nil {
+			// The upstream, not this hop, is what failed.
+			return Errorf(502, "%s: %v", what, err)
+		}
+		return nil
+	}
+	return forwardHeld(ctx, req, resp, client, what, h)
+}
+
+// forwardHeld is forward for a reply whose instance is held. The reply is read
+// into a response of its own, because a streamed body outlives this call and
+// fasthttp offers no way to take a stream back out of a response without closing
+// it: the hold has to ride the body, so the body has to be wrapped before it is
+// handed over.
+func forwardHeld(ctx context.Context, req *fasthttp.Request, resp *fasthttp.Response, client Client, what string, h hold) error {
+	got := fasthttp.AcquireResponse()
+	if err := do(ctx, client, req, got); err != nil {
+		fasthttp.ReleaseResponse(got)
+		h.release()
 		return Errorf(502, "%s: %v", what, err)
 	}
+	if !got.IsBodyStream() {
+		got.CopyTo(resp)
+		fasthttp.ReleaseResponse(got)
+		h.release()
+		return nil
+	}
+	got.Header.CopyTo(&resp.Header)
+	resp.SkipBody = got.SkipBody
+	resp.SetBodyStream(&heldBody{got: got, h: h}, got.Header.ContentLength())
 	return nil
 }
 
+// heldBody is a streamed reply that keeps its instance busy until the reply is
+// read to its end or let go: a token stream is a request in flight for as long
+// as tokens arrive. The writer closes it once, from the goroutine that reads it.
+type heldBody struct {
+	got *fasthttp.Response
+	h   hold
+}
+
+func (b *heldBody) Read(p []byte) (int, error) {
+	if b.got == nil {
+		return 0, io.EOF
+	}
+	return b.got.BodyStream().Read(p)
+}
+
+func (b *heldBody) Close() error {
+	if b.got == nil {
+		return nil
+	}
+	err := b.got.CloseBodyStream()
+	fasthttp.ReleaseResponse(b.got)
+	b.got = nil
+	b.h.release()
+	return err
+}
+
 // 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.
@@ -488,11 +549,22 @@ func upgrading(req *fasthttp.Request) bool {
 // than through the response object — fasthttp would frame a body onto a message
 // that has none. Bytes the plugin has already sent past the header are held in
 // the reader, so the copy starts there and not at the socket.
-func relay(rc *fasthttp.RequestCtx, addr, what string) error {
+//
+// h covers the whole exchange: a refused upgrade releases it on return, a
+// switched one when the tunnel closes, because a WebSocket session is a request
+// in flight for as long as it is open.
+func relay(rc *fasthttp.RequestCtx, addr, what string, h hold) error {
 	to := plain(addr)
 	if to == "" {
+		h.release()
 		return Errorf(502, "%s: no upgrade path over %s", what, NetworkOf(addr))
 	}
+	tunnel := false
+	defer func() {
+		if !tunnel {
+			h.release()
+		}
+	}()
 	up, err := net.DialTimeout(NetworkOf(to), to, upgradeWait)
 	if err != nil {
 		return Errorf(502, "%s: upgrade: %v", what, err)
@@ -525,7 +597,9 @@ func relay(rc *fasthttp.RequestCtx, addr, what string) error {
 
 	switched := head.Header()
 	rc.HijackSetNoResponse(true)
+	tunnel = true
 	rc.Hijack(func(down net.Conn) {
+		defer h.release()
 		defer func() { _ = up.Close() }()
 		defer func() { _ = down.Close() }()
 		if _, err := down.Write(switched); err != nil {

zip.go

@@ -107,6 +107,10 @@ type Config struct {
 	// so it is a ceiling rather than something a sweep restores a minute later,
 	// and again by [App.Evict] alongside each plugin's IdleAfter.
 	//
+	// A process serving a request is never the one stopped for room. When every
+	// candidate is busy, a starter waits for one to finish, for as long as its
+	// own [Plugin.Start] allows, and then answers 503 rather than kill a request.
+	//
 	// The host states it because the number is a fact about the host's memory: the
 	// processes are its own children in its own cgroup, and a library cannot know
 	// how much of it they may have.
@@ -308,6 +312,12 @@ type App struct {
 	// burst outruns. See makeRoom.
 	warm int
 
+	// roomMu makes "is there room" and "take it" one step, and starting counts
+	// the processes that have taken room but are not running yet. Without both,
+	// two starters each read 34 of 35 and each start, and the ceiling is a race.
+	roomMu   sync.Mutex
+	starting int
+
 	// mcpList is the whole tools/list array, rendered by installMCP and re-rendered
 	// by any later load: own ops ++ every plugin catalogue, sorted by name. Serving
 	// bytes is what makes the most-called MCP method free. Atomic because a Load