Skip to content

Commit b9fe771

Browse files
j15zj15z
committed
Close MCP invalidation publication races (#5754)
Publish cache invalidation tokens into database ordering and fail refresh closed when discovery is superseded. Focused MCP suites pass 48/48 and API validation passes. Note: the full app suite retains the same 41 pre-existing failures in unrelated workflow, webhook, chat, and UI tests.
1 parent 8899fdf commit b9fe771

4 files changed

Lines changed: 267 additions & 13 deletions

File tree

‎apps/sim/app/api/mcp/servers/[id]/refresh/route.test.ts‎

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -4,17 +4,19 @@
44
import type { NextRequest } from 'next/server'
55
import { beforeEach, describe, expect, it, vi } from 'vitest'
66

7-
const { mockClearCache, mockDiscoverServerTools, mockSelect, mockUpdateSet } = vi.hoisted(() => ({
8-
mockClearCache: vi.fn(),
9-
mockDiscoverServerTools: vi.fn(),
10-
mockSelect: vi.fn(),
11-
mockUpdateSet: vi.fn(),
12-
}))
7+
const { mockClearCache, mockDiscoverServerTools, mockSelect, mockUpdate, mockUpdateSet } =
8+
vi.hoisted(() => ({
9+
mockClearCache: vi.fn(),
10+
mockDiscoverServerTools: vi.fn(),
11+
mockSelect: vi.fn(),
12+
mockUpdate: vi.fn(),
13+
mockUpdateSet: vi.fn(),
14+
}))
1315

1416
vi.mock('@sim/db', () => ({
1517
db: {
1618
select: mockSelect,
17-
update: vi.fn().mockReturnValue({ set: mockUpdateSet }),
19+
update: mockUpdate.mockReturnValue({ set: mockUpdateSet }),
1820
},
1921
}))
2022

@@ -225,4 +227,30 @@ describe('MCP server refresh route', () => {
225227
})
226228
)
227229
})
230+
231+
it('fails closed without syncing workflows when discovery is superseded', async () => {
232+
mockDiscoverServerTools.mockResolvedValueOnce({
233+
tools: [],
234+
state: 'superseded',
235+
})
236+
mockSelect.mockReturnValueOnce(selectRows([initialServer]))
237+
238+
const request = new Request('http://localhost/api/mcp/servers/server-1/refresh', {
239+
method: 'POST',
240+
}) as NextRequest
241+
const response = await POST(request, { params: Promise.resolve({ id: 'server-1' }) })
242+
const body = await response.json()
243+
244+
expect(body.data).toEqual(
245+
expect.objectContaining({
246+
status: 'disconnected',
247+
error: 'Tool discovery was superseded by a newer refresh. Please retry.',
248+
toolCount: 0,
249+
workflowsUpdated: 0,
250+
updatedWorkflowIds: [],
251+
})
252+
)
253+
expect(mockSelect).toHaveBeenCalledTimes(2)
254+
expect(mockUpdate).not.toHaveBeenCalled()
255+
})
228256
})

‎apps/sim/app/api/mcp/servers/[id]/refresh/route.ts‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -230,7 +230,7 @@ export const POST = withRouteHandler(
230230
)
231231
.limit(1)
232232

233-
if (discoveryError === null) {
233+
if (discoveryError === null && discoveryState !== 'superseded') {
234234
try {
235235
syncResult = await syncToolSchemasToWorkflows(
236236
workspaceId,
@@ -256,7 +256,11 @@ export const POST = withRouteHandler(
256256
let lastError = refreshedServer ? refreshedServer.lastError : discoveryError
257257
let toolCount = refreshedServer?.toolCount ?? discoveredTools.length
258258

259-
if (
259+
if (discoveryState === 'superseded') {
260+
connectionStatus = 'disconnected'
261+
lastError = 'Tool discovery was superseded by a newer refresh. Please retry.'
262+
toolCount = 0
263+
} else if (
260264
discoveryError === null &&
261265
(discoveryState === 'unavailable' || discoveryState === 'winner-cache')
262266
) {

‎apps/sim/lib/mcp/service.test.ts‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -329,6 +329,15 @@ describe('McpService.discoverTools per-server caching', () => {
329329

330330
await mcpService.clearCache(WORKSPACE_ID)
331331

332+
const [discoveryUpdate, invalidationUpdate] = mockUpdateSet.mock.calls.map(([update]) => update)
333+
expect(invalidationUpdate).toEqual({
334+
toolCount: 0,
335+
lastToolsRefresh: expect.any(Date),
336+
})
337+
expect(invalidationUpdate.lastToolsRefresh.getTime()).toBeGreaterThan(
338+
discoveryUpdate.lastToolsRefresh.getTime()
339+
)
340+
332341
mockListTools.mockClear()
333342
mockListTools.mockResolvedValueOnce([tool('a1', 'mcp-a')])
334343
await mcpService.discoverTools(USER_ID, WORKSPACE_ID)
@@ -967,6 +976,174 @@ describe('McpService.discoverTools per-server caching', () => {
967976
await expect(discovery).resolves.toEqual([tool('a1', 'mcp-a')])
968977
})
969978

979+
it('publishes a newer invalidation barrier while successful status publication is pending', async () => {
980+
vi.useFakeTimers({ toFake: ['Date'] })
981+
try {
982+
const serverKey = `workspace:${WORKSPACE_ID}:server:mcp-a`
983+
const discoveryStartedAt = new Date('2100-02-01T00:00:00.000Z')
984+
const invalidatedAt = new Date('2100-02-01T00:00:01.000Z')
985+
mockGetWorkspaceServersRows.mockResolvedValue([dbRow('mcp-a', 'A')])
986+
mockListTools.mockResolvedValueOnce([tool('stale-tool', 'mcp-a')])
987+
988+
let releaseDiscoveryStatus: ((rows: { id: string }[]) => void) | undefined
989+
mockUpdateReturning.mockReturnValueOnce(
990+
new Promise((resolve) => {
991+
releaseDiscoveryStatus = resolve
992+
})
993+
)
994+
995+
vi.setSystemTime(discoveryStartedAt)
996+
const discovery = mcpService.discoverServerToolsWithMetadata(
997+
USER_ID,
998+
'mcp-a',
999+
WORKSPACE_ID,
1000+
true
1001+
)
1002+
await vi.waitFor(() =>
1003+
expect(mockUpdateSet).toHaveBeenCalledWith(
1004+
expect.objectContaining({
1005+
connectionStatus: 'connected',
1006+
lastToolsRefresh: discoveryStartedAt,
1007+
})
1008+
)
1009+
)
1010+
expect(cacheStore.get(serverKey)?.tools).toEqual([tool('stale-tool', 'mcp-a')])
1011+
1012+
vi.setSystemTime(invalidatedAt)
1013+
await mcpService.clearCache(WORKSPACE_ID)
1014+
1015+
const invalidationUpdate = mockUpdateSet.mock.calls
1016+
.map(([update]) => update)
1017+
.find(
1018+
(update) =>
1019+
update.toolCount === 0 && update.lastToolsRefresh?.getTime() === invalidatedAt.getTime()
1020+
)
1021+
expect(invalidationUpdate).toEqual({
1022+
toolCount: 0,
1023+
lastToolsRefresh: invalidatedAt,
1024+
})
1025+
expect(cacheStore.has(serverKey)).toBe(false)
1026+
1027+
// The real lastToolsRefresh predicate rejects this older publication
1028+
// after the invalidation barrier wins the database race.
1029+
releaseDiscoveryStatus?.([])
1030+
await expect(discovery).resolves.toEqual({ tools: [], state: 'superseded' })
1031+
} finally {
1032+
vi.useRealTimers()
1033+
}
1034+
})
1035+
1036+
it('publishes a newer invalidation barrier while failed status publication is pending', async () => {
1037+
vi.useFakeTimers({ toFake: ['Date'] })
1038+
try {
1039+
const serverKey = `workspace:${WORKSPACE_ID}:server:mcp-a`
1040+
const failureKey = `${serverKey}:failure`
1041+
const discoveryStartedAt = new Date('2100-03-01T00:00:00.000Z')
1042+
const invalidatedAt = new Date('2100-03-01T00:00:01.000Z')
1043+
mockGetWorkspaceServersRows.mockResolvedValue([
1044+
dbRow('mcp-a', 'A', {
1045+
statusConfig: { consecutiveFailures: 0, lastSuccessfulDiscovery: null },
1046+
}),
1047+
])
1048+
mockListTools.mockRejectedValueOnce(new Error('Permanent discovery failure'))
1049+
1050+
let releaseDiscoveryStatus: ((rows: { id: string }[]) => void) | undefined
1051+
mockUpdateReturning
1052+
.mockReturnValueOnce(
1053+
new Promise((resolve) => {
1054+
releaseDiscoveryStatus = resolve
1055+
})
1056+
)
1057+
.mockResolvedValueOnce([{ id: 'mcp-a' }])
1058+
.mockResolvedValueOnce([])
1059+
.mockResolvedValueOnce([])
1060+
1061+
vi.setSystemTime(discoveryStartedAt)
1062+
const discovery = mcpService.discoverServerTools(USER_ID, 'mcp-a', WORKSPACE_ID, true)
1063+
await vi.waitFor(() =>
1064+
expect(mockUpdateSet).toHaveBeenCalledWith(
1065+
expect.objectContaining({
1066+
connectionStatus: 'disconnected',
1067+
lastError: 'Connection failed',
1068+
lastToolsRefresh: discoveryStartedAt,
1069+
})
1070+
)
1071+
)
1072+
expect(cacheStore.has(failureKey)).toBe(true)
1073+
1074+
vi.setSystemTime(invalidatedAt)
1075+
await mcpService.clearCache(WORKSPACE_ID)
1076+
1077+
expect(mockUpdateSet).toHaveBeenCalledWith({
1078+
toolCount: 0,
1079+
lastToolsRefresh: invalidatedAt,
1080+
})
1081+
expect(cacheStore.has(serverKey)).toBe(false)
1082+
expect(cacheStore.has(failureKey)).toBe(false)
1083+
1084+
releaseDiscoveryStatus?.([])
1085+
await expect(discovery).rejects.toThrow('Permanent discovery failure')
1086+
expect(
1087+
mockUpdateSet.mock.calls
1088+
.map(([update]) => update)
1089+
.filter((update) => update.lastError === 'Connection failed')
1090+
.every((update) => update.lastToolsRefresh?.getTime() === discoveryStartedAt.getTime())
1091+
).toBe(true)
1092+
} finally {
1093+
vi.useRealTimers()
1094+
}
1095+
})
1096+
1097+
it('publishes a newer invalidation barrier while OAuth status publication is pending', async () => {
1098+
vi.useFakeTimers({ toFake: ['Date'] })
1099+
try {
1100+
const serverKey = `workspace:${WORKSPACE_ID}:server:mcp-a`
1101+
const discoveryStartedAt = new Date('2100-04-01T00:00:00.000Z')
1102+
const invalidatedAt = new Date('2100-04-01T00:00:01.000Z')
1103+
mockGetWorkspaceServersRows.mockResolvedValue([dbRow('mcp-a', 'A')])
1104+
mockListTools.mockRejectedValueOnce(new McpOauthAuthorizationRequiredError('mcp-a', 'A'))
1105+
1106+
let releaseDiscoveryStatus: ((rows: { id: string }[]) => void) | undefined
1107+
mockUpdateReturning.mockReturnValueOnce(
1108+
new Promise((resolve) => {
1109+
releaseDiscoveryStatus = resolve
1110+
})
1111+
)
1112+
1113+
vi.setSystemTime(discoveryStartedAt)
1114+
const discovery = mcpService.discoverServerTools(USER_ID, 'mcp-a', WORKSPACE_ID, true)
1115+
await vi.waitFor(() =>
1116+
expect(mockUpdateSet).toHaveBeenCalledWith(
1117+
expect.objectContaining({
1118+
connectionStatus: 'disconnected',
1119+
lastError: null,
1120+
lastToolsRefresh: discoveryStartedAt,
1121+
})
1122+
)
1123+
)
1124+
1125+
vi.setSystemTime(invalidatedAt)
1126+
await mcpService.clearCache(WORKSPACE_ID)
1127+
1128+
const invalidationUpdate = mockUpdateSet.mock.calls
1129+
.map(([update]) => update)
1130+
.find(
1131+
(update) =>
1132+
update.toolCount === 0 && update.lastToolsRefresh?.getTime() === invalidatedAt.getTime()
1133+
)
1134+
expect(invalidationUpdate).toEqual({
1135+
toolCount: 0,
1136+
lastToolsRefresh: invalidatedAt,
1137+
})
1138+
expect(cacheStore.has(serverKey)).toBe(false)
1139+
1140+
releaseDiscoveryStatus?.([])
1141+
await expect(discovery).rejects.toThrow('OAuth authorization required')
1142+
} finally {
1143+
vi.useRealTimers()
1144+
}
1145+
})
1146+
9701147
it('keeps a newer successful cache entry when an older failure finishes later', async () => {
9711148
mockGetWorkspaceServersRows.mockResolvedValue([dbRow('mcp-a', 'A')])
9721149

‎apps/sim/lib/mcp/service.ts‎

Lines changed: 49 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -633,6 +633,45 @@ class McpService {
633633
}
634634
}
635635

636+
/**
637+
* Publish an invalidation's cache-order token without changing connection
638+
* health. A list_changed notification comes from a live connection, while
639+
* configuration lifecycle code persists its own intended connection state.
640+
*/
641+
private async markServerCacheInvalidated(
642+
serverId: string,
643+
workspaceId: string,
644+
publicationOrder: Date
645+
): Promise<boolean> {
646+
try {
647+
const updatedServers = await db
648+
.update(mcpServers)
649+
.set({
650+
toolCount: 0,
651+
lastToolsRefresh: publicationOrder,
652+
})
653+
.where(
654+
and(
655+
eq(mcpServers.id, serverId),
656+
eq(mcpServers.workspaceId, workspaceId),
657+
isNull(mcpServers.deletedAt),
658+
or(
659+
isNull(mcpServers.lastToolsRefresh),
660+
lte(mcpServers.lastToolsRefresh, publicationOrder)
661+
)
662+
)
663+
)
664+
.returning({ id: mcpServers.id })
665+
return updatedServers.length > 0
666+
} catch (error) {
667+
logger.warn(`Failed to publish cache invalidation for server ${serverId}`, {
668+
workspaceId,
669+
error: getMcpSafeErrorDiagnostics(error),
670+
})
671+
return false
672+
}
673+
}
674+
636675
private async isServerUnhealthy(workspaceId: string, serverId: string): Promise<boolean> {
637676
try {
638677
const entry = await this.cacheAdapter.get(failureCacheKey(workspaceId, serverId))
@@ -688,10 +727,16 @@ class McpService {
688727
await this.bestEffortInvalidateServerCache(workspaceId, serverId)
689728
return
690729
}
691-
await this.applyServerCacheMutation(workspaceId, serverId, mutation, null, [
692-
serverCacheKey(workspaceId, serverId),
693-
failureCacheKey(workspaceId, serverId),
694-
])
730+
const cacheApplied = await this.applyServerCacheMutation(
731+
workspaceId,
732+
serverId,
733+
mutation,
734+
null,
735+
[serverCacheKey(workspaceId, serverId), failureCacheKey(workspaceId, serverId)]
736+
)
737+
if (cacheApplied !== 'applied') return
738+
739+
await this.markServerCacheInvalidated(serverId, workspaceId, new Date(mutation.id))
695740
}
696741

697742
private async getCurrentCachedTools(

0 commit comments

Comments
 (0)