fix(worker): use stream.pipeline for image upload streams (#187) - #189
Merged
fennifith merged 3 commits intoJul 14, 2026
Merged
Conversation
Replaces the manual source.on('error') listener patched onto
processProfileImg in playfulprogramming#186 with stream.pipeline(), which destroys every
stream in the chain on failure rather than just the one hop we patched
manually. Applies the same fix to sync-collection's processImg, which
had the identical unfixed bug playfulprogramming#187 was filed to track.
Also threads the AbortSignal that createProcessor already provides
through to pipeline's options, so a cancelled job actually stops an
in-flight upload instead of leaving it to complete silently.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughImage uploads in author and collection synchronization now use a shared abort-aware streaming utility. Job abort signals are forwarded to profile, cover, and social image uploads. Tests cover aborted signals, upload failures, stream destruction, and mocked upload draining. ChangesImage streaming and abort propagation
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant SyncProcessor
participant uploadProcessedImage
participant SharpPipeline
participant S3
SyncProcessor->>uploadProcessedImage: pass image stream, key, size, and job signal
uploadProcessedImage->>SharpPipeline: resize and encode stream
uploadProcessedImage->>S3: upload transformed stream
SharpPipeline-->>uploadProcessedImage: complete or reject
S3-->>uploadProcessedImage: complete or reject
Possibly related issues
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 (1)
apps/worker/src/tasks/sync-author/processor.ts (1)
23-42: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
processProfileImgandprocessImgare copy-paste duplicates. Both functions have identical bodies — only the constant name differs (PROFILE_IMAGE_SIZE_MAXvsIMAGE_SIZE_MAX, both2048). Extract a shared image-upload helper to a common module so the two implementations don't diverge.
apps/worker/src/tasks/sync-author/processor.ts#L23-L42: replaceprocessProfileImgbody with a call to the shared helper.apps/worker/src/tasks/sync-collection/processor.ts#L24-L43: replaceprocessImgbody with a call to the same shared helper.🤖 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-author/processor.ts` around lines 23 - 42, Extract the duplicated upload pipeline from processProfileImg in apps/worker/src/tasks/sync-author/processor.ts lines 23-42 and processImg in apps/worker/src/tasks/sync-collection/processor.ts lines 24-43 into a shared helper in a common module, preserving the existing stream, resize, bucket, abort-signal, and S3 upload behavior. Replace both function bodies with calls to that helper, passing the appropriate image-size constant.
🤖 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/sync-author/processor.ts`:
- Around line 39-42: Update the concurrent pipeline/upload handling in
apps/worker/src/tasks/sync-author/processor.ts lines 39-42 and
apps/worker/src/tasks/sync-collection/processor.ts lines 40-43 so an s3.upload
rejection aborts the shared transform and causes pipeline(source, transform,
...) to unwind immediately, using AbortSignal.any(...) or equivalent transform
destruction while preserving normal successful completion.
---
Nitpick comments:
In `@apps/worker/src/tasks/sync-author/processor.ts`:
- Around line 23-42: Extract the duplicated upload pipeline from
processProfileImg in apps/worker/src/tasks/sync-author/processor.ts lines 23-42
and processImg in apps/worker/src/tasks/sync-collection/processor.ts lines 24-43
into a shared helper in a common module, preserving the existing stream, resize,
bucket, abort-signal, and S3 upload behavior. Replace both function bodies with
calls to that helper, passing the appropriate image-size constant.
🪄 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: 18e838e0-0203-4d15-b74f-15ac709f1bc3
📒 Files selected for processing (4)
apps/worker/src/tasks/sync-author/processor.test.tsapps/worker/src/tasks/sync-author/processor.tsapps/worker/src/tasks/sync-collection/processor.test.tsapps/worker/src/tasks/sync-collection/processor.ts
…ne race CodeRabbit flagged two issues on playfulprogramming#189: processProfileImg and processImg were byte-for-byte duplicates aside from their size constant, and Promise.all([pipeline(...), s3.upload(...)]) let a rejected upload leave the pipeline running unaware, since @aws-sdk/lib-storage's Upload doesn't destroy its input stream on its own - a failed upload could leave the pipeline hanging onto a stream nobody's consuming anymore. Extracts both into a single uploadProcessedImage helper in apps/worker/src/utils, parameterized by max size, called directly from both processors with their own size constant. This stays in apps/worker rather than packages/* since it's shared within one app, not across app boundaries. Fixes the race with AbortSignal.any(), combining the job's signal with a new internal AbortController that's triggered if the upload rejects first, so pipeline unwinds immediately in that case too. The success path and the job's own signal aborting are unchanged.
fennifith
reviewed
Jul 11, 2026
|
|
||
| beforeEach(() => { | ||
| // pipeline() won't resolve until the transform's readable side is drained | ||
| vi.mocked(s3.upload).mockImplementation(async (_bucket, _key, _tag, file) => { |
Member
There was a problem hiding this comment.
Could this be moved to test-utils/setup.ts as a default so that it's shared across tests?
Otherwise LGTM!
Was duplicated as a local beforeEach in sync-author and sync-collection processor.test.ts; now the default s3.upload mock implementation in test-utils/setup.ts. Confirmed no other worker test exercises a live ReadableStream through s3.upload with the old inert no-op behavior.
fennifith
approved these changes
Jul 14, 2026
fennifith
pushed a commit
that referenced
this pull request
Jul 14, 2026
…ne race CodeRabbit flagged two issues on #189: processProfileImg and processImg were byte-for-byte duplicates aside from their size constant, and Promise.all([pipeline(...), s3.upload(...)]) let a rejected upload leave the pipeline running unaware, since @aws-sdk/lib-storage's Upload doesn't destroy its input stream on its own - a failed upload could leave the pipeline hanging onto a stream nobody's consuming anymore. Extracts both into a single uploadProcessedImage helper in apps/worker/src/utils, parameterized by max size, called directly from both processors with their own size constant. This stays in apps/worker rather than packages/* since it's shared within one app, not across app boundaries. Fixes the race with AbortSignal.any(), combining the job's signal with a new internal AbortController that's triggered if the upload rejects first, so pipeline unwinds immediately in that case too. The success path and the job's own signal aborting are unchanged.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #187. Replaces the manual
source.on('error')listener patched ontosync-author'sprocessProfileImgin #186 withpipelinefromnode:stream/promises— the real fix James pointed at on the issue rather than that stopgap. Applies the identical fix tosync-collection'sprocessImg, which had the same unfixed bug #187 was filed to track.Also threads the
AbortSignalthatcreateProcessoralready hands both processors' callbacks through toprocessProfileImg/processImgand intopipeline's options — previously unused, meaning a cancelled job wouldn't actually stop an in-flight upload.apps/worker/src/tasks/url-metadata/utils/processImage.tsis intentionally not touched here — it already has its own custom timeout/dump/comparison logic wrapping its stream handling, and deserves a separate, careful look at whetherpipelineplus the signal it already accepts could simplify that existing logic, rather than a drop-in copy of this fix.Implementation notes
A few choices worth flagging explicitly:
processProfileImgandprocessImg(previously module-private).createProcessorowns its own internalAbortControllerand never exposes it, so there was no way to drive an "already aborted" scenario through the wrapped processor export from a test. Exporting these two functions made them directly testable for the abort-rejection case.Promise.all([pipeline(source, transform, { signal }), s3.upload(...)])instead of sequential awaits.pipeline()won't resolve until both the writable and readable sides of the transform stream are settled, but nothing reads the readable side untils3.upload()starts consuming it — awaitingpipeline()first would deadlock waiting on a reader that hasn't started yet.s3.uploadmock in bothprocessor.test.tsfiles to actually drain the stream it's given. It was previously a bare no-op, which was fine for the old fire-and-forget.pipe(), but a realpipeline()now depends on that stream being consumed to resolve — an inert mock would hang the tests indefinitely.Test plan
pnpm test:unitpasses (lint, knip, publint, sherif, vitest all green)sync-author/sync-collectionprocessor tests updated for the new pipeline-based implementationSummary by CodeRabbit