feat(worker): add cleanup task for unreferenced S3 attachments - #195
feat(worker): add cleanup task for unreferenced S3 attachments#195bbornino wants to merge 6 commits into
Conversation
Sweeps S3 objects under posts/*/attachments/* on a repeatable interval and removes any with no matching post_attachments row, catching attachments orphaned by interrupted sync-post jobs.
sync-post uploads an attachment to S3 before its post_attachments row commits, so a very recent orphan-looking object may just be mid-flight. S3's list now returns lastModified alongside each key, and the cleanup task skips anything younger than a one-hour grace period.
|
Warning Review limit reached
Next review available in: 2 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughAdds attachment modification tracking and a repeatable BullMQ cleanup worker. The worker removes stale, unreferenced attachment rows and corresponding S3 objects, restores rows after S3 failures, and runs validated cleanup processing. ChangesAttachment cleanup
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Worker as Worker startup
participant Queue as BullMQ queue
participant Processor as Cleanup processor
participant DB as Attachments database
participant S3
Worker->>Queue: Schedule repeatable CLEANUP_ATTACHMENTS job
Queue->>Processor: Execute cleanup processor
Processor->>DB: Select stale unreferenced attachment row
Processor->>DB: Delete attachment row
Processor->>S3: Remove attachment object
Processor->>DB: Restore row when S3 removal fails
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
packages/s3/src/utils.ts (1)
79-107: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffConsider streaming or batching for very large attachment sets.
list()accumulates all matching S3 objects into a single in-memory array before returning. For workspaces with many posts/attachments this could consume significant memory. If scale is a concern, exposing an async iterable or accepting a callback would let the processor stream candidates without holding the full list. Not a blocker for the current scope.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/s3/src/utils.ts` around lines 79 - 107, Consider changing list() to expose matching S3 objects incrementally through an async iterable or callback instead of accumulating every object in the objects array. Preserve pagination via continuationToken and emit each valid object as soon as it is received, while updating callers to consume the streaming interface where appropriate.apps/worker/src/tasks/cleanup-attachments/processor.ts (2)
26-31: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffLoading all
post_attachmentsrows into memory may not scale.The processor selects every
attachmentKeyfrompost_attachmentsand builds an in-memorySet. For workspaces with a large number of attachments this query transfers and holds the full key set in memory on every run. Consider scoping the query to only the candidate keys (e.g.,where attachmentKey in (...)) or streaming the comparison if the table grows large.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/tasks/cleanup-attachments/processor.ts` around lines 26 - 31, Update the referenced-key lookup in the cleanup processor to scope postAttachments rows to the current candidate attachment keys, using the query builder’s attachmentKey membership filter before constructing referencedKeys. Preserve the existing Set-based comparison while avoiding retrieval of unrelated rows.
33-38: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffSequential deletion is safe but slow for large orphan sets.
The
for...ofloop awaits eachs3.removecall one at a time. This is correct for error isolation but could be slow when many orphans exist. If throughput becomes a concern, consider bounded concurrency (e.g.,p-limitor a simple chunkedPromise.all) while preserving the fail-fast behavior on the first error.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/tasks/cleanup-attachments/processor.ts` around lines 33 - 38, Update the orphan-removal loop in the cleanup processor to delete unreferenced keys with bounded concurrency instead of awaiting each s3.remove call sequentially. Preserve fail-fast behavior by propagating the first deletion error, and retain the existing skip logic and success logging for each removed key.apps/worker/src/tasks/cleanup-attachments/processor.test.ts (1)
10-121: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a test for error propagation when
s3.removethrows.The PR objective states the job "fails the job if an error occurs," but no test verifies that a throwing
s3.removecauses the processor to reject. A test likevi.mocked(s3.remove).mockRejectedValueOnce(new Error("..."))asserting the processor rejects would lock in that contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/tasks/cleanup-attachments/processor.test.ts` around lines 10 - 121, Add a test alongside the existing cleanup processor tests that configures an eligible orphaned attachment and makes s3.remove reject with an Error, then assert that processor rejects with the same error. Reuse the existing S3 listing and database mocks needed to reach removal, and verify the rejection propagates without being swallowed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/bullmq/src/tasks/types.ts`:
- Line 50: Update the TaskOutputs mapping for Tasks.CLEANUP_ATTACHMENTS from
object to void so it matches the cleanup processor’s Promise<void> return
contract and resolves the processor type mismatch.
---
Nitpick comments:
In `@apps/worker/src/tasks/cleanup-attachments/processor.test.ts`:
- Around line 10-121: Add a test alongside the existing cleanup processor tests
that configures an eligible orphaned attachment and makes s3.remove reject with
an Error, then assert that processor rejects with the same error. Reuse the
existing S3 listing and database mocks needed to reach removal, and verify the
rejection propagates without being swallowed.
In `@apps/worker/src/tasks/cleanup-attachments/processor.ts`:
- Around line 26-31: Update the referenced-key lookup in the cleanup processor
to scope postAttachments rows to the current candidate attachment keys, using
the query builder’s attachmentKey membership filter before constructing
referencedKeys. Preserve the existing Set-based comparison while avoiding
retrieval of unrelated rows.
- Around line 33-38: Update the orphan-removal loop in the cleanup processor to
delete unreferenced keys with bounded concurrency instead of awaiting each
s3.remove call sequentially. Preserve fail-fast behavior by propagating the
first deletion error, and retain the existing skip logic and success logging for
each removed key.
In `@packages/s3/src/utils.ts`:
- Around line 79-107: Consider changing list() to expose matching S3 objects
incrementally through an async iterable or callback instead of accumulating
every object in the objects array. Preserve pagination via continuationToken and
emit each valid object as soon as it is received, while updating callers to
consume the streaming interface where appropriate.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5bffafa3-3312-4144-af8f-85464415f326
📒 Files selected for processing (6)
apps/worker/src/index.tsapps/worker/src/tasks/cleanup-attachments/processor.test.tsapps/worker/src/tasks/cleanup-attachments/processor.tsapps/worker/test-utils/setup.tspackages/bullmq/src/tasks/types.tspackages/s3/src/utils.ts
TaskOutputs[Tasks.CLEANUP_ATTACHMENTS] was typed as object but the processor returns Promise<void>, breaking the build. Also adds a test locking in that an S3 removal failure rejects the job instead of being swallowed.
…ressed 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 playfulprogramming#191/playfulprogramming#195 already established for the orphan-sweep task.
| export default createProcessor(Tasks.CLEANUP_ATTACHMENTS, async () => { | ||
| const bucket = await s3.ensureBucket(env.S3_BUCKET); | ||
|
|
||
| const objects = await s3.list(bucket, ATTACHMENTS_PREFIX); |
There was a problem hiding this comment.
After merging main - can this be modified to get a list of attachments directly from the database instead of using S3's ListObjects? (i.e. find attachments rows where no corresponding postAttachments relation exists)
This job could run a loop of DELETE FROM attachments WHERE NOT EXISTS (condition above) LIMIT 1 RETURNING attachment_key, and delete each attachment from S3 when encountered.
There was a problem hiding this comment.
I'm realizing this will also conflict with #197, but we should track the lastModified time in the attachments table and use that as a condition of deletion.
…heduleS3ObjectDeletion import Matches the existing slug+branch pattern used elsewhere in this file. The removal-scheduling call sites this import backed were dropped by playfulprogramming#190's schema restructuring; per James, that's intentional going forward since attachments are now shared/reference-counted across posts and branches, and orphan cleanup belongs to playfulprogramming#191/playfulprogramming#195's sweep instead of inline detection here.
Conflicted in apps/worker/src/index.ts, apps/worker/test-utils/setup.ts, and packages/bullmq/src/tasks/types.ts: this branch's CLEANUP_ATTACHMENTS task and main's now-merged PR playfulprogramming#197 DELETE_S3_OBJECT task (playfulprogramming#188) both added entries to the same task-registry files while in flight independently. Resolved additively in all three - both tasks coexist, nothing dropped: - index.ts registers both createWorker calls (cleanup-attachments and delete-s3-object) ahead of createHealthcheck(); this branch's repeatable job scheduler for CLEANUP_ATTACHMENTS is untouched - types.ts keeps both Tasks entries, both TaskInputs entries, and both TaskOutputs entries, plus the DeleteS3ObjectInput/Output import - test-utils/setup.ts keeps both s3 mock additions in the same mock object - this branch's list() alongside playfulprogramming#188's getLastModified()/ unmodifiedSince() Full build:all and test:unit gates pass after resolution.
…w feedback Replaces the S3-list-based orphan sweep with a DB-native DELETE ... WHERE NOT EXISTS ... RETURNING loop against the attachments table, per James's review feedback on playfulprogramming#195. Adds a lastModified column to attachments, used as the grace-period condition in place of S3 object timestamps, and switches the sync-post insert from onConflictDoNothing to onConflictDoUpdate so a conflicting insert (content reused across posts/ branches) still refreshes the timestamp instead of silently no-opping.
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
packages/db/src/schema/attachments.ts (1)
8-10: 🚀 Performance & Scalability | 🔵 TrivialConsider an index for the cleanup predicate.
apps/worker/src/tasks/cleanup-attachments/processor.tsfilters onlastModifiedwithlt(...)combined with aNOT EXISTSanti-join onpost_attachments. The cleanup loop runs this query once per deleted row. If theattachmentstable grows large, add an index onlast_modifiedto keep each iteration cheap.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/db/src/schema/attachments.ts` around lines 8 - 10, Add an index for the lastModified column in the attachments schema so cleanup queries using the lt(...) predicate can efficiently find stale rows. Define it alongside the attachments table schema and preserve the existing timestamp definition and post_attachments anti-join behavior.apps/worker/src/tasks/cleanup-attachments/processor.test.ts (1)
15-46: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGrace-period behavior is not covered.
The PR objectives list grace-period tests. These tests mock the query at the
.where(expect.anything())level, so thelt(attachments.lastModified, staleBefore)predicate is never asserted.vi.setSystemTime(NOW)is set in every test, but no assertion reads the cutoff value.Add an assertion on the arguments passed to the mocked
where, or capturestaleBeforeand confirm it equalsNOWminus one hour. Without it, a change toGRACE_PERIOD_MSor an accidental removal of theltpredicate passes the suite.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/tasks/cleanup-attachments/processor.test.ts` around lines 15 - 46, Add a grace-period assertion to the cleanup processor tests around the mocked deleteAttachmentReturning query, verifying that its where predicate includes a cutoff equal to NOW minus one hour. Capture or inspect the argument passed to where rather than using only expect.anything(), while preserving the existing S3 removal and empty-result assertions.apps/worker/src/tasks/sync-post/processor.test.ts (1)
703-710: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the
onConflictDoUpdatechange.These assertions confirm the insert payload includes
lastModified. They do not confirm the conflict behavior changed fromonConflictDoNothingtoonConflictDoUpdate. That change is the mechanism that prevents the cleanup sweep from deleting a re-used attachment key. Add an assertion on the mockedonConflictDoUpdatefor theattachmentstable, similar to the assertion inapps/worker/src/tasks/cleanup-attachments/processor.test.tsat lines 105-108.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/tasks/sync-post/processor.test.ts` around lines 703 - 710, Add test coverage in the attachment processing test around the existing `db.insert(attachments).values` assertions to verify the mocked `onConflictDoUpdate` call, matching the pattern used by the cleanup-attachments test. Assert the attachments conflict-update configuration rather than only the insert payload, confirming reused attachment keys trigger the update behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/worker/src/tasks/cleanup-attachments/processor.ts`:
- Around line 10-14: Update the cleanup processor callback passed to
createProcessor to accept its abort signal and check signal.aborted before each
iteration of the for loop, returning promptly when cancellation occurs. Replace
the unbounded loop in the cleanup flow with a bounded iteration strategy if an
existing per-run limit is available, while preserving the current bucket and
staleBefore setup and deletion behavior.
- Around line 25-45: Update the delete condition in the cleanup processor to use
inArray() with attachments.attachmentKey and candidateKey instead of eq(). Leave
the correlated notExists() query and candidateKey selection unchanged.
---
Nitpick comments:
In `@apps/worker/src/tasks/cleanup-attachments/processor.test.ts`:
- Around line 15-46: Add a grace-period assertion to the cleanup processor tests
around the mocked deleteAttachmentReturning query, verifying that its where
predicate includes a cutoff equal to NOW minus one hour. Capture or inspect the
argument passed to where rather than using only expect.anything(), while
preserving the existing S3 removal and empty-result assertions.
In `@apps/worker/src/tasks/sync-post/processor.test.ts`:
- Around line 703-710: Add test coverage in the attachment processing test
around the existing `db.insert(attachments).values` assertions to verify the
mocked `onConflictDoUpdate` call, matching the pattern used by the
cleanup-attachments test. Assert the attachments conflict-update configuration
rather than only the insert payload, confirming reused attachment keys trigger
the update behavior.
In `@packages/db/src/schema/attachments.ts`:
- Around line 8-10: Add an index for the lastModified column in the attachments
schema so cleanup queries using the lt(...) predicate can efficiently find stale
rows. Define it alongside the attachments table schema and preserve the existing
timestamp definition and post_attachments anti-join behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 847a349e-6d4a-4174-8090-a27245d61872
📒 Files selected for processing (8)
apps/worker/src/tasks/cleanup-attachments/processor.test.tsapps/worker/src/tasks/cleanup-attachments/processor.tsapps/worker/src/tasks/sync-post/processor.test.tsapps/worker/src/tasks/sync-post/processor.tspackages/db/drizzle/20260807133839_sudden_micromacro/migration.sqlpackages/db/drizzle/20260807133839_sudden_micromacro/snapshot.jsonpackages/db/src/schema/attachments.tspackages/test-fixtures/src/db-mock.ts
Switches the candidate-key comparison from eq() to inArray() per CodeRabbit's suggestion - confirmed via a live toSQL() check that eq() was already correctly inlining the subquery, so this is a clarity change rather than a correctness fix. Fixes abort-signal handling so a timed-out job stops looping instead of continuing to run after BullMQ has already marked it failed. Adds a covering test for the lastModified cutoff value and the onConflictDoUpdate call in sync-post, and adds an index on attachments.lastModified.
2 participants
Description:
Closes #191.
Adds a
cleanup-attachmentsBullMQ task that removes S3 attachment objects with no matchingpost_attachmentsrow, sincesync-postcan't safely delete an attachment itself when multiple branches of the same post might still be using it.What changed
packages/s3/src/utils.ts: newlist(bucket, prefix)helper (paginatedListObjectsV2), returning each object's key andlastModified. Reuses the existing S3 client rather than adding new client logic.apps/worker/src/tasks/cleanup-attachments/processor.ts: lists S3 objects underposts/*/attachments/*, filters out anything younger than a one-hour grace period, diffs the remainder against the full set ofattachmentKeyvalues inpost_attachments, and removes any orphaned object. Plain sequential loop, no concurrency/batching, errors throw and fail the job — matching the existing worker-task conventions.packages/bullmq, wired up inapps/worker/src/index.tsalongside a repeatable job registration (see open question test: add initial unit tests #1 below).processor.test.tscovers: an unreferenced attachment getting removed, a referenced attachment being left alone, a no-op when there are no attachment objects, that the query has no per-post filter, and that an otherwise-unreferenced attachment inside the grace period is left alone (and never even triggers thepost_attachmentsquery).Redesign: DB-native cleanup instead of S3-list-based orphan detection
Per James's review feedback on the original implementation, this replaces the S3-
list()-and-diffapproach with a query against the
attachmentstable directly.Before: list every object under
posts/in S3, filter to/attachments/keys older than the graceperiod, then check each against
post_attachmentsfor references, removing whatever's left.After: loop a single query per iteration: