Merge remote-tracking branch 'origin/main'

# Conflicts:
#	load.go

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
2026-08-21 00:22:13 -07:00
co-authored by Hanzo Dev
23 changed files with 1397 additions and 35 deletions
Symlink
+1
View File
@@ -0,0 +1 @@
LLM.md
+2 -2
View File
@@ -54,7 +54,7 @@ func billingApp() *App {
//
// Its id is derived per occurrence from the ABSOLUTE path it answers at, which
// is what distinguishes the two: ID("GET", "/v1/invoices/:id") is
// get_v1_invoices_by_id, and under /admin it is get_admin_invoices_by_id.
// get_invoices_by_id, and under /admin it is get_admin_invoices_by_id.
func unnamedApp() *App {
b := quiet("billing")
Get(b, "/invoices/:id", func(_ context.Context, in *invoiceIn) (*invoiceOut, error) {
@@ -284,7 +284,7 @@ func TestDiamond_TwoOccurrencesTwoIdsOneType(t *testing.T) {
ids = append(ids, op.OperationID)
paths = append(paths, op.Path)
}
wantIDs := []string{"get_v1_invoices_by_id", "get_admin_invoices_by_id"}
wantIDs := []string{"get_invoices_by_id", "get_admin_invoices_by_id"}
wantPaths := []string{"/v1/invoices/:id", "/admin/invoices/:id"}
if strings.Join(ids, ",") != strings.Join(wantIDs, ",") {
t.Errorf("operation ids = %v, want %v", ids, wantIDs)
+1 -1
View File
@@ -104,7 +104,7 @@ func TestUndeclaredOperationIDComesFromTheAbsolutePath(t *testing.T) {
for _, op := range host.Registry() {
got[op.OperationID] = true
}
for _, want := range []string{"get_v1_invoices_by_id", "get_admin_invoices_by_id"} {
for _, want := range []string{"get_invoices_by_id", "get_admin_invoices_by_id"} {
if !got[want] {
t.Errorf("id %q missing from %v — an UNDECLARED id must be derived from the "+
"absolute path its occurrence answers at", want, got)
+11
View File
@@ -95,6 +95,17 @@ func (a *App) build() (*generation, error) {
g.serve = g.router.Handler()
g.ops = composeOps(occ)
g.hosts = computeHosts(a, occ)
// Adopt: every plugin in this composition learns which app is the ROOT of it.
// This is the only place that knows, and a plugin needs it to ask the two
// questions its own definition cannot answer — how many processes are running
// across the whole host, and what the host's ceiling is.
for _, h := range g.hosts {
h.plugMu.Lock()
for _, pl := range h.plugins {
pl.owner.Store(a)
}
h.plugMu.Unlock()
}
g.routes = len(g.router.GetRoutes(false))
return g, nil
}
+43 -13
View File
@@ -8,33 +8,33 @@ import "testing"
func TestID(t *testing.T) {
for _, c := range []struct{ method, path, want string }{
// The method is a word, the path is the rest, '_' encodes '/'.
{"GET", "/v1/agents/targets", "get_v1_agents_targets"},
{"POST", "/v1/agents/targets", "post_v1_agents_targets"},
{"GET", "/v1/agents/targets", "get_agents_targets"},
{"POST", "/v1/agents/targets", "post_agents_targets"},
// A parameter is "by_<name>", which is what keeps it apart from a
// literal segment of the same name.
{"GET", "/v1/a/{b}", "get_v1_a_by_b"},
{"GET", "/v1/a/b", "get_v1_a_b"},
{"GET", "/v1/a/{b}", "get_a_by_b"},
{"GET", "/v1/a/b", "get_a_b"},
// Every spelling of ONE parameter reaches ONE name. The router's
// constraint and optional marker say how a segment is MATCHED, not what
// it is called (see Template), so they do not reach the id.
{"GET", "/v1/invoices/:id", "get_v1_invoices_by_id"},
{"GET", "/v1/invoices/{id}", "get_v1_invoices_by_id"},
{"GET", "/v1/invoices/:id?", "get_v1_invoices_by_id"},
{"GET", "/v1/invoices/:id<guid>", "get_v1_invoices_by_id"},
{"GET", "/v1/invoices/:id", "get_invoices_by_id"},
{"GET", "/v1/invoices/{id}", "get_invoices_by_id"},
{"GET", "/v1/invoices/:id?", "get_invoices_by_id"},
{"GET", "/v1/invoices/:id<guid>", "get_invoices_by_id"},
// '-' and '.' are legal in an operationId and are PRESERVED, because '_'
// is the separator and folding them collapsed these two onto one id once.
{"GET", "/v1/pricing-policy", "get_v1_pricing-policy"},
{"GET", "/v1/pricing/policy", "get_v1_pricing_policy"},
{"GET", "/v1/pricing-policy", "get_pricing-policy"},
{"GET", "/v1/pricing/policy", "get_pricing_policy"},
// Anything else folds to '_', including a literal '_' — which is the
// residual aliasing the encoding cannot remove and the walk's uniqueness
// check exists to catch.
{"GET", "/v1/a/b_c", "get_v1_a_b_c"},
{"GET", "/v1/a/b/c", "get_v1_a_b_c"},
{"GET", "/v1/Git/Upload-Pack", "get_v1_git_upload-pack"},
{"GET", "/v1/a/b_c", "get_a_b_c"},
{"GET", "/v1/a/b/c", "get_a_b_c"},
{"GET", "/v1/Git/Upload-Pack", "get_git_upload-pack"},
// A wildcard declares no name, so it takes the positional one the
// document gives it, numbered as the document numbers it.
@@ -51,6 +51,36 @@ func TestID(t *testing.T) {
}
}
// The default version is not in the name, because it is on every address and so
// tells no two operations apart. It is dropped ONLY in the leading position and
// ONLY when it is the default: a segment that merely reads like a version deeper
// in a path is a real segment, and a later version is the exception that has to
// stay visible.
func TestIDDropsTheDefaultVersionOnly(t *testing.T) {
for _, c := range []struct{ path, want string }{
{"/v1/agents", "get_agents"}, // the default: gone
{"/agents", "get_agents"}, // an unversioned address already read this way
{"/v2/agents", "get_v2_agents"}, // the exception: kept, and so still unique
{"/v10/agents", "get_v10_agents"},
{"/api/v1/agents", "get_api_v1_agents"}, // not leading: a real segment
{"/v1/v1/agents", "get_v1_agents"}, // only the FIRST one is the version
{"/v1", "get"}, // the version alone names the root
} {
if got := ID("GET", c.path); got != c.want {
t.Errorf("ID(GET, %q) = %q, want %q", c.path, got, c.want)
}
}
}
// Dropping the default version must not make two DIFFERENT addresses share one
// name. A later version keeps its segment precisely so this holds, and the pair
// below is the one that would otherwise collapse.
func TestIDKeepsVersionsApart(t *testing.T) {
if a, b := ID("GET", "/v1/agents"), ID("GET", "/v2/agents"); a == b {
t.Fatalf("/v1/agents and /v2/agents both derive %q — two addresses, one name", a)
}
}
// The router's spelling and the document's spelling of one address are the same
// operation, so they must derive the same id — this is what lets a caller read
// the published document and reach the op by the name it found there.
+285
View File
@@ -0,0 +1,285 @@
package zip
import (
"cmp"
"slices"
"sync"
"time"
)
// Idle plugins cost what a running plugin costs, which is not nothing.
//
// A host that composes many services pays for every one it has ever served,
// forever: Lazy defers the first start, and nothing has ever stopped one. On a
// host running two dozen subsystems that is the dominant resident cost — each
// child holds its own Go heap, and the total tracks the SIZE OF THE CATALOG
// rather than the traffic. Measured on one such host: 24 children, ~150MiB of
// live heap each, ~4.4GiB resident, while three of them (the ones that do
// little) sat at 12-20MiB. The catalog was the bill.
//
// Eviction is the other half of Lazy, and it is deliberately not Unload:
// Unload means "stay down", eviction means "you may go, come back when asked".
// The next request through target() starts it again by the path that already
// exists, so the surface never changes — only the process is transient.
//
// WHY THIS IS SAFE AGAINST THE SUPERVISOR. supervise() restarts a child whose
// process exits, and distinguishes a deliberate stop by CAS-ing cur from the
// instance it supervises to nil: if that fails, someone else already claimed
// the retirement and it returns. Eviction clears cur BEFORE the child dies, so
// 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,
// while more than warm are still running, the least recently used. Zero warm
// means only the first bound applies.
//
// One pass, no goroutine — the caller decides the cadence, so a host under test
// can drive it directly and a host in production can put it on the ticker it
// already has.
//
// THE COUNT IS NOT THIS FUNCTION'S TO ENFORCE ALONE, and an earlier version of
// this comment claimed otherwise: warm was an ARGUMENT here, on the reasoning
// that the number is a fact about the host and not the library. The reasoning
// held; the placement did not. A bound consulted once a minute is outrun by
// anything that starts faster, and a fan-out door that asks every subsystem at
// once does — a hundred children inside ninety seconds, so the process held them
// all until the next sweep, stopped answering its own liveness probe, and was
// killed with the ceiling never applied. Warm is state now and makeRoom applies
// it at the start path; this still runs it, because age and count are one sweep.
//
// Draining uses the plugin's own Drain, so a request in flight when the sweep
// lands finishes on the old process exactly as it would across a Reload.
func (a *App) Evict() int {
now := time.Now()
stopped := 0
for _, p := range a.pluginSet() {
if p.evictIfIdle(now) {
stopped++
}
}
return stopped + a.evictOver(a.warm, now)
}
// host is the ROOT app of the composition this plugin belongs to, which is the
// only app that can answer how many processes are running and what the budget is.
// Falls back to the defining app for a plugin whose composition was never built —
// a Load'ed service used directly in a test, where the two are the same thing.
func (p *plugin) host() *App {
if o := p.owner.Load(); o != nil {
return o
}
return p.app
}
// makeRoom is the ceiling, and it runs where a process is about to exist rather
// than on a ticker. Called from the start path with the starter's own lock held,
// so it must never consider the starter itself: evicting p here would deadlock on
// p.mu, and p is the one plugin about to be needed anyway.
//
// This is the half a sweep cannot do. Evict trims to warm once a minute, which is
// a bound only while starts arrive slower than that. Measured on a host whose MCP
// 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) {
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)
}
}
// The starter is not running yet, so room for it means strictly under.
if live < a.warm || len(evictable) == 0 {
return
}
slices.SortFunc(evictable, func(x, y *plugin) int {
return cmp.Compare(x.lastUse.Load(), y.lastUse.Load())
})
cold := evictable[0]
if !cold.evict("room", now.Sub(time.Unix(0, cold.lastUse.Load()))) {
return // already down; re-reading would spin
}
}
}
// 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.
func (a *App) pluginSet() []*plugin {
var all []*plugin
for _, h := range a.hosts() {
h.plugMu.Lock()
for _, p := range h.plugins {
all = append(all, p)
}
h.plugMu.Unlock()
}
return all
}
// evictOver stops the least recently used evictable plugins until at most warm
// remain running. Zero means unbounded, which is the historical behaviour.
//
// AGE BOUNDS THE STEADY STATE AND COUNT BOUNDS THE BURST, which is why both
// exist. IdleAfter can only reclaim a plugin that has already gone quiet for
// its whole window, so during the minutes after a cold start — when every
// prefix that gets a request starts a child and none is old enough to be idle —
// it reclaims nothing at all. Measured on a host of this shape: 37 children at
// ~152MiB each in steady state, and an OOM kill six minutes after boot, well
// before the first plugin was 15 minutes idle. A count is a bound the burst
// cannot outrun.
//
// LEAST RECENT USE, because the plugin that has gone longest without a request
// is the one whose restart is least likely to be paid for by a caller waiting.
//
// The cap counts every RUNNING plugin, including the ones it may not stop: they
// hold the same memory, so counting only the evictable ones would authorise the
// budget twice. When the unevictable alone exceed warm this reclaims what it
// can and leaves the rest — a host is better over its budget than deprived of
// the identity or config service every other call goes through.
func (a *App) evictOver(warm int, now time.Time) int {
if warm <= 0 {
return 0
}
var (
live int
evictable []*plugin
)
for _, p := range a.pluginSet() {
if p.cur.Load() == nil {
continue
}
live++
if p.spec.Lazy && p.spec.idleAfter() > 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.
slices.SortFunc(evictable, func(x, y *plugin) int {
return cmp.Compare(x.lastUse.Load(), y.lastUse.Load())
})
stopped := 0
for _, p := range evictable {
if live-stopped <= warm {
break
}
if p.evict("lru", now.Sub(time.Unix(0, p.lastUse.Load()))) {
stopped++
}
}
return stopped
}
// 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 {
return false
}
// No lock: the swap that takes the child down happens under p.mu inside
// evict, and this read only decides whether to ask. A request landing
// between the two is safe either way — it is already being served by the
// instance retire() is about to DRAIN, or it arrives after the swap and
// starts a fresh child. So the check buys freshness, not correctness, and
// holding the lock across it would contend with the request path to learn
// something it cannot guarantee anyway.
last := p.lastUse.Load()
if last == 0 || now.Sub(time.Unix(0, last)) < after {
return false
}
return p.evict("idle", now.Sub(time.Unix(0, last)))
}
// evict stops p's current instance and reports whether it stopped one. It is
// 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) bool {
p.mu.Lock()
if p.closed || p.disabled.Load() {
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)
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",
"name", p.name, "pid", in.cmd.Process.Pid, "reason", reason,
"idle", idle.Round(time.Second).String())
}
p.retire(in, p.spec.Drain)
return true
}
// Reap runs Evict on a ticker until the returned function is called. A host
// that wants the behaviour writes one line; a host that does not is unaffected,
// because nothing here runs unless it is asked for.
func (a *App) Reap(every time.Duration) (stop func()) {
if every <= 0 {
every = time.Minute
}
done := make(chan struct{})
finished := make(chan struct{})
go func() {
defer close(finished)
t := time.NewTicker(every)
defer t.Stop()
for {
select {
case <-done:
return
case <-t.C:
a.Evict()
}
}
}()
// stop is SYNCHRONOUS: it returns only once a sweep in progress has
// finished. Signalling and returning is not enough — a sweep reads App
// state that Shutdown writes, so a reaper still running when the host began
// shutting down is a data race, and it is the one the race detector found.
//
// sync.Once, not a bool: a host that stops the reaper from two places (a
// defer and an explicit shutdown path) would otherwise race on the flag and
// double-close the channel. The second call still waits, which is correct —
// finished is closed by then, so it returns at once.
var once sync.Once
return func() {
once.Do(func() { close(done) })
<-finished
}
}
+134
View File
@@ -0,0 +1,134 @@
package zip_test
import (
"fmt"
"os"
"strconv"
"strings"
"testing"
"time"
"github.com/zap-proto/zip"
)
// TestIdle_ReclaimsRealMemory is the claim measured rather than asserted: what
// eviction is FOR is resident memory, so this reads the operating system's
// number for the child processes, not a counter the code keeps about itself.
//
// It warms N plugins, records the resident set of every child, evicts, and
// records it again. A test that only checked Running would pass on an
// implementation that leaked the process while forgetting the pointer.
//
// Run with -v to see the numbers:
//
// go test -run TestIdle_ReclaimsRealMemory -v .
func TestIdle_ReclaimsRealMemory(t *testing.T) {
const plugins = 6
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
for i := 0; i < plugins; i++ {
name := fmt.Sprintf("demo%d", i)
app.Use(must(zip.Load(
zip.Plugin{
Name: name, Bin: bin, Dir: t.TempDir(), Lazy: true,
IdleAfter: 50 * time.Millisecond,
}, "/v1/"+name,
)))
}
defer func() { _ = app.Shutdown() }()
// Warm every one, the way a host serving its whole catalog would.
for i := 0; i < plugins; i++ {
if status, _ := call(t, app, "GET", fmt.Sprintf("/v1/demo%d/version", i), ""); status != 200 {
t.Fatalf("demo%d did not serve", i)
}
}
warmPIDs := runningPIDs(t, app)
if len(warmPIDs) != plugins {
t.Fatalf("warmed %d plugins, want %d", len(warmPIDs), plugins)
}
warmRSS := totalRSS(t, warmPIDs)
time.Sleep(80 * time.Millisecond)
if n := app.Evict(); n != plugins {
t.Fatalf("evicted %d of %d", n, plugins)
}
if !settles(3*time.Second, func() bool { return len(runningPIDs(t, app)) == 0 }) {
t.Fatalf("plugins still running after eviction: %v", runningPIDs(t, app))
}
// The processes are gone from the OS, not merely forgotten by the host.
// This is the half a Running check cannot make.
//
// It has to WAIT for them. retire() is asynchronous by design — the caller
// is not blocked for the drain — so Running goes false the moment cur is
// cleared, which is strictly before the child is reaped. Asserting
// residency at that instant tests the scheduler, not the code: it passed
// on one machine and failed on the next, which is how this comment came to
// be written.
alive := len(warmPIDs)
settles(5*time.Second, func() bool {
alive = 0
for _, pid := range warmPIDs {
if rssOf(pid) > 0 {
alive++
}
}
return alive == 0
})
if alive != 0 {
t.Errorf("%d of %d evicted processes still resident after 5s — the pointer was dropped but the process was not", alive, len(warmPIDs))
}
t.Logf("%d plugins warm: %.1f MiB resident across %d child processes",
plugins, float64(warmRSS)/1024, len(warmPIDs))
t.Logf("after eviction: 0 child processes, %.1f MiB reclaimed", float64(warmRSS)/1024)
t.Logf("per plugin: %.1f MiB — multiply by a catalog to see why this matters",
float64(warmRSS)/1024/float64(plugins))
// And the catalog still works: eviction is not a teardown.
if status, _ := call(t, app, "GET", "/v1/demo0/version", ""); status != 200 {
t.Fatal("a route stopped answering after its plugin was evicted")
}
}
func runningPIDs(t *testing.T, app *zip.App) []int {
t.Helper()
var out []int
for _, p := range app.Plugins() {
if p.Running && p.PID != 0 {
out = append(out, p.PID)
}
}
return out
}
// rssOf reads VmRSS in KiB from /proc, or 0 if the process is gone. The OS is
// the authority here; anything the host reports about itself could be wrong in
// exactly the way this test exists to catch.
func rssOf(pid int) int {
b, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid))
if err != nil {
return 0
}
for _, line := range strings.Split(string(b), "\n") {
if strings.HasPrefix(line, "VmRSS:") {
f := strings.Fields(line)
if len(f) >= 2 {
kb, _ := strconv.Atoi(f[1])
return kb
}
}
}
return 0
}
func totalRSS(t *testing.T, pids []int) int {
t.Helper()
sum := 0
for _, pid := range pids {
sum += rssOf(pid)
}
return sum
}
+166
View File
@@ -0,0 +1,166 @@
package zip_test
import (
"testing"
"time"
"github.com/zap-proto/zip"
)
// TestIdle_EvictsAndComesBack is the whole contract in one pass: a lazy plugin
// that has gone unused is stopped, and the next request starts it again.
//
// The second half matters more than the first. Stopping a process is easy; the
// reason this is eviction and not Unload is that the route still works
// afterwards, and the caller cannot tell except by the pid.
func TestIdle_EvictsAndComesBack(t *testing.T) {
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
app.Use(must(zip.Load(
zip.Plugin{
Name: "demo", Bin: bin, Dir: t.TempDir(), Lazy: true,
IdleAfter: 50 * time.Millisecond,
}, "/v1/demo",
)))
defer func() { _ = app.Shutdown() }()
if status, body := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
t.Fatalf("first request: status=%d body=%q", status, body)
}
first := app.Plugins()[0]
if !first.Running || first.PID == 0 {
t.Fatalf("plugin not running after a served request: %+v", first)
}
// Fresh from a request, so it is NOT idle. An evictor that stops a plugin
// that just served would be a request-latency bug that only shows in prod.
if n := app.Evict(); n != 0 {
t.Fatalf("evicted %d plugins that had just served", n)
}
if !app.Plugins()[0].Running {
t.Fatal("plugin stopped despite serving a moment ago")
}
time.Sleep(80 * time.Millisecond)
if n := app.Evict(); n != 1 {
t.Fatalf("Evict stopped %d, want 1 after idling past IdleAfter", n)
}
// retire() drains asynchronously; with Drain unset there is no grace to
// wait out, so this settles immediately in practice.
if !settles(2*time.Second, func() bool { return !app.Plugins()[0].Running }) {
t.Fatalf("plugin still running after eviction: %+v", app.Plugins()[0])
}
// The route still answers, and answers from a DIFFERENT process.
status, body := call(t, app, "GET", "/v1/demo/version", "")
if status != 200 {
t.Fatalf("request after eviction: status=%d body=%q — eviction must not take the route down", status, body)
}
second := app.Plugins()[0]
if !second.Running || second.PID == 0 {
t.Fatalf("plugin did not restart on demand: %+v", second)
}
if second.PID == first.PID {
t.Fatalf("same pid %d before and after eviction — nothing was actually stopped", first.PID)
}
}
// TestIdle_ZeroNeverEvicts is the default, and the setting an always-hot plugin
// wants: identity or config that every other call goes through must never make
// a caller pay a cold start.
func TestIdle_ZeroNeverEvicts(t *testing.T) {
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
app.Use(must(zip.Load(
zip.Plugin{Name: "demo", Bin: bin, Dir: t.TempDir(), Lazy: true}, // IdleAfter unset
"/v1/demo",
)))
defer func() { _ = app.Shutdown() }()
if status, _ := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
t.Fatal("first request failed")
}
pid := app.Plugins()[0].PID
time.Sleep(60 * time.Millisecond)
if n := app.Evict(); n != 0 {
t.Fatalf("evicted %d with IdleAfter unset — zero must mean never", n)
}
if got := app.Plugins()[0]; !got.Running || got.PID != pid {
t.Fatalf("plugin disturbed despite IdleAfter unset: %+v (was pid %d)", got, pid)
}
}
// TestIdle_EagerIsNeverEvicted: an eager plugin was started at Load on purpose.
// Stopping it would contradict the instruction that started it, and its next
// request would pay a start it was configured never to pay.
func TestIdle_EagerIsNeverEvicted(t *testing.T) {
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
app.Use(must(zip.Load(
zip.Plugin{
Name: "demo", Bin: bin, Dir: t.TempDir(), // Lazy false
IdleAfter: 10 * time.Millisecond,
}, "/v1/demo",
)))
defer func() { _ = app.Shutdown() }()
pid := app.Plugins()[0].PID
if pid == 0 {
t.Fatal("eager plugin did not start at Load")
}
// Serve one request first. Without it lastUse stays zero and evictIfIdle
// returns on that check before it ever consults Lazy — the test would pass
// whether or not the eager guard exists, which a mutation proved it did.
if status, _ := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
t.Fatal("eager plugin did not serve")
}
time.Sleep(40 * time.Millisecond)
if n := app.Evict(); n != 0 {
t.Fatalf("evicted %d eager plugins — IdleAfter applies to lazy only", n)
}
if got := app.Plugins()[0]; !got.Running || got.PID != pid {
t.Fatalf("eager plugin stopped: %+v (was pid %d)", got, pid)
}
}
// TestIdle_ReapRunsAndStops covers the ticker wrapper, including that its
// stop function is idempotent — a host calling it twice (defer plus an explicit
// shutdown path) must not panic on a closed channel.
func TestIdle_ReapRunsAndStops(t *testing.T) {
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
app.Use(must(zip.Load(
zip.Plugin{
Name: "demo", Bin: bin, Dir: t.TempDir(), Lazy: true,
IdleAfter: 20 * time.Millisecond,
}, "/v1/demo",
)))
defer func() { _ = app.Shutdown() }()
if status, _ := call(t, app, "GET", "/v1/demo/version", ""); status != 200 {
t.Fatal("first request failed")
}
stop := app.Reap(10 * time.Millisecond)
if !settles(3*time.Second, func() bool { return !app.Plugins()[0].Running }) {
t.Fatal("Reap never evicted an idle plugin")
}
stop()
stop() // idempotent
}
func settles(limit time.Duration, ok func() bool) bool {
deadline := time.Now().Add(limit)
for time.Now().Before(deadline) {
if ok() {
return true
}
time.Sleep(5 * time.Millisecond)
}
return ok()
}
+30 -2
View File
@@ -127,6 +127,13 @@ type Plugin struct {
// requests already in flight on it finish. Zero means 5s.
Drain time.Duration
// IdleAfter stops a LAZY plugin that has not served for this long, to be
// started again by the next request that needs it. Zero means never, which
// is the historical behaviour. See idle.go: without it a host pays for
// every plugin it has ever served, so resident memory tracks the size of
// the catalog rather than the traffic.
IdleAfter time.Duration
// Lazy defers starting the child until the first request actually reaches
// one of its prefixes. Routes register at Load either way, so the surface
// is identical — only the process is deferred.
@@ -201,12 +208,23 @@ type plugin struct {
prefixes []string // every subtree this plugin answers
spec Plugin
app *App // for logging and supervision when started on demand
app *App // for logging and supervision when started on demand
// 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
// the wrong thing to ask "how many are running" or "what is the budget".
owner atomic.Pointer[App]
cur atomic.Pointer[instance]
mu sync.Mutex
reloads atomic.Int64
restarts atomic.Int64
closed bool // Shutdown ran; the supervisor must stop resurrecting it
// lastUse is the unix-nano of the most recent target() resolve, and
// evictions counts idle stops. Together they are what Evict reads: a
// plugin nothing has asked for is one nothing needs running.
lastUse atomic.Int64
evictions atomic.Int64
closed bool // Shutdown ran; the supervisor must stop resurrecting it
// disabled means Unload took it down deliberately. Without this a LAZY
// plugin — the affordable default for a host composing many services — is
@@ -257,6 +275,11 @@ type instance struct {
// keeps the steady-state cost to one load once it is running.
func (p *plugin) target() (Client, string) {
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())
}
return in.client, in.sock
}
if p.app == nil || !p.spec.Lazy || p.disabled.Load() {
@@ -310,6 +333,10 @@ func (p *plugin) startOnDemand() (Client, string) {
if p.closed || p.disabled.Load() {
return nil, ""
}
// 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)
in, err := start(p.spec)
if err != nil {
p.app.logger.Error("zip lazy plugin failed to start", "name", p.name, "err", err)
@@ -339,6 +366,7 @@ func (p *plugin) startOnDemand() (Client, string) {
}
return nil, ""
}
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)
+29 -2
View File
@@ -103,7 +103,20 @@ type Source interface {
const mimeJSON = "application/json"
// mcpProtocolVersion is the MCP spec revision zip implements.
const mcpProtocolVersion = "2025-06-18"
//
// 2026-07-28 made MCP STATELESS, and zip already was: there is no
// Mcp-Session-Id anywhere in this file, nothing is remembered between requests,
// and tools/list answers without a preceding initialize. So this door met the
// new revision's central requirement before the revision existed, and went on
// announcing the old one — which is the version a client negotiates against.
//
// Announcing it is therefore a statement about what is implemented, not an
// upgrade: stateless transport, no mandatory handshake, server/discover
// answered, and initialize kept for older clients.
//
// What this revision DEPRECATES, zip never had — roots, sampling, logging, and
// the HTTP+SSE transport are absent here, so nothing needs a migration window.
const mcpProtocolVersion = "2026-07-28"
// defaultMCPPath is where an app serves its own MCP door unless MCPConfig moves
// it, and therefore where a host forwards a composed tools/call.
@@ -325,7 +338,21 @@ func (a *App) MCP(ctx context.Context, f *zapmcp.Frame) *zapmcp.Frame {
// possible way for an inversion to be incomplete.
a.mcpOnce.Do(a.renderTools)
switch f.Method {
case "initialize":
// initialize and server/discover answer the SAME thing, because under a
// stateless protocol they ask the same question.
//
// 2026-07-28 removed the mandatory handshake: there is no session to open, so
// `initialize` stopped being a lifecycle step and became one more way to ask
// "what are you". `server/discover` is the name the new spec gives that
// question, and it is optional — which is exactly why it must be answered
// rather than skipped. A client that uses it to inspect a server before
// committing gets `method not found` from a server that is otherwise fully
// stateless, and reads the whole door as pre-2026.
//
// `initialize` stays for every client still speaking the older revision. It
// costs one case and keeps them working, which is the compatibility the spec's
// own migration path assumes.
case "initialize", "server/discover":
return f.Answer(mcpJSON(map[string]any{
"protocolVersion": mcpProtocolVersion,
"capabilities": map[string]any{"tools": map[string]any{"listChanged": false}},
+4 -4
View File
@@ -85,7 +85,7 @@ func running(t *testing.T, app *zip.App) []string {
func TestMCP_ComposedListWakesNothing(t *testing.T) {
bin := buildPluginBin(t, "v1")
tools := catalogue(t, bin)
if !strings.Contains(string(tools), "get_v1_demo_version") {
if !strings.Contains(string(tools), "get_demo_version") {
t.Fatalf("plugin catalogue does not name its own op: %s", tools)
}
@@ -125,7 +125,7 @@ func TestMCP_ComposedListWakesNothing(t *testing.T) {
}
var found bool
for _, n := range names {
if n == "get_v1_demo_version" {
if n == "get_demo_version" {
found = true
}
}
@@ -150,7 +150,7 @@ func TestMCP_ComposedListWakesNothing(t *testing.T) {
// tools/call — the ONE trigger, and it wakes ONE child.
status, body := call(t, app, "POST", "/v1/mcp",
`{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"get_v1_demo_version","arguments":{}}}`)
`{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"get_demo_version","arguments":{}}}`)
if status != 200 {
t.Fatalf("tools/call: status %d body=%s", status, body)
}
@@ -182,7 +182,7 @@ func TestMCP_DuplicateToolNameRefusedAtBuild(t *testing.T) {
if err == nil {
t.Fatal("two plugins claiming one tool name must be refused")
}
if !strings.Contains(err.Error(), "get_v1_demo_") ||
if !strings.Contains(err.Error(), "get_demo_") ||
!strings.Contains(err.Error(), `"a"`) || !strings.Contains(err.Error(), `"b"`) {
t.Fatalf("the refusal must name the tool and BOTH claimants: %v", err)
}
+44
View File
@@ -0,0 +1,44 @@
package zip
import (
"context"
"encoding/json"
"testing"
zapmcp "github.com/zap-proto/mcp"
)
// 2026-07-28 removed the mandatory handshake, so a client may inspect a server
// with server/discover instead of initialize. Answering one and not the other
// reads as a pre-2026 door however stateless the transport already is.
func TestDiscoverAnswersLikeInitialize(t *testing.T) {
a := New(Config{})
get := func(method string) map[string]any {
f := a.MCP(context.Background(), &zapmcp.Frame{Method: method, Kind: zapmcp.Request, ID: "1"})
if f == nil {
t.Fatalf("%s: nil frame", method)
}
b, err := json.Marshal(f)
if err != nil {
t.Fatalf("%s: marshal: %v", method, err)
}
var env struct {
Result map[string]any `json:"result"`
Error any `json:"error"`
}
if err := json.Unmarshal(b, &env); err != nil {
t.Fatalf("%s: unmarshal %s: %v", method, b, err)
}
if env.Error != nil {
t.Fatalf("%s: error %v", method, env.Error)
}
return env.Result
}
in, disc := get("initialize"), get("server/discover")
if in["protocolVersion"] != "2026-07-28" {
t.Fatalf("protocolVersion = %v, want 2026-07-28", in["protocolVersion"])
}
if disc["protocolVersion"] != in["protocolVersion"] {
t.Fatalf("server/discover %v != initialize %v", disc["protocolVersion"], in["protocolVersion"])
}
}
+1 -1
View File
@@ -85,7 +85,7 @@ func TestOpen_NamedCallerGetsItsOwnTools(t *testing.T) {
switch n {
case "acme_own":
mine = true
case "get_v1_demo_version":
case "get_demo_version":
projected = true
}
}
+25 -2
View File
@@ -284,8 +284,8 @@ func (a *App) buildOpenAPI() map[string]any {
// "everything non-alphanumeric becomes _" rule, collapsed the two onto one id.
// Hyphenated addresses are permanent (…/git-upload-pack, …/documents/delete-batch),
// so '-' and '.' — both legal in an operationId — are PRESERVED rather than
// folded, which keeps such a pair distinct: get_v1_pricing-policy is not
// get_v1_pricing_policy.
// folded, which keeps such a pair distinct: get_pricing-policy is not
// get_pricing_policy.
//
// A parameter contributes "by_<name>", so /v1/a/{b} and /v1/a/b do not collapse
// either. Every spelling of one parameter reaches the same name: the router's
@@ -297,14 +297,32 @@ func (a *App) buildOpenAPI() map[string]any {
// '/' (/v1/a/b_c against /v1/a/b/c). The walk VERIFIES uniqueness across the
// whole program and refuses to publish a duplicate — that check, not this
// encoding, is what makes the ids trustworthy.
//
// THE LEADING v1 IS DROPPED, because it distinguishes nothing. A segment earns
// its place in an id by telling two operations apart, and this one is on every
// address in the program: it is a constant, and a constant repeated 2,000 times
// is noise carried by four surfaces at once — the document, the tool name an
// agent reads on every turn, the by-name key and the CLI's spelling.
//
// Only v1 goes. A later version is the EXCEPTION and keeps its segment, which
// is both how it stays honest and how it stays unique: get_agents is /v1/agents
// and get_v2_agents is /v2/agents, so the two cannot collapse onto one name the
// walk would then refuse. The default is implicit; a departure from it is not.
func ID(method, path string) string {
var b strings.Builder
b.WriteString(strings.ToLower(method))
stars := 0
first := true
for _, seg := range strings.Split(path, "/") {
if seg == "" {
continue
}
if first {
first = false
if seg == defaultVersion {
continue // says nothing: every address carries it
}
}
b.WriteByte('_')
switch {
case seg == "*" || seg == "+":
@@ -324,6 +342,11 @@ func ID(method, path string) string {
return b.String()
}
// defaultVersion is the version every address is expected to carry, and so the
// one [ID] leaves out of a name. It is a value here rather than a rule spread
// across callers: one place decides what "unversioned by default" means.
const defaultVersion = "v1"
// sanitize reduces a path segment to [a-z0-9.-], the characters that are legal
// in an operationId and cannot be confused with the '_' path separator.
func sanitize(s string) string {
+1 -1
View File
@@ -255,7 +255,7 @@ func TestMCPCallIsUnaffectedByQueryBinding(t *testing.T) {
t.Fatalf("Build: %v", err)
}
req, _ := http.NewRequest("POST", "/mcp", stringReader(
`{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_v1_t_search","arguments":{"q":"viajson","limit":9}}}`))
`{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_t_search","arguments":{"q":"viajson","limit":9}}}`))
req.Header.Set("Content-Type", "application/json")
resp, err := a.Fiber().Test(req)
if err != nil {
+1 -1
View File
@@ -140,7 +140,7 @@ func TestSchema_SelfReferentialInputTerminates(t *testing.T) {
func TestSchema_MCPToolSchemaIsSelfContained(t *testing.T) {
var tree map[string]any
for _, tool := range schemaApp(t).MCPTools() {
if tool["name"] == "post_v1_s_tree" {
if tool["name"] == "post_s_tree" {
tree, _ = tool["inputSchema"].(map[string]any)
}
}
+1 -1
View File
@@ -32,7 +32,7 @@ func TestToolNames_RideTheOperationIDRule(t *testing.T) {
if len(got) != 2 {
t.Fatalf("two occurrences produced %d tool names, want 2: %v", len(got), got)
}
for _, want := range []string{"get_v1_invoices_by_id", "get_admin_invoices_by_id"} {
for _, want := range []string{"get_invoices_by_id", "get_admin_invoices_by_id"} {
if !names[want] {
t.Errorf("tool name %q missing — own-op tool names must be the occurrence's operation id: %v", want, got)
}
+186 -2
View File
@@ -1,11 +1,16 @@
package zip
import (
"bufio"
"bytes"
"fmt"
"io"
"net"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/valyala/fasthttp"
"github.com/zap-proto/http"
@@ -221,6 +226,14 @@ func (a *App) listenOn(addrs []string) error {
t.applyConfig(a.cfg)
}
servers = append(servers, s)
// The plain-HTTP listener that lets an upgrade reach this app. [plain]
// derives its address and the host dials it; this is the half that
// answers. Appended to servers so it shuts down with the rest, and NOT
// bound: it is the same app at a derived address, not a second place to
// find it.
if p := a.plainSibling(addr, h); p != nil {
servers = append(servers, p)
}
// This app now answers at this address, and a caller in THIS process can
// reach its ops without the address (see [Serving] and [Here]). Recorded
// here rather than at Serve because binding is what makes it true.
@@ -289,6 +302,9 @@ func (a *App) mount(prefix, addr string) error {
func (a *App) mountVia(prefix string, to func() (Client, string)) {
h := func(c *Ctx) error {
client, host := to()
if upgrading(c.fc.Request()) {
return relay(c.fc.RequestCtx(), host, "mount "+prefix)
}
return forward(c.fc.Request(), c.fc.Response(), client, host, "", "mount "+prefix)
}
prefix = strings.TrimSuffix(normPath(prefix), "/")
@@ -337,6 +353,153 @@ func forward(req *fasthttp.Request, resp *fasthttp.Response, client Client, host
return nil
}
// 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.
const upgradeWait = time.Second
// plain is where a mounted plugin speaks ordinary HTTP, derived from the address
// it speaks ZAP on.
//
// A plugin's socket carries ZAP frames: one length-prefixed request, one
// length-prefixed reply. That shape has no room for a connection that STOPS being
// request-and-reply, which is exactly what an upgrade is — so a websocket cannot
// cross it, and the frames a client sends after the handshake arrive at an HTTP
// parser as nonsense ("error when reading request headers", the first byte being
// 0x81). The plugin therefore listens a second time, in plain HTTP, beside the
// first. This DERIVES that address instead of configuring it, so a mount is still
// one address and there is no second value to keep in step with the first.
//
// Only over a unix socket, because that is the one place the sibling is known to
// be the same process. A tcp mount would need a port nobody has agreed on.
func plain(addr string) string {
if networkOf(addr) != "unix" {
return ""
}
return addr + ".http"
}
// plainSibling is the listener that answers at [plain] — the second, ordinary
// HTTP listener a mounted app needs because an upgrade cannot cross ZAP framing.
// Without it the host dials a socket nobody created and every websocket through
// a mount fails the handshake.
//
// nil when the address has no sibling. A tcp address has none (no port both ends
// have agreed on), and neither does an app that is ITSELF a sibling — ops and
// peer are reached by name from inside the fleet, never mounted, so a second
// socket beside them would be one nothing dials.
func (a *App) plainSibling(addr string, h fasthttp.RequestHandler) Server {
if a.sibling {
return nil
}
to := plain(addr)
if to == "" {
return nil
}
// Resolved through the registry rather than constructed here, so the sibling
// is the same HTTP transport a caller gets from an http:// address and there
// is one implementation of "serve plain HTTP" to keep correct.
_, at, t, err := transportFor("http://" + to)
if err != nil || t.Serve == nil {
a.logger.Error("zip: no http transport for the upgrade sibling", "addr", to, "err", err)
return nil
}
s := t.Serve(at, h)
if t, ok := s.(tunableServer); ok {
t.applyConfig(a.cfg)
}
a.logger.Info("zip listening", "transport", "http", "addr", at, "for", "upgrade")
return s
}
// upgrading reports whether a request is asking to stop being HTTP.
func upgrading(req *fasthttp.Request) bool {
return len(req.Header.Peek(fasthttp.HeaderUpgrade)) > 0 &&
bytes.Contains(bytes.ToLower(req.Header.Peek(fasthttp.HeaderConnection)),
[]byte("upgrade"))
}
// relay carries an upgraded connection between a caller and a mounted plugin.
//
// Everything up to the switch is ordinary — the request goes up as it arrived,
// and a plugin that declines (an unauthorized socket is refused BEFORE it opens)
// answers an ordinary reply that comes straight back. What is different is what
// happens after 101: neither end is speaking HTTP any more, so there is nothing
// left to parse and the only correct act is to move bytes until one side stops.
//
// The switch reply is written verbatim, from the bytes the plugin sent, rather
// 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 {
to := plain(addr)
if to == "" {
return Errorf(502, "%s: no upgrade path over %s", what, networkOf(addr))
}
up, err := net.DialTimeout(networkOf(to), to, upgradeWait)
if err != nil {
return Errorf(502, "%s: upgrade: %v", what, err)
}
bw := bufio.NewWriter(up)
if err := rc.Request.Write(bw); err == nil {
err = bw.Flush()
}
if err != nil {
_ = up.Close()
return Errorf(502, "%s: upgrade: %v", what, err)
}
br := bufio.NewReader(up)
var head fasthttp.ResponseHeader
if err := head.Read(br); err != nil {
_ = up.Close()
return Errorf(502, "%s: upgrade: %v", what, err)
}
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)
}
rc.Response.SetBody(body)
return nil
}
switched := head.Header()
rc.HijackSetNoResponse(true)
rc.Hijack(func(down net.Conn) {
defer func() { _ = up.Close() }()
defer func() { _ = down.Close() }()
if _, err := down.Write(switched); err != nil {
return
}
go func() {
_, _ = io.Copy(up, down)
// The caller is gone. Close write so the plugin's read sees EOF and
// ends its session rather than holding a shell nobody is watching.
if c, ok := up.(interface{ CloseWrite() error }); ok {
_ = c.CloseWrite()
}
}()
_, _ = io.Copy(down, br)
})
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.
@@ -392,8 +555,21 @@ type httpServer struct {
srv *fasthttp.Server
}
func (h *httpServer) ListenAndServe() error { return h.srv.ListenAndServe(h.addr) }
func (h *httpServer) Close() error { return h.srv.Shutdown() }
// ListenAndServe binds the address, over tcp or over a unix socket.
//
// A PATH IS A SOCKET on this transport for the same reason it is on the ZAP one:
// the address names where the bytes are spoken and the shape of the address says
// which kind of place that is. It earns its keep next to a plugin — an upgraded
// connection cannot cross ZAP framing, so a plugin listens HTTP beside its socket
// and the host relays to it — and a path there must not mean "a tcp host called
// /var/lib/…", which is what binding it as tcp amounts to.
func (h *httpServer) ListenAndServe() error {
if networkOf(h.addr) == "unix" {
return h.srv.ListenAndServeUNIX(h.addr, 0o600)
}
return h.srv.ListenAndServe(h.addr)
}
func (h *httpServer) Close() error { return h.srv.Shutdown() }
// tunableServer is a transport whose underlying server accepts the App's
// per-conn wire tuning. Listen applies it after construction so zip.Config's
@@ -410,6 +586,14 @@ func (h *httpServer) applyConfig(cfg Config) {
if cfg.ReadBufferSize > 0 {
h.srv.ReadBufferSize = cfg.ReadBufferSize
}
// The body ceiling reaches the wire only here. fiber binds its own BodyLimit
// when fiber owns the listener, and on this transport it does not, so without
// this the socket keeps fasthttp's 4 MiB default. fasthttp refuses an
// oversized body before any handler runs, answering 400 "Error when parsing
// request".
if cfg.BodyLimit > 0 {
h.srv.MaxRequestBodySize = cfg.BodyLimit
}
if cfg.WriteBufferSize > 0 {
h.srv.WriteBufferSize = cfg.WriteBufferSize
}
+50
View File
@@ -130,6 +130,56 @@ func TestHTTPTransport_ReadBufferSize_Raises431Ceiling(t *testing.T) {
}
}
// TestHTTPTransport_BodyLimitReachesTheSocket asserts zip.Config.BodyLimit binds
// the socket in both directions: raised, a body past fasthttp's 4 MiB default is
// served; lowered, a body past the App's own ceiling is refused. It drives a real
// listener, because fiber's config and this transport's server are separate
// objects and only the second one answers the wire.
func TestHTTPTransport_BodyLimitReachesTheSocket(t *testing.T) {
post := func(n int) string {
return fmt.Sprintf("POST /v1/echo HTTP/1.1\r\nHost: x\r\nContent-Type: application/octet-stream\r\n"+
"Content-Length: %d\r\nConnection: close\r\n\r\n%s", n, strings.Repeat("a", n))
}
serve := func(name string, limit int) string {
app := zip.New(zip.Config{AppName: name, DisableStartupMessage: true, BodyLimit: limit})
app.Post("/v1/echo", func(c *zip.Ctx) error { return c.JSON(200, map[string]int{"n": len(c.Body())}) })
addr := freeAddr(t)
go func() { _ = app.Listen("http://" + addr) }()
t.Cleanup(func() { _ = app.Shutdown() })
waitDialable(t, addr)
return addr
}
// Raised: 5 MiB is past fasthttp's 4 MiB default and inside an 8 MiB App.
if !answered200(t, serve("raised", 8<<20), post(5<<20)) {
t.Errorf("BodyLimit 8 MiB: 5 MiB body was refused, want 200")
}
// Lowered: 2 MiB is past a 1 MiB App and inside fasthttp's default, so a 200
// here means the socket is on the default rather than on Config.
if answered200(t, serve("lowered", 1<<20), post(2<<20)) {
t.Errorf("BodyLimit 1 MiB: 2 MiB body was served, want a refusal")
}
}
// answered200 reports whether the handler answered 200. A write or read failure is a
// refusal, not a test failure: fasthttp drops the connection mid-body when the
// limit is crossed, so the peer never gets a status line to parse.
func answered200(t *testing.T, addr, rawRequest string) bool {
t.Helper()
c, err := net.DialTimeout("tcp", addr, 2*time.Second)
if err != nil {
t.Fatalf("dial %s: %v", addr, err)
}
defer func() { _ = c.Close() }()
_ = c.SetDeadline(time.Now().Add(5 * time.Second))
if _, err := io.WriteString(c, rawRequest); err != nil {
return false
}
line, err := bufio.NewReader(c).ReadString('\n')
return err == nil && strings.HasPrefix(line, "HTTP/1.1 200")
}
// TestHTTPTransport_ServerHeaderCoversPreRoutingErrors proves fasthttp's OWN
// pre-routing error responses (431 on header overflow) carry the App's
// ServerHeader and NEVER the framework default "fasthttp"/"zip". Those bytes are
+121
View File
@@ -0,0 +1,121 @@
package zip_test
import (
"bytes"
"net"
"testing"
"time"
"github.com/fasthttp/websocket"
"github.com/zap-proto/zip"
"github.com/zap-proto/zip/wsx"
)
// A MOUNTED PLUGIN'S WEBSOCKET REACHES THE CALLER.
//
// A plugin's socket carries ZAP frames — one request, one reply — and an
// upgraded connection is neither. The host wrote the 101 through and then went
// on parsing HTTP, so the first frame the client sent arrived at the header
// parser as nonsense (first byte 0x81) and the connection died. Every websocket
// behind a mounted plugin failed on its first message, and the session the
// plugin meant to run after the handshake never started.
//
// These pin both halves: the switch is relayed and the bytes flow both ways,
// and a plugin that REFUSES to switch still answers its own reply.
// box is a plugin in the shape production runs: ONE address, the way a plugin
// main is written. The plain-HTTP sibling an upgrade needs is derived from that
// address, not passed alongside it — so this also pins that Listen creates it.
// Naming the sibling here instead would test the relay while assuming away the
// wiring, and a mounted websocket would still fail everywhere it is really used.
func box(t *testing.T) string {
t.Helper()
dir := sockDir(t)
sock := dir + "/box.sock"
child := zip.New(zip.Config{AppName: "box", DisableStartupMessage: true})
child.Get("/v1/box/ws", wsx.Upgrade(func(c *wsx.Conn) error {
for {
typ, msg, err := c.ReadMessage()
if err != nil {
return err
}
if err := c.WriteMessage(typ, bytes.ToUpper(msg)); err != nil {
return err
}
}
}))
// Refused BEFORE the upgrade, which is how an unticketed terminal is refused.
child.Get("/v1/box/shut", func(c *zip.Ctx) error {
return zip.ErrUnauthorized("no ticket")
})
go func() { _ = child.Listen(sock) }()
waitSock(t, sock)
waitSock(t, sock+".http")
t.Cleanup(func() { _ = child.Shutdown() })
host := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
loaded, err := zip.Load(zip.Plugin{Name: "box", Addr: sock}, "/v1/box")
if err != nil {
t.Fatalf("load: %v", err)
}
host.Use(loaded)
probe, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("port: %v", err)
}
addr := probe.Addr().String()
_ = probe.Close()
go func() { _ = host.Listen("http://" + addr) }()
for i := 0; i < 200; i++ {
if c, err := net.Dial("tcp", addr); err == nil {
_ = c.Close()
break
}
time.Sleep(20 * time.Millisecond)
}
t.Cleanup(func() { _ = host.Shutdown() })
return addr
}
func TestMountedPluginServesAWebsocket(t *testing.T) {
addr := box(t)
d := websocket.Dialer{HandshakeTimeout: 10 * time.Second}
conn, _, err := d.Dial("ws://"+addr+"/v1/box/ws", nil)
if err != nil {
t.Fatalf("dial through the mount: %v", err)
}
defer func() { _ = conn.Close() }()
_ = conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
// PAST THE SWITCH: the frame that used to reach an HTTP header parser.
if err := conn.WriteMessage(websocket.TextMessage, []byte("hello")); err != nil {
t.Fatalf("send: %v", err)
}
_, got, err := conn.ReadMessage()
if err != nil {
t.Fatalf("recv: %v", err)
}
if string(got) != "HELLO" {
t.Fatalf("want HELLO, got %q", got)
}
}
func TestMountedPluginMayRefuseToUpgrade(t *testing.T) {
addr := box(t)
d := websocket.Dialer{HandshakeTimeout: 10 * time.Second}
_, resp, err := d.Dial("ws://"+addr+"/v1/box/shut", nil)
if err == nil {
t.Fatal("want the plugin's refusal, got a socket")
}
if resp == nil {
t.Fatalf("want a reply to read, got only %v", err)
}
if resp.StatusCode != 401 {
t.Fatalf("want 401, got %d", resp.StatusCode)
}
}
+202
View File
@@ -0,0 +1,202 @@
package zip_test
import (
"fmt"
"testing"
"time"
"github.com/zap-proto/zip"
)
// The ceiling is enforced where a plugin STARTS, so most of these tests never
// call Evict: they ask for plugins and read how many processes exist. That is the
// design — a bound restored by a sweep is not a bound, because the burst that
// matters is over before the sweep runs.
//
// Dir is sockDir(t) and not t.TempDir(): a unix socket path is bounded at ~108
// bytes and t.TempDir() spells the TEST'S NAME into it, so a descriptive name
// pushes the child's listen past the limit and it dies with `bind: invalid
// argument` — which surfaces as a 503 that reads exactly like a lazy plugin
// declining to start.
func warmHost(t *testing.T, n, warm int, idle time.Duration) *zip.App {
t.Helper()
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true, Warm: warm})
for i := range n {
name := fmt.Sprintf("demo%d", i)
app.Use(must(zip.Load(
zip.Plugin{
Name: name, Bin: bin, Dir: sockDir(t), Lazy: true,
IdleAfter: idle,
}, "/v1/"+name,
)))
}
t.Cleanup(func() { _ = app.Shutdown() })
return app
}
// ask serves one request per plugin, oldest first with a gap, so last-use order is
// demo0 < demo1 < ... and "the coldest" is deterministic rather than whatever the
// clock did. It returns the PEAK number of processes seen along the way, which is
// the figure a trim-afterwards implementation gets wrong.
func ask(t *testing.T, app *zip.App, n int) (peak int) {
t.Helper()
for i := range n {
if status, body := call(t, app, "GET", fmt.Sprintf("/v1/demo%d/version", i), ""); status != 200 {
t.Fatalf("demo%d did not serve: status=%d body=%q", i, status, body)
}
if live := len(runningPIDs(t, app)); live > peak {
peak = live
}
time.Sleep(2 * time.Millisecond)
}
return peak
}
// TestWarm_CeilingHoldsDuringABurst is the property a sweep cannot provide, and
// the one whose absence took a production pod down. NO sweep runs here — no Reap,
// no Evict — so the only thing that can hold the line is the start path. The count
// is read after EVERY request, because the end is exactly where a
// trim-afterwards implementation also looks correct.
func TestWarm_CeilingHoldsDuringABurst(t *testing.T) {
const plugins, warm = 12, 4
app := warmHost(t, plugins, warm, time.Hour) // an hour: age can reclaim nothing
peak := ask(t, app, plugins)
if peak > warm {
t.Fatalf("peak %d processes against a ceiling of %d — the bound is being "+
"restored by a sweep instead of enforced at the start", peak, warm)
}
if peak != warm { // or this proves only that twelve fit
t.Errorf("peak was %d, want exactly the ceiling %d", peak, warm)
}
t.Logf("%d sequential cold starts, peak live %d, ceiling %d", plugins, peak, warm)
// A ceiling is not an Unload: the route evicted first still answers.
if status, _ := call(t, app, "GET", "/v1/demo0/version", ""); status != 200 {
t.Fatal("the coldest plugin stopped answering after being evicted for room")
}
}
// TestWarm_KeepsTheMostRecentlyUsed checks WHICH survive, not just how many: a
// policy that shed the hottest would satisfy a count assertion and be exactly
// backwards.
func TestWarm_KeepsTheMostRecentlyUsed(t *testing.T) {
const plugins, warm = 6, 2
app := warmHost(t, plugins, warm, time.Hour)
ask(t, app, plugins)
live := map[string]bool{}
for _, p := range app.Plugins() {
if p.Running {
live[p.Name] = true
}
}
for _, n := range []string{"demo4", "demo5"} {
if !live[n] {
t.Errorf("%s is not running, but it is among the %d most recently used", n, warm)
}
}
for _, n := range []string{"demo0", "demo1", "demo2", "demo3"} {
if live[n] {
t.Errorf("%s survived, but it is colder than the survivors", n)
}
}
}
// TestWarm_ZeroIsUnbounded pins the default: a host that states no ceiling keeps
// the behaviour it had before one existed.
func TestWarm_ZeroIsUnbounded(t *testing.T) {
const plugins = 4
app := warmHost(t, plugins, 0, time.Hour)
if peak := ask(t, app, plugins); peak != plugins {
t.Fatalf("peak %d of %d with no ceiling — zero must mean unbounded", peak, plugins)
}
if n := app.Evict(); n != 0 {
t.Fatalf("Evict stopped %d with no ceiling and nothing idle", n)
}
}
// TestWarm_UnderTheCeilingStopsNothing: it is a ceiling, not a target. A host
// holding fewer processes than it may must not pay a cold start to reach it.
func TestWarm_UnderTheCeilingStopsNothing(t *testing.T) {
app := warmHost(t, 2, 5, time.Hour)
if peak := ask(t, app, 2); peak != 2 {
t.Fatalf("peak %d, want 2", peak)
}
if n := app.Evict(); n != 0 {
t.Fatalf("Evict stopped %d with 2 running under a ceiling of 5", n)
}
}
// TestWarm_ExemptPluginsCountButAreNotStopped is the honest half. IdleAfter zero
// means never-evict, and such a plugin holds memory like any other — so it must
// COUNT against the ceiling (or the budget is authorised twice) and must still not
// be stopped (or "never" meant nothing). When the exempt alone fill the ceiling the
// host goes OVER it rather than shedding the identity service every other call
// goes through.
func TestWarm_ExemptPluginsCountButAreNotStopped(t *testing.T) {
bin := buildPlugin(t, "v1")
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true, Warm: 2})
for _, p := range []zip.Plugin{
{Name: "keep0", Bin: bin, Dir: sockDir(t), Lazy: true}, // IdleAfter unset: exempt
{Name: "keep1", Bin: bin, Dir: sockDir(t), Lazy: true},
{Name: "drop0", Bin: bin, Dir: sockDir(t), Lazy: true, IdleAfter: time.Hour},
{Name: "drop1", Bin: bin, Dir: sockDir(t), Lazy: true, IdleAfter: time.Hour},
} {
app.Use(must(zip.Load(p, "/v1/"+p.Name)))
}
defer func() { _ = app.Shutdown() }()
for _, n := range []string{"keep0", "keep1", "drop0", "drop1"} {
if status, body := call(t, app, "GET", "/v1/"+n+"/version", ""); status != 200 {
t.Fatalf("%s did not serve: status=%d body=%q", n, status, body)
}
time.Sleep(2 * time.Millisecond)
}
for _, p := range app.Plugins() {
switch p.Name {
case "keep0", "keep1":
if !p.Running {
t.Errorf("%s was evicted, but IdleAfter unset means never", p.Name)
}
case "drop0":
if p.Running {
t.Error("drop0 survived: drop1's start should have taken the coldest evictable")
}
}
}
// Two exempt plus the one evictable that just served: over the ceiling of 2,
// and correctly so.
if live := len(runningPIDs(t, app)); live != 3 {
t.Fatalf("%d running, want 3 — two exempt over the ceiling plus the newest", live)
}
// A sweep does not shed the exempt pair either.
app.Evict()
if !settles(2*time.Second, func() bool { return len(runningPIDs(t, app)) == 2 }) {
t.Fatalf("after a sweep: %d running, want the 2 exempt", len(runningPIDs(t, app)))
}
for _, p := range app.Plugins() {
if (p.Name == "keep0" || p.Name == "keep1") && !p.Running {
t.Errorf("%s was evicted by the sweep, but it is exempt", p.Name)
}
}
}
// TestWarm_AgeStillReclaimsUnderTheCeiling: the two bounds are one sweep, and age
// reclaims plugins the ceiling had no reason to touch.
func TestWarm_AgeStillReclaimsUnderTheCeiling(t *testing.T) {
const plugins, warm = 3, 5 // ceiling never reached
app := warmHost(t, plugins, warm, 40*time.Millisecond)
ask(t, app, plugins)
time.Sleep(80 * time.Millisecond) // every one is now past IdleAfter
if n := app.Evict(); n != plugins {
t.Fatalf("Evict stopped %d, want all %d on age alone", n, plugins)
}
if !settles(3*time.Second, func() bool { return len(runningPIDs(t, app)) == 0 }) {
t.Fatalf("%d still running", len(runningPIDs(t, app)))
}
}
+17
View File
@@ -103,6 +103,16 @@ type Config struct {
// DisableStartupMessage suppresses Fiber's startup banner.
DisableStartupMessage bool
// Warm is how many plugin processes this app may hold at once. Zero means
// unbounded, the historical behaviour. It is enforced where a plugin STARTS,
// so it is a ceiling rather than something a sweep restores a minute later,
// and again by [App.Evict] alongside each plugin's IdleAfter.
//
// 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.
Warm int
// ErrorHandler is the catch-all error handler. Defaults to zip.errorHandler
// which renders {error, code, status} JSON.
ErrorHandler fiber.ErrorHandler
@@ -258,6 +268,12 @@ type App struct {
open *plugin
plugMu sync.Mutex
// warm is Config.Warm: how many plugin processes may run at once. It is state
// rather than an argument to Evict because the ceiling must be known where a
// process is STARTED — a bound only a periodic sweep consults is a bound a
// burst outruns. See makeRoom.
warm 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
@@ -343,6 +359,7 @@ func newApp(cfg Config) *App {
logger: cfg.Logger,
loader: cfg.Loader,
born: time.Now(),
warm: cfg.Warm,
telemetry: newTelemetry(cfg),
}
}
+42 -3
View File
@@ -200,7 +200,27 @@ func TestSetLogReachesDownstream(t *testing.T) {
// Each iteration calls ResetUserValues, which is exactly what both transports
// do before dispatching (zap-proto/http's serveConn; fasthttp's keep-alive
// loop via Request.Reset), so one iteration is one real request.
//
// SO THE MEASURE IS THE DELTA, NOT THE TOTAL. This asserted `total <= 1` and was
// therefore red on every commit for as long as anyone can remember, reporting 34
// at every depth — which is the invariant PASSING and the bound failing. A total
// counts everything a request pays, and the wrapper does not own most of it: with
// the logger replaced by luxlog.NewNoOpLogger the figure is 32, so the access log
// is 2, and an alloc profile of the remaining 32 attributes them to TELEMETRY —
// zip.describe with its strings.Builder (~47% of objects, cumulative), the metric
// library's labelsKeyFromLabels (~27%), plus the hex trace/span ids, net.IP.String
// and a strconv.FormatInt. Per-request metrics and spans are a feature, so no
// arrangement of the wrapper reaches one, and a bound nothing can satisfy gates
// nothing: it trains a reader to skip a red line.
//
// The delta is the claim above, it is exactly measurable, and it holds: five
// middlewares cost the same as none. The total is held separately as a RATCHET so
// the constant cannot grow unnoticed — lower it when describe or the label key
// gets cheaper, and raise it only with a reason written down.
func TestServePathAllocsAreChainDepthInvariant(t *testing.T) {
// Indexed by middleware depth, so the assertions below compare depths rather
// than each depth against a literal.
allocs := map[int]float64{}
for _, mw := range []int{0, 1, 5} {
app := zip.New(zip.Config{DisableStartupMessage: true, ServerHeader: "-"})
for i := 0; i < mw; i++ {
@@ -220,11 +240,30 @@ func TestServePathAllocsAreChainDepthInvariant(t *testing.T) {
if fctx.Response.StatusCode() != 204 {
t.Fatalf("mw=%d: status %d", mw, fctx.Response.StatusCode())
}
if got > 1 {
t.Errorf("mw=%d: %.0f allocs/request, want <= 1 (the *Ctx); "+
"the wrapper must not allocate per handler", mw, got)
allocs[mw] = got
t.Logf("mw=%d: %.0f allocs/request", mw, got)
}
// THE CLAIM: the wrapper allocates per REQUEST, not per HANDLER. Before
// one-Ctx-per-request a five-middleware stack paid six, so a regression here
// shows up as a depth-proportional climb.
for _, mw := range []int{1, 5} {
if grew := allocs[mw] - allocs[0]; grew > 0.5 {
t.Errorf("mw=%d costs %.0f allocs against %.0f at mw=0 — the wrapper is "+
"allocating per handler again (%.0f more for %d handlers)",
mw, allocs[mw], allocs[0], grew, mw)
}
}
// THE RATCHET: sync.Pool's pinSlow shows up in the profile, so the figure can
// wobble with GOMAXPROCS; the margin is for that and not for drift.
const ceiling = 36
if allocs[0] > ceiling {
t.Errorf("a bare request costs %.0f allocs, over the %d ceiling — profile it "+
"(go test -bench . -memprofile) before raising this: the constant is "+
"telemetry, chiefly zip.describe and the metric label key",
allocs[0], ceiling)
}
}
// TestIdentityHeaders confirms c.Org/User/Email map to X-* headers.