fix(worker): use stream.pipeline for image upload streams (#187)#189
fix(worker): use stream.pipeline for image upload streams (#187)#189bbornino wants to merge 2 commits into
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.
|
Warning Review limit reached
Next review available in: 17 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. 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 Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughImage uploads in author and collection synchronization now use abort-aware streaming pipelines. Job abort signals are propagated to image helpers, which upload transformed streams through S3. Tests cover aborted signals and drain mocked upload streams. ChangesImage streaming and abort propagation
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant SyncProcessor
participant ImageHelper
participant Pipeline
participant S3
SyncProcessor->>ImageHelper: pass image stream, upload key, and job signal
ImageHelper->>Pipeline: process stream through Sharp with signal
ImageHelper->>S3: upload transformed stream
Pipeline-->>ImageHelper: complete or reject on abort
S3-->>ImageHelper: complete upload
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.
|
|
||
| beforeEach(() => { | ||
| // pipeline() won't resolve until the transform's readable side is drained | ||
| vi.mocked(s3.upload).mockImplementation(async (_bucket, _key, _tag, file) => { |
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!
2 participants
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