diff --git a/docs/services/mcp.md b/docs/services/mcp.md index 6a603d58..3a5ba0fb 100644 --- a/docs/services/mcp.md +++ b/docs/services/mcp.md @@ -130,9 +130,9 @@ Every request is authorized afresh from its bearer; a session never replaces the The `read` and `write` scopes are a ceiling: a grant with only `read` sees tools whose read-only hint is true, `write` unlocks the rest, and permissions always filter on top. The `ai` and `offline_access` scopes play no part in it. -With `mcp.allow_personal_tokens` true, personal API tokens also open sessions, each capped by the token's own scope exactly as over HTTP: an administrator's `g15_` token yields backend tools only, with the token's scope limit computed per dispatched route method; a frontend user's token, resolved by the user plugin's published `pact.MCPPersonalTokenResolver`, yields frontend tools only. With the flag off (the default), any bearer that is not an OAuth token is refused without a lookup. +With `mcp.allow_personal_tokens` true, personal API tokens also open sessions, each capped by the token's own scope exactly as over HTTP: an administrator's `g15_` token yields backend tools only, with the token's scope limit computed per dispatched route method; a frontend user's token, resolved by the user plugin's published `pact.MCPPersonalTokenResolver`, yields frontend tools only. An application has one such resolver; a subject it answers must carry the token's `TokenID`, which keys the session binding, and a subject without one is refused. With the flag off (the default), any bearer that is not an OAuth token is refused without a lookup. -Sessions follow the streamable HTTP transport: `initialize` opens one, `Mcp-Session-Id` names it, DELETE ends it. A session is bound to the client and user that opened it (or to the personal token), so it survives access-token rotation, and another identity presenting its id is refused with 403. Sessions carry a sliding idle TTL and a cap per binding. With `mcp.sessions.store: redis` several instances share them: a session opened on one instance is served by another, which rehydrates it from the stored session state, and a permission change, a link change or a revocation fired on any instance reaches every session it concerns, as `notifications/tools/list_changed` or as the end of the session. Clients of protocol version 2026-07-28 and later, which have no sessions, are served statelessly from the same path with the same identity, dispatch and filtering. +Sessions follow the streamable HTTP transport: `initialize` opens one, `Mcp-Session-Id` names it, DELETE ends it. A session is bound to the client and user that opened it (or to the personal token), so it survives access-token rotation, and another identity presenting its id is refused with 403. Sessions carry a sliding idle TTL and a cap per binding; a session whose client went away without DELETE is closed on every instance once its idle TTL passes. When the same client and user come back with a wider grant (more scopes, the backend scope with a link), the session gains the newly allowed tools on that request. With `mcp.sessions.store: redis` several instances share them: a session opened on one instance is served by another, which rehydrates it from the stored session state, and a permission change, a link change or a revocation fired on any instance reaches every session it concerns, as `notifications/tools/list_changed` or as the end of the session. Clients of protocol version 2026-07-28 and later, which have no sessions, are served statelessly from the same path with the same identity, dispatch and filtering. ## Protected resource metadata diff --git a/modules/pact/README.md b/modules/pact/README.md index 0fcced40..9ded316e 100644 --- a/modules/pact/README.md +++ b/modules/pact/README.md @@ -157,7 +157,7 @@ func (p *Plugin) Schedule() []pact.ScheduledCommand { | `pact.MCPSubject` | The identity an MCP session acts as under one guard: guard name, user id, granted scopes, OAuth client id and personal token id. | | `pact.MCPDispatch` | A guard provider's answer for one call: the principal (as `any`), an optional credential and the bouncer guard names that honour the dispatch marker. | | `pact.MCPTokenVerifier` | Published by the OAuth server plugin: verifies an access token at the MCP endpoint into a `pact.MCPGrant` (user, client, token, scopes, expiry) and names its resource. A refused token is `pact.ErrMCPTokenInvalid`. | -| `pact.MCPPersonalTokenResolver` | Published by a plugin whose personal tokens may open an MCP session: resolves a raw token into a `pact.MCPSubject`, false when the prefix is not its own. | +| `pact.MCPPersonalTokenResolver` | Published by the one plugin whose personal tokens may open an MCP session (one resolver per application): resolves a raw token into a `pact.MCPSubject` whose `TokenID` identifies the token row (it keys the session binding; a subject without one is refused); false when the prefix is not its own, which the endpoint answers as unauthorized. | | `pact.MCPLinkedBackendResolver` | Published by the plugin that owns the frontend-to-backend account link: the `pact.LinkedBackendUser` (id, login, display name) of a frontend user, false when unlinked. | | `pact.MCPSubjectChanged` | Event fired on the festival bus when a subject's permissions, link, grants or existence change (`Kind` is `pact.MCPSubjectPermissions`, `pact.MCPSubjectLink`, `pact.MCPSubjectRevoked` or `pact.MCPSubjectDeleted`), so the MCP endpoint re-lists tools or ends sessions. | | `pact.FrontendPermission` | One frontend permission code a plugin declares: `Code`, `Label`, `Tab` and `Comment`, offered in the user plugin's permission editors next to the seeded codes. | diff --git a/modules/pact/mcp.go b/modules/pact/mcp.go index cdd1d897..082574cf 100644 --- a/modules/pact/mcp.go +++ b/modules/pact/mcp.go @@ -236,9 +236,13 @@ var ErrMCPTokenInvalid = errors.New("pact: mcp token invalid") // MCPPersonalTokenResolver resolves a personal API token (a frontend user // token; the admin's g15_ tokens are resolved by the framework) into an MCP -// subject. The boolean is false when the token's prefix is not the -// resolver's, so the next resolver is tried; an error means the token has -// this resolver's prefix but is not usable. +// subject. An application has one resolver, published through app.Lookup +// by the plugin that owns the frontend tokens. The boolean is false when +// the token's prefix is not the resolver's, and the endpoint then answers +// unauthorized; an error means the token has this resolver's prefix but is +// not usable. The subject's UserID and TokenID must both be set: TokenID +// identifies the token row and keys the session binding +// (pat:user:), so a subject without a TokenID is refused. type MCPPersonalTokenResolver interface { MCPResolvePersonalToken(ctx context.Context, raw string) (MCPSubject, bool, error) } diff --git a/modules/pier/README.md b/modules/pier/README.md index 1bdde9ce..ab9ac04d 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -33,13 +33,14 @@ Use cases: - In-process dispatch through `pier.Host.Handler`, the route's compiled chain, with `bouncer.WithDispatch` carrying the guard provider's `MCPDispatch` principal and credential, path values set as the mux would, `RemoteAddr`, `X-Forwarded-For`, `X-Real-IP` and `Accept-Language` copied from the MCP request, `mcp.call_timeout` on the context and the answer captured up to `mcp.max_result_bytes` (D-17). - Results (D-13): 2xx is text plus structured content (an object as is, any other JSON wrapped as `{"data": …}`); a 4xx or 5xx is a tool error whose text is the route's body; a too-large answer is a tool error with a fixed text. - 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. +- 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. One `pact.MCPPersonalTokenResolver` serves the application; a subject it answers without a `TokenID` is refused, so personal-token users never share a binding. - 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 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. +- 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. -- 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. +- 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. +- 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. - 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. diff --git a/modules/pier/events.go b/modules/pier/events.go index 0b688f8c..8da5bc02 100644 --- a/modules/pier/events.go +++ b/modules/pier/events.go @@ -79,6 +79,7 @@ func (p *Pier) recomputeSession(ctx context.Context, ls *localSession, relink bo } tools, prompts := p.catalog.allowed(ctx, id) applyDiff(ls.server, p, id, tools, prompts, ls.tools, ls.prompts) + ls.fingerprint = id.fingerprint() ls.mu.Unlock() } @@ -92,9 +93,28 @@ func (p *Pier) recomputeStateless(ctx context.Context, e *statelessEntry, relink } tools, prompts := p.catalog.allowed(ctx, id) applyDiff(e.server, p, id, tools, prompts, e.tools, e.prompts) + e.fingerprint = id.fingerprint() e.mu.Unlock() } +// widen re-filters a cached server for the identity of the request that +// reached it when that identity's fingerprint differs from the one the +// server was last filtered for (A-WR-04): a binding that returns with more +// scope, or with a backend subject added, gains the newly allowed tools and +// prompts at once, and one that returns with less loses them. The caller +// holds the session's or entry's own mutex, never a router-wide one. +func (p *Pier) widen(ctx context.Context, srv serverFeatures, id Identity, fp *string, haveTools, havePrompts map[string]bool) { + next := id.fingerprint() + if *fp == next { + return + } + // A client that hangs up mid-check must not leave the server filtered + // by cancelled guard checks (an error hides the item). + tools, prompts := p.catalog.allowed(context.WithoutCancel(ctx), id) + applyDiff(srv, p, id, tools, prompts, haveTools, havePrompts) + *fp = next +} + // applyDiff brings a server's tool and prompt sets to the allowed ones, // updating the tracking maps in place. func applyDiff(srv serverFeatures, p *Pier, id Identity, tools []*tool, prompts []*prompt, haveTools, havePrompts map[string]bool) { diff --git a/modules/pier/identity.go b/modules/pier/identity.go index b6932fe5..e04a689d 100644 --- a/modules/pier/identity.go +++ b/modules/pier/identity.go @@ -77,6 +77,44 @@ func (id Identity) writes() bool { return !id.ceiling || slices.Contains(id.Scopes, "write") } +// fingerprint encodes the authorization-relevant parts of the identity +// (P-26): the sorted scopes, both subjects (guard, user, client, token and +// their sorted scopes), the ceiling and the token kind. Two requests of one +// binding with different fingerprints may see different tools, so a cached +// server is re-filtered when it changes (A-WR-04). +func (id Identity) fingerprint() string { + var b strings.Builder + sorted := func(in []string) string { + out := slices.Clone(in) + slices.Sort(out) + return strings.Join(out, ",") + } + subject := func(s *pact.MCPSubject) { + if s == nil { + b.WriteString("-|") + return + } + b.WriteString(s.Guard) + b.WriteByte(':') + b.WriteString(strconv.FormatUint(uint64(s.UserID), 10)) + b.WriteByte(':') + b.WriteString(s.ClientID) + b.WriteByte(':') + b.WriteString(strconv.FormatUint(uint64(s.TokenID), 10)) + b.WriteByte(':') + b.WriteString(sorted(s.Scopes)) + b.WriteByte('|') + } + b.WriteString(sorted(id.Scopes)) + b.WriteByte('|') + subject(id.Frontend) + subject(id.Backend) + b.WriteString(strconv.FormatBool(id.ceiling)) + b.WriteByte('|') + b.WriteString(id.TokenKind) + return b.String() +} + // subjects lists the identity's subjects (frontend first). func (id Identity) subjects() []pact.MCPSubject { var out []pact.MCPSubject @@ -227,7 +265,9 @@ func (r *identityResolver) resolvePersonalToken(ctx context.Context, raw string) return Identity{}, ErrUnauthorized } subject, ok, err := resolver.MCPResolvePersonalToken(ctx, raw) - if err != nil || !ok || subject.UserID == 0 { + // TokenID keys the binding (pat:user:): a subject without one + // would put every user of the resolver on one binding (A-WR-09). + if err != nil || !ok || subject.UserID == 0 || subject.TokenID == 0 { return Identity{}, ErrUnauthorized } if subject.Guard == "" { diff --git a/modules/pier/pier.go b/modules/pier/pier.go index 035910ec..1284bc77 100644 --- a/modules/pier/pier.go +++ b/modules/pier/pier.go @@ -58,10 +58,10 @@ type Pier struct { 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 ends the session reaper (and a Redis store's warm-up) 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 } @@ -72,8 +72,10 @@ type Pier struct { // behind the inline throttle. Nothing is mounted when the flag is off. A // bad binding, a duplicate name, an unknown guard or a route already // registered at mcp.path fails with an error naming the plugin, so boot -// (and route:list) fail. Nothing dials Redis here. The mounted *Pier is -// published on the application. +// (and route:list) fail. Mount never waits for Redis: a Redis store is +// opened by a background warm-up with a backoff, so the event subscription +// exists before traffic arrives. Mount also starts the session reaper. The +// mounted *Pier is published on the application. func Mount(host Host, app *backpack.App, plugins []party.Plugin) error { if host == nil || app == nil { return errors.New("pier: a host and an application are required") @@ -139,13 +141,17 @@ func Mount(host Host, app *backpack.App, plugins []party.Plugin) error { reapCtx, stop := context.WithCancel(context.Background()) p.stopReap = stop go p.sessions.reap(reapCtx, reapInterval(cfg.IdleTTL)) + if ls, ok := p.store.(*lazyStore); ok { + go ls.warm(reapCtx) + } return nil } // newStore builds the session store mcp.sessions.store selects: the memory // store, subscribed to its own events at once, or a Redis store opened on // first use from redis.url under cooler's key prefix (nothing dials at -// mount; a missing URL is a 503 on the first request, logged once). +// mount itself; a background warm-up opens it with a backoff, and a missing +// URL is a 503 on stateful requests, logged). func (p *Pier) newStore() Store { if p.cfg.SessionStore != StoreRedis { ms := newMemoryStore() diff --git a/modules/pier/session.go b/modules/pier/session.go index 5e2289b5..e92c6bbb 100644 --- a/modules/pier/session.go +++ b/modules/pier/session.go @@ -33,6 +33,7 @@ const ( methodListPrompts = "prompts/list" methodGetPrompt = "prompts/get" methodDiscover = "server/discover" + methodListen = "subscriptions/listen" ) // ephemeralSession is the session value of the audit line of a tool call @@ -64,10 +65,11 @@ type localSession struct { transport *mcp.StreamableServerTransport ss *mcp.ServerSession - mu sync.Mutex - identity Identity - tools map[string]bool - prompts map[string]bool + mu sync.Mutex + identity Identity + fingerprint string // of the identity the tool set was last filtered for + tools map[string]bool + prompts map[string]bool initialized atomic.Bool persisted atomic.Bool @@ -329,6 +331,7 @@ func (sr *sessionRouter) lookup(w http.ResponseWriter, r *http.Request, id Ident } ls.mu.Lock() ls.identity = id + sr.p.widen(ctx, ls.server, id, &ls.fingerprint, ls.tools, ls.prompts) ls.mu.Unlock() if err := sr.p.store.Touch(ctx, sid, sr.p.cfg.IdleTTL); err != nil { sr.p.storeUnavailable(w, err) @@ -341,7 +344,7 @@ func (sr *sessionRouter) lookup(w http.ResponseWriter, r *http.Request, id Ident // with the session id, from the stored state when rehydrating. A goroutine // drops the local entry when the session closes. func (sr *sessionRouter) connect(ctx context.Context, sid string, id Identity, state *mcp.ServerSessionState) (*localSession, error) { - ls := &localSession{id: sid, binding: id.Binding, created: time.Now().UTC(), identity: id} + ls := &localSession{id: sid, binding: id.Binding, created: time.Now().UTC(), identity: id, fingerprint: id.fingerprint()} srv, tools, prompts := sr.p.newServer(ctx, id, func() { ls.initialized.Store(true) }) ls.server, ls.tools, ls.prompts = srv, tools, prompts ls.transport = &mcp.StreamableServerTransport{SessionID: sid} diff --git a/modules/pier/stateless.go b/modules/pier/stateless.go index dc294463..8cdf1778 100644 --- a/modules/pier/stateless.go +++ b/modules/pier/stateless.go @@ -3,11 +3,13 @@ package pier import ( "bytes" "container/list" + "context" "errors" "io" "net/http" "slices" "sync" + "time" "github.com/modelcontextprotocol/go-sdk/mcp" ) @@ -27,10 +29,11 @@ type statelessEntry struct { server *mcp.Server elem *list.Element - mu sync.Mutex - identity Identity - tools map[string]bool - prompts map[string]bool + mu sync.Mutex + identity Identity + fingerprint string // of the identity the tool set was last filtered for + tools map[string]bool + prompts map[string]bool } // statelessPath serves clients of protocol 2026-07-28 and later (D-32, @@ -71,8 +74,53 @@ func newStatelessPath(p *Pier) *statelessPath { return sp } +// ServeHTTP serves one stateless request. A subscriptions/listen POST is +// long-lived, so like the stateful GET stream it re-verifies its bearer +// 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) { - sp.handler.ServeHTTP(w, r) + if r.Method != http.MethodPost || r.Body == nil { + sp.handler.ServeHTTP(w, r) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + var mbe *http.MaxBytesError + if errors.As(err, &mbe) { + http.Error(w, "request body too large", http.StatusRequestEntityTooLarge) + return + } + http.Error(w, "failed to read body", http.StatusBadRequest) + return + } + _ = r.Body.Close() + r.Body = io.NopCloser(bytes.NewReader(body)) + id, ok := identityFromRequest(r) + if !ok || !slices.Contains(peekMethods(body), methodListen) { + sp.handler.ServeHTTP(w, r) + return + } + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + raw := bearerOf(r) + go func() { + ticker := time.NewTicker(sp.p.recheck) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + cur, err := sp.p.identities.resolve(ctx, raw) + if err != nil || cur.Binding != id.Binding { + logSession(logger(), "stream_closed", "", id.Binding) + cancel() + return + } + } + } + }() + sp.handler.ServeHTTP(w, r.WithContext(ctx)) } // getServer returns the per-identity server of the request's binding, @@ -83,23 +131,24 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { return nil } if e := sp.hit(id.Binding); e != nil { - sp.refresh(e, id) + sp.refresh(r.Context(), e, id) return e.server } // Built with no lock held: the guard checks may query a database // (A-WR-07). A concurrent request of the same binding may have // inserted an entry meanwhile; that one is adopted. srv, tools, prompts := sp.p.newServer(r.Context(), id, nil) - e := &statelessEntry{binding: id.Binding, server: srv, identity: id, tools: tools, prompts: prompts} + e := &statelessEntry{binding: id.Binding, server: srv, identity: id, fingerprint: id.fingerprint(), tools: tools, prompts: prompts} sp.mu.Lock() if cur, ok := sp.entries[id.Binding]; ok { sp.order.MoveToFront(cur.elem) sp.mu.Unlock() - sp.refresh(cur, id) + sp.refresh(r.Context(), cur, id) return cur.server } e.elem = sp.order.PushFront(e) sp.entries[id.Binding] = e + var evicted []*statelessEntry for sp.order.Len() > sp.capacity { last := sp.order.Back() if last == nil { @@ -108,8 +157,14 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { old := last.Value.(*statelessEntry) sp.order.Remove(last) delete(sp.entries, old.binding) + evicted = append(evicted, old) } 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 } @@ -125,13 +180,22 @@ func (sp *statelessPath) hit(binding string) *statelessEntry { return e } -// refresh stores the request's identity on an entry. -func (sp *statelessPath) refresh(e *statelessEntry, id Identity) { +// refresh stores the request's identity on an entry and re-filters its +// server when the identity's fingerprint changed (A-WR-04). +func (sp *statelessPath) refresh(ctx context.Context, e *statelessEntry, id Identity) { e.mu.Lock() e.identity = id + sp.p.widen(ctx, e.server, id, &e.fingerprint, e.tools, e.prompts) e.mu.Unlock() } +// closeSessions closes every session of a server. +func closeSessions(srv *mcp.Server) { + for ss := range srv.Sessions() { + _ = ss.Close() + } +} + // matching lists the entries whose identity includes the subject. func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry { sp.mu.Lock() @@ -158,9 +222,7 @@ func (sp *statelessPath) evict(guard string, userID uint) { delete(sp.entries, e.binding) } sp.mu.Unlock() - for ss := range e.server.Sessions() { - _ = ss.Close() - } + closeSessions(e.server) } } @@ -169,6 +231,12 @@ func (sp *statelessPath) evict(guard string, userID uint) { // POST is server/discover; everything else goes to the stateful router // (P-08). func (p *Pier) route(w http.ResponseWriter, r *http.Request) { + // The first request on any path opens a Redis store that the mount + // warm-up has not opened yet, without waiting for it, so an instance + // serving only stateless clients still subscribes to fan-out (A-WR-05). + if ls, ok := p.store.(*lazyStore); ok { + ls.kick() + } if p.stateless == nil { p.sessions.ServeHTTP(w, r) return diff --git a/modules/pier/store_redis.go b/modules/pier/store_redis.go index 8d9cf7ec..5453c058 100644 --- a/modules/pier/store_redis.go +++ b/modules/pier/store_redis.go @@ -232,10 +232,12 @@ func (s *redisStore) Subscribe(ctx context.Context, fn func(Event)) error { // (route:list and tests never dial) and a missing redis.url is a 503 on the // first request, logged once, not a boot failure (cooler's rule). type lazyStore struct { - open func(ctx context.Context) (Store, error) - mu sync.Mutex - store Store - warned atomic.Bool + open func(ctx context.Context) (Store, error) + mu sync.Mutex + store Store + warned atomic.Bool + ready atomic.Bool // the store is open + opening atomic.Bool // a kick is opening it } // get returns the store, opening it on the first call and retrying on a @@ -254,9 +256,43 @@ func (l *lazyStore) get(ctx context.Context) (Store, error) { return nil, err } l.store = s + l.ready.Store(true) return s, nil } +// kick opens the store in the background when it is not open yet and no +// other kick is opening it; the caller never waits. +func (l *lazyStore) kick() { + if l.ready.Load() || !l.opening.CompareAndSwap(false, true) { + return + } + go func() { + defer l.opening.Store(false) + _, _ = l.get(context.Background()) + }() +} + +// warm opens the store at mount in the background, retrying with a backoff +// from one second doubling to thirty, so the event subscription exists +// before traffic arrives (A-WR-05). It returns once the store is open or +// ctx ends; each failure is logged once per backoff step. +func (l *lazyStore) warm(ctx context.Context) { + backoff := time.Second + for { + if _, err := l.get(ctx); err == nil { + return + } else if ctx.Err() == nil { + logger().Warn("pier: session store not open yet, retrying", "retry_in", backoff.String(), "error", err) + } + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff = min(backoff*2, 30*time.Second) + } +} + func (l *lazyStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error { s, err := l.get(ctx) if err != nil {