From 897a0e03b72f1df19cc1325d557319d8f02870d7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 18 Oct 2026 16:19:36 -0700 Subject: [PATCH 1/2] fix(api): answer upstream MCP, OAuth, and id-conflict failures with 4xx; keep PII keep-alive past the load balancer - MCP tool discovery answers 422 (warn) when the registered server is unreachable or cooling down - MCP OAuth start answers 422 (warn) with the server's message for any authorization-server refusal - POST /api/workflows answers 409 when a caller-supplied id already exists - PII service keeps idle connections open longer than the load balancer's idle timeout - PII mask client replaces NUL with U+FFFD so the route does not reject the whole chunk --- .../sim/app/api/mcp/oauth/start/route.test.ts | 96 +++++++++++++++++++ apps/sim/app/api/mcp/oauth/start/route.ts | 84 ++++++++-------- .../app/api/mcp/tools/discover/route.test.ts | 29 ++++++ apps/sim/app/api/mcp/tools/discover/route.ts | 40 ++++++-- apps/sim/lib/guardrails/mask-client.test.ts | 48 ++++++++++ apps/sim/lib/guardrails/mask-client.ts | 11 ++- apps/sim/lib/mcp/client.test.ts | 2 +- apps/sim/lib/mcp/client.ts | 4 +- .../create-workflow-with-state.integration.ts | 56 ++++++++++- .../orchestration/workflow-lifecycle.ts | 17 +++- docker/pii-entrypoint.sh | 6 ++ 11 files changed, 337 insertions(+), 56 deletions(-) diff --git a/apps/sim/app/api/mcp/oauth/start/route.test.ts b/apps/sim/app/api/mcp/oauth/start/route.test.ts index a5d1bf3b79c..9eabc6486a8 100644 --- a/apps/sim/app/api/mcp/oauth/start/route.test.ts +++ b/apps/sim/app/api/mcp/oauth/start/route.test.ts @@ -148,6 +148,102 @@ describe('MCP OAuth start route', () => { }) }) +/** + * Runs the SDK's real dynamic client registration against an authorization server that + * answers with `status` and `body`, so the route sees exactly the error class the SDK builds. + */ +async function registerAgainst(status: number, body: string) { + const { registerClient } = await import('@modelcontextprotocol/sdk/client/auth.js') + return registerClient('https://auth.example.com', { + clientMetadata: { redirect_uris: ['https://sim.example.com/api/mcp/oauth/callback'] }, + fetchFn: async () => new Response(body, { status }), + }) +} + +describe('MCP OAuth start route: authorization-server refusals', () => { + beforeEach(() => { + resetDbChainMock() + hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ + success: true, + userId: 'user-1', + userName: 'User One', + userEmail: 'user1@example.com', + authType: 'session', + }) + permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('write') + dbChainMockFns.limit.mockResolvedValue([ + { + id: 'server-1', + name: 'Docs', + url: 'https://mcp.example.com/mcp', + workspaceId: 'workspace-1', + authType: 'oauth', + deletedAt: null, + }, + ]) + mcpOauthMockFns.mockGetOrCreateOauthRow.mockResolvedValue({ + id: 'oauth-row-1', + mcpServerId: 'server-1', + userId: 'user-1', + workspaceId: 'workspace-1', + clientInformation: null, + tokens: null, + codeVerifier: null, + state: null, + stateCreatedAt: null, + updatedAt: new Date(), + }) + mcpOauthMockFns.mockLoadPreregisteredClient.mockResolvedValue(undefined) + }) + + const start = () => + GET( + new NextRequest( + 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' + ) + ) + + it.each([ + [ + 'an RFC 7591 code the SDK does not know, which it raises as a ServerError', + 400, + '{"error":"invalid_redirect_uri","error_description":"Only loopback redirect URIs are supported."}', + 'Only loopback redirect URIs are supported.', + ], + [ + 'a registered OAuth code, raised as its typed subclass', + 400, + '{"error":"invalid_client_metadata","error_description":"client_name is required"}', + 'invalid_client_metadata: client_name is required', + ], + [ + 'a non-spec body, wrapped by the SDK in a ServerError', + 403, + '{"message":"redirect URI https://sim.example.com/api/mcp/oauth/callback is not allowed"}', + 'Authorization server: redirect URI https://sim.example.com/api/mcp/oauth/callback is not allowed', + ], + ] as const)( + 'answers 422 with the server message for %s', + async (_name, status, body, message) => { + mcpOauthMockFns.mockMcpAuthGuarded.mockImplementationOnce(() => registerAgainst(status, body)) + + const response = await start() + + expect(response.status).toBe(422) + expect((await response.json()).error).toBe(message) + } + ) + + it('keeps a 500 for a failure that is not an authorization-server answer', async () => { + mcpOauthMockFns.mockMcpAuthGuarded.mockRejectedValueOnce(new Error('decrypt failed')) + + const response = await start() + + expect(response.status).toBe(500) + expect((await response.json()).error).toBe('Failed to start OAuth flow') + }) +}) + describe('surfaceOauthError', () => { it('parses Raw body envelope for ServerError fallbacks (non-spec vendors)', async () => { const { ServerError } = await import('@modelcontextprotocol/sdk/server/auth/errors.js') diff --git a/apps/sim/app/api/mcp/oauth/start/route.ts b/apps/sim/app/api/mcp/oauth/start/route.ts index df0455632c1..7d84573a5f2 100644 --- a/apps/sim/app/api/mcp/oauth/start/route.ts +++ b/apps/sim/app/api/mcp/oauth/start/route.ts @@ -49,7 +49,7 @@ const DCR_UNSUPPORTED_MESSAGE = /** * The MCP SDK throws a plain `Error` (no typed class or code) when an auth server lacks a * `registration_endpoint`, so this string-match is the only available signal. Pinned to - * `@modelcontextprotocol/sdk` v1.29.0 `registerClient` (client/auth.js): "Incompatible auth + * `@modelcontextprotocol/sdk` v1.31.0 `registerClient` (client/auth.js): "Incompatible auth * server: does not support dynamic client registration". If a version bump rephrases this, the * 422 branch silently stops firing (users get a generic 500) — update the substring here. */ @@ -170,43 +170,53 @@ export const GET = withRouteHandler( ) const provider = new SimMcpOauthProvider({ row, preregistered }) - try { - // OAuth discovery + DCR through the guarded fetch, bounded so a transient stall fails - // fast with a labeled log instead of hanging the popup. `McpOauthRedirectRequired` is - // the SUCCESS signal (a throw carrying the authorize URL), so we catch it INSIDE the - // bounded step and return it as a normal value — otherwise timedStep would error-log - // every successful authorize. Only a real error or a timeout escapes as a throw. - const authOutcome = await timedStep('mcpAuthGuarded', MCP_AUTH_STEP_MS, async () => { - try { - return { kind: 'result' as const, value: await mcpAuthGuarded(provider, { serverUrl }) } - } catch (e) { - if (e instanceof McpOauthRedirectRequired) { - return { kind: 'redirect' as const, authorizationUrl: e.authorizationUrl } + // OAuth discovery + DCR through the guarded fetch, bounded so a transient stall fails + // fast with a labeled log instead of hanging the popup. `McpOauthRedirectRequired` is + // the SUCCESS signal (a throw carrying the authorize URL), and an authorization-server + // refusal is an expected outcome rather than a Sim fault, so both are caught INSIDE the + // bounded step and returned as values — otherwise timedStep would error-log them. Only + // a real error or a timeout escapes as a throw. + const authOutcome = await timedStep('mcpAuthGuarded', MCP_AUTH_STEP_MS, async () => { + try { + return { kind: 'result' as const, value: await mcpAuthGuarded(provider, { serverUrl }) } + } catch (e) { + if (e instanceof McpOauthRedirectRequired) { + return { kind: 'redirect' as const, authorizationUrl: e.authorizationUrl } + } + if (isDynamicClientRegistrationUnsupported(e)) { + return { + kind: 'rejected' as const, + error: toError(e), + message: DCR_UNSUPPORTED_MESSAGE, } - throw e } - }) - if (authOutcome.kind === 'redirect') { - logger.info(`OAuth redirect for server ${serverId}`) - return NextResponse.json({ - status: 'redirect', - authorizationUrl: authOutcome.authorizationUrl, - }) - } - if (authOutcome.value === 'AUTHORIZED') { - return NextResponse.json({ status: 'already_authorized' }) - } - return createMcpErrorResponse( - new Error('Provider did not capture redirect URL'), - 'Failed to start OAuth flow', - 500 - ) - } catch (e) { - if (isDynamicClientRegistrationUnsupported(e)) { - return createMcpErrorResponse(toError(e), DCR_UNSUPPORTED_MESSAGE, 422) + if (e instanceof OAuthError) { + return { kind: 'rejected' as const, error: e, message: surfaceOauthError(e) } + } + throw e } - throw e + }) + if (authOutcome.kind === 'redirect') { + logger.info(`OAuth redirect for server ${serverId}`) + return NextResponse.json({ + status: 'redirect', + authorizationUrl: authOutcome.authorizationUrl, + }) } + if (authOutcome.kind === 'rejected') { + logger.warn(`Authorization server refused the MCP OAuth flow for server ${serverId}`, { + error: authOutcome.error.message, + }) + return createMcpErrorResponse(authOutcome.error, authOutcome.message, 422) + } + if (authOutcome.value === 'AUTHORIZED') { + return NextResponse.json({ status: 'already_authorized' }) + } + return createMcpErrorResponse( + new Error('Provider did not capture redirect URL'), + 'Failed to start OAuth flow', + 500 + ) } catch (error) { // Any bounded step timing out (DB reads or the auth step) is a stall, not a bug — // surface it as a fast 504 so the popup closes with a clear "try again" rather than a @@ -223,11 +233,7 @@ export const GET = withRouteHandler( ) } logger.error('Error starting MCP OAuth flow:', error) - // Only surface OAuth-flow errors verbatim; everything else (DB, decryption, - // network) gets a generic message to avoid leaking internal details. - const userMessage = - error instanceof OAuthError ? surfaceOauthError(error) : 'Failed to start OAuth flow' - return createMcpErrorResponse(toError(error), userMessage, 500) + return createMcpErrorResponse(toError(error), 'Failed to start OAuth flow', 500) } }) ) diff --git a/apps/sim/app/api/mcp/tools/discover/route.test.ts b/apps/sim/app/api/mcp/tools/discover/route.test.ts index cdfa1ce30d4..03d72326663 100644 --- a/apps/sim/app/api/mcp/tools/discover/route.test.ts +++ b/apps/sim/app/api/mcp/tools/discover/route.test.ts @@ -25,6 +25,7 @@ vi.mock('@/lib/mcp/middleware', () => ({ vi.mock('@/lib/mcp/service', () => mcpServiceMock) import { McpOauthRedirectRequired } from '@/lib/mcp/oauth' +import { McpConnectionError, McpServerCooldownError } from '@/lib/mcp/types' import { GET } from '@/app/api/mcp/tools/discover/route' describe('GET /api/mcp/tools/discover', () => { @@ -43,4 +44,32 @@ describe('GET /api/mcp/tools/discover', () => { expect(response.status).toBe(401) expect(JSON.stringify(await response.json())).not.toContain('secret-state') }) + + const discover = () => + GET( + new NextRequest( + 'http://localhost:3000/api/mcp/tools/discover?workspaceId=workspace-1&serverId=mcp-1' + ), + createRouteContext({}) + ) + + it.each([ + ['cannot be reached', () => new McpConnectionError('connect ECONNREFUSED', 'Docs')], + ['is cooling down after a recent failure', () => new McpServerCooldownError('mcp-1')], + ])('answers 422 when the registered server %s', async (_name, makeError) => { + mcpServiceMockFns.mockDiscoverServerTools.mockRejectedValueOnce(makeError()) + + const response = await discover() + + expect(response.status).toBe(422) + expect((await response.json()).error).toBe('Failed to discover MCP tools') + }) + + it('keeps a 500 for a failure that is not the registered server', async () => { + mcpServiceMockFns.mockDiscoverServerTools.mockRejectedValueOnce(new Error('pool exhausted')) + + const response = await discover() + + expect(response.status).toBe(500) + }) }) diff --git a/apps/sim/app/api/mcp/tools/discover/route.ts b/apps/sim/app/api/mcp/tools/discover/route.ts index 659ee84ad5e..51901490865 100644 --- a/apps/sim/app/api/mcp/tools/discover/route.ts +++ b/apps/sim/app/api/mcp/tools/discover/route.ts @@ -12,7 +12,11 @@ import { } from '@/lib/mcp/middleware' import { McpOauthRedirectRequired } from '@/lib/mcp/oauth' import { mcpService } from '@/lib/mcp/service' -import { McpOauthAuthorizationRequiredError, type McpToolDiscoveryResponse } from '@/lib/mcp/types' +import { + McpConnectionError, + McpOauthAuthorizationRequiredError, + type McpToolDiscoveryResponse, +} from '@/lib/mcp/types' import { categorizeError, createMcpErrorResponse, createMcpSuccessResponse } from '@/lib/mcp/utils' const logger = createLogger('McpToolDiscoveryAPI') @@ -20,6 +24,32 @@ const MCP_REFRESH_DISCOVERY_CONCURRENCY = 5 export const dynamic = 'force-dynamic' +/** + * Renders a discovery failure that is not an authorization problem. + * + * Discovery talks to a server the workspace registered, so a server that cannot be + * reached, times out, or is cooling down after a recent failure is a fault of that + * server rather than of Sim: it answers 422 (the request was valid; the dependency it + * names cannot serve it), the status this API family already uses for an upstream that + * cannot complete the operation, and logs at warn. Anything else keeps + * {@link categorizeError}'s status and logs at error only when that status is a 5xx. + */ +function discoveryFailureResponse(requestId: string, error: unknown, defaultMessage: string) { + if (error instanceof McpConnectionError) { + logger.warn(`[${requestId}] MCP server unavailable for tool discovery`, { + error: getErrorMessage(error), + }) + return createMcpErrorResponse(error, defaultMessage, 422) + } + const { message, status } = categorizeError(error) + if (status >= 500) { + logger.error(`[${requestId}] ${defaultMessage}:`, error) + } else { + logger.warn(`[${requestId}] ${defaultMessage}`, { error: getErrorMessage(error), status }) + } + return createMcpErrorResponse(new Error(message), defaultMessage, status) +} + async function settleWithConcurrency( items: T[], concurrency: number, @@ -104,9 +134,7 @@ export const GET = withRouteHandler( ) { return createMcpErrorResponse(error, 'OAuth re-authorization required', 401) } - logger.error(`[${requestId}] Error discovering MCP tools:`, error) - const { message, status } = categorizeError(error) - return createMcpErrorResponse(new Error(message), 'Failed to discover MCP tools', status) + return discoveryFailureResponse(requestId, error, 'Failed to discover MCP tools') } }) ) @@ -171,9 +199,7 @@ export const POST = withRouteHandler( ) { return createMcpErrorResponse(error, 'OAuth re-authorization required', 401) } - logger.error(`[${requestId}] Error refreshing tool discovery:`, error) - const { message, status } = categorizeError(error) - return createMcpErrorResponse(new Error(message), 'Failed to refresh tool discovery', status) + return discoveryFailureResponse(requestId, error, 'Failed to refresh tool discovery') } }) ) diff --git a/apps/sim/lib/guardrails/mask-client.test.ts b/apps/sim/lib/guardrails/mask-client.test.ts index 06b63bd1a39..892e56645e8 100644 --- a/apps/sim/lib/guardrails/mask-client.test.ts +++ b/apps/sim/lib/guardrails/mask-client.test.ts @@ -1,6 +1,8 @@ import { resetUrlsMock, urlsMockFns } from '@sim/testing' import { authInternalMock, authInternalMockFns } from '@sim/testing/mocks/auth-internal.mock' +import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock' import { utilsHelpersMock, utilsHelpersMockFns } from '@sim/testing/mocks/utils-helpers.mock' +import { NextRequest } from 'next/server' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' afterAll(resetUrlsMock) @@ -13,6 +15,7 @@ vi.mock('@/lib/auth/internal', () => authInternalMock) vi.mock('@sim/utils/helpers', () => utilsHelpersMock) import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client' +import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route' const mockSleep = utilsHelpersMockFns.mockSleep @@ -119,4 +122,49 @@ describe('maskPIIBatchViaHttp', () => { expect(fetchMock).toHaveBeenCalledTimes(1) expect(mockSleep).not.toHaveBeenCalled() }) + + describe('through the real mask-batch route', () => { + let presidioTexts: string[][] + + beforeEach(() => { + presidioTexts = [] + hybridAuthMockFns.mockCheckInternalAuth.mockResolvedValue({ success: true }) + fetchMock.mockImplementation( + async (url: string, init: { body: string; headers: Record }) => { + if (url.startsWith('http://app.internal:3000')) { + return maskBatchRoute( + new NextRequest(url, { method: 'POST', body: init.body, headers: init.headers }) + ) + } + const { texts } = JSON.parse(init.body) as { texts: string[] } + presidioTexts.push(texts) + return Response.json({ + texts: texts.map((t) => t.replace('ada@example.com', '')), + }) + } + ) + }) + + it('masks a chunk whose strings carry a NUL instead of failing the whole chunk', async () => { + const { masked, scrubbedCount } = await maskPIIBatchViaHttp( + ['mail ada@example.com\u0000end', 'plain text'], + ['EMAIL_ADDRESS'], + undefined, + undefined, + { failedChunkPlaceholder: '[FAILED]' } + ) + + expect(scrubbedCount).toBe(0) + expect(masked).toEqual(['mail \uFFFDend', 'plain text']) + }) + + it('sends text without a NUL to Presidio unchanged', async () => { + const text = 'mail ada@example.com \u{1F600} done' + + const { masked } = await maskPIIBatchViaHttp([text], ['EMAIL_ADDRESS']) + + expect(presidioTexts).toEqual([[text]]) + expect(masked).toEqual(['mail \u{1F600} done']) + }) + }) }) diff --git a/apps/sim/lib/guardrails/mask-client.ts b/apps/sim/lib/guardrails/mask-client.ts index f503abf3451..07725cbe3d7 100644 --- a/apps/sim/lib/guardrails/mask-client.ts +++ b/apps/sim/lib/guardrails/mask-client.ts @@ -2,6 +2,7 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { sleep } from '@sim/utils/helpers' import { backoffWithJitter, parseRetryAfter } from '@sim/utils/retry' +import { sanitizeForJsonb } from '@sim/utils/string' import type { GuardrailsMaskBatchResult } from '@/lib/api/contracts' import { generateInternalToken } from '@/lib/auth/internal' import { env } from '@/lib/core/config/env' @@ -223,7 +224,15 @@ async function postChunkOnce( 'content-type': 'application/json', authorization: `Bearer ${token}`, }, - body: JSON.stringify({ texts, entityTypes, language, customPatterns }), + body: JSON.stringify({ + // The route refuses any NUL in a request body, so a log string carrying one would + // fail its whole chunk. Replacing it with U+FFFD (the substitution Presidio already + // gets for unpaired surrogates) is length-preserving, so no detected span moves. + texts: texts.map(sanitizeForJsonb), + entityTypes, + language, + customPatterns, + }), }) if (!response.ok) { diff --git a/apps/sim/lib/mcp/client.test.ts b/apps/sim/lib/mcp/client.test.ts index a3bde1c723d..b6e004faa17 100644 --- a/apps/sim/lib/mcp/client.test.ts +++ b/apps/sim/lib/mcp/client.test.ts @@ -273,7 +273,7 @@ describe('McpClient notification handler', () => { await expect(client.connect()).rejects.toBeInstanceOf(UnauthorizedError) - expect(mockLogger.error).toHaveBeenCalledWith( + expect(mockLogger.warn).toHaveBeenCalledWith( expect.stringContaining('Failed to connect'), expect.objectContaining({ outcome: 'unauthorized' }) ) diff --git a/apps/sim/lib/mcp/client.ts b/apps/sim/lib/mcp/client.ts index 329bc639f21..0c466bbfe3d 100644 --- a/apps/sim/lib/mcp/client.ts +++ b/apps/sim/lib/mcp/client.ts @@ -249,7 +249,7 @@ export class McpClient { await this.closeTransportAgent() const errorMessage = getErrorMessage(error, 'Unknown error') const outcome = classifyConnectionOutcome(error, this.config.authType) - logger.error(`Failed to connect to MCP server ${this.config.name}`, { + logger.warn(`Failed to connect to MCP server ${this.config.name}`, { ...diagnostics, durationMs: Date.now() - startedAt, error: getMcpSafeErrorDiagnostics(error), @@ -511,7 +511,7 @@ export class McpClient { logger.info(`[${this.config.name}] Ping successful`) return response } catch (error) { - logger.error(`[${this.config.name}] Ping failed:`, error) + logger.warn(`[${this.config.name}] Ping failed:`, error) throw error } } diff --git a/apps/sim/lib/workflows/orchestration/create-workflow-with-state.integration.ts b/apps/sim/lib/workflows/orchestration/create-workflow-with-state.integration.ts index 864925778ef..c6801de8c18 100644 --- a/apps/sim/lib/workflows/orchestration/create-workflow-with-state.integration.ts +++ b/apps/sim/lib/workflows/orchestration/create-workflow-with-state.integration.ts @@ -1,4 +1,4 @@ -/** Real PostgreSQL coverage for the atomic import primitive shared by the admin imports. */ +/** Real PostgreSQL coverage for the workflow create primitives: the atomic import and a plain create. */ import { mkdir, writeFile } from 'node:fs/promises' import { dirname, resolve } from 'node:path' import { db } from '@sim/db' @@ -9,7 +9,10 @@ import { generateId } from '@sim/utils/id' import type { WorkflowState } from '@sim/workflow-types/workflow' import { eq } from 'drizzle-orm' import { afterAll, beforeAll, describe, expect, it } from 'vitest' -import { createWorkflowWithState } from '@/lib/workflows/orchestration/workflow-lifecycle' +import { + createWorkflowWithState, + performCreateWorkflow, +} from '@/lib/workflows/orchestration/workflow-lifecycle' const userId = generateId() const workspaceId = generateId() @@ -109,6 +112,55 @@ describe('createWorkflowWithState against PostgreSQL', () => { expect(second.workflow.name).not.toBe(first.workflow.name) }) + check('reports a caller-supplied id that already exists as a conflict', async () => { + const id = generateId() + const first = await performCreateWorkflow({ id, userId, workspaceId, name: 'Original' }) + if (!first.success) throw new Error(`Expected the first create to succeed: ${first.error}`) + + const second = await performCreateWorkflow({ id, userId, workspaceId, name: 'Replay' }) + + expect(second).toEqual({ + success: false, + error: 'A workflow with this ID already exists', + errorCode: 'conflict', + }) + expect( + await db.select({ name: workflow.name }).from(workflow).where(eq(workflow.id, id)) + ).toEqual([{ name: 'Original' }]) + }) + + check( + 'reports a deduplicated create with a taken id as a conflict, not a name retry', + async () => { + const id = generateId() + const first = await performCreateWorkflow({ id, userId, workspaceId, name: 'Retried' }) + if (!first.success) throw new Error(`Expected the first create to succeed: ${first.error}`) + + const second = await performCreateWorkflow({ + id, + userId, + workspaceId, + name: 'Retried', + deduplicate: true, + }) + + expect(second).toMatchObject({ success: false, errorCode: 'conflict' }) + } + ) + + check('still reports an exact-name collision as a name conflict', async () => { + const first = await performCreateWorkflow({ userId, workspaceId, name: 'Taken name' }) + if (!first.success) throw new Error(`Expected the first create to succeed: ${first.error}`) + + const second = await performCreateWorkflow({ userId, workspaceId, name: 'Taken name' }) + + expect(second).toEqual({ + success: false, + error: 'A workflow named "Taken name" already exists in this folder', + errorCode: 'conflict', + }) + }) + check('refuses an archived workspace without writing a row', async () => { await expect(create('Late import', emptyState, archivedWorkspaceId)).rejects.toMatchObject({ code: 'not_found', diff --git a/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts b/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts index 65736d6c7b1..6f4c519d729 100644 --- a/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts +++ b/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts @@ -31,6 +31,8 @@ const logger = createLogger('WorkflowLifecycle') /** Partial unique index on `(workspace_id, coalesce(folder_id, ''), name) WHERE archived_at IS NULL`. */ const WORKFLOW_NAME_UNIQUE_INDEX = 'workflow_workspace_folder_name_active_unique' +/** Primary key of `workflow`; only a caller-supplied `id` can collide with it. */ +const WORKFLOW_PRIMARY_KEY = 'workflow_pkey' const WORKFLOW_NAME_DEDUPLICATION_ATTEMPTS = 8 export interface PerformCreateWorkflowParams { @@ -354,10 +356,17 @@ export async function performCreateWorkflowTransition( * pre-check reports. Matching the constraint avoids relabeling a `23505` from * normalized workflow tables as a name collision. */ - const isNameConflict = - getPostgresErrorCode(error) === '23505' && - getPostgresConstraintName(error) === WORKFLOW_NAME_UNIQUE_INDEX - if (!isNameConflict) { + const conflictingConstraint = + getPostgresErrorCode(error) === '23505' ? getPostgresConstraintName(error) : undefined + if (conflictingConstraint === WORKFLOW_PRIMARY_KEY) { + logger.warn(`[${requestId}] Workflow id is already taken`, { workflowId }) + return { + success: false, + error: 'A workflow with this ID already exists', + errorCode: 'conflict', + } + } + if (conflictingConstraint !== WORKFLOW_NAME_UNIQUE_INDEX) { throw error } diff --git a/docker/pii-entrypoint.sh b/docker/pii-entrypoint.sh index 31ecb1fcb9d..7d8fad38647 100755 --- a/docker/pii-entrypoint.sh +++ b/docker/pii-entrypoint.sh @@ -3,6 +3,12 @@ # make a malformed PII_WORKERS fail uvicorn's arg parsing, not the shell. set -- --host 0.0.0.0 --port 5001 --workers "${PII_WORKERS:-1}" --timeout-graceful-shutdown 300 +# The load balancer in front keeps idle connections to this service open for 300s and +# reuses them. uvicorn's 5s default closed them first, so a request sent on a connection +# uvicorn was closing came back as an instant load-balancer 502. Keep-alive must outlast +# the load balancer's idle timeout so the load balancer is always the side that closes. +set -- "$@" --timeout-keep-alive "${PII_TIMEOUT_KEEP_ALIVE:-310}" + # Worker memory grows with uptime, so with a supervisor (more than one worker) each # worker exits after a jittered request count, health checks included, and is # respawned. A single worker has no supervisor to respawn it, so it never recycles. From cf47f894656f36ee5b7f3575603f0ad8a3b7537d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 18 Oct 2026 16:29:53 -0700 Subject: [PATCH 2/2] fix(mcp): treat tools/list timeouts and HTTP errors as an unavailable server; fix test route contexts - Share the v2 discovery classification (timeout, connection, protocol, HTTP error) as isMcpServerUnavailableError and use it in the internal discover route - Restore the original POST refresh catch, which never sees per-server failures - Pass a route context in the new and existing oauth start and mask client tests; drop the oauth start test-types baseline entry --- .../sim/app/api/mcp/oauth/start/route.test.ts | 14 +++--- .../app/api/mcp/tools/discover/route.test.ts | 7 +++ apps/sim/app/api/mcp/tools/discover/route.ts | 43 +++++++++++-------- apps/sim/app/api/v2/mcp-servers/utils.ts | 16 ++----- apps/sim/lib/guardrails/mask-client.test.ts | 4 +- apps/sim/lib/mcp/utils.ts | 21 ++++++++- scripts/check-test-types.baseline.json | 3 -- 7 files changed, 66 insertions(+), 42 deletions(-) diff --git a/apps/sim/app/api/mcp/oauth/start/route.test.ts b/apps/sim/app/api/mcp/oauth/start/route.test.ts index 9eabc6486a8..1520a1741e6 100644 --- a/apps/sim/app/api/mcp/oauth/start/route.test.ts +++ b/apps/sim/app/api/mcp/oauth/start/route.test.ts @@ -9,6 +9,7 @@ import { permissionsMockFns, resetDbChainMock, } from '@sim/testing' +import { createRouteContext } from '@sim/testing/helpers/http' import { NextRequest } from 'next/server' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -61,7 +62,7 @@ describe('MCP OAuth start route', () => { 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' ) - await GET(request) + await GET(request, createRouteContext({})) // The route must call the guarded wrapper (which defaults fetchFn to the // SSRF-guarded fetch internally) rather than the raw SDK `auth()` — see @@ -82,7 +83,7 @@ describe('MCP OAuth start route', () => { 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' ) - const response = await GET(request) + const response = await GET(request, createRouteContext({})) expect(mcpOauthMockFns.mockMcpAuthGuarded).toHaveBeenCalledTimes(1) expect(response.status).toBe(504) @@ -105,7 +106,7 @@ describe('MCP OAuth start route', () => { 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' ) - const response = await GET(request) + const response = await GET(request, createRouteContext({})) const body = await response.json() expect(response.status).toBe(409) @@ -121,7 +122,7 @@ describe('MCP OAuth start route', () => { 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' ) - const response = await GET(request) + const response = await GET(request, createRouteContext({})) const body = await response.json() expect(response.status).toBe(500) @@ -138,7 +139,7 @@ describe('MCP OAuth start route', () => { 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' ) - const response = await GET(request) + const response = await GET(request, createRouteContext({})) const body = await response.json() expect(response.status).toBe(422) @@ -200,7 +201,8 @@ describe('MCP OAuth start route: authorization-server refusals', () => { GET( new NextRequest( 'http://localhost:3000/api/mcp/oauth/start?workspaceId=workspace-1&serverId=server-1' - ) + ), + createRouteContext({}) ) it.each([ diff --git a/apps/sim/app/api/mcp/tools/discover/route.test.ts b/apps/sim/app/api/mcp/tools/discover/route.test.ts index 03d72326663..7d0eed851e0 100644 --- a/apps/sim/app/api/mcp/tools/discover/route.test.ts +++ b/apps/sim/app/api/mcp/tools/discover/route.test.ts @@ -24,6 +24,8 @@ vi.mock('@/lib/mcp/middleware', () => ({ vi.mock('@/lib/mcp/service', () => mcpServiceMock) +import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js' +import { ErrorCode, McpError } from '@modelcontextprotocol/sdk/types.js' import { McpOauthRedirectRequired } from '@/lib/mcp/oauth' import { McpConnectionError, McpServerCooldownError } from '@/lib/mcp/types' import { GET } from '@/app/api/mcp/tools/discover/route' @@ -56,6 +58,11 @@ describe('GET /api/mcp/tools/discover', () => { it.each([ ['cannot be reached', () => new McpConnectionError('connect ECONNREFUSED', 'Docs')], ['is cooling down after a recent failure', () => new McpServerCooldownError('mcp-1')], + ['times out on tools/list', () => new McpError(ErrorCode.RequestTimeout, 'Request timed out')], + [ + 'answers tools/list with an HTTP error', + () => new StreamableHTTPError(503, 'Service Unavailable'), + ], ])('answers 422 when the registered server %s', async (_name, makeError) => { mcpServiceMockFns.mockDiscoverServerTools.mockRejectedValueOnce(makeError()) diff --git a/apps/sim/app/api/mcp/tools/discover/route.ts b/apps/sim/app/api/mcp/tools/discover/route.ts index 51901490865..c6ed21a0dd1 100644 --- a/apps/sim/app/api/mcp/tools/discover/route.ts +++ b/apps/sim/app/api/mcp/tools/discover/route.ts @@ -12,12 +12,13 @@ import { } from '@/lib/mcp/middleware' import { McpOauthRedirectRequired } from '@/lib/mcp/oauth' import { mcpService } from '@/lib/mcp/service' +import { McpOauthAuthorizationRequiredError, type McpToolDiscoveryResponse } from '@/lib/mcp/types' import { - McpConnectionError, - McpOauthAuthorizationRequiredError, - type McpToolDiscoveryResponse, -} from '@/lib/mcp/types' -import { categorizeError, createMcpErrorResponse, createMcpSuccessResponse } from '@/lib/mcp/utils' + categorizeError, + createMcpErrorResponse, + createMcpSuccessResponse, + isMcpServerUnavailableError, +} from '@/lib/mcp/utils' const logger = createLogger('McpToolDiscoveryAPI') const MCP_REFRESH_DISCOVERY_CONCURRENCY = 5 @@ -28,26 +29,30 @@ export const dynamic = 'force-dynamic' * Renders a discovery failure that is not an authorization problem. * * Discovery talks to a server the workspace registered, so a server that cannot be - * reached, times out, or is cooling down after a recent failure is a fault of that - * server rather than of Sim: it answers 422 (the request was valid; the dependency it - * names cannot serve it), the status this API family already uses for an upstream that - * cannot complete the operation, and logs at warn. Anything else keeps - * {@link categorizeError}'s status and logs at error only when that status is a 5xx. + * reached, times out on connect or on `tools/list`, answers with a protocol or HTTP + * error, or is cooling down after a recent failure is a fault of that server rather + * than of Sim: it answers 422 (the request was valid; the dependency it names cannot + * serve it), the status this API family already uses for an upstream that cannot + * complete the operation, and logs at warn. Anything else keeps {@link categorizeError}'s + * status and logs at error only when that status is a 5xx. */ -function discoveryFailureResponse(requestId: string, error: unknown, defaultMessage: string) { - if (error instanceof McpConnectionError) { +function discoveryFailureResponse(requestId: string, error: unknown) { + if (isMcpServerUnavailableError(error)) { logger.warn(`[${requestId}] MCP server unavailable for tool discovery`, { error: getErrorMessage(error), }) - return createMcpErrorResponse(error, defaultMessage, 422) + return createMcpErrorResponse(error, 'Failed to discover MCP tools', 422) } const { message, status } = categorizeError(error) if (status >= 500) { - logger.error(`[${requestId}] ${defaultMessage}:`, error) + logger.error(`[${requestId}] Error discovering MCP tools:`, error) } else { - logger.warn(`[${requestId}] ${defaultMessage}`, { error: getErrorMessage(error), status }) + logger.warn(`[${requestId}] Failed to discover MCP tools`, { + error: getErrorMessage(error), + status, + }) } - return createMcpErrorResponse(new Error(message), defaultMessage, status) + return createMcpErrorResponse(new Error(message), 'Failed to discover MCP tools', status) } async function settleWithConcurrency( @@ -134,7 +139,7 @@ export const GET = withRouteHandler( ) { return createMcpErrorResponse(error, 'OAuth re-authorization required', 401) } - return discoveryFailureResponse(requestId, error, 'Failed to discover MCP tools') + return discoveryFailureResponse(requestId, error) } }) ) @@ -199,7 +204,9 @@ export const POST = withRouteHandler( ) { return createMcpErrorResponse(error, 'OAuth re-authorization required', 401) } - return discoveryFailureResponse(requestId, error, 'Failed to refresh tool discovery') + logger.error(`[${requestId}] Error refreshing tool discovery:`, error) + const { message, status } = categorizeError(error) + return createMcpErrorResponse(new Error(message), 'Failed to refresh tool discovery', status) } }) ) diff --git a/apps/sim/app/api/v2/mcp-servers/utils.ts b/apps/sim/app/api/v2/mcp-servers/utils.ts index a47cdef6d65..4ac6c52e319 100644 --- a/apps/sim/app/api/v2/mcp-servers/utils.ts +++ b/apps/sim/app/api/v2/mcp-servers/utils.ts @@ -1,17 +1,12 @@ import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js' -import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js' -import { McpError as McpSdkError } from '@modelcontextprotocol/sdk/types.js' import { type V2McpServer, v2McpServerSchema } from '@/lib/api/contracts/v2/mcp-servers' import { createV2ResourceConcealmentPolicy, type V2ErrorPolicy } from '@/lib/api/server/routes' import { isTimeoutError } from '@/lib/core/execution-limits' import { McpOauthRedirectRequired } from '@/lib/mcp/oauth' import { projectMcpHeaders } from '@/lib/mcp/projection' import type { McpServerRow } from '@/lib/mcp/queries' -import { - McpConnectionError, - McpOauthAuthorizationRequiredError, - McpServerCooldownError, -} from '@/lib/mcp/types' +import { McpOauthAuthorizationRequiredError, McpServerCooldownError } from '@/lib/mcp/types' +import { isMcpServerUnavailableError } from '@/lib/mcp/utils' import { v2Error } from '@/app/api/v2/lib/response' /** @@ -116,12 +111,7 @@ export const v2McpToolDiscoveryErrorPolicy = { ) } - if ( - isTimeoutError(error) || - error instanceof McpConnectionError || - error instanceof McpSdkError || - error instanceof StreamableHTTPError - ) { + if (isMcpServerUnavailableError(error)) { return v2Error('SERVICE_UNAVAILABLE', unreachableServerMessage(error)) } diff --git a/apps/sim/lib/guardrails/mask-client.test.ts b/apps/sim/lib/guardrails/mask-client.test.ts index 892e56645e8..48ed3feedde 100644 --- a/apps/sim/lib/guardrails/mask-client.test.ts +++ b/apps/sim/lib/guardrails/mask-client.test.ts @@ -1,4 +1,5 @@ import { resetUrlsMock, urlsMockFns } from '@sim/testing' +import { createRouteContext } from '@sim/testing/helpers/http' import { authInternalMock, authInternalMockFns } from '@sim/testing/mocks/auth-internal.mock' import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock' import { utilsHelpersMock, utilsHelpersMockFns } from '@sim/testing/mocks/utils-helpers.mock' @@ -133,7 +134,8 @@ describe('maskPIIBatchViaHttp', () => { async (url: string, init: { body: string; headers: Record }) => { if (url.startsWith('http://app.internal:3000')) { return maskBatchRoute( - new NextRequest(url, { method: 'POST', body: init.body, headers: init.headers }) + new NextRequest(url, { method: 'POST', body: init.body, headers: init.headers }), + createRouteContext({}) ) } const { texts } = JSON.parse(init.body) as { texts: string[] } diff --git a/apps/sim/lib/mcp/utils.ts b/apps/sim/lib/mcp/utils.ts index 23f5637594a..72a9ba7c5e7 100644 --- a/apps/sim/lib/mcp/utils.ts +++ b/apps/sim/lib/mcp/utils.ts @@ -1,7 +1,9 @@ import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js' +import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js' +import { McpError as McpSdkError } from '@modelcontextprotocol/sdk/types.js' import { generateShortId } from '@sim/utils/id' import { NextResponse } from 'next/server' -import { DEFAULT_EXECUTION_TIMEOUT_MS } from '@/lib/core/execution-limits' +import { DEFAULT_EXECUTION_TIMEOUT_MS, isTimeoutError } from '@/lib/core/execution-limits' import { type McpApiResponse, McpConnectionError, @@ -136,6 +138,23 @@ export function mcpOrchestrationStatus(errorCode: string | undefined): number { return 500 } +/** + * Reports whether a registered MCP server failed to answer usefully: it could not be + * reached, timed out, is cooling down after a recent failure, or answered `tools/list` + * with a protocol or HTTP error. Typed only, never matched on message text, because + * `McpConnectionError` interpolates the server's display name into its message. Check + * authorization errors first: `UnauthorizedError` is not one of these, but a caller that + * wants to answer reauthorization separately must do so before asking. + */ +export function isMcpServerUnavailableError(error: unknown): boolean { + return ( + isTimeoutError(error) || + error instanceof McpConnectionError || + error instanceof McpSdkError || + error instanceof StreamableHTTPError + ) +} + /** * Enhanced error categorization for more specific HTTP status codes. * Returns safe, generic messages to prevent leaking internal details. diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json index cb2b56fbc55..91b042d3f49 100644 --- a/scripts/check-test-types.baseline.json +++ b/scripts/check-test-types.baseline.json @@ -210,9 +210,6 @@ "apps/sim/app/api/mcp/oauth/callback/route.test.ts": { "TS2554": 7 }, - "apps/sim/app/api/mcp/oauth/start/route.test.ts": { - "TS2554": 5 - }, "apps/sim/app/api/mcp/serve/[serverId]/route.test.ts": { "TS2322": 2 },