diff --git a/.gitignore b/.gitignore index 0f1a46a..f71ac69 100644 --- a/.gitignore +++ b/.gitignore @@ -75,4 +75,6 @@ Thumbs.db # Logs logs -*.log \ No newline at end of file +*.log +# Playwright MCP artifacts +.playwright-mcp/ diff --git a/__tests__/integration/concurrency-queue.test.ts b/__tests__/integration/concurrency-queue.test.ts new file mode 100644 index 0000000..2fbec79 --- /dev/null +++ b/__tests__/integration/concurrency-queue.test.ts @@ -0,0 +1,107 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { Nock, PloneMockServer, sampleDocument } from "plone-mcp/__tests__/utils/test-helpers"; +import { ploneRemoveSingleBlock } from "plone-mcp/tools/plone_remove_single_block"; +import { PloneClient } from "plone-mcp/plone-client"; +import { sessionManager } from "plone-mcp/session-manager"; + +describe("concurrency queue for single-block mutations", () => { + let mockServer: PloneMockServer; + const testBaseUrl = "http://localhost:8080/Plone"; + const testPath = "/my-page"; + + const blockToRemoveId1 = "block-to-remove-1"; + const blockToRemoveId2 = "block-to-remove-2"; + + const mockContentWithBothBlocks = { + ...sampleDocument, + "@id": `${testBaseUrl}/++api++${testPath}`, + id: "my-page", + blocks: { + [blockToRemoveId1]: { "@type": "text", plaintext: "Block to remove 1" }, + [blockToRemoveId2]: { "@type": "text", plaintext: "Block to remove 2" }, + }, + blocks_layout: { + items: [blockToRemoveId1, blockToRemoveId2], + }, + }; + + const mockContentAfterFirstRemoval = { + ...sampleDocument, + "@id": `${testBaseUrl}/++api++${testPath}`, + id: "my-page", + blocks: { + [blockToRemoveId2]: { "@type": "text", plaintext: "Block to remove 2" }, + }, + blocks_layout: { + items: [blockToRemoveId2], + }, + }; + + const mockContentAfterBothRemovals = { + ...sampleDocument, + "@id": `${testBaseUrl}/++api++${testPath}`, + id: "my-page", + blocks: {}, + blocks_layout: { + items: [], + }, + }; + + const sessionId = "test-session-id"; + const mockExtra = { + sessionId, + signal: new AbortController().signal, + requestId: "test-request-id", + } as any; + + beforeEach(() => { + mockServer = new PloneMockServer(testBaseUrl); + const service = sessionManager.getSession(sessionId); + service.client = new PloneClient({ baseUrl: testBaseUrl }); + }); + + afterEach(() => { + Nock.cleanAll(); + sessionManager.clearSession(sessionId); + }); + + it("should serialize two concurrent removals on the same path", async () => { + // First GET returns the initial state; second GET returns the state after + // the first removal, proving the second operation waited for the first. + mockServer.mockContentGet(testPath, mockContentWithBothBlocks); + mockServer.mockContentGet(testPath, mockContentAfterFirstRemoval); + + // First PATCH must only remove block 1; second PATCH must remove block 2 + // from the already-updated state (so neither block remains). + Nock(testBaseUrl) + .patch(`/++api++${testPath}`, (body: { blocks: Record }) => { + expect(body.blocks).not.toHaveProperty(blockToRemoveId1); + expect(body.blocks).toHaveProperty(blockToRemoveId2); + return true; + }) + .reply(200, mockContentAfterFirstRemoval); + + Nock(testBaseUrl) + .patch(`/++api++${testPath}`, (body: { blocks: Record }) => { + expect(body.blocks).not.toHaveProperty(blockToRemoveId1); + expect(body.blocks).not.toHaveProperty(blockToRemoveId2); + return true; + }) + .reply(200, mockContentAfterBothRemovals); + + const promiseA = ploneRemoveSingleBlock.handler( + { path: testPath, blockId: blockToRemoveId1 }, + mockExtra, + ); + const promiseB = ploneRemoveSingleBlock.handler( + { path: testPath, blockId: blockToRemoveId2 }, + mockExtra, + ); + + const [resultA, resultB] = await Promise.all([promiseA, promiseB]); + + expect(JSON.parse(resultA.content[0].text)).toEqual(mockContentAfterFirstRemoval); + expect(JSON.parse(resultB.content[0].text)).toEqual(mockContentAfterBothRemovals); + expect(Nock.isDone()).toBe(true); + }); +}); diff --git a/news/+concurrency.bugfix b/news/+concurrency.bugfix new file mode 100644 index 0000000..ef98d08 --- /dev/null +++ b/news/+concurrency.bugfix @@ -0,0 +1,5 @@ +Fix read-modify-write race in single-block tools. Concurrent calls to +`plone_add_single_block`, `plone_remove_single_block`, or +`plone_update_single_block` on the same content path now serialize through a +module-level per-path queue, preventing lost updates when two callers read the +same snapshot and the last PATCH overwrites the earlier change. @nileshgulia1 diff --git a/src/tools/plone_add_single_block.ts b/src/tools/plone_add_single_block.ts index 5a16544..9d4493f 100644 --- a/src/tools/plone_add_single_block.ts +++ b/src/tools/plone_add_single_block.ts @@ -9,6 +9,7 @@ import { processBlock, validateImageURL, } from "../utils/block-utils.js"; +import { withContentPathQueue } from "../utils/concurrency-queue.js"; import { PloneContent } from "../plone-client.js"; const inputSchema = z.object({ @@ -40,65 +41,67 @@ export const ploneAddSingleBlock = { const { path, blockType, position, blockData } = args; const client = service.getClient(); - // First get the current content - const content = (await client.get(path)) as PloneContent; + const updatedContent = await withContentPathQueue(path, async () => { + // First get the current content + const content = (await client.get(path)) as PloneContent; - const blocks = content.blocks || {}; - const blocks_layout = (content.blocks_layout as { items: string[] }) || { - items: [], - }; + const blocks = content.blocks || {}; + const blocks_layout = (content.blocks_layout as { items: string[] }) || { + items: [], + }; - // Generate new block ID - const blockId = generateBlockId(); + // Generate new block ID + const blockId = generateBlockId(); - // Validate image URLs asynchronously before processing - if ( - blockType === "image" && - typeof blockData.url === "string" && - blockData.url - ) { - const isValid = await validateImageURL( - blockData.url, - client.config.baseUrl, - ); - if (!isValid) { - throw wrapError( - "AddBlock", - `Invalid or inaccessible image URL: ${blockData.url} `, + // Validate image URLs asynchronously before processing + if ( + blockType === "image" && + typeof blockData.url === "string" && + blockData.url + ) { + const isValid = await validateImageURL( + blockData.url, + client.config.baseUrl, ); + if (!isValid) { + throw wrapError( + "AddBlock", + `Invalid or inaccessible image URL: ${blockData.url} `, + ); + } } - } - // Process block using centralized logic - try { - blocks[blockId] = processBlock( - blockType, - blockData, - client.config.baseUrl, - ); - } catch (error) { - throw new Error( - `Error processing block data: ${ - error instanceof Error ? error.message : String(error) - } `, - ); - } + // Process block using centralized logic + try { + blocks[blockId] = processBlock( + blockType, + blockData, + client.config.baseUrl, + ); + } catch (error) { + throw new Error( + `Error processing block data: ${ + error instanceof Error ? error.message : String(error) + } `, + ); + } - // Insert at specified position or at the end - if ( - position !== undefined && - position >= 0 && - position <= blocks_layout.items.length - ) { - blocks_layout.items.splice(position, 0, blockId); - } else { - blocks_layout.items.push(blockId); - } + // Insert at specified position or at the end + if ( + position !== undefined && + position >= 0 && + position <= blocks_layout.items.length + ) { + blocks_layout.items.splice(position, 0, blockId); + } else { + blocks_layout.items.push(blockId); + } - // Update the content - const updatedContent = await client.patch(path, { - blocks, - blocks_layout, + // Update the content + return await client.patch(path, { + blocks, + blocks_layout, + }); }); return { diff --git a/src/tools/plone_remove_single_block.ts b/src/tools/plone_remove_single_block.ts index 5b4f164..e6b4ec8 100644 --- a/src/tools/plone_remove_single_block.ts +++ b/src/tools/plone_remove_single_block.ts @@ -3,6 +3,7 @@ import { RequestHandlerExtra } from "@modelcontextprotocol/sdk/shared/protocol.j import { ServerRequest, ServerNotification } from "@modelcontextprotocol/sdk/types.js"; import { sessionManager } from "../session-manager.js"; import { wrapError } from "../utils/block-utils.js"; +import { withContentPathQueue } from "../utils/concurrency-queue.js"; import { PloneContent } from "../plone-client.js"; const inputSchema = z.object({ @@ -28,36 +29,38 @@ export const ploneRemoveSingleBlock = { const { path, blockId } = args; const client = service.getClient(); - // First get the current content - const content = (await client.get(path)) as PloneContent; + const updatedContent = await withContentPathQueue(path, async () => { + // First get the current content + const content = (await client.get(path)) as PloneContent; - const blocks = content.blocks || {}; - const blocks_layout = content.blocks_layout || { items: [] }; + const blocks = content.blocks || {}; + const blocks_layout = content.blocks_layout || { items: [] }; - if (!blocks[blockId]) { - const availableBlockIds = Object.keys(blocks); - throw new Error( - `Block with ID '${blockId}' not found. Available block IDs: ${availableBlockIds.join( - ", ", - )}`, - ); - } + if (!blocks[blockId]) { + const availableBlockIds = Object.keys(blocks); + throw new Error( + `Block with ID '${blockId}' not found. Available block IDs: ${availableBlockIds.join( + ", ", + )}`, + ); + } - // Remove the block - const updatedBlocks = Object.fromEntries( - Object.entries(blocks).filter(([key]) => key !== blockId), - ); + // Remove the block + const updatedBlocks = Object.fromEntries( + Object.entries(blocks).filter(([key]) => key !== blockId), + ); - // Remove from layout - const updatedLayoutItems = blocks_layout.items.filter( - (id: string) => id !== blockId, - ); + // Remove from layout + const updatedLayoutItems = blocks_layout.items.filter( + (id: string) => id !== blockId, + ); - // Update the content - const updatedContent = (await client.patch(path, { - blocks: updatedBlocks, - blocks_layout: { items: updatedLayoutItems }, - })) as PloneContent; + // Update the content + return (await client.patch(path, { + blocks: updatedBlocks, + blocks_layout: { items: updatedLayoutItems }, + })) as PloneContent; + }); return { content: [ diff --git a/src/tools/plone_update_single_block.ts b/src/tools/plone_update_single_block.ts index 3887770..a9c10fc 100644 --- a/src/tools/plone_update_single_block.ts +++ b/src/tools/plone_update_single_block.ts @@ -7,6 +7,7 @@ import { processBlock, validateImageURL, } from "../utils/block-utils.js"; +import { withContentPathQueue } from "../utils/concurrency-queue.js"; import { PloneContent } from "../plone-client.js"; const inputSchema = z.object({ @@ -33,54 +34,56 @@ export const ploneUpdateSingleBlock = { const { path, blockId, blockData } = args; const client = service.getClient(); - // First get the current content - const content = (await client.get(path)) as PloneContent; + const updatedContent = await withContentPathQueue(path, async () => { + // First get the current content + const content = (await client.get(path)) as PloneContent; - const blocks = content.blocks || {}; + const blocks = content.blocks || {}; - if (!blocks[blockId]) { - const availableBlockIds = Object.keys(blocks); - throw new Error( - `Block with ID '${blockId}' not found. Available block IDs: ${availableBlockIds.join( - ", ", - )}`, - ); - } + if (!blocks[blockId]) { + const availableBlockIds = Object.keys(blocks); + throw new Error( + `Block with ID '${blockId}' not found. Available block IDs: ${availableBlockIds.join( + ", ", + )}`, + ); + } - // Update the specific block - const existingBlock = blocks[blockId] as Record; - const blockType = - (blockData["@type"] as string) || (existingBlock["@type"] as string); - const mergedData = { ...existingBlock, ...blockData }; + // Update the specific block + const existingBlock = blocks[blockId] as Record; + const blockType = + (blockData["@type"] as string) || (existingBlock["@type"] as string); + const mergedData = { ...existingBlock, ...blockData }; - // Validate image URLs before processing - if ( - blockType === "image" && - typeof mergedData.url === "string" && - mergedData.url - ) { - const isValid = await validateImageURL( - mergedData.url, - client.config.baseUrl, - ); - if (!isValid) { - throw wrapError( - "UpdateBlock", - `Invalid or inaccessible image URL: ${mergedData.url}`, + // Validate image URLs before processing + if ( + blockType === "image" && + typeof mergedData.url === "string" && + mergedData.url + ) { + const isValid = await validateImageURL( + mergedData.url, + client.config.baseUrl, ); + if (!isValid) { + throw wrapError( + "UpdateBlock", + `Invalid or inaccessible image URL: ${mergedData.url}`, + ); + } } - } - blocks[blockId] = processBlock( - blockType, - mergedData, - client.config.baseUrl, - ); + blocks[blockId] = processBlock( + blockType, + mergedData, + client.config.baseUrl, + ); - // Update the content - const updatedContent = (await client.patch(path, { - blocks, - })) as PloneContent; + // Update the content + return (await client.patch(path, { + blocks, + })) as PloneContent; + }); return { content: [ diff --git a/src/utils/concurrency-queue.ts b/src/utils/concurrency-queue.ts new file mode 100644 index 0000000..60259ec --- /dev/null +++ b/src/utils/concurrency-queue.ts @@ -0,0 +1,51 @@ +/** + * Module-level per-path serialization queue for Plone content mutations. + * + * Multiple MCP sessions (stdio or HTTP) share this map because it is defined + * at module scope. Operations targeting the same content path are chained so + * that a later operation only starts after the previous one has settled, + * preventing read-modify-write races where two callers read the same snapshot + * and the last PATCH overwrites an earlier change. + */ + +const pathQueues = new Map>(); + +function normalizeQueuePath(path: string): string { + if (!path || path === "/") { + return "/"; + } + + let normalized = path.replace(/\/$/, ""); + if (!normalized.startsWith("/")) { + normalized = `/${normalized}`; + } + + return normalized; +} + +/** + * Run `operation` for `path` after any previously queued operation for the + * same path has settled. Rejected operations do not block subsequent ones. + */ +export async function withContentPathQueue( + path: string, + operation: () => Promise, +): Promise { + const key = normalizeQueuePath(path); + const previous = pathQueues.get(key); + + const current: Promise = previous + ? previous.then(() => operation(), () => operation()) + : operation(); + + pathQueues.set(key, current); + + try { + return await current; + } finally { + // Only clear the queue if no newer operation has chained onto it. + if (pathQueues.get(key) === current) { + pathQueues.delete(key); + } + } +}