fix(pier): widen returning identities, warm the Redis store, end stateless listen streams, refuse personal tokens without a TokenID
- Identity.fingerprint covers scopes, both subjects, the ceiling and the token kind; a stateful lookup or stateless hit whose fingerprint changed re-filters the server with the event diff under its own mutex (A-WR-04) - Mount warms a Redis store in the background with a 1 s to 30 s backoff and every request kicks a pending open without waiting, so stateless-only instances subscribe to fan-out (A-WR-05) - a stateless subscriptions/listen POST re-verifies its bearer every recheck interval; an LRU-evicted stateless server closes its sessions (A-WR-06) - resolvePersonalToken refuses a subject with TokenID 0, and the pact.MCPPersonalTokenResolver contract says one resolver per application, false means unauthorized and TokenID keys the binding (A-WR-09, A-IN-03)
This commit is contained in:
Jakub Zych
@@ -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
|
||||
|
||||
|
||||
@@ -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. |
|
||||
|
||||
@@ -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:<TokenID>), so a subject without a TokenID is refused.
|
||||
type MCPPersonalTokenResolver interface {
|
||||
MCPResolvePersonalToken(ctx context.Context, raw string) (MCPSubject, bool, error)
|
||||
}
|
||||
|
||||
@@ -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:<client_id>:<user_id>`, `pat:user:<token_id>` or `pat:backend:<token_id>` 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:<client_id>:<user_id>`, `pat:user:<token_id>` or `pat:backend:<token_id>` 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 `<prefix>mcp:s:<id>` with `PX` TTLs, per-binding sets `<prefix>mcp:b:<binding>`, events on channel `<prefix>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 `<prefix>mcp:s:<id>` with `PX` TTLs, per-binding sets `<prefix>mcp:b:<binding>`, events on channel `<prefix>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<mcp.path>` 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="<app.url>/.well-known/oauth-protected-resource<mcp.path>", scope="read"`; a browser request whose `Origin` is not in `mcp.allowed_origins` is refused with 403 before any body is read.
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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:<token_id>): 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 == "" {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user