From 4ae92a89ff05d0929f583071dc9794ae92d5ec5f Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Wed, 15 Jul 2026 10:05:33 -0700 Subject: [PATCH 1/4] feat(worker): delay S3 object deletion instead of removing immediately Immediate s3.remove() calls risked breaking in-flight or cached requests for a key the frontend was still pointing at. Deletion is now scheduled as a delayed BullMQ job (24h grace period) via a shared scheduleS3ObjectDeletion() helper, reusing the existing delete-object primitive rather than depending on S3-native lifecycle rules, since Tigris (prod) only supports prefix-based lifecycle filtering and can't act on object tags. --- apps/worker/src/index.ts | 1 + .../tasks/delete-s3-object/processor.test.ts | 18 +++++++++ .../src/tasks/delete-s3-object/processor.ts | 9 +++++ .../src/tasks/sync-post/processor.test.ts | 40 +++++++++++-------- apps/worker/src/tasks/sync-post/processor.ts | 18 ++++++--- .../url-metadata/getEmbedDataFromGist.test.ts | 34 ++++++++++++++++ .../url-metadata/getEmbedDataFromGist.ts | 6 ++- apps/worker/test-utils/setup.ts | 4 ++ packages/bullmq/src/queues.ts | 2 + packages/bullmq/src/tasks/delete-s3-object.ts | 25 ++++++++++++ packages/bullmq/src/tasks/index.ts | 1 + packages/bullmq/src/tasks/types.ts | 7 ++++ 12 files changed, 141 insertions(+), 24 deletions(-) create mode 100644 apps/worker/src/tasks/delete-s3-object/processor.test.ts create mode 100644 apps/worker/src/tasks/delete-s3-object/processor.ts create mode 100644 packages/bullmq/src/tasks/delete-s3-object.ts diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 5114b521..89d250e3 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -12,4 +12,5 @@ createWorker( Tasks.GRANT_AUTHOR_ACHIEVEMENTS, "./tasks/grant-author-achievements/processor.ts", ); +createWorker(Tasks.DELETE_S3_OBJECT, "./tasks/delete-s3-object/processor.ts"); createHealthcheck(); diff --git a/apps/worker/src/tasks/delete-s3-object/processor.test.ts b/apps/worker/src/tasks/delete-s3-object/processor.test.ts new file mode 100644 index 00000000..b3047ba3 --- /dev/null +++ b/apps/worker/src/tasks/delete-s3-object/processor.test.ts @@ -0,0 +1,18 @@ +import processor from "./processor.ts"; +import type { TaskInputs } from "@playfulprogramming/bullmq"; +import type { Job } from "bullmq"; +import { s3 } from "@playfulprogramming/s3"; + +test("removes the object at the given bucket/key from S3", async () => { + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + }, + } as unknown as Job); + + expect(s3.remove).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + ); +}); diff --git a/apps/worker/src/tasks/delete-s3-object/processor.ts b/apps/worker/src/tasks/delete-s3-object/processor.ts new file mode 100644 index 00000000..1bb4e429 --- /dev/null +++ b/apps/worker/src/tasks/delete-s3-object/processor.ts @@ -0,0 +1,9 @@ +import { Tasks } from "@playfulprogramming/bullmq"; +import { s3 } from "@playfulprogramming/s3"; +import { createProcessor } from "../../createProcessor.ts"; + +export default createProcessor(Tasks.DELETE_S3_OBJECT, async (job) => { + const { bucket, key } = job.data; + await s3.remove(bucket, key); + console.log(`Removed ${bucket}/${key} from S3 after grace period`); +}); diff --git a/apps/worker/src/tasks/sync-post/processor.test.ts b/apps/worker/src/tasks/sync-post/processor.test.ts index 43f9ae22..e31c38d5 100644 --- a/apps/worker/src/tasks/sync-post/processor.test.ts +++ b/apps/worker/src/tasks/sync-post/processor.test.ts @@ -1,5 +1,10 @@ import processor from "./processor.ts"; -import { createJob, type TaskInputs, Tasks } from "@playfulprogramming/bullmq"; +import { + createJob, + scheduleS3ObjectDeletion, + type TaskInputs, + Tasks, +} from "@playfulprogramming/bullmq"; import type { Job } from "bullmq"; import { posts, @@ -289,12 +294,12 @@ test("Deletes a post record if it no longer exists", async () => { }, } as unknown as Job); - // Assert: Post's attachments were removed from S3 before the cascading delete - expect(s3.remove).toBeCalledWith( + // Assert: Post's attachments had their S3 removal scheduled before the cascading delete + expect(scheduleS3ObjectDeletion).toBeCalledWith( "example-bucket", "posts/example-post/attachments/notes.pdf", ); - expect(s3.remove).toBeCalledWith( + expect(scheduleS3ObjectDeletion).toBeCalledWith( "example-bucket", "posts/example-post/attachments/banner.jpeg", ); @@ -1143,15 +1148,15 @@ published: "2024-01-15T00:00:00Z" }, } as unknown as Job); - // Assert: attachment no longer in the repo was removed from S3 - expect(s3.remove).toBeCalledWith( + // Assert: attachment no longer in the repo had its S3 removal scheduled + expect(scheduleS3ObjectDeletion).toBeCalledWith( "example-bucket", "posts/diffing-post/attachments/old-file-sha.txt", ); - // Assert: changed attachment's old sha-keyed object was removed, and the - // new sha-keyed object was uploaded in its place - expect(s3.remove).toBeCalledWith( + // Assert: changed attachment's old sha-keyed object had its removal + // scheduled, and the new sha-keyed object was uploaded in its place + expect(scheduleS3ObjectDeletion).toBeCalledWith( "example-bucket", "posts/diffing-post/attachments/old-changed-sha.txt", ); @@ -1163,8 +1168,9 @@ published: "2024-01-15T00:00:00Z" "text/plain", ); - // Assert: the new object is uploaded before the old one is removed, so a - // failed upload can't leave the persisted row pointing at a deleted key + // Assert: the new object is uploaded before the old one's removal is + // scheduled, so a failed upload can't leave the persisted row pointing at + // a deleted key const newKeyUploadOrder = vi .mocked(s3.upload) .mock.calls.findIndex( @@ -1172,7 +1178,7 @@ published: "2024-01-15T00:00:00Z" call[1] === "posts/diffing-post/attachments/new-changed-sha.txt", ); const oldKeyRemoveOrder = vi - .mocked(s3.remove) + .mocked(scheduleS3ObjectDeletion) .mock.calls.findIndex( (call) => call[1] === "posts/diffing-post/attachments/old-changed-sha.txt", @@ -1180,10 +1186,12 @@ published: "2024-01-15T00:00:00Z" expect( vi.mocked(s3.upload).mock.invocationCallOrder[newKeyUploadOrder], ).toBeLessThan( - vi.mocked(s3.remove).mock.invocationCallOrder[oldKeyRemoveOrder], + vi.mocked(scheduleS3ObjectDeletion).mock.invocationCallOrder[ + oldKeyRemoveOrder + ], ); - // Assert: unchanged attachment was NOT re-uploaded or removed + // Assert: unchanged attachment was NOT re-uploaded or scheduled for removal expect(s3.upload).not.toBeCalledWith( "example-bucket", "posts/diffing-post/attachments/unchanged-sha.txt", @@ -1191,7 +1199,7 @@ published: "2024-01-15T00:00:00Z" expect.anything(), expect.anything(), ); - expect(s3.remove).not.toBeCalledWith( + expect(scheduleS3ObjectDeletion).not.toBeCalledWith( "example-bucket", "posts/diffing-post/attachments/unchanged-sha.txt", ); @@ -1337,7 +1345,7 @@ published: "2024-01-15T00:00:00Z" expect.anything(), expect.anything(), ); - expect(s3.remove).not.toBeCalled(); + expect(scheduleS3ObjectDeletion).not.toBeCalled(); // Assert: the existing row was carried forward unchanged expect(insertPostAttachmentsValues).toBeCalledWith([ diff --git a/apps/worker/src/tasks/sync-post/processor.ts b/apps/worker/src/tasks/sync-post/processor.ts index 2cf26c73..3a1a1c3e 100644 --- a/apps/worker/src/tasks/sync-post/processor.ts +++ b/apps/worker/src/tasks/sync-post/processor.ts @@ -1,5 +1,9 @@ import { env, PostMetaSchema } from "@playfulprogramming/common"; -import { Tasks, createJob } from "@playfulprogramming/bullmq"; +import { + Tasks, + createJob, + scheduleS3ObjectDeletion, +} from "@playfulprogramming/bullmq"; import { db, posts, @@ -146,8 +150,10 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => { const bucket = await s3.ensureBucket(env.S3_BUCKET); await Promise.all( removedAttachmentRows.map(async ({ attachmentKey }) => { - await s3.remove(bucket, attachmentKey); - console.log(`Removed attachment ${attachmentKey} from S3`); + await scheduleS3ObjectDeletion(bucket, attachmentKey); + console.log( + `Scheduled removal of attachment ${attachmentKey} from S3`, + ); }), ); } @@ -286,9 +292,9 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => { for (const [attachmentName, row] of previousAttachmentsByName) { if (!discoveredAttachmentNames.has(attachmentName)) { - await s3.remove(bucket, row.attachmentKey); + await scheduleS3ObjectDeletion(bucket, row.attachmentKey); console.log( - `Removed attachment ${row.attachmentKey} from S3 (no longer in repo)`, + `Scheduled removal of attachment ${row.attachmentKey} from S3 (no longer in repo)`, ); } } @@ -356,7 +362,7 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => { console.log(`Uploaded attachment ${attachmentKey} to S3`); if (previous !== undefined) { - await s3.remove(bucket, previous.attachmentKey); + await scheduleS3ObjectDeletion(bucket, previous.attachmentKey); } attachmentRows.push({ diff --git a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts index 59a92f44..e8b70278 100644 --- a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts +++ b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts @@ -7,6 +7,7 @@ import { urlMetadataGist, urlMetadataGistFile, } from "@playfulprogramming/db"; +import { scheduleS3ObjectDeletion } from "@playfulprogramming/bullmq"; test("fetches the expected information for a successful gist response", async () => { const gistUrl = new URL( @@ -59,3 +60,36 @@ test("fetches the expected information for a successful gist response", async () language: "text", }); }); + +test("schedules S3 removal for gist files that were deleted from the gist", async () => { + const gistUrl = new URL( + "https://gist.github.com/crutchcorn/36fe5553219c05ea38bacf1c7396085b", + ); + + (getGistById as Mock).mockReturnValueOnce( + Promise.resolve({ + description: "This is a description of the gist.", + files: {}, + }), + ); + + ( + db.delete(urlMetadataGistFile).where(undefined).returning as Mock + ).mockReturnValueOnce(Promise.resolve([{ filename: "old-file.txt" }])); + + const result = await getEmbedDataFromGist( + gistUrl, + new AbortController().signal, + ); + expect(result).toEqual({ + error: false, + gistId: "36fe5553219c05ea38bacf1c7396085b", + }); + + // Assert: S3 removal was scheduled (not performed immediately), keyed the + // same way getFileKey derives it - a hash of the filename under the gist's ID + expect(scheduleS3ObjectDeletion).toBeCalledWith( + "example-bucket", + "remote-gist/36fe5553219c05ea38bacf1c7396085b/775d94f3d7c5ee0d18ee08d4b65152b5", + ); +}); diff --git a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts index d56118de..2673d450 100644 --- a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts +++ b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts @@ -3,6 +3,7 @@ import { urlMetadataGist, urlMetadataGistFile, } from "@playfulprogramming/db"; +import { scheduleS3ObjectDeletion } from "@playfulprogramming/bullmq"; import { s3 } from "@playfulprogramming/s3"; import { fetchAsBot } from "../../utils/fetchAsBot.ts"; import * as github from "@playfulprogramming/github-api"; @@ -103,9 +104,10 @@ export async function getEmbedDataFromGist( }); }); - // Clean up deleted files from S3 + // Schedule cleanup of deleted files from S3, after a grace period so any + // in-flight or cached request for the old key doesn't 404 immediately for (const { filename } of deletedFilesResult) { - await s3.remove(BUCKET, getFileKey(filename)); + await scheduleS3ObjectDeletion(BUCKET, getFileKey(filename)); } return { diff --git a/apps/worker/test-utils/setup.ts b/apps/worker/test-utils/setup.ts index f7c9052c..ebfa8ffe 100644 --- a/apps/worker/test-utils/setup.ts +++ b/apps/worker/test-utils/setup.ts @@ -27,6 +27,10 @@ vi.mock("@playfulprogramming/bullmq", async () => { flowProducer: { add: vi.fn() }, createQueue: vi.fn(), createJob: vi.fn(), + // scheduleS3ObjectDeletion calls the real createJob internally via a + // relative import, which bypasses the createJob mock above - it needs + // its own override so tests don't hit a real BullMQ queue/Redis. + scheduleS3ObjectDeletion: vi.fn(), }; }); diff --git a/packages/bullmq/src/queues.ts b/packages/bullmq/src/queues.ts index 3de3695e..8f89f793 100644 --- a/packages/bullmq/src/queues.ts +++ b/packages/bullmq/src/queues.ts @@ -35,6 +35,7 @@ export async function createJob( task: T, id: string, data: TaskInputs[T], + opts?: { delay?: number }, ) { // eslint-disable-next-line @typescript-eslint/no-explicit-any const queue = createQueue(task) as Queue; @@ -42,5 +43,6 @@ export async function createJob( deduplication: { id: id, }, + delay: opts?.delay, }); } diff --git a/packages/bullmq/src/tasks/delete-s3-object.ts b/packages/bullmq/src/tasks/delete-s3-object.ts new file mode 100644 index 00000000..cb3fff44 --- /dev/null +++ b/packages/bullmq/src/tasks/delete-s3-object.ts @@ -0,0 +1,25 @@ +import { createJob } from "../queues.ts"; +import { Tasks } from "./types.ts"; + +export interface DeleteS3ObjectInput { + bucket: string; + key: string; +} + +export type DeleteS3ObjectOutput = void; + +// Grace period before a scheduled S3 deletion actually runs, so the frontend +// or CDN doesn't hit a 404 for a key it just fetched or cached. +export const DELETE_S3_OBJECT_GRACE_PERIOD_MS = 24 * 60 * 60 * 1000; + +export async function scheduleS3ObjectDeletion( + bucket: string, + key: string, +): Promise { + await createJob( + Tasks.DELETE_S3_OBJECT, + `delete-s3-object:${bucket}:${key}`, + { bucket, key }, + { delay: DELETE_S3_OBJECT_GRACE_PERIOD_MS }, + ); +} diff --git a/packages/bullmq/src/tasks/index.ts b/packages/bullmq/src/tasks/index.ts index 1f56e3ef..d47e91b8 100644 --- a/packages/bullmq/src/tasks/index.ts +++ b/packages/bullmq/src/tasks/index.ts @@ -1,3 +1,4 @@ +export * from "./delete-s3-object.ts"; export * from "./grant-author-achievements.ts"; export * from "./post-image.ts"; export * from "./sync-all.ts"; diff --git a/packages/bullmq/src/tasks/types.ts b/packages/bullmq/src/tasks/types.ts index 465ff1f3..efadacc7 100644 --- a/packages/bullmq/src/tasks/types.ts +++ b/packages/bullmq/src/tasks/types.ts @@ -11,6 +11,10 @@ import type { GrantAuthorAchievementsInput, GrantAuthorAchievementsOutput, } from "./grant-author-achievements.ts"; +import type { + DeleteS3ObjectInput, + DeleteS3ObjectOutput, +} from "./delete-s3-object.ts"; export const Tasks = { SYNC_ALL: "sync-all", @@ -20,6 +24,7 @@ export const Tasks = { URL_METADATA: "url-metadata", POST_IMAGES: "post-images", GRANT_AUTHOR_ACHIEVEMENTS: "grant-author-achievements", + DELETE_S3_OBJECT: "delete-s3-object", } as const; export type TasksKeys = keyof typeof Tasks; @@ -33,6 +38,7 @@ export interface TaskInputs { [Tasks.URL_METADATA]: UrlMetadataInput; [Tasks.POST_IMAGES]: PostImageInput; [Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsInput; + [Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectInput; } export type TaskInputsValues = TaskInputs[TasksValues]; @@ -45,6 +51,7 @@ export interface TaskOutputs { [Tasks.URL_METADATA]: UrlMetadataOutput; [Tasks.POST_IMAGES]: PostImageOutput; [Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsOutput; + [Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectOutput; } export type TaskOutputsValues = TaskOutputs[TasksValues]; From 8e9913efb52ef0c9f0bba9d65126e19d1544aeb9 Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Wed, 15 Jul 2026 11:42:45 -0700 Subject: [PATCH 2/4] fix(worker): guard delete-s3-object against re-referenced content-addressed keys CodeRabbit found that a scheduled deletion's stable job ID lets a content-addressed attachment key get deleted even after it's been legitimately re-referenced (e.g. identical content reappearing with the same sha). ETag-based staleness checking can't catch this, since identical content always produces an identical ETag whether it's the original upload or a fresh one. Capturing LastModified at scheduling time and verifying it's unchanged before deleting does catch it, since S3 bumps LastModified on every write regardless of content match - mirroring the same staleness-check pattern #191/#195 already established for the orphan-sweep task. --- .../tasks/delete-s3-object/processor.test.ts | 39 ++++++++++++++++++- .../src/tasks/delete-s3-object/processor.ts | 18 ++++++++- apps/worker/test-utils/setup.ts | 2 + packages/bullmq/package.json | 1 + packages/bullmq/src/tasks/delete-s3-object.ts | 10 ++++- packages/s3/src/utils.ts | 33 ++++++++++++++++ pnpm-lock.yaml | 3 ++ 7 files changed, 103 insertions(+), 3 deletions(-) diff --git a/apps/worker/src/tasks/delete-s3-object/processor.test.ts b/apps/worker/src/tasks/delete-s3-object/processor.test.ts index b3047ba3..63d43f3d 100644 --- a/apps/worker/src/tasks/delete-s3-object/processor.test.ts +++ b/apps/worker/src/tasks/delete-s3-object/processor.test.ts @@ -3,7 +3,7 @@ import type { TaskInputs } from "@playfulprogramming/bullmq"; import type { Job } from "bullmq"; import { s3 } from "@playfulprogramming/s3"; -test("removes the object at the given bucket/key from S3", async () => { +test("removes the object when no lastModified was captured at scheduling time", async () => { await processor({ data: { bucket: "example-bucket", @@ -11,8 +11,45 @@ test("removes the object at the given bucket/key from S3", async () => { }, } as unknown as Job); + expect(s3.unmodifiedSince).not.toBeCalled(); expect(s3.remove).toBeCalledWith( "example-bucket", "posts/example-post/attachments/notes.pdf", ); }); + +test("removes the object when it hasn't been modified since scheduling", async () => { + vi.mocked(s3.unmodifiedSince).mockResolvedValueOnce(true); + + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + lastModified: "2025-05-05T00:00:00.000Z", + }, + } as unknown as Job); + + expect(s3.unmodifiedSince).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + new Date("2025-05-05T00:00:00.000Z"), + ); + expect(s3.remove).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + ); +}); + +test("skips removal when the object was rewritten since scheduling", async () => { + vi.mocked(s3.unmodifiedSince).mockResolvedValueOnce(false); + + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + lastModified: "2025-05-05T00:00:00.000Z", + }, + } as unknown as Job); + + expect(s3.remove).not.toBeCalled(); +}); diff --git a/apps/worker/src/tasks/delete-s3-object/processor.ts b/apps/worker/src/tasks/delete-s3-object/processor.ts index 1bb4e429..c9c9fffb 100644 --- a/apps/worker/src/tasks/delete-s3-object/processor.ts +++ b/apps/worker/src/tasks/delete-s3-object/processor.ts @@ -3,7 +3,23 @@ import { s3 } from "@playfulprogramming/s3"; import { createProcessor } from "../../createProcessor.ts"; export default createProcessor(Tasks.DELETE_S3_OBJECT, async (job) => { - const { bucket, key } = job.data; + const { bucket, key, lastModified } = job.data; + + if (lastModified !== undefined) { + const stillUnmodified = await s3.unmodifiedSince( + bucket, + key, + new Date(lastModified), + ); + + if (!stillUnmodified) { + console.log( + `Skipped removal of ${bucket}/${key} - object was rewritten since deletion was scheduled`, + ); + return; + } + } + await s3.remove(bucket, key); console.log(`Removed ${bucket}/${key} from S3 after grace period`); }); diff --git a/apps/worker/test-utils/setup.ts b/apps/worker/test-utils/setup.ts index ebfa8ffe..2bce2ba9 100644 --- a/apps/worker/test-utils/setup.ts +++ b/apps/worker/test-utils/setup.ts @@ -40,6 +40,8 @@ vi.mock("@playfulprogramming/s3", () => { ensureBucket: vi.fn(() => "example-bucket"), upload: vi.fn(), remove: vi.fn(), + getLastModified: vi.fn(), + unmodifiedSince: vi.fn(() => true), }, }; }); diff --git a/packages/bullmq/package.json b/packages/bullmq/package.json index db8d2502..ee62e2a8 100644 --- a/packages/bullmq/package.json +++ b/packages/bullmq/package.json @@ -11,6 +11,7 @@ }, "dependencies": { "@playfulprogramming/redis": "workspace:*", + "@playfulprogramming/s3": "workspace:*", "bullmq": "catalog:", "typebox": "catalog:" } diff --git a/packages/bullmq/src/tasks/delete-s3-object.ts b/packages/bullmq/src/tasks/delete-s3-object.ts index cb3fff44..3ba6f05b 100644 --- a/packages/bullmq/src/tasks/delete-s3-object.ts +++ b/packages/bullmq/src/tasks/delete-s3-object.ts @@ -1,9 +1,15 @@ +import { s3 } from "@playfulprogramming/s3"; import { createJob } from "../queues.ts"; import { Tasks } from "./types.ts"; export interface DeleteS3ObjectInput { bucket: string; key: string; + // ISO timestamp of the object's LastModified at scheduling time. The + // processor re-checks this before deleting, so a key that gets rewritten + // in the meantime (even with byte-identical content, which leaves its + // ETag unchanged) doesn't get deleted out from under its new reference. + lastModified?: string; } export type DeleteS3ObjectOutput = void; @@ -16,10 +22,12 @@ export async function scheduleS3ObjectDeletion( bucket: string, key: string, ): Promise { + const lastModified = await s3.getLastModified(bucket, key); + await createJob( Tasks.DELETE_S3_OBJECT, `delete-s3-object:${bucket}:${key}`, - { bucket, key }, + { bucket, key, lastModified: lastModified?.toISOString() }, { delay: DELETE_S3_OBJECT_GRACE_PERIOD_MS }, ); } diff --git a/packages/s3/src/utils.ts b/packages/s3/src/utils.ts index 7c4eed77..3c40c9f0 100644 --- a/packages/s3/src/utils.ts +++ b/packages/s3/src/utils.ts @@ -85,6 +85,39 @@ export async function matchesEtag( } } +export async function getLastModified( + bucket: string, + key: string, +): Promise { + try { + const obj = await client.send( + new HeadObjectCommand({ Bucket: bucket, Key: key }), + ); + return obj.LastModified; + } catch (_e) { + return undefined; + } +} + +export async function unmodifiedSince( + bucket: string, + key: string, + since: Date, +): Promise { + try { + const obj = await client.send( + new HeadObjectCommand({ + Bucket: bucket, + Key: key, + IfUnmodifiedSince: since, + }), + ); + return !!obj; + } catch (_e) { + return false; + } +} + export async function upload( bucket: string, key: string, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 72487e8c..a00fc6d8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -266,6 +266,9 @@ importers: '@playfulprogramming/redis': specifier: workspace:* version: link:../redis + '@playfulprogramming/s3': + specifier: workspace:* + version: link:../s3 bullmq: specifier: 'catalog:' version: 5.76.10 From 4cb935979d76f159e48de8b6f6bf4a947f6a8328 Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Wed, 15 Jul 2026 13:05:42 -0700 Subject: [PATCH 3/4] fix(worker): scope delete-s3-object job ids to the object's generation BullMQ's deduplication silently drops a second queue.add() call under the same job id while the first job is still pending, keeping no record of the dropped call's data. Since the job id was a stable bucket:key string, a key that got deleted, legitimately re-added, then deleted again within the 24h grace period would have its second deletion silently absorbed into the first (now-stale) job - which then skips deleting on the safety check, leaking the object in S3 permanently. Fold the object's LastModified (or a random id when it can't be read) into the job id so each generation of a key gets its own job, while calls for an unchanged object still dedupe as before. --- packages/bullmq/package.json | 4 ++ .../bullmq/src/tasks/delete-s3-object.test.ts | 55 +++++++++++++++++++ packages/bullmq/src/tasks/delete-s3-object.ts | 10 +++- packages/bullmq/tsconfig.json | 3 + packages/bullmq/vitest.config.ts | 7 +++ pnpm-lock.yaml | 4 ++ 6 files changed, 81 insertions(+), 2 deletions(-) create mode 100644 packages/bullmq/src/tasks/delete-s3-object.test.ts create mode 100644 packages/bullmq/vitest.config.ts diff --git a/packages/bullmq/package.json b/packages/bullmq/package.json index ee62e2a8..518c12fc 100644 --- a/packages/bullmq/package.json +++ b/packages/bullmq/package.json @@ -7,6 +7,7 @@ "scripts": { "test:eslint": "eslint ./src", "test:build": "publint --strict", + "test": "vitest run", "build": "tsc --noEmit" }, "dependencies": { @@ -14,5 +15,8 @@ "@playfulprogramming/s3": "workspace:*", "bullmq": "catalog:", "typebox": "catalog:" + }, + "devDependencies": { + "vitest": "catalog:" } } diff --git a/packages/bullmq/src/tasks/delete-s3-object.test.ts b/packages/bullmq/src/tasks/delete-s3-object.test.ts new file mode 100644 index 00000000..45aaf73d --- /dev/null +++ b/packages/bullmq/src/tasks/delete-s3-object.test.ts @@ -0,0 +1,55 @@ +import { scheduleS3ObjectDeletion } from "./delete-s3-object.ts"; +import { createJob } from "../queues.ts"; +import { s3 } from "@playfulprogramming/s3"; + +vi.mock("../queues.ts", () => ({ + createJob: vi.fn(), +})); + +vi.mock("@playfulprogramming/s3", () => ({ + s3: { + getLastModified: vi.fn(), + }, +})); + +afterEach(() => { + vi.clearAllMocks(); +}); + +function jobIdFromCall(callIndex: number): string { + return vi.mocked(createJob).mock.calls[callIndex][1] as string; +} + +test("reuses the same job id when lastModified is unchanged across calls", async () => { + vi.mocked(s3.getLastModified).mockResolvedValue( + new Date("2026-01-01T00:00:00.000Z"), + ); + + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + expect(jobIdFromCall(0)).toEqual(jobIdFromCall(1)); +}); + +test("uses a different job id when lastModified changes between calls", async () => { + vi.mocked(s3.getLastModified).mockResolvedValueOnce( + new Date("2026-01-01T00:00:00.000Z"), + ); + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + vi.mocked(s3.getLastModified).mockResolvedValueOnce( + new Date("2026-01-02T00:00:00.000Z"), + ); + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + expect(jobIdFromCall(0)).not.toEqual(jobIdFromCall(1)); +}); + +test("always uses a unique job id when lastModified can't be determined", async () => { + vi.mocked(s3.getLastModified).mockResolvedValue(undefined); + + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + expect(jobIdFromCall(0)).not.toEqual(jobIdFromCall(1)); +}); diff --git a/packages/bullmq/src/tasks/delete-s3-object.ts b/packages/bullmq/src/tasks/delete-s3-object.ts index 3ba6f05b..4efd0016 100644 --- a/packages/bullmq/src/tasks/delete-s3-object.ts +++ b/packages/bullmq/src/tasks/delete-s3-object.ts @@ -23,11 +23,17 @@ export async function scheduleS3ObjectDeletion( key: string, ): Promise { const lastModified = await s3.getLastModified(bucket, key); + const lastModifiedIso = lastModified?.toISOString(); + // The job ID includes a generation marker (the object's LastModified, or + // a random ID when that couldn't be read) so that scheduling a deletion + // for a key that's since been rewritten gets its own job instead of + // silently deduplicating against - and being dropped in favor of - a + // still-pending job for the previous generation of that key. await createJob( Tasks.DELETE_S3_OBJECT, - `delete-s3-object:${bucket}:${key}`, - { bucket, key, lastModified: lastModified?.toISOString() }, + `delete-s3-object:${bucket}:${key}:${lastModifiedIso ?? crypto.randomUUID()}`, + { bucket, key, lastModified: lastModifiedIso }, { delay: DELETE_S3_OBJECT_GRACE_PERIOD_MS }, ); } diff --git a/packages/bullmq/tsconfig.json b/packages/bullmq/tsconfig.json index f27b9ce9..42296d79 100644 --- a/packages/bullmq/tsconfig.json +++ b/packages/bullmq/tsconfig.json @@ -1,4 +1,7 @@ { "extends": "../../tsconfig.json", + "compilerOptions": { + "types": ["vitest/globals"] + }, "include": ["src", "eslint.config.mjs"] } diff --git a/packages/bullmq/vitest.config.ts b/packages/bullmq/vitest.config.ts new file mode 100644 index 00000000..076c92fe --- /dev/null +++ b/packages/bullmq/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + globals: true, + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a00fc6d8..28b6599d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -275,6 +275,10 @@ importers: typebox: specifier: 'catalog:' version: 1.1.38 + devDependencies: + vitest: + specifier: 'catalog:' + version: 4.1.7(@types/node@25.9.1)(@vitest/coverage-v8@4.1.7)(vite@8.0.13(@types/node@25.9.1)(jiti@2.7.0)(yaml@2.9.0)) packages/common: dependencies: From b80ce7b4a2f9b0d625028dcb45a9783288552d1b Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Wed, 15 Jul 2026 13:35:50 -0700 Subject: [PATCH 4/4] fix(worker): don't schedule delete-s3-object when LastModified is unknown Scheduling a deletion without a LastModified left the processor with no safety check to run at execution time, so it would unconditionally delete whatever was at that key 24h later - including a legitimate new upload made during the grace period. Bail out of scheduling instead, with a warning log so a transient metadata-read failure doesn't silently leak the object. lastModified on DeleteS3ObjectInput is now required, and the random-id job ID fallback is gone along with it - every scheduled job is guaranteed to carry a real generation marker. --- .../bullmq/src/tasks/delete-s3-object.test.ts | 5 ++- packages/bullmq/src/tasks/delete-s3-object.ts | 31 ++++++++++++++----- 2 files changed, 25 insertions(+), 11 deletions(-) diff --git a/packages/bullmq/src/tasks/delete-s3-object.test.ts b/packages/bullmq/src/tasks/delete-s3-object.test.ts index 45aaf73d..026df988 100644 --- a/packages/bullmq/src/tasks/delete-s3-object.test.ts +++ b/packages/bullmq/src/tasks/delete-s3-object.test.ts @@ -45,11 +45,10 @@ test("uses a different job id when lastModified changes between calls", async () expect(jobIdFromCall(0)).not.toEqual(jobIdFromCall(1)); }); -test("always uses a unique job id when lastModified can't be determined", async () => { +test("does not schedule a deletion when lastModified can't be determined", async () => { vi.mocked(s3.getLastModified).mockResolvedValue(undefined); - await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); - expect(jobIdFromCall(0)).not.toEqual(jobIdFromCall(1)); + expect(createJob).not.toBeCalled(); }); diff --git a/packages/bullmq/src/tasks/delete-s3-object.ts b/packages/bullmq/src/tasks/delete-s3-object.ts index 4efd0016..dbc7cb9c 100644 --- a/packages/bullmq/src/tasks/delete-s3-object.ts +++ b/packages/bullmq/src/tasks/delete-s3-object.ts @@ -9,7 +9,7 @@ export interface DeleteS3ObjectInput { // processor re-checks this before deleting, so a key that gets rewritten // in the meantime (even with byte-identical content, which leaves its // ETag unchanged) doesn't get deleted out from under its new reference. - lastModified?: string; + lastModified: string; } export type DeleteS3ObjectOutput = void; @@ -23,16 +23,31 @@ export async function scheduleS3ObjectDeletion( key: string, ): Promise { const lastModified = await s3.getLastModified(bucket, key); - const lastModifiedIso = lastModified?.toISOString(); - // The job ID includes a generation marker (the object's LastModified, or - // a random ID when that couldn't be read) so that scheduling a deletion - // for a key that's since been rewritten gets its own job instead of - // silently deduplicating against - and being dropped in favor of - a - // still-pending job for the previous generation of that key. + if (lastModified === undefined) { + // Without a LastModified to check at execution time, the processor + // would have no way to detect a rewrite during the grace period and + // would unconditionally delete whatever's at this key 24h from now - + // including a legitimate new upload. Bail out instead of scheduling + // an unsafe deletion, but log it: a genuine transient failure to read + // the object's metadata here means this object never gets scheduled + // for cleanup at all, so it'd otherwise leak in S3 with no trace. + console.warn( + `Skipped scheduling deletion of ${bucket}/${key} - could not read its LastModified`, + ); + return; + } + + const lastModifiedIso = lastModified.toISOString(); + + // The job ID includes a generation marker (the object's LastModified) so + // that scheduling a deletion for a key that's since been rewritten gets + // its own job instead of silently deduplicating against - and being + // dropped in favor of - a still-pending job for the previous generation + // of that key. await createJob( Tasks.DELETE_S3_OBJECT, - `delete-s3-object:${bucket}:${key}:${lastModifiedIso ?? crypto.randomUUID()}`, + `delete-s3-object:${bucket}:${key}:${lastModifiedIso}`, { bucket, key, lastModified: lastModifiedIso }, { delay: DELETE_S3_OBJECT_GRACE_PERIOD_MS }, );