From 2bf9dca603b6f53d818a75d2f5701365cf58ef35 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Fri, 23 Oct 2026 15:50:04 +0200 Subject: [PATCH] fix(pier): reap stateful sessions whose store record expired - a reaper started at Mount checks every local session against the store every min(idle_ttl/2, 1m) and closes each copy whose record is gone, logging the expired step (A-CR-01, A-IN-06) - the GET stream re-check ends the stream when the record is gone (A-IN-01) - tool calls served without a session on the stateful path log the session as ephemeral (A-IN-06) - TestPierReaperSmoke and the ReapNow, LocalSessions, StopReaper seams --- modules/pier/README.md | 7 +-- modules/pier/audit.go | 15 ++++-- modules/pier/export_test.go | 18 +++++++ modules/pier/pier.go | 9 ++++ modules/pier/reaper_smoke_test.go | 66 ++++++++++++++++++++++++ modules/pier/session.go | 83 ++++++++++++++++++++++++++++++- 6 files changed, 190 insertions(+), 8 deletions(-) create mode 100644 modules/pier/reaper_smoke_test.go diff --git a/modules/pier/README.md b/modules/pier/README.md index 80ed7b97..1bdde9ce 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -35,14 +35,15 @@ Use cases: - Built-in `backend` guard provider (`pier.BackendGuard`): routes behind cabana's `backend` middleware, listing through `cabana.Allows` on the administrator's principal from `cabana.BackendPrincipal`, an admin personal token's scope limit computed per dispatched route method with `cabana.ScopeLimitFor`; a plugin that registers its own `backend` provider wins. - Identity per request (`pier.Identity`): `oas_at_` tokens through the published verifier, personal tokens only with `mcp.allow_personal_tokens`, the backend subject through the published link resolver with an in-process cache of at most `pier.LinkCacheTTL`; the binding key `oauth::`, `pat:user:` or `pat:backend:` survives access-token rotation. - Per-identity SDK server plus a receiving middleware that re-filters `tools/list` and `prompts/list` and refuses `tools/call` and `prompts/get` for the identity of the current request, so a permission change takes effect on the next request (D-08, D-25). -- Stateful sessions (D-25, D-27): `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`. A standalone GET stream re-verifies its bearer every minute and ends when it no longer resolves to the session's binding. +- Stateful sessions (D-25, D-27): `initialize` opens a session (429 above `mcp.sessions.max_per_binding`), a request with an unknown id answers 404 (the client re-initialises), another identity presenting the id answers 403 `session_binding`, DELETE ends the session everywhere, the record's TTL slides on every request, a session initialised on one instance is rehydrated on another from the stored `ServerSessionState`. A standalone GET stream re-verifies its bearer every minute and ends when it no longer resolves to the session's binding or the session's record is gone. +- Session reaper (A-CR-01): most clients exit without DELETE, so a janitor started at mount checks every local session against the store every `min(mcp.sessions.idle_ttl/2, 1m)` and closes each local copy whose record has expired (its SDK server session ends and it leaves the fan-out set), logging the `expired` step. The store's TTL decides on every instance: a session another instance keeps alive still has its record, so no instance closes it. No configuration key; the reaper runs for the life of the process. - 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`); the Redis client is opened on the first request from `redis.url`, never at mount, so `route:list` and tests never dial, and a missing URL is a 503 logged once. - 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. - 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. - 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 401 carries `WWW-Authenticate: Bearer resource_metadata="/.well-known/oauth-protected-resource", scope="read"`; a browser request whose `Origin` is not in `mcp.allowed_origins` is refused with 403 before any body is read. -- One audit line per tool call (`pier: tool call`: tool, guard, user id, admin id, client id, token kind, session prefix, status, duration) and one per session step (`pier: session`), never arguments, bodies or tokens. +- One audit line per tool call (`pier: tool call`: tool, guard, user id, admin id, client id, token kind, session prefix, status, duration) and one per session step (`pier: session`: `open`, `rehydrate`, `end`, `binding_refused`, `revoked`, `stream_closed`, `expired`), never arguments, bodies or tokens. A tool call served on the stateful path without a session logs the session `ephemeral`, one on the stateless path `stateless`. ## Usage @@ -141,4 +142,4 @@ Only this module (and surf's tagged `pier_on.go`) imports the SDK; `pact` stays ## Testing -`go test -short ./modules/pier/...` runs without Docker, a real Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route; `TestPierSessionsSmoke` runs two instances on one miniredis (rehydration, the binding refusal, DELETE, the idle TTL, the per-binding cap); `TestPierFanOutSmoke` receives `list_changed` on one instance for an event fired on the other and sees a revocation end the session; `TestPierStatelessSmoke` connects the newest SDK client statelessly next to a 2025-11-25 client with a session. The tracer's subtests cover the read/write ceiling (a read-only grant lists the GET tool only), prompts (`prompts/list`, `prompts/get`, the required-argument refusal), the audit line (tool, user and status present, arguments and token absent) and personal tokens (401 without the flag and no lookup; with the flag an injected admin token resolver yields a backend-only identity bound to `pat:backend:`). `TestExampleBlogPlugin` keeps the `acme.blog` example of `docs/services/mcp.md` mounting. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf. +`go test -short ./modules/pier/...` runs without Docker, a real Redis or a build tag: `TestPierTracerSmoke` builds a real surf router with a fake plugin, mounts pier through an in-test `pier.Host` adapter, connects the SDK client over `httptest` and proves `tools/list`, `tools/call` through the compiled guard chain with the dispatch marker (no `Authorization` or `Cookie` reaches the route), the 401 challenge, and the boot errors for a duplicate name and an unknown route; `TestPierSessionsSmoke` runs two instances on one miniredis (rehydration, the binding refusal, DELETE, the idle TTL, the per-binding cap); `TestPierFanOutSmoke` receives `list_changed` on one instance for an event fired on the other and sees a revocation end the session; `TestPierStatelessSmoke` connects the newest SDK client statelessly next to a 2025-11-25 client with a session. `TestPierReaperSmoke` lets a session expire on the memory store and sees one reaper pass close it (404 next) while a session kept active survives. The tracer's subtests cover the read/write ceiling (a read-only grant lists the GET tool only), prompts (`prompts/list`, `prompts/get`, the required-argument refusal), the audit line (tool, user and status present, arguments and token absent) and personal tokens (401 without the flag and no lookup; with the flag an injected admin token resolver yields a backend-only identity bound to `pat:backend:`). `TestExampleBlogPlugin` keeps the `acme.blog` example of `docs/services/mcp.md` mounting. `go test -short -tags summercms_mcp ./modules/pier/... ./modules/surf/...` runs the same tests with the SDK linked into surf. diff --git a/modules/pier/audit.go b/modules/pier/audit.go index f09e6c82..0f57516d 100644 --- a/modules/pier/audit.go +++ b/modules/pier/audit.go @@ -22,11 +22,15 @@ func logger() *slog.Logger { } // sessionPrefix shortens a session id for logging; a stateless call logs -// "stateless". +// "stateless" and a call served without a session on the stateful path +// "ephemeral". func sessionPrefix(id string) string { if id == "" { return stateless } + if id == ephemeralSession { + return ephemeralSession + } if len(id) > sessionPrefixLen { return id[:sessionPrefixLen] } @@ -63,9 +67,12 @@ func logToolCall(log *slog.Logger, id Identity, t *tool, session string, status ) } -// logSession writes one line per session lifecycle step (open, rehydrate, -// end, binding_refused, expired) with the session prefix and the binding, -// never the token. +// logSession writes one line per session lifecycle step with the session +// prefix and the binding, never the token. The steps: open (serveNew), +// rehydrate (lookup), end (DELETE), binding_refused (another binding +// presented the id), revoked (a revocation or deletion event), +// stream_closed (a GET stream whose bearer or record is gone) and expired +// (the reaper closed a local copy whose record expired). func logSession(log *slog.Logger, step, session, binding string) { log.Info("pier: session", "step", step, "session", sessionPrefix(session), "binding", binding) } diff --git a/modules/pier/export_test.go b/modules/pier/export_test.go index 9d8598ef..98f51e15 100644 --- a/modules/pier/export_test.go +++ b/modules/pier/export_test.go @@ -49,3 +49,21 @@ func (p *Pier) StatelessServers() int { // SetRecheck sets how often a long GET stream re-verifies its bearer. func (p *Pier) SetRecheck(d time.Duration) { p.recheck = d } + +// ReapNow runs one session reaper pass and answers how many local sessions +// it closed. +func (p *Pier) ReapNow(ctx context.Context) int { return p.sessions.reapOnce(ctx) } + +// LocalSessions counts the stateful sessions connected on this instance. +func (p *Pier) LocalSessions() int { + p.sessions.mu.Lock() + defer p.sessions.mu.Unlock() + return len(p.sessions.local) +} + +// StopReaper ends the session reaper goroutine Mount started. +func (p *Pier) StopReaper() { + if p.stopReap != nil { + p.stopReap() + } +} diff --git a/modules/pier/pier.go b/modules/pier/pier.go index 54a4a4ca..035910ec 100644 --- a/modules/pier/pier.go +++ b/modules/pier/pier.go @@ -57,6 +57,12 @@ type Pier struct { appName string version string instance string + + // stopReap ends the session reaper Mount starts. backpack.App has no + // shutdown hook yet, so in production the reaper runs for the life of + // the process, like the store subscription; when backpack gains one, + // Mount registers stopReap there. + stopReap context.CancelFunc } // Mount reads the mcp.* keys and, when mcp.enabled is true, compiles every @@ -130,6 +136,9 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error { if err := app.Publish(p); err != nil { return fmt.Errorf("pier: publishing the endpoint: %w", err) } + reapCtx, stop := context.WithCancel(context.Background()) + p.stopReap = stop + go p.sessions.reap(reapCtx, reapInterval(cfg.IdleTTL)) return nil } diff --git a/modules/pier/reaper_smoke_test.go b/modules/pier/reaper_smoke_test.go new file mode 100644 index 00000000..f151eed9 --- /dev/null +++ b/modules/pier/reaper_smoke_test.go @@ -0,0 +1,66 @@ +package pier_test + +import ( + "context" + "net/http" + "testing" + "time" + + "git.golem15.com/golem15/summercms/modules/pact" +) + +// TestPierReaperSmoke is the tracer of phase 19.7 plan 06 (A-CR-01): on the +// memory store, a stateful session whose record expired is closed by the +// reaper (it leaves the local map, so fan-out no longer reaches it), its id +// answers 404 next, and a session kept active by a request survives a pass. +func TestPierReaperSmoke(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + in := startInstance(t, newApp(t, appSpec{ + mcp: "enabled: false\nsessions:\n store: memory\n idle_ttl: 30m\n", + guards: []pact.MCPGuard{&acmeGuard{}}, + }), acmePlugin{tools: []pact.MCPTool{echoTool()}}) + p := in.pier() + t.Cleanup(p.StopReaper) + + // 1. An SDK client opens a session on this instance. + abandoned := sdkClient(t, ctx, in.url(), "oas_at_test", clientOpts{}) + if n := p.LocalSessions(); n != 1 { + t.Fatalf("local sessions after connect = %d, want 1", n) + } + + // 2. The client goes away without DELETE; past the idle TTL its record + // is gone and one reaper pass closes the local copy. + if !p.AdvanceStoreClock(31 * time.Minute) { + t.Fatal("the instance has no memory store") + } + if n := p.ReapNow(ctx); n != 1 { + t.Fatalf("reaped = %d, want 1", n) + } + if n := p.LocalSessions(); n != 0 { + t.Fatalf("local sessions after the reap = %d, want 0", n) + } + + // 3. The old session id answers 404, so the client re-initialises. + if gone := rpc(t, ctx, in.url(), "oas_at_test", abandoned.ID(), callBody); gone.status != http.StatusNotFound { + t.Fatalf("call with the reaped session = %d %s", gone.status, gone.body) + } + + // 4. A session touched by a request within the TTL survives a pass. + active := openSession(t, ctx, in, "oas_at_test") + if !p.AdvanceStoreClock(20 * time.Minute) { + t.Fatal("the instance has no memory store") + } + if call := rpc(t, ctx, in.url(), "oas_at_test", active, callBody); call.status != http.StatusOK { + t.Fatalf("call on the active session = %d %s", call.status, call.body) + } + if !p.AdvanceStoreClock(20 * time.Minute) { + t.Fatal("the instance has no memory store") + } + if n := p.ReapNow(ctx); n != 0 { + t.Fatalf("reaped the active session (%d closed)", n) + } + if n := p.LocalSessions(); n != 1 { + t.Fatalf("local sessions after the second pass = %d, want 1", n) + } +} diff --git a/modules/pier/session.go b/modules/pier/session.go index e9a6f5a6..a153200c 100644 --- a/modules/pier/session.go +++ b/modules/pier/session.go @@ -15,6 +15,7 @@ import ( "sync/atomic" "time" + "github.com/modelcontextprotocol/go-sdk/auth" "github.com/modelcontextprotocol/go-sdk/jsonrpc" "github.com/modelcontextprotocol/go-sdk/mcp" ) @@ -34,6 +35,10 @@ const ( methodDiscover = "server/discover" ) +// ephemeralSession is the session value of the audit line of a tool call +// served on the stateful path without a session (serveEphemeral). +const ephemeralSession = "ephemeral" + // Keys of auth.TokenInfo.Extra set by the bearer middleware. const ( extraIdentity = "pier.identity" @@ -171,6 +176,14 @@ func (sr *sessionRouter) serveNew(w http.ResponseWriter, r *http.Request, id Ide // serveEphemeral answers one request outside any session with the state // the SDK synthesises for a pre-initialize request. func (sr *sessionRouter) serveEphemeral(w http.ResponseWriter, r *http.Request, id Identity) { + // The audit line of a tool call served here names the session + // "ephemeral", not "stateless": it ran on the stateful path without a + // session. The token info is built per request, so it is ours to mark. + if ti := auth.TokenInfoFromContext(r.Context()); ti != nil && ti.Extra != nil { + info, _ := ti.Extra[extraCall].(callInfo) + info.session = ephemeralSession + ti.Extra[extraCall] = info + } pv := r.Header.Get(protocolVersionHeader) if pv == "" || pv >= protocolVersion202607 { pv = protocolVersion202503 @@ -384,6 +397,14 @@ func (sr *sessionRouter) serveStream(w http.ResponseWriter, r *http.Request, ls cancel() return } + // A record that is gone (an ended event lost between instances, + // an expiry) ends the stream and the local copy. + if _, ok, err := sr.p.store.Get(ctx, ls.id); err == nil && !ok { + logSession(logger(), "stream_closed", ls.id, ls.binding) + cancel() + sr.closeLocal(ls.id) + return + } _ = sr.p.store.Touch(ctx, ls.id, sr.p.cfg.IdleTTL) } } @@ -411,6 +432,66 @@ func (sr *sessionRouter) closeLocal(sid string) { } } +// reap closes, every interval until ctx ends, the local sessions whose +// store record has expired (P-24, A-CR-01). The store's TTL is the single +// source of truth across instances: a session another instance keeps alive +// still has its record, so this instance keeps its copy. +func (sr *sessionRouter) reap(ctx context.Context, every time.Duration) { + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + sr.reapOnce(ctx) + } + } +} + +// reapOnce runs one reaper pass: each local session whose record is gone is +// closed (its SDK server session ends, it leaves the local map and so the +// fan-out matching set) and logged as expired. A store error keeps the +// local copy (never close on doubt) and is logged once per pass. It returns +// how many sessions it closed. +func (sr *sessionRouter) reapOnce(ctx context.Context) int { + type entry struct{ id, binding string } + sr.mu.Lock() + entries := make([]entry, 0, len(sr.local)) + for id, ls := range sr.local { + entries = append(entries, entry{id, ls.binding}) + } + sr.mu.Unlock() + closed := 0 + warned := false + for _, e := range entries { + if ctx.Err() != nil { + break + } + _, ok, err := sr.p.store.Get(ctx, e.id) + if err != nil { + if !warned { + logger().Warn("pier: session reaper could not read the store", "error", err) + warned = true + } + continue + } + if ok { + continue + } + sr.closeLocal(e.id) + logSession(logger(), "expired", e.id, e.binding) + closed++ + } + return closed +} + +// reapInterval is how often the reaper runs: half the idle TTL, at most a +// minute and at least a second. +func reapInterval(idleTTL time.Duration) time.Duration { + return max(min(idleTTL/2, time.Minute), time.Second) +} + // negotiatedVersion mirrors the SDK's handshake rule: the client's version // when the SDK supports it and it predates 2026-07-28, else the newest // handshake-era version. @@ -570,7 +651,7 @@ func containsPrompt(prompts []*prompt, name string) bool { func (p *Pier) toolHandler(t *tool, session Identity) mcp.ToolHandler { return func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { id, info := identityFromExtra(req.Extra, session) - if req.Session != nil { + if req.Session != nil && req.Session.ID() != "" { info.session = req.Session.ID() } return p.dispatch.call(ctx, id, info, t, req.Params.Arguments), nil