From cccf4025eaba8f06742230945ee0cc85a8b6f856 Mon Sep 17 00:00:00 2001 From: Jakub Zych Date: Fri, 23 Oct 2026 20:02:13 +0200 Subject: [PATCH] fix(19.7): RA-WR-01 never wait on a session lock under the router lock Fan-out matching snapshots the sessions (and stateless entries) under the router-wide lock, releases it, and reads each identity through a small identity box that is never held across I/O, so an event no longer waits on a session that is re-filtering while it holds sr.mu or sp.mu. Widening and the fan-out recomputes run their guard checks with guardContext: not cancelled with the triggering request, but bounded by mcp.call_timeout, so a provider that never answers cannot hold a session lock forever. README and docs/services/mcp.md describe the wider meaning of mcp.call_timeout. TestLockScopeUnderWidening blocks a widening guard check, fires a permission event for that session and asserts that another session's call still finishes, and that a never-answering check ends at call_timeout. --- docs/services/mcp.md | 2 +- modules/pier/README.md | 2 +- modules/pier/config.go | 3 +- modules/pier/events.go | 65 +++++++++++++---- modules/pier/lock_scope_test.go | 120 ++++++++++++++++++++++++++++++++ modules/pier/session.go | 10 ++- modules/pier/stateless.go | 24 ++++--- 7 files changed, 198 insertions(+), 28 deletions(-) diff --git a/docs/services/mcp.md b/docs/services/mcp.md index 31f66c3b..04c806e4 100644 --- a/docs/services/mcp.md +++ b/docs/services/mcp.md @@ -39,7 +39,7 @@ path: /mcp | `mcp.resource` | `app.url` + `mcp.path` | The resource indicator; must equal the OAuth server's MCP resource. | | `mcp.throttle` | `120,1` | The endpoint's rate limit, `,`. | | `mcp.allowed_origins` | the origin of `app.url` | Browser origins allowed to call the endpoint; another `Origin` is refused with 403. | -| `mcp.call_timeout` | `30s` | Bound on one tool call. | +| `mcp.call_timeout` | `30s` | Bound on one tool call, and on the guard checks that re-filter a session when its grant or permissions change. | | `mcp.max_result_bytes` | `1048576` | Cap on a route answer returned as a tool result. | ## Declaring tools and prompts diff --git a/modules/pier/README.md b/modules/pier/README.md index fe2d4168..0347e37c 100644 --- a/modules/pier/README.md +++ b/modules/pier/README.md @@ -127,7 +127,7 @@ Configuration keys are exported as constants: `pier.EnabledKey`, `pier.PathKey`, | `mcp.resource` | `app.url` + `mcp.path` | The resource indicator the endpoint identifies itself by; must equal the OAuth server's MCP resource. Required (or `app.url`) when enabled. | | `mcp.throttle` | `120,1` | The endpoint's inline throttle, surf's `,` spelling. | | `mcp.allowed_origins` | the origin of `app.url` | Browser origins (`scheme://host[:port]`) allowed to call the endpoint; a request with another `Origin` is refused with 403 before its body is read. Requests without `Origin` pass. | -| `mcp.call_timeout` | `30s` | Bound on one in-process tool call. | +| `mcp.call_timeout` | `30s` | Bound on one in-process tool call, and on the guard checks pier runs while it re-filters a session or stateless server (widening on return, a fan-out recompute). | | `mcp.max_result_bytes` | `1048576` | Cap on the route answer captured for a tool result; a larger answer is a tool error. | Environment: `SUMMER_MCP__ENABLED`, `SUMMER_MCP__PATH`, `SUMMER_MCP__SESSIONS__STORE` and so on. diff --git a/modules/pier/config.go b/modules/pier/config.go index 2d704198..af2a28cd 100644 --- a/modules/pier/config.go +++ b/modules/pier/config.go @@ -40,7 +40,8 @@ const ( // endpoint; default the origin of app.url. A request without an Origin // header is never refused by it. AllowedOriginsKey = "mcp.allowed_origins" - // CallTimeoutKey bounds one in-process tool call, default 30s. + // CallTimeoutKey bounds one in-process tool call, and the guard checks + // run while a session or stateless server is re-filtered, default 30s. CallTimeoutKey = "mcp.call_timeout" // MaxResultBytesKey caps the route answer captured for a tool result, // default 1048576 (1 MiB); a larger answer is a tool error. diff --git a/modules/pier/events.go b/modules/pier/events.go index 470cda5d..08f1f0dc 100644 --- a/modules/pier/events.go +++ b/modules/pier/events.go @@ -2,6 +2,7 @@ package pier import ( "context" + "sync" "git.golem15.com/golem15/summercms/modules/pact" ) @@ -96,11 +97,13 @@ func revokedMatch(ev Event) func(Identity) bool { // the backend link after a link event) and applies the difference with // AddTool, RemoveTools, AddPrompt and RemovePrompts. func (p *Pier) recomputeSession(ctx context.Context, ls *localSession, relink bool) { + ctx, cancel := p.guardContext(ctx) + defer cancel() ls.mu.Lock() - id := ls.identity + id := ls.identity.get() if relink { p.identities.refreshBackend(ctx, &id) - ls.identity = id + ls.identity.set(id) } tools, prompts := p.catalog.allowed(ctx, id) applyDiff(ls.server, p, id, tools, prompts, ls.tools, ls.prompts) @@ -110,11 +113,13 @@ func (p *Pier) recomputeSession(ctx context.Context, ls *localSession, relink bo // recomputeStateless does the same for one per-identity stateless server. func (p *Pier) recomputeStateless(ctx context.Context, e *statelessEntry, relink bool) { + ctx, cancel := p.guardContext(ctx) + defer cancel() e.mu.Lock() - id := e.identity + id := e.identity.get() if relink { p.identities.refreshBackend(ctx, &id) - e.identity = id + e.identity.set(id) } tools, prompts := p.catalog.allowed(ctx, id) applyDiff(e.server, p, id, tools, prompts, e.tools, e.prompts) @@ -134,12 +139,23 @@ func (p *Pier) widen(ctx context.Context, srv serverFeatures, id Identity, fp *s 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) + // by cancelled guard checks (an error hides the item), but the checks + // still end within mcp.call_timeout: the caller holds a lock. + ctx, cancel := p.guardContext(ctx) + defer cancel() + tools, prompts := p.catalog.allowed(ctx, id) applyDiff(srv, p, id, tools, prompts, haveTools, havePrompts) *fp = next } +// guardContext is the context of guard checks run while a session's or +// entry's mutex is held (widening and fan-out recomputes): not cancelled +// with the request that triggered them, but bounded by mcp.call_timeout, +// so a slow provider never holds that mutex indefinitely (RA-WR-01). +func (p *Pier) guardContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), p.cfg.CallTimeout) +} + // 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) { @@ -187,21 +203,46 @@ func (sr *sessionRouter) matching(guard string, userID uint) []*localSession { } // matchingFunc lists the local sessions whose identity satisfies match. +// The sessions are snapshotted under the router lock and matched after it +// is released, through each session's identity box, which is never held +// across I/O: an event never waits on a session that is re-filtering, and +// never while it holds the router lock (RA-WR-01). func (sr *sessionRouter) matchingFunc(match func(Identity) bool) []*localSession { sr.mu.Lock() - defer sr.mu.Unlock() - var out []*localSession + all := make([]*localSession, 0, len(sr.local)) for _, ls := range sr.local { - ls.mu.Lock() - hit := match(ls.identity) - ls.mu.Unlock() - if hit { + all = append(all, ls) + } + sr.mu.Unlock() + var out []*localSession + for _, ls := range all { + if match(ls.identity.get()) { out = append(out, ls) } } return out } +// identityBox holds the identity of a session or stateless entry behind a +// mutex of its own that is never held across I/O, so fan-out matching +// reads it without waiting for a guard check (RA-WR-01). +type identityBox struct { + mu sync.Mutex + id Identity +} + +func (b *identityBox) get() Identity { + b.mu.Lock() + defer b.mu.Unlock() + return b.id +} + +func (b *identityBox) set(id Identity) { + b.mu.Lock() + b.id = id + b.mu.Unlock() +} + // refreshBackend re-resolves an OAuth identity's backend subject through // the link resolver (after a link event), adding or dropping it. func (r *identityResolver) refreshBackend(ctx context.Context, id *Identity) { diff --git a/modules/pier/lock_scope_test.go b/modules/pier/lock_scope_test.go index a602cc08..3637ea67 100644 --- a/modules/pier/lock_scope_test.go +++ b/modules/pier/lock_scope_test.go @@ -159,3 +159,123 @@ type statusError struct{ got, want int } func (e *statusError) Error() string { return "status " + http.StatusText(e.got) + ", want " + http.StatusText(e.want) } + +// widenGuard is the acme guard whose first MCPAllows for user 8 after arm +// blocks until release or until the check's context ends (then it errors, +// as a database call would). +type widenGuard struct { + acmeGuard + armed atomic.Bool + entered chan struct{} + release chan struct{} +} + +func (g *widenGuard) MCPAllows(ctx context.Context, s pact.MCPSubject, _ pact.MCPRequirements, _ []string) (bool, error) { + if s.UserID == 8 && g.armed.CompareAndSwap(true, false) { + select { + case g.entered <- struct{}{}: + default: + } + select { + case <-g.release: + return true, nil + case <-ctx.Done(): + return false, ctx.Err() + } + } + return true, nil +} + +const listToolsBody = `{"jsonrpc":"2.0","id":3,"method":"tools/list","params":{}}` + +// TestLockScopeUnderWidening pins RA-WR-01: a session re-filtered on return +// (its identity's fingerprint changed) runs its guard checks under its own +// lock only, with mcp.call_timeout as their deadline. A fan-out event that +// concerns the blocked session never stalls the router, so another +// session's call still finishes, and a check that never answers ends at the +// deadline instead of holding the session forever. +func TestLockScopeUnderWidening(t *testing.T) { + setup := func(t *testing.T, mcpYAML string) (*instance, *fakeVerifier, *widenGuard) { + t.Helper() + guard := &widenGuard{entered: make(chan struct{}, 1), release: make(chan struct{})} + verifier := newFakeVerifier(nil) + in := startInstance(t, newApp(t, appSpec{ + mcp: mcpYAML, + guards: []pact.MCPGuard{guard}, + verifier: verifier, + }), acmePlugin{tools: []pact.MCPTool{echoTool(), showTool()}}) + t.Cleanup(in.pier().StopReaper) + return in, verifier, guard + } + // widenBlocked narrows user 8's grant, so the next request on its + // session re-filters, arms the guard and sends that request; it answers + // the request's status channel once the guard check has started. + widenBlocked := func(t *testing.T, ctx context.Context, in *instance, verifier *fakeVerifier, guard *widenGuard, sid string) <-chan int { + t.Helper() + verifier.setGrant("oas_at_other", grant(8, "c2", 2, "read")) + guard.armed.Store(true) + done := make(chan int, 1) + go func() { + status, _ := rawStatus(ctx, http.MethodPost, in.url(), "oas_at_other", sid, "2025-11-25", listToolsBody) + done <- status + }() + select { + case <-guard.entered: + case <-time.After(10 * time.Second): + t.Fatal("the widening request never reached the guard") + } + return done + } + + t.Run("an event for the blocked session does not stall another session", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + in, verifier, guard := setup(t, "enabled: false\nsessions:\n store: memory\n") + active := openSession(t, ctx, in, "oas_at_test") + slow := openSession(t, ctx, in, "oas_at_other") + blocked := widenBlocked(t, ctx, in, verifier, guard, slow) + defer func() { + close(guard.release) + <-blocked + }() + + // The event recomputes user 8's session, so it waits for that + // session's lock; it must not hold the router lock meanwhile. + fireSubjectChanged(t, in.app, "acme", 8, pact.MCPSubjectPermissions) + time.Sleep(200 * time.Millisecond) + + done := make(chan error, 1) + go func() { + got, err := rawStatus(ctx, http.MethodPost, in.url(), "oas_at_test", active, "2025-11-25", callBody) + if err == nil && got != http.StatusOK { + err = &statusError{got: got, want: http.StatusOK} + } + done <- err + }() + select { + case err := <-done: + if err != nil { + t.Fatalf("tools/call on another session: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("another session's call waited for the blocked widening") + } + }) + + t.Run("a blocked widening ends at mcp.call_timeout", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + in, verifier, guard := setup(t, "enabled: false\ncall_timeout: 1s\nsessions:\n store: memory\n") + slow := openSession(t, ctx, in, "oas_at_other") + blocked := widenBlocked(t, ctx, in, verifier, guard, slow) + defer close(guard.release) + select { + case status := <-blocked: + if status != http.StatusOK { + t.Fatalf("the widening request = %d", status) + } + case <-time.After(10 * time.Second): + t.Fatal("the widening guard check outlived mcp.call_timeout") + } + }) +} diff --git a/modules/pier/session.go b/modules/pier/session.go index e92c6bbb..cfe8903f 100644 --- a/modules/pier/session.go +++ b/modules/pier/session.go @@ -65,8 +65,12 @@ type localSession struct { transport *mcp.StreamableServerTransport ss *mcp.ServerSession + // mu serializes re-filtering (widening, fan-out recomputes) and guards + // fingerprint, tools and prompts; it may be held across guard checks, + // bounded by mcp.call_timeout. identity has its own box, so fan-out + // matching never waits on mu. mu sync.Mutex - identity Identity + identity identityBox fingerprint string // of the identity the tool set was last filtered for tools map[string]bool prompts map[string]bool @@ -330,7 +334,7 @@ func (sr *sessionRouter) lookup(w http.ResponseWriter, r *http.Request, id Ident } } ls.mu.Lock() - ls.identity = id + ls.identity.set(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 { @@ -344,7 +348,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, fingerprint: id.fingerprint()} + ls := &localSession{id: sid, binding: id.Binding, created: time.Now().UTC(), identity: identityBox{id: 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 6cb51812..be00c282 100644 --- a/modules/pier/stateless.go +++ b/modules/pier/stateless.go @@ -29,8 +29,10 @@ type statelessEntry struct { server *mcp.Server elem *list.Element + // mu serializes re-filtering and guards fingerprint, tools and prompts + // (see localSession); identity has its own box. mu sync.Mutex - identity Identity + identity identityBox fingerprint string // of the identity the tool set was last filtered for tools map[string]bool prompts map[string]bool @@ -138,7 +140,7 @@ func (sp *statelessPath) getServer(r *http.Request) *mcp.Server { // (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, fingerprint: id.fingerprint(), tools: tools, prompts: prompts} + e := &statelessEntry{binding: id.Binding, server: srv, identity: identityBox{id: id}, fingerprint: id.fingerprint(), tools: tools, prompts: prompts} sp.mu.Lock() if cur, ok := sp.entries[id.Binding]; ok { sp.order.MoveToFront(cur.elem) @@ -184,7 +186,7 @@ func (sp *statelessPath) hit(binding string) *statelessEntry { // 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 + e.identity.set(id) sp.p.widen(ctx, e.server, id, &e.fingerprint, e.tools, e.prompts) e.mu.Unlock() } @@ -201,16 +203,18 @@ func (sp *statelessPath) matching(guard string, userID uint) []*statelessEntry { return sp.matchingFunc(func(id Identity) bool { return id.has(guard, userID) }) } -// matchingFunc lists the entries whose identity satisfies match. +// matchingFunc lists the entries whose identity satisfies match, matched +// after the path lock is released (see sessionRouter.matchingFunc). func (sp *statelessPath) matchingFunc(match func(Identity) bool) []*statelessEntry { sp.mu.Lock() - defer sp.mu.Unlock() - var out []*statelessEntry + all := make([]*statelessEntry, 0, len(sp.entries)) for _, e := range sp.entries { - e.mu.Lock() - hit := match(e.identity) - e.mu.Unlock() - if hit { + all = append(all, e) + } + sp.mu.Unlock() + var out []*statelessEntry + for _, e := range all { + if match(e.identity.get()) { out = append(out, e) } }