feat: add configurable limits for sessions and resource subscriptions

README.md

@@ -263,6 +263,31 @@ await app.register(mcpPlugin, {
 })
 ```
 
+### Session and subscription limits
+
+By default there is no bound on how many sessions one caller may keep open or on how many
+resources one session may subscribe to. Set `maxSessionsPerPrincipal` and
+`maxResourceSubscriptionsPerSession` to bound retained state; each defaults to `Infinity`.
+
+```typescript
+await app.register(mcpPlugin, {
+  maxSessionsPerPrincipal: 10,        // active sessions per caller
+  maxResourceSubscriptionsPerSession: 50
+})
+```
+
+- **`maxSessionsPerPrincipal`** caps the non-expired sessions a caller may keep active. When
+authorization identifies a caller the cap is keyed by the token's (issuer, subject) pair, so
+each principal gets its own quota; without authorization, or for requests that identify no
+caller, every session shares one anonymous bucket. At the cap a new session is refused with
+`SESSION_LIMIT_REACHED` (`-32035`) — an existing session is never torn down, and sessions
+still expire on their own via `sessionTtlMs`. The Redis store enforces the cap atomically, so
+concurrent session creation across instances cannot overshoot it.
+- **`maxResourceSubscriptionsPerSession`** caps the distinct resource URIs one session may
+subscribe to via `resources/subscribe`. Subscribing to an already-subscribed URI stays
+idempotent; only at the cap does a new URI fail with `SUBSCRIPTION_LIMIT_REACHED` (`-32036`).
+The Redis store enforces the cap atomically, so concurrent subscribes cannot overshoot it.
+
 ## Request Body Errors
 
 Body-parser failures and bodies that are not JSON-RPC messages on `POST /mcp` are answered
@@ -916,9 +941,11 @@ const store: SessionStore = new MemorySessionStore()
 ```
 
 - **`SessionStore`**: the interface the plugin uses to persist SSE session metadata, message history, token-to-session mappings, and per-session resource subscriptions (`addResourceSubscription` / `removeResourceSubscription`, exposed as `resourceSubscriptions` on the metadata). The optional `touch(sessionId)` records activity so the session is not expired while in use.
-- **`MemorySessionStore`**: in-process implementation used automatically when no `redis` option is passed. Takes an optional `maxMessages` history cap (default 100) and `{ sessionTtlMs }` idle expiry (default 1 hour); call `cleanup()` periodically to free expired sessions when using it outside the plugin.
+- **`MemorySessionStore`**: in-process implementation used automatically when no `redis` option is passed. Takes an optional `maxMessages` history cap (default 100) and `{ sessionTtlMs }` idle expiry (default 1 hour); call `cleanup()` periodically to free expired sessions when using it outside the plugin. Both stores also accept `maxSessionsPerPrincipal` and `maxResourceSubscriptionsPerSession` to bound active sessions and per-session resource subscriptions.
 - **`RedisSessionStore`**: distributed implementation used automatically when the `redis` option is passed; sessions and message history live in Redis so any instance can serve any client.
 
+`SessionLimitError` and `SubscriptionLimitError` (with `SESSION_LIMIT_REACHED` and `SUBSCRIPTION_LIMIT_REACHED`) are also exported, so you can catch them when calling a store directly or distinguish those failures from an internal error.
+
 These are the same store classes the plugin instantiates internally based on the `redis` option. Exporting them lets you construct or inspect one directly in tests and scripts.
 
 ### Multi-Instance Deployment

src/handlers.ts

@@ -37,6 +37,7 @@ import type { RequestId } from './schema.ts'
 
 import type { MCPTool, MCPResource, MCPPrompt, MCPPluginOptions, ResourceHandlers, McpCallToolOutcome, ToolAccessOperation, MCPToolCallCompleteEvent, TracerLike } from './types.ts'
 import type { SessionStore } from './stores/session-store.ts'
+import { SubscriptionLimitError, SUBSCRIPTION_LIMIT_REACHED } from './stores/session-store.ts'
 import type { TaskStore, TaskRecord, TaskOwner, TaskWaiters, TaskLeases } from './stores/task-store.ts'
 import { InvalidTaskCursorError, TaskLimitError, TASK_LIMIT_REACHED, TASK_OWNER_LOST, decodeTaskCursor, isTaskOwner, isTerminal, ownerLostOutcome, toWireTask } from './stores/task-store.ts'
 import type { AuthorizationContext } from './types/auth-types.ts'
@@ -1474,7 +1475,14 @@ async function handleResourcesSubscribe (
     // Remember the subscription so resources/updated broadcasts only reach
     // the sessions that asked for this URI
     if (sessionId && dependencies.sessionStore) {
-      await dependencies.sessionStore.addResourceSubscription(sessionId, params.uri)
+      try {
+        await dependencies.sessionStore.addResourceSubscription(sessionId, params.uri)
+      } catch (error: any) {
+        if (error instanceof SubscriptionLimitError) {
+          return createError(request.id, SUBSCRIPTION_LIMIT_REACHED, `Resource subscription limit reached (${error.limit})`)
+        }
+        throw error
+      }
     }
     return createResponse(request.id, result)
   } catch (error: any) {

src/index.ts

@@ -2,6 +2,7 @@ import type { FastifyInstance, preHandlerHookHandler } from 'fastify'
 import fp from 'fastify-plugin'
 import { Redis } from 'ioredis'
 import type { SessionStore } from './stores/session-store.ts'
+import { validateStoreLimit } from './stores/session-store.ts'
 import type { MessageBroker } from './brokers/message-broker.ts'
 import { MemorySessionStore } from './stores/memory-session-store.ts'
 import { MemoryMessageBroker } from './brokers/memory-message-broker.ts'
@@ -98,6 +99,17 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
   if (opts.sessionTtlMs !== undefined && !(Number.isFinite(opts.sessionTtlMs) && opts.sessionTtlMs > 0)) {
     throw new Error('sessionTtlMs must be a positive number of milliseconds')
   }
+  const validateSessionLimit = (name: string, value: unknown): void => {
+    // Validation happens here (before a Redis connection is opened) and inside
+    // the stores, so a bad value cannot leak a connection.
+    validateStoreLimit(name, value)
+  }
+  if (opts.maxSessionsPerPrincipal !== undefined) {
+    validateSessionLimit('maxSessionsPerPrincipal', opts.maxSessionsPerPrincipal)
+  }
+  if (opts.maxResourceSubscriptionsPerSession !== undefined) {
+    validateSessionLimit('maxResourceSubscriptionsPerSession', opts.maxResourceSubscriptionsPerSession)
+  }
   if (opts.taskMaxRetainedPerPrincipal !== undefined) {
     // Checked before any Redis connection is opened, so a bad value cannot leak one
     validateTaskLimit('taskMaxRetainedPerPrincipal', opts.taskMaxRetainedPerPrincipal)
@@ -121,7 +133,13 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
     redis.on('error', (err) => {
       app.log.error({ err }, 'Redis connection error')
     })
-    sessionStore = new RedisSessionStore({ redis, maxMessages: 100, sessionTtlMs: opts.sessionTtlMs })
+    sessionStore = new RedisSessionStore({
+      redis,
+      maxMessages: 100,
+      sessionTtlMs: opts.sessionTtlMs,
+      maxSessionsPerPrincipal: opts.maxSessionsPerPrincipal,
+      maxResourceSubscriptionsPerSession: opts.maxResourceSubscriptionsPerSession
+    })
     messageBroker = new RedisMessageBroker(redis, {
       onCloseTimeout: (closeTimeoutMs) => {
         app.log.warn({ closeTimeoutMs }, 'Redis message broker close timed out; forcing disconnect')
@@ -139,7 +157,11 @@ const mcpPlugin = fp(async function (app: FastifyInstance, opts: MCPPluginOption
     }
   } else {
     // Memory implementations for single-instance deployment
-    sessionStore = new MemorySessionStore(100, { sessionTtlMs: opts.sessionTtlMs })
+    sessionStore = new MemorySessionStore(100, {
+      sessionTtlMs: opts.sessionTtlMs,
+      maxSessionsPerPrincipal: opts.maxSessionsPerPrincipal,
+      maxResourceSubscriptionsPerSession: opts.maxResourceSubscriptionsPerSession
+    })
     messageBroker = new MemoryMessageBroker()
     if (enableTasks) {
       taskStore = new MemoryTaskStore({ maxTasksPerPrincipal: opts.taskMaxRetainedPerPrincipal })
@@ -462,8 +484,18 @@ export { RedisTaskStore } from './stores/redis-task-store.ts'
 
 // Session storage, for callers that want to supply or inspect a backend
 export type { SessionStore, SessionMetadata } from './stores/session-store.ts'
-export { MemorySessionStore } from './stores/memory-session-store.ts'
+export {
+  MemorySessionStore
+} from './stores/memory-session-store.ts'
 export { RedisSessionStore } from './stores/redis-session-store.ts'
+export {
+  SessionLimitError,
+  SubscriptionLimitError,
+  SESSION_LIMIT_REACHED,
+  SUBSCRIPTION_LIMIT_REACHED,
+  DEFAULT_MAX_SESSIONS_PER_PRINCIPAL,
+  DEFAULT_MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION
+} from './stores/session-store.ts'
 
 // In-process MCP client, for exercising a registered server via app.inject()
 export { createMcpClient } from './client.ts'

src/routes/mcp.ts

@@ -14,12 +14,13 @@ import type {
   MCPRouteSchemaTransformer
 } from '../types.ts'
 import type { SessionStore, SessionMetadata } from '../stores/session-store.ts'
+import { SessionLimitError, SESSION_LIMIT_REACHED } from '../stores/session-store.ts'
 import { DEFAULT_SESSION_TTL_MS } from '../stores/memory-session-store.ts'
 import type { TaskStore, TaskWaiters, TaskLeases } from '../stores/task-store.ts'
 import type { MessageBroker } from '../brokers/message-broker.ts'
 import type { AuthorizationContext } from '../types/auth-types.ts'
 import type { JsonSchemaValidator } from '../validation/json-schema-validator.ts'
-import { processMessage, checkRequestId } from '../handlers.ts'
+import { processMessage, checkRequestId, createError } from '../handlers.ts'
 import { isStdioRequest } from '../stdio-marker.ts'
 
 const MAX_TIMER_DELAY_MS = 2 ** 31 - 1
@@ -493,6 +494,15 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
           reply.code(202)
         }
       } catch (error) {
+        if (error instanceof SessionLimitError) {
+          // A stable protocol error rather than an opaque 500, so a client that
+          // hit the active-session cap can react (e.g. free a session) instead
+          // of treating it as a transient server failure.
+          const id = Array.isArray(request.body)
+            ? null
+            : (request.body as { id?: string | number | null })?.id ?? null
+          return reply.type('application/json').code(429).send(createError(id, SESSION_LIMIT_REACHED, `Session limit reached (${error.limit})`))
+        }
         app.log.error({ err: error }, 'Error processing MCP message')
         reply.type('application/json').code(500).send({
           jsonrpc: JSONRPC_VERSION,
@@ -542,27 +552,34 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
 
         request.log.info({ sessionId }, 'Handling SSE request')
 
-        // We are opting out of Fastify proper
-        reply.hijack()
-
-        const raw = reply.raw
-
-        // Set up SSE stream
-        raw.setHeader('Content-type', 'text/event-stream')
-        raw.setHeader('Cache-Control', 'no-cache')
-
+        // Resolve (or create) the session before hijacking the reply, so a
+        // failed session creation (e.g. the active-session cap) can still be
+        // answered with a normal HTTP response rather than a hung stream.
         let session: SessionMetadata
+        let isNewSession = false
         if (sessionId) {
           const existingSession = await sessionStore.get(sessionId)
           if (existingSession) {
             session = existingSession
             await sessionStore.touch?.(session.id)
           } else {
             session = await createSession(request)
-            raw.setHeader('Mcp-Session-Id', session.id)
+            isNewSession = true
           }
         } else {
           session = await createSession(request)
+          isNewSession = true
+        }
+
+        // We are opting out of Fastify proper
+        reply.hijack()
+
+        const raw = reply.raw
+
+        // Set up SSE stream
+        raw.setHeader('Content-type', 'text/event-stream')
+        raw.setHeader('Cache-Control', 'no-cache')
+        if (isNewSession) {
           raw.setHeader('Mcp-Session-Id', session.id)
         }
 
@@ -685,6 +702,10 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
           clearInterval(heartbeatInterval)
         })
       } catch (error) {
+        if (error instanceof SessionLimitError) {
+          reply.type('application/json').code(429).send({ error: `Session limit reached (${error.limit})` })
+          return
+        }
         app.log.error({ err: error }, 'Error setting up SSE stream')
         reply.type('application/json').code(500).send({ error: 'Internal server error' })
       }

src/stores/memory-session-store.ts

@@ -1,5 +1,6 @@
 import type { JSONRPCMessage } from '../schema.ts'
 import type { SessionStore, SessionMetadata } from './session-store.ts'
+import { SessionLimitError, SubscriptionLimitError, DEFAULT_MAX_SESSIONS_PER_PRINCIPAL, DEFAULT_MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION, validateStoreLimit } from './session-store.ts'
 import type { AuthorizationContext, TokenRefreshInfo } from '../types/auth-types.ts'
 
 interface MessageHistoryEntry {
@@ -16,6 +17,17 @@ export interface MemorySessionStoreOptions {
    * (default 1 hour, like the Redis store's TTL).
    */
   sessionTtlMs?: number
+  /**
+   * Most non-expired sessions one principal may keep active (default
+   * Infinity). Keyed by the session's (ownerIss, ownerSub) pair; sessions with
+   * no owner share one anonymous bucket.
+   */
+  maxSessionsPerPrincipal?: number
+  /**
+   * Most distinct resource URIs one session may subscribe to (default
+   * Infinity). Duplicate subscriptions are idempotent.
+   */
+  maxResourceSubscriptionsPerSession?: number
 }
 
 export class MemorySessionStore implements SessionStore {
@@ -24,10 +36,40 @@ export class MemorySessionStore implements SessionStore {
   private tokenToSession = new Map<string, string>() // tokenHash -> sessionId
   private maxMessages: number
   readonly sessionTtlMs: number
+  private readonly maxSessionsPerPrincipal: number
+  private readonly maxResourceSubscriptionsPerSession: number
 
   constructor (maxMessages: number = 100, options: MemorySessionStoreOptions = {}) {
     this.maxMessages = maxMessages
     this.sessionTtlMs = options.sessionTtlMs ?? DEFAULT_SESSION_TTL_MS
+    this.maxSessionsPerPrincipal = options.maxSessionsPerPrincipal ?? DEFAULT_MAX_SESSIONS_PER_PRINCIPAL
+    this.maxResourceSubscriptionsPerSession = options.maxResourceSubscriptionsPerSession ?? DEFAULT_MAX_RESOURCE_SUBSCRIPTIONS_PER_SESSION
+    validateStoreLimit('maxSessionsPerPrincipal', this.maxSessionsPerPrincipal)
+    validateStoreLimit('maxResourceSubscriptionsPerSession', this.maxResourceSubscriptionsPerSession)
+  }
+
+  /**
+   * The bucket a session counts against for the per-principal active-session cap.
+   * A session is owned when both the issuer and subject are known; otherwise all
+   * ownerless sessions share one anonymous bucket.
+   */
+  private static ownerKey (metadata: SessionMetadata): string {
+    if (metadata.ownerSub === undefined && metadata.ownerIss === undefined) {
+      return '__anonymous__'
+    }
+    return `${metadata.ownerIss ?? ''}:${metadata.ownerSub ?? ''}`
+  }
+
+  /** Count non-expired sessions in the same owner bucket. */
+  private activeSessionCountFor (metadata: SessionMetadata): number {
+    const key = MemorySessionStore.ownerKey(metadata)
+    let count = 0
+    for (const [otherId, other] of this.sessions) {
+      if (otherId === metadata.id) continue
+      if (this.isExpired(other)) continue
+      if (MemorySessionStore.ownerKey(other) === key) count++
+    }
+    return count
   }
 
   private isExpired (session: SessionMetadata): boolean {
@@ -50,6 +92,10 @@ export class MemorySessionStore implements SessionStore {
   }
 
   async create (metadata: SessionMetadata): Promise<void> {
+    if (this.maxSessionsPerPrincipal !== Infinity &&
+        this.activeSessionCountFor(metadata) >= this.maxSessionsPerPrincipal) {
+      throw new SessionLimitError(this.maxSessionsPerPrincipal)
+    }
     this.sessions.set(metadata.id, { ...metadata })
     this.messageHistory.set(metadata.id, [])
   }
@@ -209,9 +255,15 @@ export class MemorySessionStore implements SessionStore {
     const session = this.live(sessionId)
     if (!session) return
     const subscriptions = session.resourceSubscriptions ??= []
-    if (!subscriptions.includes(uri)) {
-      subscriptions.push(uri)
+    // Duplicate subscriptions stay idempotent: the URI is already present.
+    if (subscriptions.includes(uri)) {
+      return
+    }
+    if (this.maxResourceSubscriptionsPerSession !== Infinity &&
+        subscriptions.length >= this.maxResourceSubscriptionsPerSession) {
+      throw new SubscriptionLimitError(this.maxResourceSubscriptionsPerSession)
     }
+    subscriptions.push(uri)
   }
 
   async removeResourceSubscription (sessionId: string, uri: string): Promise<void> {