feat(finalizer): run Worker finalization with Effect - #24
Conversation
Review SummaryThis PR migrates Worker finalization to Effect-based services and adapters, adding scoped lease renewal, bounded retries, publication verification, cleanup, and test layers. The refactor preserves the queue contract but introduces a retry bug where publication attempts reuse a consumed output stream, risking empty or truncated R2 objects after transient failures. Key Changes:
Confidence Score: 5/5Recommendation: Review findings before merge. Rationale: The identified stream-reuse issue is directly tied to the changed publication retry logic and can cause corrupted or missing finalized recordings in production. The broader migration is sufficiently represented by the supplied changed-file context and validation commands. Special Attention- Re-fetch or recreate the container output stream for every publication retry. - Verify behavior when R2 put fails transiently or head returns inconclusive metadata. - Review lease renewal scope, retry boundaries, and cleanup behavior during failed finalization.Important Files Changed
Sequence DiagramsequenceDiagram
participant W as Worker
participant P as Finalization Process
participant C as Container
participant R as R2 Recordings
participant D as Database
W->>P: Start finalization
P->>D: Claim job and renew lease
P->>C: Upload parts and finalize job
P->>C: Fetch output stream
loop Publication retries
P->>R: Put output stream
R-->>P: Success or transient failure
P->>R: Head and verify publication
end
Note over P,C: Retrying the same one-shot stream can publish empty or truncated data
P->>D: Complete or release finalization
Audit history |
![access-time-code-pr-reviewer[bot]](https://avatars.githubusercontent.com/u/173532333?s=60&v=4){kind=small}
There was a problem hiding this comment.
🤖 review-bot · openai/gpt-5.6-luna · standard · 159.5s · 1,225,888 tokens · diff: 10 files · preloaded context: 10 changed + 10 related · policy: github-pr-review, thermo-nuclear-code-quality-review · optional skills invoked: vercel-composition-patterns, vercel-react-best-practices
| ); | ||
| const outputKey = `recordings/${encodeURIComponent(sessionId)}/finalizations/attempt-${attempt}/output.${outputMediaType(manifest) === "video/mp4" ? "mp4" : "webm"}`; | ||
|
|
||
| yield* Effect.forkScoped( |
There was a problem hiding this comment.
🟡 Warning
The lease-renewal fiber can fail with LeaseLost without interrupting the parent finalization effect, so processing may continue after losing ownership. Make renewal failure fail/interupt the attempt, and handle the resulting retry explicitly.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 62
Severity: warning
Comment: The lease-renewal fiber can fail with LeaseLost without interrupting the parent finalization effect, so processing may continue after losing ownership. Make renewal failure fail/interupt the attempt, and handle the resulting retry explicitly.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
| @@ -443,7 +235,9 @@ export async function reconciliationBatch(db: FinalizerDb) { | |||
| } | |||
| export default { | |||
There was a problem hiding this comment.
🟡 Warning
The new Effect-based worker modules coexist with duplicate legacy orchestration in this file, while the production entrypoint uses only the new path. Remove the duplicate implementation or make these exports thin adapters to one canonical flow; otherwise fixes can diverge between paths.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker.ts
Line: 236
Severity: warning
Comment: The new Effect-based worker modules coexist with duplicate legacy orchestration in this file, while the production entrypoint uses only the new path. Remove the duplicate implementation or make these exports thin adapters to one canonical flow; otherwise fixes can diverge between paths.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
mrevanzak
changed the title
ba69d8f to
e9f5786
Compare
| size: output.size, | ||
| }; | ||
| const published = yield* Effect.gen(function* () { | ||
| const written = yield* recordings.put(outputKey, output.body, { |
There was a problem hiding this comment.
🟡 Warning
output.body is a one-shot ReadableStream, but the surrounding publication operation retries on OutputPublicationNotProven. After the first failed upload consumes or disturbs the stream, retries reuse the same body and can fail or upload invalid data. Re-fetch the output stream for each attempt or provide a replayable body before applying the retry schedule.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 104
Severity: warning
Comment: `output.body` is a one-shot ReadableStream, but the surrounding publication operation retries on `OutputPublicationNotProven`. After the first failed upload consumes or disturbs the stream, retries reuse the same body and can fail or upload invalid data. Re-fetch the output stream for each attempt or provide a replayable body before applying the retry schedule.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
| ); | ||
| } | ||
| return true; | ||
| }).pipe(Effect.retry(retrySchedule)); |
There was a problem hiding this comment.
🟡 Warning
The publication retry reuses output.body, a one-shot ReadableStream. If the first R2 put fails after consuming any bytes, retries pass a locked/exhausted stream and finalization fails instead of recovering. Retry by reacquiring the container output stream for each publication attempt, or otherwise provide a fresh body per retry.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 128
Severity: warning
Comment: The publication retry reuses `output.body`, a one-shot ReadableStream. If the first R2 put fails after consuming any bytes, retries pass a locked/exhausted stream and finalization fails instead of recovering. Retry by reacquiring the container output stream for each publication attempt, or otherwise provide a fresh body per retry.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
e9f5786 to
7733ad8
Compare
| outputMediaType: outputMediaType(manifest), | ||
| segments: manifest.segments.map((segment) => ({ | ||
| partIndexes: segment.parts.map((part) => part.sequence), | ||
| segmentIndex: segment.index, |
There was a problem hiding this comment.
🟡 Warning
The publication effect retries recordings.put while reusing the same one-shot ReadableStream. If R2 consumes the stream and a transient upload error occurs, retrying with the disturbed stream fails and finalization is released instead of retried. Fetch a fresh container output stream inside each publication retry attempt, or otherwise recreate the body per attempt.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 91
Severity: warning
Comment: The publication effect retries `recordings.put` while reusing the same one-shot `ReadableStream`. If R2 consumes the stream and a transient upload error occurs, retrying with the disturbed stream fails and finalization is released instead of retried. Fetch a fresh container output stream inside each publication retry attempt, or otherwise recreate the body per attempt.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
7733ad8 to
81d1bef
Compare
There was a problem hiding this comment.
🤖 review-bot · openai/gpt-5.6-luna · standard · 110.1s · 536,617 tokens · diff: 10 files · preloaded context: 10 changed + 10 related · policy: github-pr-review, thermo-nuclear-code-quality-review · optional skills invoked: vercel-composition-patterns, vercel-react-best-practices
| size: output.size, | ||
| }; | ||
| const published = yield* Effect.gen(function* () { | ||
| const written = yield* recordings.put(outputKey, output.body, { |
There was a problem hiding this comment.
🟡 Warning
output.body is a one-shot ReadableStream, but this entire publication block is retried at line 128. If recordings.put fails after consuming the stream, retries reuse an exhausted body and can repeatedly fail, releasing an otherwise valid finalization. Re-fetch/create a fresh output stream for each publication attempt, or move only replayable operations inside the retry.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 104
Severity: warning
Comment: `output.body` is a one-shot ReadableStream, but this entire publication block is retried at line 128. If `recordings.put` fails after consuming the stream, retries reuse an exhausted body and can repeatedly fail, releasing an otherwise valid finalization. Re-fetch/create a fresh output stream for each publication attempt, or move only replayable operations inside the retry.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
| size: output.size, | ||
| }; | ||
| const published = yield* Effect.gen(function* () { | ||
| const written = yield* recordings.put(outputKey, output.body, { |
There was a problem hiding this comment.
🟡 Warning
output.body is a one-shot ReadableStream, but this recordings.put runs inside a retry. After a transient R2 failure, retrying reuses the consumed/disturbed stream and publication will fail permanently. Re-fetch a fresh container output stream for each attempt or make the body replayable before applying the retry.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 104
Severity: warning
Comment: `output.body` is a one-shot ReadableStream, but this `recordings.put` runs inside a retry. After a transient R2 failure, retrying reuses the consumed/disturbed stream and publication will fail permanently. Re-fetch a fresh container output stream for each attempt or make the body replayable before applying the retry.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
R2 verifies the container checksum incrementally. Stop buffering the full remux in getOutput and hashing it again before put.
81d1bef to
07f0bd1
Compare
| size: output.size, | ||
| }; | ||
| const published = yield* Effect.gen(function* () { | ||
| const written = yield* recordings.put(outputKey, output.body, { |
There was a problem hiding this comment.
🟡 Warning
The same ReadableStream is reused across the publication Effect.retry at line 128. If R2 put fails after consuming or locking the stream, retries pass the disturbed stream again and fail instead of recovering. Buffer or tee/recreate the stream per retry, or retry only before consuming it.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 104
Severity: warning
Comment: The same ReadableStream is reused across the publication Effect.retry at line 128. If R2 put fails after consuming or locking the stream, retries pass the disturbed stream again and fail instead of recovering. Buffer or tee/recreate the stream per retry, or retry only before consuming it.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
| size: output.size, | ||
| }; | ||
| const published = yield* Effect.gen(function* () { | ||
| const written = yield* recordings.put(outputKey, output.body, { |
There was a problem hiding this comment.
🟡 Warning
The publication retry reuses output.body, a one-shot ReadableStream. After a transient R2 error or an inconclusive head, subsequent retries pass the already-consumed stream to recordings.put, so retries can fail or publish an empty/truncated object. Re-fetch the container output for each publication attempt (or otherwise provide a fresh stream) instead of retrying the same stream.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/finalizer/src/worker/process.ts
Line: 104
Severity: warning
Comment: The publication retry reuses `output.body`, a one-shot ReadableStream. After a transient R2 error or an inconclusive `head`, subsequent retries pass the already-consumed stream to `recordings.put`, so retries can fail or publish an empty/truncated object. Re-fetch the container output for each publication attempt (or otherwise provide a fresh stream) instead of retrying the same stream.
How can I resolve this? If you propose a fix, please make it concise.
Think this is incorrect? Reply false positive: <reason>.
1 participant
Summary
Context.Tagservices, Cloudflare live adapters, and test layers.Validation
pnpm --filter @interview-web/finalizer testpnpm --filter @interview-web/finalizer check-types