diff --git a/components/mcp/lifecycle.ts b/components/mcp/lifecycle.ts index aafcca9ef9..b64cf98c26 100644 --- a/components/mcp/lifecycle.ts +++ b/components/mcp/lifecycle.ts @@ -7,7 +7,7 @@ * server SHOULD respond with its preferred supported version so the client * can decide whether to connect on the older version or disconnect. */ -import { createSession, saveSession, type McpSessionRecord } from './session.ts'; +import { createSession, patchSession, type McpSessionRecord } from './session.ts'; import { packageJson } from '../../utility/packageUtils.js'; export const PROTOCOL_VERSION_PREFERRED = '2025-06-18'; @@ -106,6 +106,6 @@ export async function handleInitialize( export async function handleInitialized(session: McpSessionRecord): Promise { if (session.initialized) return session; const updated: McpSessionRecord = { ...session, initialized: true }; - await saveSession(updated); + await patchSession(session.id, { initialized: true }); return updated; } diff --git a/components/mcp/session.ts b/components/mcp/session.ts index b537cc34ae..4d23f7715e 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -3,7 +3,7 @@ * * Eviction is delegated to Harper's native TTL (`Table.setTTLExpiration`): * every write to a session record updates its `version`, which Harper uses - * to determine expiration. So calling `saveSession(record)` on each request + * to determine expiration. So calling `patchSession(id, changes)` on each request * gives sliding-window idle semantics for free — no custom timer, no sweep. * * Spec: when a request bears an `Mcp-Session-Id` the server doesn't @@ -34,6 +34,7 @@ const DEFAULT_IDLE_TIMEOUT_SECONDS = 1800; * is identical either way.) */ const EVICTION_WINDOW_SECONDS = 60; +const DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS = 10_000; export interface McpSessionRecord { id: string; @@ -63,9 +64,12 @@ export interface McpSessionRecord { * to clients that declared support. Undefined = client declared none. */ clientCapabilities?: Record; + /** Node-local hint for the worker that owns the current GET-SSE stream. */ + streamOwner?: { threadId: number; token: string }; } let _sessionTable: Table | undefined; +let subscriptionLockTimeoutMs = DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS; /** * Lazily declare the system table. Called by `ensureSessionTable()` at @@ -90,6 +94,7 @@ function declareSessionTable(): Table { { name: 'logLevel' }, { name: 'subscriptions' }, { name: 'clientCapabilities' }, + { name: 'streamOwner' }, ], }); } @@ -111,6 +116,10 @@ export function _setSessionTableForTest(fake: Table | undefined): void { _sessionTable = fake; } +export function _setSubscriptionLockTimeoutForTest(value: number | undefined): void { + subscriptionLockTimeoutMs = value ?? DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS; +} + function getTable(): Table { if (!_sessionTable) throw new Error('MCP session table not initialized'); return _sessionTable; @@ -145,16 +154,70 @@ export async function createSession({ */ export async function loadSession(id: string): Promise { const record = (await (getTable() as any).get(id)) as McpSessionRecord | undefined | null; - if (!record) return null; + if ( + !record || + typeof record.protocolVersion !== 'string' || + typeof record.initialized !== 'boolean' || + typeof record.user !== 'string' || + typeof record.createdAt !== 'number' || + typeof record.lastActivity !== 'number' + ) + return null; return record; } -/** - * Persist updated session state. Used to bump `lastActivity` (sliding-window - * idle reset) and to flip `initialized` after `notifications/initialized`. - */ -export async function saveSession(record: McpSessionRecord): Promise { - await (getTable() as any).put(record); +export async function patchSession(id: string, changes: Partial>): Promise { + await (getTable() as any).patch({ id, ...changes }); +} + +function subscriptionLockKey(id: string): string { + return `mcp-subscriptions:${id}`; +} + +function acquireSessionSubscriptionLock(store: Table['primaryStore'], key: string): Promise { + return new Promise((resolve, reject) => { + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + reject(new Error('Timed out acquiring MCP subscription lock')); + }, subscriptionLockTimeoutMs); + const attempt = () => { + if (settled) return; + try { + if (store.tryLock(key, attempt)) { + settled = true; + clearTimeout(timer); + resolve(); + } + } catch (error) { + settled = true; + clearTimeout(timer); + reject(error); + } + }; + attempt(); + }); +} + +/** Serialize durable subscription read-modify-writes across HTTP workers. */ +export async function updateSessionSubscriptions( + id: string, + update: (subscriptions: string[]) => string[] | Promise +): Promise { + const store = getTable().primaryStore; + const key = subscriptionLockKey(id); + await acquireSessionSubscriptionLock(store, key); + try { + const session = await loadSession(id); + if (!session) return null; + const subscriptions = session.subscriptions ?? []; + const updated = await update(subscriptions); + if (updated !== subscriptions) await patchSession(id, { subscriptions: updated }); + return updated === subscriptions ? session : { ...session, subscriptions: updated }; + } finally { + store.unlock(key); + } } export async function deleteSession(id: string): Promise { @@ -174,6 +237,6 @@ export async function deleteSession(id: string): Promise { */ export async function touchSession(record: McpSessionRecord): Promise { const touched: McpSessionRecord = { ...record, lastActivity: Date.now() }; - await saveSession(touched); + await patchSession(record.id, { lastActivity: touched.lastActivity }); return touched; } diff --git a/components/mcp/sessionRegistry.ts b/components/mcp/sessionRegistry.ts index 9c10840930..f72110e14a 100644 --- a/components/mcp/sessionRegistry.ts +++ b/components/mcp/sessionRegistry.ts @@ -17,6 +17,7 @@ * without a graceful close). */ import { IterableEventQueue } from '../../resources/IterableEventQueue.ts'; +import { randomUUID } from 'node:crypto'; import type { McpLogLevel } from './logging.ts'; import type { AuthedUser } from './toolRegistry.ts'; import type { McpProfile } from './transport.ts'; @@ -29,6 +30,7 @@ export interface SseEvent { export interface RegisteredSession { sessionId: string; + streamToken: string; profile: McpProfile; user: AuthedUser; queue: IterableEventQueue; @@ -98,6 +100,7 @@ export function registerSession(sessionId: string, profile: McpProfile, user: Au const queue = new IterableEventQueue(); const record: RegisteredSession = { sessionId, + streamToken: randomUUID(), profile, user, queue, diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts new file mode 100644 index 0000000000..7fd2e94e07 --- /dev/null +++ b/components/mcp/subscriptionRouting.ts @@ -0,0 +1,321 @@ +/** Route MCP resource-subscription commands to the worker owning the GET-SSE stream. */ +import { randomUUID } from 'node:crypto'; +import { threadId } from 'node:worker_threads'; +import { ITC_EVENT_TYPES } from '../../utility/hdbTerms.ts'; +import harperLogger from '../../utility/logging/harper_logger.ts'; +import { patchSession, updateSessionSubscriptions, type McpSessionRecord } from './session.ts'; +import { getRegisteredSession } from './sessionRegistry.ts'; +import { addResourceSubscription, removeResourceSubscription } from './subscriptions.ts'; +import type { AuthedUser } from './toolRegistry.ts'; + +const DEFAULT_RESPONSE_TIMEOUT_MS = 30_000; +const MAX_PENDING = 100; +const MAX_PENDING_PER_SESSION = 25; + +export type SubscriptionRouteResult = 'success' | 'not-subscribable' | 'no-live-stream' | 'timeout' | 'internal-error'; +type Operation = 'subscribe' | 'unsubscribe'; + +interface Command { + requestId: string; + originator: number; + sessionId: string; + streamToken: string; + operation: Operation; + uri: string; + user?: AuthedUser; +} + +interface Response { + requestId: string; + originator: number; + result: SubscriptionRouteResult; +} + +function isCommand(value: unknown): value is Command { + if (!value || typeof value !== 'object') return false; + const command = value as Partial; + return ( + typeof command.requestId === 'string' && + typeof command.originator === 'number' && + typeof command.sessionId === 'string' && + typeof command.streamToken === 'string' && + (command.operation === 'subscribe' || command.operation === 'unsubscribe') && + typeof command.uri === 'string' && + (command.user === undefined || (command.user !== null && typeof command.user === 'object')) + ); +} + +interface ItcBridge { + available?: boolean; + sendToThread(threadId: number, event: { type: string; message: unknown }): boolean; + onMessageByType(type: string, listener: (event: { message?: unknown }) => void): void; +} + +// manageThreads assigns this connected-port array as the package-global `threads` export. Its +// direct-send helper lives on that array, while typed listener registration is a module export. +declare const threads: { sendToThread(threadId: number, event: { type: string; message: unknown }): boolean }; + +interface Pending { + sessionId: string; + targetThreadId: number; + resolve: (result: SubscriptionRouteResult) => void; + timer: ReturnType; +} + +const pending = new Map(); +const operationChains = new Map>(); +let wired = false; +let bridgeOverride: ItcBridge | undefined; +let currentThreadId = (): number => threadId; +let responseTimeoutMs = DEFAULT_RESPONSE_TIMEOUT_MS; + +function bridge(): ItcBridge { + if (bridgeOverride) return bridgeOverride; + try { + const { onMessageByType } = require('../../server/threads/manageThreads.js'); + if (typeof threads !== 'undefined' && typeof threads.sendToThread === 'function') { + return { available: true, sendToThread: threads.sendToThread.bind(threads), onMessageByType }; + } + } catch (error) { + harperLogger.trace(`MCP subscription routing is unavailable: ${(error as Error).message}`); + return { available: false, sendToThread: () => false, onMessageByType: () => {} }; + } + harperLogger.trace('MCP subscription routing is unavailable: thread bridge is not initialized'); + return { available: false, sendToThread: () => false, onMessageByType: () => {} }; +} + +export function _setSubscriptionItcForTest(fake: ItcBridge | undefined): void { + bridgeOverride = fake; + wired = false; +} + +export function _setSubscriptionThreadIdForTest(value: number | undefined): void { + currentThreadId = value === undefined ? () => threadId : () => value; +} + +export function _setSubscriptionTimeoutForTest(value: number | undefined): void { + responseTimeoutMs = value ?? DEFAULT_RESPONSE_TIMEOUT_MS; +} + +export function _resetSubscriptionRoutingForTest(): void { + for (const entry of pending.values()) clearTimeout(entry.timer); + pending.clear(); + operationChains.clear(); + wired = false; + currentThreadId = () => threadId; + responseTimeoutMs = DEFAULT_RESPONSE_TIMEOUT_MS; +} + +export function _pendingSubscriptionRouteCount(): number { + return pending.size; +} + +function ensureWired(): boolean { + if (wired) return true; + const itc = bridge(); + if (itc.available === false) return false; + itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { + const command = event.message; + if (!isCommand(command)) { + harperLogger.warn('Ignoring malformed MCP subscription command'); + return; + } + void handleCommand(command).catch((error) => { + harperLogger.error('MCP subscription command failed', error); + sendResponse(command, 'internal-error'); + }); + }); + itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, (event) => { + const response = event.message as Response; + const entry = pending.get(response?.requestId); + if (!entry || response.originator !== entry.targetThreadId) return; + if (!['success', 'not-subscribable', 'no-live-stream', 'timeout', 'internal-error'].includes(response.result)) + return; + clearTimeout(entry.timer); + pending.delete(response.requestId); + entry.resolve(response.result); + }); + wired = true; + return true; +} + +export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { + if (!ensureWired()) { + await patchSession(sessionId, { streamOwner: undefined }); + return false; + } + await patchSession(sessionId, { streamOwner: { threadId: currentThreadId(), token: streamToken } }); + return true; +} + +function countPendingForSession(sessionId: string): number { + let count = 0; + for (const entry of pending.values()) if (entry.sessionId === sessionId) count++; + return count; +} + +export function subscriptionUser(user: AuthedUser): AuthedUser { + // Preserve role permissions and scoped-token expiry for authorization rechecks; + // credentials and mutable user state must not cross the worker boundary. + return { + ...(user.username !== undefined ? { username: user.username } : {}), + ...(user.authExpiresAt !== undefined ? { authExpiresAt: user.authExpiresAt } : {}), + ...(user._scopedToken ? { _scopedToken: true } : {}), + ...(user.role + ? { + role: { + ...(user.role.role !== undefined ? { role: user.role.role } : {}), + ...(user.role.permission ? { permission: user.role.permission } : {}), + }, + } + : {}), + }; +} + +function routeRemote( + owner: NonNullable, + command: Omit +): Promise { + if (!ensureWired()) return Promise.resolve('no-live-stream'); + if (pending.size >= MAX_PENDING || countPendingForSession(command.sessionId) >= MAX_PENDING_PER_SESSION) { + return Promise.resolve('internal-error'); + } + const requestId = randomUUID(); + return new Promise((resolve) => { + const timer = setTimeout(() => { + pending.delete(requestId); + resolve('timeout'); + }, responseTimeoutMs); + timer.unref(); + pending.set(requestId, { sessionId: command.sessionId, targetThreadId: owner.threadId, resolve, timer }); + let sent = false; + let sendFailed = false; + try { + sent = bridge().sendToThread(owner.threadId, { + type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, + message: { + ...command, + ...(command.user ? { user: subscriptionUser(command.user) } : {}), + requestId, + originator: currentThreadId(), + streamToken: owner.token, + }, + }); + } catch (error) { + sendFailed = true; + harperLogger.error('Unable to route MCP subscription command', error); + } + if (!sent) { + clearTimeout(timer); + pending.delete(requestId); + resolve(sendFailed ? 'internal-error' : 'no-live-stream'); + } + }); +} + +export function withSessionSubscriptionLock(sessionId: string, operation: () => Promise): Promise { + const previous = operationChains.get(sessionId) ?? Promise.resolve(); + const current = previous.then(operation, operation); + const tail = current + .then( + () => undefined, + () => undefined + ) + .finally(() => { + if (operationChains.get(sessionId) === tail) operationChains.delete(sessionId); + }); + operationChains.set(sessionId, tail); + return current; +} + +async function executeLocal(command: Command): Promise { + const registered = getRegisteredSession(command.sessionId); + if (!registered || registered.streamToken !== command.streamToken) return 'no-live-stream'; + return withSessionSubscriptionLock(command.sessionId, async () => { + const current = getRegisteredSession(command.sessionId); + if (!current || current.streamToken !== command.streamToken) return 'no-live-stream'; + if (command.operation === 'subscribe') { + if (!command.user) return 'internal-error'; + const added = await addResourceSubscription(command.sessionId, command.uri, command.user); + if (!added) return 'not-subscribable'; + try { + const session = await updateSessionSubscriptions(command.sessionId, (subscriptions) => + subscriptions.includes(command.uri) ? subscriptions : [...subscriptions, command.uri] + ); + if (!session) { + removeResourceSubscription(command.sessionId, command.uri); + return 'no-live-stream'; + } + return 'success'; + } catch (error) { + removeResourceSubscription(command.sessionId, command.uri); + throw error; + } + } + await updateSessionSubscriptions(command.sessionId, (subscriptions) => + subscriptions.includes(command.uri) ? subscriptions.filter((uri) => uri !== command.uri) : subscriptions + ); + removeResourceSubscription(command.sessionId, command.uri); + return 'success'; + }); +} + +async function executeLocalContained(command: Command): Promise { + try { + return await executeLocal(command); + } catch (error) { + harperLogger.error('MCP subscription owner failed to execute command', error); + return 'internal-error'; + } +} + +async function handleCommand(command: Command): Promise { + const result = await executeLocalContained(command); + sendResponse(command, result); +} + +function sendResponse(command: Command, result: SubscriptionRouteResult): void { + try { + bridge().sendToThread(command.originator, { + type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, + message: { requestId: command.requestId, originator: currentThreadId(), result } satisfies Response, + }); + } catch (error) { + harperLogger.trace(`Unable to return MCP subscription response: ${(error as Error).message}`); + } +} + +export async function routeResourceSubscription(args: { + session: McpSessionRecord; + operation: Operation; + uri: string; + user?: AuthedUser; +}): Promise { + const owner = args.session.streamOwner; + const command = { + sessionId: args.session.id, + operation: args.operation, + uri: args.uri, + ...(args.user ? { user: args.user } : {}), + }; + if (!owner) { + if (ensureWired()) return 'no-live-stream'; + const registered = getRegisteredSession(args.session.id); + if (!registered) return 'no-live-stream'; + return executeLocalContained({ + ...command, + requestId: '', + originator: currentThreadId(), + streamToken: registered.streamToken, + }); + } + if (owner.threadId === currentThreadId()) { + return executeLocalContained({ + ...command, + requestId: '', + originator: currentThreadId(), + streamToken: owner.token, + }); + } + return routeRemote(owner, command); +} diff --git a/components/mcp/subscriptions.ts b/components/mcp/subscriptions.ts index 40bc22bdb2..190b651c6d 100644 --- a/components/mcp/subscriptions.ts +++ b/components/mcp/subscriptions.ts @@ -74,23 +74,24 @@ export function dropSessionSubscriptions(sessionId: string): void { /** * Re-establish subscriptions on SSE reconnect from the durable URI list. Each is * best-effort: a URI that's no longer subscribable (resource removed) is skipped. - * Returns the URIs that were successfully restored (the caller prunes the rest - * from the durable record). + * Returns the URIs that should remain durable: successful subscriptions and + * retryable failures are retained, while confirmed non-subscribable URIs are omitted. */ export async function restoreResourceSubscriptions( sessionId: string, uris: ReadonlyArray, user: AuthedUser ): Promise { - const restored: string[] = []; + const retained: string[] = []; for (const uri of uris) { try { - if (await addResourceSubscription(sessionId, uri, user)) restored.push(uri); + if (await addResourceSubscription(sessionId, uri, user)) retained.push(uri); } catch (err) { harperLogger.trace(`MCP subscription restore ${uri}: ${(err as Error).message}`); + retained.push(uri); } } - return restored; + return retained; } /** Test seam — count of live subscriptions for a session. */ diff --git a/components/mcp/toolRegistry.ts b/components/mcp/toolRegistry.ts index fc0131ba4f..35b41c229d 100644 --- a/components/mcp/toolRegistry.ts +++ b/components/mcp/toolRegistry.ts @@ -60,6 +60,7 @@ export interface ToolDescriptor { /** Authenticated user object as Harper builds it (subset we touch). */ export interface AuthedUser { username?: string; + authExpiresAt?: number; // Attribution-only principal that must not be re-resolved against hdb_user (see refreshSessionUser). _scopedToken?: boolean; role?: { diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index f5187ec2bb..78fa3107ad 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -32,16 +32,23 @@ import { decodeCursor } from './pagination.ts'; import { seedSessionSnapshot } from './listChanged.ts'; import { tryAdmit, resolveClientIdentity } from './rateLimit.ts'; import { checkDurableQuota } from './quota.ts'; -import { deleteSession, loadSession, saveSession, touchSession, type McpSessionRecord } from './session.ts'; +import { + deleteSession, + loadSession, + patchSession, + touchSession, + updateSessionSubscriptions, + type McpSessionRecord, +} from './session.ts'; import { listResources, listResourceTemplates, readResource, completeResourceArgument } from './resources.ts'; import { ensureApplicationToolsFresh } from './tools/application.ts'; import { getPrompt, listPrompts, completePromptArgument } from './promptRegistry.ts'; +import { dropSessionSubscriptions, restoreResourceSubscriptions } from './subscriptions.ts'; import { - addResourceSubscription, - removeResourceSubscription, - dropSessionSubscriptions, - restoreResourceSubscriptions, -} from './subscriptions.ts'; + claimSubscriptionOwner, + routeResourceSubscription, + withSessionSubscriptionLock, +} from './subscriptionRouting.ts'; import { sendServerRequest, routeClientResponse, @@ -50,8 +57,8 @@ import { } from './serverRequests.ts'; import { registerSession, - touchRegisteredSession, getRegisteredSession, + touchRegisteredSession, replaySince, type SseEvent, } from './sessionRegistry.ts'; @@ -256,11 +263,8 @@ async function handlePost(request: NormRequest): Promise { return { status: 403, headers: {} }; } - // Sliding-window idle reset. Awaited (not fire-and-forget) so a concurrent - // DELETE that arrives mid-request can't be resurrected by a late put. Adopt - // the touched copy (fresh `lastActivity`) so any later save in this request - // — `handleInitialized`, `dispatchSetLevel` — persists the current activity - // time instead of rolling it back to the load-time value. + // Adopt the persisted sliding-window update so later writes cannot roll + // lastActivity back; loadSession rejects partial rows from DELETE/patch races. session = await touchSession(session); // A client's response to a server→client request (#3.7): route it to the @@ -370,7 +374,7 @@ async function dispatchSetLevel( // on the worker where the change fires. Cross-worker push is a separate, // subsystem-wide design item (tracked in the MCP design-doc issue). session.logLevel = level; - await saveSession(session); + await patchSession(session.id, { logLevel: level }); setSessionLogLevel(session.id, level); return jsonResponse(200, buildSuccess(messageId, {})); } @@ -403,6 +407,12 @@ async function handleGet(request: NormRequest): Promise { }; } const record = registerSession(sessionId, request.profile, effectiveUser(request)); + try { + await claimSubscriptionOwner(sessionId, record.streamToken); + } catch (error) { + record.queue.emit('close'); + throw error; + } // Seed the live record with any previously-set logging level so a reconnect // (or a setLevel that preceded this stream) keeps delivering notifications/message. // (A fresh record's logLevel is already undefined, so a direct assign is safe.) @@ -416,13 +426,31 @@ async function handleGet(request: NormRequest): Promise { record.queue.once('close', () => dropSessionSubscriptions(sessionId)); // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. - if (session.subscriptions?.length) { - const restored = await restoreResourceSubscriptions(sessionId, session.subscriptions, effectiveUser(request)); - if (restored.length !== session.subscriptions.length) { - session.subscriptions = restored; - await saveSession(session); + try { + const sessionPresent = await withSessionSubscriptionLock(sessionId, async () => { + const snapshot = await updateSessionSubscriptions(sessionId, (subscriptions) => subscriptions); + if (!snapshot) return false; + const attempted = snapshot.subscriptions ?? []; + if (!attempted.length) return true; + const retained = await restoreResourceSubscriptions(sessionId, attempted, effectiveUser(request)); + if (retained.length === attempted.length) return true; + const attemptedSet = new Set(attempted); + const retainedSet = new Set(retained); + await updateSessionSubscriptions(sessionId, (subscriptions) => { + const updated = subscriptions.filter((uri) => !attemptedSet.has(uri) || retainedSet.has(uri)); + return updated.length === subscriptions.length ? subscriptions : updated; + }); + return true; + }); + if (!sessionPresent) { + record.queue.emit('close'); + return { status: 404, headers: {} }; } + } catch (error) { + record.queue.emit('close'); + throw error; } + if (getRegisteredSession(sessionId) !== record) dropSessionSubscriptions(sessionId); // Resumability (#3.8): on reconnect with Last-Event-ID, replay buffered frames // the client missed (those with a higher id) before live frames flow. Re-sent // raw so their original event ids are preserved. Best-effort + per-worker: the @@ -917,26 +945,31 @@ async function dispatchResourcesSubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/subscribe requires params.uri') ); } - // Require a live GET SSE stream: that's where notifications/resources/updated is - // delivered, and its 'close' is the only teardown hook for the subscription. A - // subscription opened without a stream would leak its audit-log iterator and - // drop every update silently. - if (!getRegisteredSession(session.id)) { + const result = await routeResourceSubscription({ + session, + operation: 'subscribe', + uri, + user: effectiveUser(request), + }); + if (result === 'no-live-stream') { return jsonResponse( 200, buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'open the GET SSE stream before subscribing to resources') ); } - const ok = await addResourceSubscription(session.id, uri, effectiveUser(request)); - if (!ok) { + if (result === 'not-subscribable') { // Only row-backed application resources are subscribable; synthetic harper://* // URIs (and unknown URIs) have no change source. return jsonResponse(200, buildError(messageId, ERROR_CODES.INVALID_PARAMS, `resource is not subscribable: ${uri}`)); } - // Persist the URI on the durable record so it survives an SSE reconnect. - if (!session.subscriptions?.includes(uri)) { - session.subscriptions = [...(session.subscriptions ?? []), uri]; - await saveSession(session); + if (result === 'timeout') { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource subscription timed out; retry the request') + ); + } + if (result === 'internal-error') { + return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource subscription failed')); } return jsonResponse(200, buildSuccess(messageId, {})); } @@ -955,10 +988,47 @@ async function dispatchResourcesUnsubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/unsubscribe requires params.uri') ); } - removeResourceSubscription(session.id, uri); - if (session.subscriptions?.includes(uri)) { - session.subscriptions = session.subscriptions.filter((u) => u !== uri); - await saveSession(session); + let routedSession = session; + let result: Awaited> = 'no-live-stream'; + let ownerChangedAfterFinalAttempt = false; + for (let attempt = 0; attempt < 3; attempt++) { + result = await routeResourceSubscription({ session: routedSession, operation: 'unsubscribe', uri }); + if (result !== 'no-live-stream') break; + const currentSession = await loadSession(session.id); + const owner = routedSession.streamOwner; + const currentOwner = currentSession?.streamOwner; + if (!currentSession || !currentOwner) break; + if (owner && currentOwner.threadId === owner.threadId && currentOwner.token === owner.token) break; + routedSession = currentSession; + ownerChangedAfterFinalAttempt = attempt === 2; + } + if (result === 'no-live-stream' && ownerChangedAfterFinalAttempt) { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe owner changed; retry the request') + ); + } + if (result === 'no-live-stream') { + // The live owner is already gone. Remove durable state locally so a later + // reconnect cannot restore the cancelled subscription. + try { + await updateSessionSubscriptions(session.id, (subscriptions) => + subscriptions.includes(uri) ? subscriptions.filter((subscription) => subscription !== uri) : subscriptions + ); + } catch (error) { + harperLogger.error('MCP resource unsubscribe durable update failed', error); + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe failed; retry the request') + ); + } + } else if (result === 'timeout') { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe timed out; retry the request') + ); + } else if (result === 'internal-error') { + return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe failed')); } return jsonResponse(200, buildSuccess(messageId, {})); } diff --git a/integrationTests/fixtures/custom-resources/resources.js b/integrationTests/fixtures/custom-resources/resources.js index 547c278df6..f40db0703a 100644 --- a/integrationTests/fixtures/custom-resources/resources.js +++ b/integrationTests/fixtures/custom-resources/resources.js @@ -1,3 +1,13 @@ +import { threadId } from 'node:worker_threads'; + +/** Test-only endpoint used to pin HTTP requests to distinct keep-alive worker connections. */ +export class WorkerIdentity extends Resource { + static loadAsInstance = false; + get() { + return { threadId }; + } +} + // WorkItem: async write-then-patch pattern (CDI RT enqueueing + AI inference result attachment) export class WorkItem extends tables.WorkItem { // Author-opt-in custom MCP surface declared on a subclass of an exported @table (#1448). diff --git a/integrationTests/mcp/sse-listchanged.test.ts b/integrationTests/mcp/sse-listchanged.test.ts index cc9d1ef48c..4025d7b8fa 100644 --- a/integrationTests/mcp/sse-listchanged.test.ts +++ b/integrationTests/mcp/sse-listchanged.test.ts @@ -20,6 +20,7 @@ import { suite, test, before, after } from 'node:test'; import { ok, strictEqual } from 'node:assert'; import { resolve } from 'node:path'; +import { Agent, request as httpRequest } from 'node:http'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; @@ -114,6 +115,65 @@ async function mcpCall( return res.json(); } +async function requestWithAgent( + baseUrl: string, + agent: Agent, + path: string, + options: { method?: string; headers?: Record; body?: string; localPort?: number } = {} +): Promise<{ status: number; body: string }> { + return new Promise((resolveRequest, reject) => { + const request = httpRequest( + new URL(path, baseUrl), + { method: options.method ?? 'GET', headers: options.headers, agent, localPort: options.localPort }, + (response) => { + const chunks: Buffer[] = []; + response.on('data', (chunk) => chunks.push(Buffer.from(chunk))); + response.on('end', () => + resolveRequest({ status: response.statusCode ?? 0, body: Buffer.concat(chunks).toString() }) + ); + } + ); + request.on('error', reject); + if (options.body) request.end(options.body); + else request.end(); + }); +} + +async function workerIdentity(baseUrl: string, auth: string, agent: Agent, localPort: number): Promise { + const response = await requestWithAgent(baseUrl, agent, '/WorkerIdentity', { + headers: { authorization: auth }, + localPort, + }); + strictEqual(response.status, 200, `WorkerIdentity failed: ${response.status} ${response.body}`); + return JSON.parse(response.body).threadId; +} + +async function mcpCallWithAgent( + baseUrl: string, + auth: string, + agent: Agent, + session: { sessionId: string; protocolVersion: string }, + method: string, + params: Record, + id: number, + localPort: number +): Promise { + const response = await requestWithAgent(baseUrl, agent, '/mcp', { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'accept': 'application/json, text/event-stream', + 'authorization': auth, + 'mcp-session-id': session.sessionId, + 'mcp-protocol-version': session.protocolVersion, + }, + body: JSON.stringify({ jsonrpc: '2.0', id, method, params }), + localPort, + }); + strictEqual(response.status, 200, `MCP ${method} failed: ${response.status} ${response.body}`); + return JSON.parse(response.body); +} + /** Upsert a WorkItem record by id through the application REST API. */ async function putWorkItem(ctx: ContextWithHarper, id: string, body: Record): Promise { const res = await fetch(new URL(`/WorkItem/${id}`, ctx.harper.httpURL), { @@ -223,10 +283,102 @@ async function openSse( }; } +/** Open SSE on the sole socket in `agent`, pinning the stream to that socket's worker. */ +async function openSseWithAgent( + baseUrl: string, + auth: string, + agent: Agent, + session: { sessionId: string; protocolVersion: string }, + localPort: number, + headerTimeoutMs = 2500 +): Promise<{ + status: number; + next: (predicate: (msg: any) => boolean, timeoutMs: number) => Promise; + close: () => void; +}> { + let request; + const response = (await Promise.race([ + new Promise((resolveResponse, reject) => { + request = httpRequest( + new URL('/mcp', baseUrl), + { + method: 'GET', + agent, + localPort, + headers: { + 'accept': 'text/event-stream', + 'authorization': auth, + 'mcp-session-id': session.sessionId, + 'mcp-protocol-version': session.protocolVersion, + }, + }, + resolveResponse + ); + request.on('error', reject); + request.end(); + }), + new Promise((_, reject) => + setTimeout(() => reject(new Error('pinned SSE headers never flushed (hung)')), headerTimeoutMs) + ), + ])) as import('node:http').IncomingMessage; + + let buffer = ''; + const parsed: any[] = []; + const waiters: Array<{ predicate: (message: any) => boolean; resolve: (message: any) => void }> = []; + response.on('data', (chunk) => { + buffer += chunk.toString(); + let index; + while ((index = buffer.indexOf('\n\n')) !== -1) { + const frame = buffer.slice(0, index); + buffer = buffer.slice(index + 2); + const dataLine = frame.split('\n').find((line) => line.startsWith('data:')); + if (!dataLine) continue; + let message; + try { + message = JSON.parse(dataLine.slice(5).trim()); + } catch { + continue; + } + parsed.push(message); + for (let i = waiters.length - 1; i >= 0; i--) { + if (waiters[i].predicate(message)) { + waiters[i].resolve(message); + waiters.splice(i, 1); + } + } + } + }); + + return { + status: response.statusCode ?? 0, + next(predicate, timeoutMs) { + const existing = parsed.find(predicate); + if (existing) return Promise.resolve(existing); + return new Promise((resolveNext) => { + const timer = setTimeout(() => resolveNext(undefined), timeoutMs); + waiters.push({ + predicate, + resolve: (message) => { + clearTimeout(timer); + resolveNext(message); + }, + }); + }); + }, + close() { + response.destroy(); + request?.destroy(); + }, + }; +} + suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => { before(async () => { await setupHarperWithFixture(ctx, FIXTURE_PATH, { - config: { mcp: { operations: { mountPath: '/mcp' }, application: { mountPath: '/mcp' } } }, + config: { + threads: { count: 4 }, + mcp: { operations: { mountPath: '/mcp' }, application: { mountPath: '/mcp' } }, + }, env: {}, }); }); @@ -318,4 +470,87 @@ suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => sse.close(); } }); + + test('N5: resource subscription routes from a sibling POST worker to the GET-SSE owner', async (t) => { + if (process.platform === 'win32' || process.env.HARPER_RUNTIME === 'bun') { + t.skip( + process.platform === 'win32' + ? 'Harper forces one HTTP worker on Windows' + : 'Bun assigns pinned socket probes to one HTTP worker; Node CI covers cross-worker routing' + ); + return; + } + const auth = adminAuth(ctx); + const session = await initialize(ctx.harper.httpURL, auth); + const firstLocalPort = 20000 + (process.pid % 9000); + let getAgent; + let getThreadId; + let getLocalPort; + for (let attempt = 0; attempt < 25; attempt++) { + const candidateAgent = new Agent({ keepAlive: true, maxSockets: 1 }); + try { + getLocalPort = firstLocalPort + attempt; + getThreadId = await workerIdentity(ctx.harper.httpURL, auth, candidateAgent, getLocalPort); + getAgent = candidateAgent; + break; + } catch (error) { + candidateAgent.destroy(); + if ((error as NodeJS.ErrnoException).code !== 'EADDRINUSE') throw error; + } + } + ok(getAgent && getThreadId !== undefined && getLocalPort !== undefined, 'found a free pinned GET socket'); + const sse = await openSseWithAgent(ctx.harper.httpURL, auth, getAgent, session, getLocalPort); + let postAgent; + let postThreadId = getThreadId; + let postLocalPort = getLocalPort; + let postConnected = false; + try { + strictEqual(sse.status, 200, 'pinned GET SSE establishes'); + for (let attempt = 0; attempt < 24 && postThreadId === getThreadId; attempt++) { + postAgent?.destroy(); + postAgent = new Agent({ keepAlive: true, maxSockets: 1 }); + postLocalPort = getLocalPort + attempt + 1; + try { + postThreadId = await workerIdentity(ctx.harper.httpURL, auth, postAgent, postLocalPort); + postConnected = true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EADDRINUSE') throw error; + } + } + ok(postAgent && postConnected, 'created a POST keep-alive connection on a free local port'); + ok( + postThreadId !== getThreadId, + `expected a second application HTTP worker after 24 socket probes; all used thread ${getThreadId}` + ); + + const id = `cross_worker_${Date.now().toString(36)}`; + const uri = new URL(`/WorkItem/${id}`, ctx.harper.httpURL).href; + await putWorkItem(ctx, id, { state: 'pending', payload: 'cross-worker' }); + const subscription = await mcpCallWithAgent( + ctx.harper.httpURL, + auth, + postAgent!, + session, + 'resources/subscribe', + { uri }, + 20, + postLocalPort + ); + strictEqual( + subscription.error, + undefined, + `cross-worker resources/subscribe should succeed: ${JSON.stringify(subscription.error)}` + ); + await putWorkItem(ctx, id, { state: 'done', payload: 'cross-worker' }); + const event = await sse.next( + (message) => message?.method === 'notifications/resources/updated' && message?.params?.uri === uri, + 5000 + ); + ok(event, 'the owner worker delivered resources/updated on the pinned GET stream'); + } finally { + sse.close(); + getAgent.destroy(); + postAgent?.destroy(); + } + }); }); diff --git a/unitTests/components/mcp/adapters/fastify.test.js b/unitTests/components/mcp/adapters/fastify.test.js index 4a96d0288b..fa5812272e 100644 --- a/unitTests/components/mcp/adapters/fastify.test.js +++ b/unitTests/components/mcp/adapters/fastify.test.js @@ -8,6 +8,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/adapters/harperHttp.test.js b/unitTests/components/mcp/adapters/harperHttp.test.js index c80fc25c57..c445e563b9 100644 --- a/unitTests/components/mcp/adapters/harperHttp.test.js +++ b/unitTests/components/mcp/adapters/harperHttp.test.js @@ -11,6 +11,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/lifecycle.test.js b/unitTests/components/mcp/lifecycle.test.js index 573f7f2cd0..e7c0c60eda 100644 --- a/unitTests/components/mcp/lifecycle.test.js +++ b/unitTests/components/mcp/lifecycle.test.js @@ -16,6 +16,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/session.test.js b/unitTests/components/mcp/session.test.js index fab9989451..e072ea7c18 100644 --- a/unitTests/components/mcp/session.test.js +++ b/unitTests/components/mcp/session.test.js @@ -2,19 +2,42 @@ const assert = require('node:assert'); const { createSession, loadSession, - saveSession, deleteSession, touchSession, + updateSessionSubscriptions, _setSessionTableForTest, + _setSubscriptionLockTimeoutForTest, } = require('#src/components/mcp/session'); function makeFakeTable() { const store = new Map(); + const locks = new Set(); + const waiters = new Map(); return { store, + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; @@ -33,6 +56,7 @@ describe('mcp/session', () => { }); afterEach(() => { _setSessionTableForTest(undefined); + _setSubscriptionLockTimeoutForTest(undefined); }); describe('createSession', () => { @@ -65,14 +89,10 @@ describe('mcp/session', () => { const loaded = await loadSession('not-a-session'); assert.equal(loaded, null); }); - }); - describe('saveSession', () => { - it('persists changes', async () => { - const created = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); - await saveSession({ ...created, initialized: true }); - const reloaded = await loadSession(created.id); - assert.equal(reloaded.initialized, true); + it('returns null for a partial record left by a late patch after deletion', async () => { + fake.store.set('deleted-session', { id: 'deleted-session', lastActivity: Date.now() }); + assert.equal(await loadSession('deleted-session'), null); }); }); @@ -104,4 +124,43 @@ describe('mcp/session', () => { assert.equal(touched.protocolVersion, '2025-06-18'); }); }); + + describe('updateSessionSubscriptions', () => { + it('times out without acquiring a lock later from a stale wake callback', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const key = `mcp-subscriptions:${session.id}`; + assert.equal( + fake.primaryStore.tryLock(key, () => {}), + true + ); + _setSubscriptionLockTimeoutForTest(5); + + await assert.rejects( + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions), + /Timed out acquiring MCP subscription lock/ + ); + fake.primaryStore.unlock(key); + await new Promise(setImmediate); + + await updateSessionSubscriptions(session.id, (subscriptions) => [...subscriptions, 'recovered']); + assert.deepEqual((await loadSession(session.id)).subscriptions, ['recovered']); + }); + + it('serializes concurrent read-modify-writes for the same session', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + fake.store.get(session.id).subscriptions = ['first', 'second']; + const originalPatch = fake.patch; + fake.patch = async (record) => { + if (record.subscriptions) await new Promise((resolve) => setImmediate(resolve)); + return originalPatch.call(fake, record); + }; + + await Promise.all([ + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions.filter((uri) => uri !== 'first')), + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions.filter((uri) => uri !== 'second')), + ]); + + assert.deepEqual((await loadSession(session.id)).subscriptions, []); + }); + }); }); diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js new file mode 100644 index 0000000000..300c7dec1f --- /dev/null +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -0,0 +1,407 @@ +const assert = require('node:assert'); +const { + claimSubscriptionOwner, + routeResourceSubscription, + withSessionSubscriptionLock, + _setSubscriptionItcForTest, + _setSubscriptionThreadIdForTest, + _setSubscriptionTimeoutForTest, + _resetSubscriptionRoutingForTest, + _pendingSubscriptionRouteCount, +} = require('#src/components/mcp/subscriptionRouting'); +const { ITC_EVENT_TYPES } = require('#src/utility/hdbTerms'); +const { createSession, loadSession, patchSession, _setSessionTableForTest } = require('#src/components/mcp/session'); +const { registerSession, _resetSessionRegistryForTest } = require('#src/components/mcp/sessionRegistry'); +const { _setSubscribeImplForTest } = require('#src/components/mcp/resources'); +const { _resetSubscriptionsForTest } = require('#src/components/mcp/subscriptions'); + +const USER = { username: 'alice', role: { permission: { super_user: true } } }; + +function fakeTable() { + const store = new Map(); + const locks = new Set(); + const waiters = new Map(); + return { + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, + async put(record) { + store.set(record.id, { ...record }); + }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, + async get(id) { + const record = store.get(id); + return record && { ...record }; + }, + async delete(id) { + store.delete(id); + }, + }; +} + +function fakeBridge(send) { + const listeners = new Map(); + return { + listeners, + onMessageByType(type, listener) { + listeners.set(type, listener); + }, + sendToThread(target, event) { + return send?.(target, event, listeners) ?? true; + }, + }; +} + +describe('mcp/subscriptionRouting', () => { + beforeEach(() => { + _setSessionTableForTest(fakeTable()); + _setSubscriptionThreadIdForTest(1); + _setSubscriptionItcForTest(fakeBridge()); + }); + + afterEach(() => { + _resetSubscriptionsForTest(); + _resetSessionRegistryForTest(); + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); + _setSessionTableForTest(undefined); + _setSubscribeImplForTest(undefined); + }); + + async function remoteSession() { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await patchSession(session.id, { streamOwner: { threadId: 7, token: 'owner-token' } }); + return loadSession(session.id); + } + + it('accepts a correlated response only from the expected owner thread', async () => { + const bridge = fakeBridge((_target, event, listeners) => { + assert.equal(event.message.user.password, undefined, 'credentials must not cross the worker boundary'); + assert.equal(event.message.user.customClaim, undefined, 'custom principal state must remain local'); + assert.equal(event.message.user.username, ''); + assert.equal(event.message.user.authExpiresAt, 12345); + assert.equal(event.message.user.role.role, ''); + const requestId = event.message.requestId; + setImmediate(() => { + listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ + message: { requestId, originator: 8, result: 'not-subscribable' }, + }); + listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ + message: { requestId, originator: 7, result: 'success' }, + }); + }); + return true; + }); + _setSubscriptionItcForTest(bridge); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: { + ...USER, + username: '', + authExpiresAt: 12345, + role: { ...USER.role, role: '' }, + password: 'do-not-forward', + customClaim: 'local-only', + }, + }); + assert.equal(result, 'success'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('fails quickly when the persisted owner thread is unreachable', async () => { + _setSubscriptionItcForTest(fakeBridge(() => false)); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'no-live-stream'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('retries listener wiring after the thread bridge becomes available', async () => { + const bridge = fakeBridge(); + bridge.available = false; + _setSubscriptionItcForTest(bridge); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await patchSession(session.id, { streamOwner: { threadId: 7, token: 'stale-owner' } }); + assert.equal(await claimSubscriptionOwner(session.id, 'first-stream'), false); + assert.equal((await loadSession(session.id)).streamOwner, undefined); + assert.equal(bridge.listeners.size, 0); + bridge.available = true; + assert.equal(await claimSubscriptionOwner(session.id, 'second-stream'), true); + assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND), true); + assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE), true); + }); + + it('routes locally without publishing an owner when the thread bridge is unavailable', async () => { + const bridge = fakeBridge(); + bridge.available = false; + _setSubscriptionItcForTest(bridge); + _setSubscribeImplForTest(async () => ({ + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + })); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const record = registerSession(session.id, 'application', USER); + assert.equal(await claimSubscriptionOwner(session.id, record.streamToken), false); + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }), + 'success' + ); + assert.deepEqual((await loadSession(session.id)).subscriptions, ['https://app.test/Product/1']); + }); + + it('does not guess a local owner when worker routing is wired', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + registerSession(session.id, 'application', USER); + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }), + 'no-live-stream' + ); + }); + + it('reports a send exception as an internal error rather than a missing stream', async () => { + _setSubscriptionItcForTest( + fakeBridge(() => { + throw new Error('could not clone command'); + }) + ); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'internal-error'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('bounds a sent command when the owner never responds', async () => { + _setSubscriptionTimeoutForTest(5); + _setSubscriptionItcForTest(fakeBridge(() => true)); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'timeout'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('rejects a command whose stream token no longer owns the local registry', async () => { + let response; + const bridge = fakeBridge((_target, event) => { + if (event.type === ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE) response = event.message; + return true; + }); + _setSubscriptionItcForTest(bridge); + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + await bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r1', + originator: 1, + sessionId: session.id, + streamToken: 'stale-token', + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }, + }); + await new Promise(setImmediate); + assert.equal(response.result, 'no-live-stream'); + }); + + it('ignores malformed commands without attempting a response', async () => { + let sent = 0; + const bridge = fakeBridge(() => { + sent++; + return true; + }); + _setSubscriptionItcForTest(bridge); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await claimSubscriptionOwner(session.id, 'owner-token'); + const listener = bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND); + assert.doesNotThrow(() => listener({ message: undefined })); + assert.doesNotThrow(() => listener({ message: { requestId: 'r1', originator: 1 } })); + await new Promise(setImmediate); + assert.equal(sent, 0); + }); + + it('executes subscribe and unsubscribe on the owner and updates the durable URI list', async () => { + let response; + const bridge = fakeBridge((_target, event) => { + if (event.type === ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE) response = event.message; + return true; + }); + _setSubscriptionItcForTest(bridge); + _setSubscriptionThreadIdForTest(7); + _setSubscribeImplForTest(async (_path, user) => { + assert.deepEqual(user, USER); + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + const uri = 'https://app.test/Product/1'; + bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r2', + originator: 1, + sessionId: session.id, + streamToken: registered.streamToken, + operation: 'subscribe', + uri, + user: USER, + }, + }); + for (let i = 0; i < 20 && !response; i++) await new Promise(setImmediate); + assert.equal(response.result, 'success'); + assert.deepEqual((await loadSession(session.id)).subscriptions, [uri]); + + response = undefined; + bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r3', + originator: 1, + sessionId: session.id, + streamToken: registered.streamToken, + operation: 'unsubscribe', + uri, + }, + }); + for (let i = 0; i < 20 && !response; i++) await new Promise(setImmediate); + assert.equal(response.result, 'success'); + assert.deepEqual((await loadSession(session.id)).subscriptions, []); + }); + + it('serializes owner commands behind reconnect restoration for the same session', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + let releaseRestore; + const restoreBlocked = new Promise((resolve) => (releaseRestore = resolve)); + const restoring = withSessionSubscriptionLock(session.id, () => restoreBlocked); + let subscribeStarted = false; + _setSubscribeImplForTest(async () => { + subscribeStarted = true; + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + const subscribing = routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/2', + user: USER, + }); + await new Promise(setImmediate); + assert.equal(subscribeStarted, false); + releaseRestore(); + await restoring; + assert.equal(await subscribing, 'success'); + assert.equal(subscribeStarted, true); + }); + + it('rechecks stream ownership after waiting for an earlier session operation', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + let releaseEarlierOperation; + const earlierOperation = withSessionSubscriptionLock( + session.id, + () => new Promise((resolve) => (releaseEarlierOperation = resolve)) + ); + let subscribeStarted = false; + _setSubscribeImplForTest(async () => { + subscribeStarted = true; + }); + const subscribing = routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/3', + user: USER, + }); + await new Promise(setImmediate); + registerSession(session.id, 'application', USER); + releaseEarlierOperation(); + await earlierOperation; + + assert.equal(await subscribing, 'no-live-stream'); + assert.equal(subscribeStarted, false); + }); + + it('normalizes local owner failures to an internal-error result', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + _setSubscribeImplForTest(async () => ({ + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + })); + const table = fakeTable(); + await table.put(await loadSession(session.id)); + table.patch = async () => { + throw new Error('write failed'); + }; + _setSessionTableForTest(table); + + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/4', + user: USER, + }), + 'internal-error' + ); + }); +}); diff --git a/unitTests/components/mcp/subscriptions.test.js b/unitTests/components/mcp/subscriptions.test.js index 41c2dc5dc1..8f39ebca96 100644 --- a/unitTests/components/mcp/subscriptions.test.js +++ b/unitTests/components/mcp/subscriptions.test.js @@ -132,4 +132,13 @@ describe('mcp/subscriptions', () => { assert.deepEqual(restored, [URI], 'only the subscribable URI is restored'); assert.equal(_liveSubscriptionCount('s1'), 1); }); + + it('keeps a durable URI when restoration throws so a later reconnect can retry it', async () => { + _setSubscribeImplForTest(async () => { + throw new Error('authorization service unavailable'); + }); + const retained = await restoreResourceSubscriptions('s1', [URI], USER); + assert.deepEqual(retained, [URI]); + assert.equal(_liveSubscriptionCount('s1'), 0); + }); }); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 2d7ede1898..fe1cc7b2c5 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -1,8 +1,10 @@ const assert = require('node:assert'); +const { threadId } = require('node:worker_threads'); const rewire = require('rewire'); const transport_mod = rewire('#src/components/mcp/transport'); const { handleMcpRequest } = transport_mod; -const { _setSessionTableForTest, createSession, loadSession, saveSession } = require('#src/components/mcp/session'); +const sessionModule = require('#src/components/mcp/session'); +const { _setSessionTableForTest, createSession, loadSession, patchSession } = sessionModule; const { getRegisteredSession, pushSessionFrame, @@ -22,6 +24,9 @@ const { _setHttpUrlPrefixForTest, _setSubscribeImplForTest, } = require('#src/components/mcp/resources'); +const subscriptionRouting = require('#src/components/mcp/subscriptionRouting'); +const { _setSubscriptionItcForTest, _setSubscriptionTimeoutForTest, _resetSubscriptionRoutingForTest } = + subscriptionRouting; function makeFakeResources(entries) { const map = new Map(); @@ -42,11 +47,33 @@ function makeFakeResources(entries) { function makeFakeTable() { const store = new Map(); + const locks = new Set(); + const waiters = new Map(); return { store, + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; @@ -92,6 +119,7 @@ describe('mcp/transport', () => { _setResourcesForTest(makeFakeResources([])); _setOpenApiGeneratorForTest(() => ({ openapi: '3.0.3', info: { title: 'fake' }, paths: {} })); _setHttpUrlPrefixForTest(''); + _setSubscriptionItcForTest({ onMessageByType() {}, sendToThread: () => true }); }); afterEach(() => { @@ -101,6 +129,8 @@ describe('mcp/transport', () => { _setResourcesForTest(undefined); _setOpenApiGeneratorForTest(undefined); _setHttpUrlPrefixForTest(undefined); + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); }); describe('POST initialize', () => { @@ -222,6 +252,117 @@ describe('mcp/transport', () => { assert.ok(res.sseIterable, 'SSE stream returned'); }); + it('does not close a superseding GET stream when an earlier owner claim fails', async () => { + let replacement; + const originalClaim = subscriptionRouting.claimSubscriptionOwner; + subscriptionRouting.claimSubscriptionOwner = async () => { + replacement = registerSession(sessionId, 'application', { + username: 'alice', + role: { permission: { super_user: true } }, + }); + throw new Error('claim failed'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 500); + assert.equal(getRegisteredSession(sessionId), replacement); + } finally { + subscriptionRouting.claimSubscriptionOwner = originalClaim; + } + }); + + it('serves the GET stream when cross-worker routing is unavailable', async () => { + const originalClaim = subscriptionRouting.claimSubscriptionOwner; + subscriptionRouting.claimSubscriptionOwner = async () => false; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 200); + assert.ok(res.sseIterable); + res.sseIterable.emit('close'); + } finally { + subscriptionRouting.claimSubscriptionOwner = originalClaim; + } + }); + + it('returns 404 when the durable session disappears during reconnect', async () => { + const originalUpdate = sessionModule.updateSessionSubscriptions; + sessionModule.updateSessionSubscriptions = async () => null; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 404); + assert.equal(getRegisteredSession(sessionId), undefined); + } finally { + sessionModule.updateSessionSubscriptions = originalUpdate; + } + }); + + it('closes the registered GET stream when subscription restoration fails', async () => { + const originalLock = subscriptionRouting.withSessionSubscriptionLock; + subscriptionRouting.withSessionSubscriptionLock = async () => { + throw new Error('restore failed'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 500); + assert.equal(getRegisteredSession(sessionId), undefined); + } finally { + subscriptionRouting.withSessionSubscriptionLock = originalLock; + } + }); + + it('restores subscriptions with the full local principal', async () => { + const uri = 'https://app.test:9926/Product/restore'; + await patchSession(sessionId, { subscriptions: [uri] }); + let restoredUser; + _setSubscribeImplForTest(async (_path, user) => { + restoredUser = user; + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + userObject: { + username: 'alice', + password: 'do-not-forward', + customClaim: 'not-in-contract', + role: { permission: { super_user: true } }, + }, + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(restoredUser.username, 'alice'); + assert.equal(restoredUser.password, 'do-not-forward'); + assert.equal(restoredUser.customClaim, 'not-in-contract'); + res.sseIterable.emit('close'); + }); + it('accepts a matching MCP-Protocol-Version and returns Method-not-found for unknown methods', async () => { const res = await handleMcpRequest( makeReq({ @@ -273,10 +414,9 @@ describe('mcp/transport', () => { it('does not roll back lastActivity when persisting the level (touchSession adopted)', async () => { // Force a known-old lastActivity, then setLevel: the request's touchSession - // must advance it, and the level-persisting saveSession must NOT write the + // must advance it, and the level patch must not write the // stale load-time value back (root fix — handlePost adopts the touched copy). - const stale = await loadSession(sessionId); - await saveSession({ ...stale, lastActivity: 1 }); + await patchSession(sessionId, { lastActivity: 1 }); await handleMcpRequest( makeReq({ body: jsonRpc(2, 'logging/setLevel', { level: 'info' }), @@ -1113,9 +1253,13 @@ describe('mcp/transport', () => { }); describe('resources/subscribe + resources/unsubscribe', () => { - beforeEach(() => { + beforeEach(async () => { // A live GET stream is required to subscribe — register one for the session. - registerSession(sessionId, 'application', { username: 'alice', role: { permission: { super_user: true } } }); + const registered = registerSession(sessionId, 'application', { + username: 'alice', + role: { permission: { super_user: true } }, + }); + await patchSession(sessionId, { streamOwner: { threadId, token: registered.streamToken } }); // Inject a fake change stream so dispatch doesn't need the real audit log. // `null` for the sentinel path makes the resource non-subscribable. _setSubscribeImplForTest(async (path) => @@ -1179,6 +1323,30 @@ describe('mcp/transport', () => { assert.equal(res.jsonBody.error.code, -32602); }); + it('returns a retryable internal error when subscription routing times out', async () => { + _setSubscriptionItcForTest({ + onMessageByType() {}, + sendToThread() { + return true; + }, + }); + _setSubscriptionTimeoutForTest(5); + await patchSession(sessionId, { streamOwner: { threadId: threadId + 1, token: 'remote-owner' } }); + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(73, 'resources/subscribe', { uri: 'https://app.test:9926/Product/1' }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /timed out; retry/); + } finally { + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); + } + }); + it('unsubscribe removes the URI from the durable record', async () => { const uri = 'https://app.test:9926/Product/2'; await handleMcpRequest( @@ -1198,6 +1366,89 @@ describe('mcp/transport', () => { const saved = await loadSession(sessionId); assert.ok(!(saved.subscriptions ?? []).includes(uri), 'URI dropped from the record'); }); + + it('reroutes unsubscribe when the stream owner changes during the first attempt', async () => { + const uri = 'https://app.test:9926/Product/3'; + await patchSession(sessionId, { subscriptions: [uri] }); + let calls = 0; + const originalRoute = subscriptionRouting.routeResourceSubscription; + subscriptionRouting.routeResourceSubscription = async ({ session }) => { + calls++; + if (calls === 1) { + await patchSession(sessionId, { streamOwner: { threadId: threadId + 1, token: 'new-owner' } }); + return 'no-live-stream'; + } + assert.deepEqual(session.streamOwner, { threadId: threadId + 1, token: 'new-owner' }); + await patchSession(sessionId, { subscriptions: [] }); + return 'success'; + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(76, 'resources/unsubscribe', { uri }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.deepEqual(res.jsonBody.result, {}); + assert.equal(calls, 2); + assert.deepEqual((await loadSession(sessionId)).subscriptions, []); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + } + }); + + it('retains durable state when the unsubscribe owner keeps changing', async () => { + const uri = 'https://app.test:9926/Product/5'; + await patchSession(sessionId, { subscriptions: [uri] }); + let calls = 0; + const originalRoute = subscriptionRouting.routeResourceSubscription; + subscriptionRouting.routeResourceSubscription = async () => { + calls++; + await patchSession(sessionId, { streamOwner: { threadId: threadId + calls, token: `owner-${calls}` } }); + return 'no-live-stream'; + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(78, 'resources/unsubscribe', { uri }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(res.jsonBody.id, 78); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /retry/); + assert.equal(calls, 3); + assert.deepEqual((await loadSession(sessionId)).subscriptions, [uri]); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + } + }); + + it('contains a durable unsubscribe failure in the request JSON-RPC response', async () => { + const originalRoute = subscriptionRouting.routeResourceSubscription; + const originalUpdate = sessionModule.updateSessionSubscriptions; + subscriptionRouting.routeResourceSubscription = async () => 'no-live-stream'; + sessionModule.updateSessionSubscriptions = async () => { + throw new Error('lock timed out'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(77, 'resources/unsubscribe', { uri: 'https://app.test:9926/Product/4' }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(res.jsonBody.id, 77); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /retry/); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + sessionModule.updateSessionSubscriptions = originalUpdate; + } + }); }); describe('completion/complete', () => { diff --git a/utility/hdbTerms.ts b/utility/hdbTerms.ts index 775e6c7e3b..d27005698c 100644 --- a/utility/hdbTerms.ts +++ b/utility/hdbTerms.ts @@ -952,6 +952,8 @@ export const ITC_EVENT_TYPES = { // MCP §3.7: route a client's response to a server→client request back to the // worker awaiting it (the response POST can land on any worker). MCP_CLIENT_RESPONSE: 'mcp_client_response', + MCP_SUBSCRIPTION_COMMAND: 'mcp_subscription_command', + MCP_SUBSCRIPTION_RESPONSE: 'mcp_subscription_response', // #1736: components load per-worker, so a `server.registerOperation()` made there lands in // a worker-local OPERATION_FUNCTION_MAP the main-thread ops-API dispatcher can't see. A // worker announces each registration (OPERATION_REGISTERED) so the main thread can forward