diff --git a/modules/pier/README.md b/modules/pier/README.md index e82c731f..eb0b278c 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -40,7 +40,7 @@ Use cases: - Session stores: `memory` (one instance; records in the process) or `redis` (records `mcp:s:` with `PX` TTLs, per-binding sets `mcp:b:`, events on channel `mcp:events`, under `cooler.KeyPrefix`); Mount never waits for Redis: a background warm-up opens the client from `redis.url` with a backoff (one second doubling to thirty, each failure logged) and subscribes to the event channel, so an instance that serves only stateless clients still receives fan-out, and the first request on any path also starts the open without waiting for it; while the store is down stateful requests answer 503 and stateless requests are served. - Fan-out (D-26): every `pact.MCPSubjectChanged` fired on the application is published through the store and applied on every instance: a permission or link change recomputes the tools and prompts of the subject's sessions and stateless servers (the SDK then sends `notifications/tools/list_changed` and `notifications/prompts/list_changed`), a revocation or deletion closes those sessions, drops their records and evicts the stateless servers; a link event also drops the link cache at once. A revocation naming `TokenID` ends only the sessions whose subject under the event's guard is that user with that token id and no client id (one personal token); one naming `ClientID` ends only that OAuth client's sessions of the user; one naming neither, and every deletion, ends all sessions of the subject. - Client address: before resolving a bearer, pier puts the client address on the resolvers' context with `towel.WithClientIP`, taken from the host when it offers one (surf applies `http.trusted_proxies`) or from the request's `RemoteAddr`, so a resolver can log a failed attempt with it. -- Hybrid protocol (D-32): a request whose `Mcp-Protocol-Version` is 2026-07-28 or later, or whose first JSON-RPC method is `server/discover`, goes to the SDK's stateless handler over an in-process LRU of per-identity servers keyed by binding (capacity `mcp.sessions.max_per_binding` times 16); every other request goes to the stateful router. Both run behind the same bearer verification. A server evicted from the LRU closes its sessions, and a stateless `subscriptions/listen` stream re-verifies its bearer every minute and ends when it no longer resolves to its binding, as the stateful GET stream does. +- Hybrid protocol (D-32): a request whose `Mcp-Protocol-Version` is 2026-07-28 or later, or whose first JSON-RPC method is `server/discover`, goes to the SDK's stateless handler over an in-process LRU of per-identity servers keyed by binding (capacity `mcp.sessions.max_per_binding` times 16); every other request goes to the stateful router. Both run behind the same bearer verification. Past capacity the least recently used idle server is evicted (and its sessions closed); a server that is serving a request, a `subscriptions/listen` stream included, is never evicted, so capacity never ends a quiet listener or a tool call in flight, and the LRU stays above capacity while every server is busy. A stateless `subscriptions/listen` stream re-verifies its bearer every minute and ends when it no longer resolves to its binding, as the stateful GET stream does. - Widening on return (A-WR-04): when a binding comes back with a different identity (more or fewer scopes, a backend subject added or dropped), its stateful session or stateless server is re-filtered on that request, so newly allowed tools and prompts appear at once (with `list_changed`) instead of waiting for an event. Token ids play no part in this, so an access-token rotation alone re-filters nothing. A stateless server is shared by every live token of its binding, so a return with less reach only adds there and never removes: the receiving middleware hides, per request, what that request's identity may not use, and a call of a wider token is never cut off by a narrower one. A guard provider error during this re-filtering is not a refusal: the cached server stays as it was and the next request retries (logged as `pier: widening deferred`); a fan-out recompute that meets one hides the failing items and leaves the server to be re-filtered by the binding's next request. - Protected resource metadata (RFC 9728, `pier.PRMScopes`): `GET /.well-known/oauth-protected-resource` and `GET /.well-known/oauth-protected-resource` answer the resource, the authorization server (the published verifier's resource origin, else `app.url`), the supported scopes and the bearer method; a `mcp.resource` that differs from the verifier's resource is a boot warning. - Prompts (D-28): prefixed and filtered like tools, required arguments checked before `Render` runs; a prompt never dispatches a route. A prompt declared with `Writes` (it directs tools that are not read-only) sits under the read/write ceiling and is hidden from an identity without `write`. diff --git a/modules/pier/stateless.go b/modules/pier/stateless.go index 20325677..87171246 100644 --- a/modules/pier/stateless.go +++ b/modules/pier/stateless.go @@ -28,6 +28,10 @@ type statelessEntry struct { binding string server *mcp.Server elem *list.Element + // active counts the requests being served on server (a + // subscriptions/listen stream for its whole life); guarded by + // statelessPath.mu. The LRU never evicts an active entry (RA-WR-04). + active int // mu serializes re-filtering and guards fingerprint, widened, tools and // prompts (see localSession); identity has its own box. @@ -61,6 +65,16 @@ type statelessPath struct { // LRU capacity. const statelessCapacityFactor = 16 +// entryUseKey keys the entryUse of one stateless request on its context. +type entryUseKey struct{} + +// entryUse lists the entries one stateless request is served by, so +// ServeHTTP releases them when the request ends; guarded by +// statelessPath.mu. +type entryUse struct { + entries []*statelessEntry +} + func newStatelessPath(p *Pier) *statelessPath { sp := &statelessPath{ p: p, @@ -84,6 +98,9 @@ func newStatelessPath(p *Pier) *statelessPath { // every recheck interval and ends when the bearer no longer resolves to // the binding that opened it (A-WR-06). func (sp *statelessPath) ServeHTTP(w http.ResponseWriter, r *http.Request) { + use := &entryUse{} + r = r.WithContext(context.WithValue(r.Context(), entryUseKey{}, use)) + defer sp.release(use) if r.Method != http.MethodPost || r.Body == nil { sp.handler.ServeHTTP(w, r) return @@ -135,7 +152,7 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { if !ok { return nil } - if e := sp.hit(id.Binding); e != nil { + if e := sp.hit(r, id.Binding); e != nil { sp.refresh(r.Context(), e, id) return e.server } @@ -147,34 +164,39 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { sp.mu.Lock() if cur, ok := sp.entries[id.Binding]; ok { sp.order.MoveToFront(cur.elem) + sp.track(r, cur) sp.mu.Unlock() sp.refresh(r.Context(), cur, id) return cur.server } e.elem = sp.order.PushFront(e) sp.entries[id.Binding] = e + sp.track(r, e) + // Past capacity the least recently used idle entries go. An entry that + // is serving a request (a quiet subscriptions/listen stream, a tool call + // in flight) is never evicted, so capacity never cuts a stream or a call + // off and a revocation still finds every open stream (A-WR-06, RA-WR-04); + // while every entry is busy the LRU stays above capacity. var evicted []*statelessEntry - for sp.order.Len() > sp.capacity { - last := sp.order.Back() - if last == nil { - break + for el := sp.order.Back(); el != nil && sp.order.Len() > sp.capacity; { + prev := el.Prev() + if old := el.Value.(*statelessEntry); old.active == 0 { + sp.order.Remove(el) + delete(sp.entries, old.binding) + evicted = append(evicted, old) } - old := last.Value.(*statelessEntry) - sp.order.Remove(last) - delete(sp.entries, old.binding) - evicted = append(evicted, old) + el = prev } sp.mu.Unlock() - // An evicted server's listen streams end with it, so none outlives the - // entry a later revocation would look for (A-WR-06). for _, old := range evicted { closeSessions(old.server) } return srv } -// hit answers the entry of a binding, most recently used now, or nil. -func (sp *statelessPath) hit(binding string) *statelessEntry { +// hit answers the entry of a binding, most recently used now and counted as +// serving r, or nil. +func (sp *statelessPath) hit(r *http.Request, binding string) *statelessEntry { sp.mu.Lock() defer sp.mu.Unlock() e, ok := sp.entries[binding] @@ -182,9 +204,32 @@ func (sp *statelessPath) hit(binding string) *statelessEntry { return nil } sp.order.MoveToFront(e.elem) + sp.track(r, e) return e } +// track counts e as serving r until ServeHTTP releases r's entries; the +// caller holds sp.mu. A request reaching getServer without ServeHTTP's +// entryUse is not counted. +func (sp *statelessPath) track(r *http.Request, e *statelessEntry) { + use, _ := r.Context().Value(entryUseKey{}).(*entryUse) + if use == nil || slices.Contains(use.entries, e) { + return + } + e.active++ + use.entries = append(use.entries, e) +} + +// release ends the request's use of its entries. +func (sp *statelessPath) release(use *entryUse) { + sp.mu.Lock() + for _, e := range use.entries { + e.active-- + } + use.entries = nil + sp.mu.Unlock() +} + // refresh stores the request's identity on an entry and, when the // identity's fingerprint is new to the entry, adds the tools and prompts // it allows (A-WR-04). It never removes any (RA-WR-02): one server serves diff --git a/modules/pier/stateless_fanout_test.go b/modules/pier/stateless_fanout_test.go index 6c2e8b8a..e0555dbc 100644 --- a/modules/pier/stateless_fanout_test.go +++ b/modules/pier/stateless_fanout_test.go @@ -91,8 +91,9 @@ func waitEnded(t *testing.T, ended <-chan struct{}, within time.Duration, what s // A-WR-06): an instance that served only stateless requests receives a // revocation published by the other instance through Redis and evicts its // entry; a subscriptions/listen stream ends within the re-check interval -// once its bearer stops resolving; and an LRU eviction at capacity closes -// the evicted server's listen stream. +// once its bearer stops resolving; and an LRU eviction at capacity spares a +// server with an open listen stream (RA-WR-04), evicting the least recently +// used idle one, while a revocation still ends that stream. func TestStatelessFanOut(t *testing.T) { plugin := acmePlugin{tools: []pact.MCPTool{echoTool(), showTool()}} @@ -138,7 +139,7 @@ func TestStatelessFanOut(t *testing.T) { } }) - t.Run("an LRU eviction closes the evicted server's listen stream", func(t *testing.T) { + t.Run("an LRU eviction spares a server with a listen stream", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() const capacity = 16 // max_per_binding 1 times the LRU factor @@ -164,10 +165,20 @@ func TestStatelessFanOut(t *testing.T) { if n := in.pier().StatelessServers(); n != capacity { t.Fatalf("stateless servers at capacity = %d, want %d", n, capacity) } + // One binding past capacity: the listener's server is the least + // recently used, but it is serving the stream, so the oldest idle + // server goes instead. listNames(t, ctx, sdkClient(t, ctx, in.url(), fmt.Sprintf("oas_at_lru_%d", capacity-1), clientOpts{version: "newest"})) - waitEnded(t, ended, 5*time.Second, "LRU eviction") + select { + case <-ended: + t.Fatal("the LRU eviction ended a live listen stream") + case <-time.After(300 * time.Millisecond): + } if n := in.pier().StatelessServers(); n != capacity { t.Fatalf("stateless servers after the eviction = %d, want %d", n, capacity) } + // The spared entry is still where a revocation looks for it. + fireSubjectChanged(t, in.app, "acme", 20, pact.MCPSubjectRevoked) + waitEnded(t, ended, 5*time.Second, "revocation after the eviction") }) }