From 592bce69c6da02a411f06bedf74aebf415e092b0 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 11:49:44 -0600 Subject: [PATCH] Advance bounded MCP cleanup past retained failures Keep failed uploads unverified while traversing subsequent candidates and wrapping for retry. Join concurrent sweeps and advance only after commit. Co-Authored-By: Paperclip --- .../__tests__/idle-completed-history.test.ts | 170 ++++++++++++++++++ .../src/services/public-mcp/file-transfers.ts | 20 ++- 2 files changed, 185 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts index 7573d84dbf7..389422f543c 100644 --- a/server/src/__tests__/idle-completed-history.test.ts +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -517,6 +517,176 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos } }, 60_000); + async function abandonedMcpBacklog() { + const { createPublicMcpTransfers } = await import("../services/public-mcp/file-transfers.js"); + const { default: express } = await import("express"); + const { default: request } = await import("supertest"); + const f = await seed(), p = await connectedMcp(f), other = await seed(), q = await connectedMcp(other); + try { + const principal = await p.oauth.authenticate(p.token), unrelated = await q.oauth.authenticate(q.token); + const hooks: { afterRead?: () => void } = {}; + const api: typeof p.api = async (...args) => { const result = await p.api(...args); hooks.afterRead?.(); return result; }; + const transfers = createPublicMcpTransfers(db, p.oauth, api, p.storage); + const app = express(); app.use(transfers.router); + const bytes = Buffer.from("real abandoned upload bytes"); + const args = (owner: typeof f) => ({ companyId: owner.companyId, taskId: owner.issueId, requestId: randomUUID(), + filename: "retained.txt", contentType: "text/plain", byteSize: bytes.length, + sha256: createHash("sha256").update(bytes).digest("hex") }); + const anchorArgs = args(other), link = await transfers.getUploadUrl(unrelated, anchorArgs); + if (!("url" in link)) throw new Error("Expected a new anchor upload link"); + const url = new URL(String(link.url)); + const uploaded = await request(app).put(url.pathname + url.search).set("Content-Type", "text/plain").send(bytes) + .timeout({ response: 5000, deadline: 10_000 }); + expect(uploaded.status).toBe(200); + for (let index = 0; index < 22; index++) { + const allocated = await transfers.getUploadUrl(principal, args(f)); + if (!("uploadId" in allocated)) throw new Error("Expected a new abandoned upload identity"); + const [upload] = await db.select().from(tables.mcpAttachmentUploads) + .where(eq(tables.mcpAttachmentUploads.id, String(allocated.uploadId))); + await p.storage.putFile({ companyId: f.companyId, namespace: "mcp-transfers", objectKey: upload.objectKey, + originalFilename: upload.originalFilename, contentType: upload.contentType, body: bytes }); + } + await db.update(tables.mcpAttachmentUploads).set({ expiresAt: new Date(0) }) + .where(eq(tables.mcpAttachmentUploads.companyId, f.companyId)); + const read = () => db.select().from(tables.mcpAttachmentUploads) + .where(eq(tables.mcpAttachmentUploads.companyId, f.companyId)).orderBy(tables.mcpAttachmentUploads.id); + const uploads = await read(); expect(uploads).toHaveLength(22); + const trigger = (kind: "upload" | "download", owner = transfers) => kind === "upload" + ? owner.getUploadUrl(unrelated, anchorArgs) + : owner.getDownloadUrl(unrelated, uploaded.body.attachment.id); + return { f, p, q, unrelated, api, hooks, transfers, createPublicMcpTransfers, trigger, uploads, read, + cleanup: async () => { await p.cleanup(); await q.cleanup(); } }; + } catch (error) { await p.cleanup(); await q.cleanup(); throw error; } + } + + it.each(["delete_failed", "object_survived", "head_failed"] as const)("advances bounded MCP cleanup through failed rows and retries after wrap: %s", async failure => { + const b = await abandonedMcpBacklog(); + const originalDelete = b.p.storage.deleteObject, originalHead = b.p.storage.headObject; + const good = b.uploads.at(-1)!, failed = b.uploads.slice(0, -1); + const failedKeys = new Set(failed.map(upload => upload.objectKey)); + const idsByKey = new Map(b.uploads.map(upload => [upload.objectKey, upload.id])); + const attempted: string[] = []; + try { + b.p.storage.deleteObject = async (companyId, objectKey) => { + attempted.push(idsByKey.get(objectKey)!); + if (failedKeys.has(objectKey)) { + if (failure === "delete_failed") throw new Error("controlled backlog DELETE failure"); + if (failure === "object_survived") return; + } + return originalDelete(companyId, objectKey); + }; + b.p.storage.headObject = async (companyId, objectKey) => { + if (failedKeys.has(objectKey) && failure === "head_failed") throw new Error("controlled backlog HEAD failure"); + return originalHead(companyId, objectKey); + }; + for (const kind of ["upload", "download"] as const) { + const before = attempted.length; + await b.trigger(kind); + expect(attempted.length - before).toBeLessThanOrEqual(20); + } + expect(new Set(attempted)).toEqual(new Set(b.uploads.map(upload => upload.id))); + const after = await b.read(); + expect(after.find(upload => upload.id === good.id)).toMatchObject({ cleanedAt: expect.any(Date), cleanupVerifiedAt: expect.any(Date) }); + expect(after.filter(upload => upload.id !== good.id)).toEqual(failed); + expect(await originalHead(b.f.companyId, good.objectKey)).toEqual({ exists: false }); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + // Restore the same storage owner. Future real calls must wrap and retry + // every failed identity; cursor movement alone cannot complete a row. + b.p.storage.deleteObject = async (companyId, objectKey) => { attempted.push(idsByKey.get(objectKey)!); return originalDelete(companyId, objectKey); }; + b.p.storage.headObject = originalHead; + for (const kind of ["download", "upload"] as const) { + const before = attempted.length; + await b.trigger(kind); + expect(attempted.length - before).toBeLessThanOrEqual(20); + } + for (const upload of await b.read()) { + expect(upload).toMatchObject({ cleanedAt: expect.any(Date), cleanupVerifiedAt: expect.any(Date) }); + expect(await originalHead(b.f.companyId, upload.objectKey)).toEqual({ exists: false }); + } + expect(await blocked("mcp_attachment_uploads")).toBe(false); + } finally { b.p.storage.deleteObject = originalDelete; b.p.storage.headObject = originalHead; await b.cleanup(); } + }, 60_000); + + it("coalesces concurrent MCP cleanup and revisits a locked row after wrap", async () => { + const b = await abandonedMcpBacklog(), lockedUpload = b.uploads[0]!; + const originalDelete = b.p.storage.deleteObject, originalHead = b.p.storage.headObject; + let releaseLock!: () => void, enteredLock!: () => void, releaseStorage!: () => void, enteredStorage!: () => void; + const locked = new Promise(resolve => { enteredLock = resolve; }); + const unlock = new Promise(resolve => { releaseLock = resolve; }); + const storageEntered = new Promise(resolve => { enteredStorage = resolve; }); + const continueStorage = new Promise(resolve => { releaseStorage = resolve; }); + let first: Promise | undefined, second: Promise | undefined; + const lock = db.transaction(async tx => { + await tx.select().from(tables.mcpAttachmentUploads).where(eq(tables.mcpAttachmentUploads.id, lockedUpload.id)).for("update"); + enteredLock(); await unlock; + }); + const attempted: string[] = []; + try { + await locked; + b.p.storage.deleteObject = async (companyId, objectKey) => { + attempted.push(objectKey); + if (attempted.length === 1) { enteredStorage(); await continueStorage; } + return originalDelete(companyId, objectKey); + }; + first = b.trigger("download"); await storageEntered; + let secondAuthorized!: () => void; + const authorized = new Promise(resolve => { secondAuthorized = resolve; }); + b.hooks.afterRead = secondAuthorized; + second = b.trigger("upload"); await authorized; + // Authority has returned. Drain its promise continuations before + // releasing the real storage operation; this is not a wall-clock wait. + await new Promise(resolve => setImmediate(resolve)); + releaseStorage(); await Promise.all([first, second]); + expect(attempted).toHaveLength(20); + const pending = (await b.read()).filter(upload => !upload.cleanupVerifiedAt); + expect(pending).toHaveLength(2); + expect(pending).toContainEqual(lockedUpload); + expect(await originalHead(b.f.companyId, lockedUpload.objectKey)).toMatchObject({ exists: true }); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + releaseLock(); await lock; + await b.trigger("download"); await b.trigger("upload"); + expect((await b.read()).every(upload => upload.cleanupVerifiedAt !== null)).toBe(true); + expect(await originalHead(b.f.companyId, lockedUpload.objectKey)).toEqual({ exists: false }); + expect(await blocked("mcp_attachment_uploads")).toBe(false); + } finally { + releaseStorage(); releaseLock(); await Promise.allSettled([lock, first, second]); + b.p.storage.deleteObject = originalDelete; await b.cleanup(); + } + }, 60_000); + + it("retries the same MCP cleanup batch after its real transaction rolls back", async () => { + const b = await abandonedMcpBacklog(); + const originalDelete = b.p.storage.deleteObject; + let rejectCommit = true; + const transaction: typeof db.transaction = (callback, config) => db.transaction(async tx => { + const result = await callback(tx); + if (rejectCommit) { rejectCommit = false; throw new Error("controlled cleanup transaction rollback"); } + return result; + }, config); + const faultDb = new Proxy(db, { get(target, property, receiver) { + if (property === "transaction") return transaction; + const value = Reflect.get(target, property, receiver); + return typeof value === "function" ? value.bind(target) : value; + } }); + const transfers = b.createPublicMcpTransfers(faultDb, b.p.oauth, b.api, b.p.storage); + const attempted: string[] = []; + try { + b.p.storage.deleteObject = async (companyId, objectKey) => { attempted.push(objectKey); return originalDelete(companyId, objectKey); }; + await expect(b.trigger("download", transfers)).rejects.toThrow("controlled cleanup transaction rollback"); + expect(attempted).toHaveLength(20); + expect(await b.read()).toEqual(b.uploads); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + const abortedBatch = attempted.slice(); + await b.trigger("upload", transfers); + expect(attempted.slice(abortedBatch.length)).toEqual(abortedBatch); + expect((await b.read()).filter(upload => upload.cleanupVerifiedAt !== null)).toHaveLength(20); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + await b.trigger("download", transfers); + expect((await b.read()).every(upload => upload.cleanupVerifiedAt !== null)).toBe(true); + expect(await blocked("mcp_attachment_uploads")).toBe(false); + } finally { b.p.storage.deleteObject = originalDelete; await b.cleanup(); } + }, 60_000); + it.each(["delete_failed", "object_survived", "head_failed"] as const)("retains MCP cleanup uncertainty after actual attachment removal: %s", async failure => { const { createPublicMcpTransfers } = await import("../services/public-mcp/file-transfers.js"); const { default: express } = await import("express"); diff --git a/server/src/services/public-mcp/file-transfers.ts b/server/src/services/public-mcp/file-transfers.ts index 72870c3f889..108fd3be52c 100644 --- a/server/src/services/public-mcp/file-transfers.ts +++ b/server/src/services/public-mcp/file-transfers.ts @@ -1,7 +1,7 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; import { pipeline } from "node:stream/promises"; import express, { Router } from "express"; -import { and, eq, gt, isNull, lt, sql } from "drizzle-orm"; +import { and, asc, eq, gt, isNull, lt, sql } from "drizzle-orm"; import { type Db, mcpAttachmentUploads, mcpFileTickets, mcpOauthGrants, issues } from "@paperclipai/db"; import { z } from "zod"; import { MAX_ATTACHMENT_BYTES, isAllowedContentType } from "../../attachment-types.js"; @@ -79,18 +79,27 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api await taskAccess(p, attachment.issueId, false, queryDb, sessionId); return attachment; } - async function sweep() { + // Each transfer service lives for the app lifetime. Advance this bounded + // traversal only after commit, and wrap so uncertain rows are retried. + let cleanupAfterId: string | undefined; + let pendingSweep: Promise | undefined; + function sweep() { + if (!pendingSweep) pendingSweep = runSweep().finally(() => { pendingSweep = undefined; }); + return pendingSweep; + } + async function runSweep() { await db.delete(mcpFileTickets).where(lt(mcpFileTickets.expiresAt, new Date())); // Serialize cleanup against completion/reissue. Requalify old optimistic // cleanup and orphaned removed attachments, never a still-linked file. - await db.transaction(async tx => { + cleanupAfterId = await db.transaction(async tx => { const abandoned = await tx.select().from(mcpAttachmentUploads).where(and( + eq(mcpAttachmentUploads.storageProvider, storage.provider), isNull(mcpAttachmentUploads.cleanupVerifiedAt), lt(mcpAttachmentUploads.expiresAt, new Date()), + cleanupAfterId ? gt(mcpAttachmentUploads.id, cleanupAfterId) : undefined, sql`(${mcpAttachmentUploads.attachmentId} IS NULL OR NOT EXISTS ( SELECT 1 FROM issue_attachments attachment WHERE attachment.id = ${mcpAttachmentUploads.attachmentId}))`, - )).limit(20).for("update", { skipLocked: true }); + )).orderBy(asc(mcpAttachmentUploads.id)).limit(20).for("update", { skipLocked: true }); for (const upload of abandoned) { - if (upload.storageProvider !== storage.provider) continue; try { await storage.deleteObject(upload.companyId, upload.objectKey); if ((await storage.headObject(upload.companyId, upload.objectKey)).exists !== false) { @@ -105,6 +114,7 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api const verifiedAt = new Date(); await tx.update(mcpAttachmentUploads).set({ cleanedAt: verifiedAt, cleanupVerifiedAt: verifiedAt }).where(eq(mcpAttachmentUploads.id, upload.id)); } + return abandoned.length === 20 ? abandoned.at(-1)!.id : undefined; }); } async function ticket(p: McpPrincipal, target: { uploadId?: string; attachmentId?: string }, sessionId?: string) {