R2 + FFmpeg fallback encoding path#101
Conversation
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (12)
✨ 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 |
|
ECC bundle files are already tracked in this repository. Skipping generation of another bundle PR. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an R2 and FFmpeg-based fallback encoding path, enabling video transcoding via an external worker service. The changes include a dispatcher for sending HMAC-signed encoding requests and a new webhook handler to process completion callbacks and update video metadata. Review feedback focused on improving the robustness of the implementation by suggesting atomic database updates to prevent race conditions and verifying status transitions before initiating encoding jobs.
| if (target === 'ready' && payload.playbackHlsUrl) { | ||
| // Persist the HLS manifest URL alongside the status flip. | ||
| const result = await c.env.DB.prepare( | ||
| `UPDATE videos | ||
| SET playback_hls_url = ?, | ||
| updated_at = CURRENT_TIMESTAMP | ||
| WHERE id = ?`, | ||
| ) | ||
| .bind(payload.playbackHlsUrl, payload.videoId) | ||
| .run(); | ||
| if ((result.meta?.changes ?? 0) === 0) { | ||
| return c.json({ ok: true, matched: 0, status: target }, 202); | ||
| } | ||
| } | ||
|
|
||
| const transition = await transitionVideoStatus(c.env.DB, payload.videoId, target); | ||
| if (!transition.ok) { | ||
| return c.json({ ok: true, matched: 0, status: target }, 202); | ||
| } | ||
|
|
||
| return c.json({ ok: true, matched: transition.changes, status: target }); |
There was a problem hiding this comment.
The status transition and HLS URL update are currently performed in two separate, non-atomic database calls. This can lead to race conditions where metadata is updated even if the status transition fails (e.g., if the video is already in a terminal state).
Combining these into a single query guarded by the current status (using the 'WHERE status IN (...)' pattern) ensures atomicity and consistency, following the same pattern used in the Stream webhook handler. Additionally, logging the error message from the payload when the state is 'failed' will improve observability.
if (target === 'failed' && payload.error) {
console.error('FFmpeg encoding failed for video ' + payload.videoId + ': ' + payload.error);
}
const allowedFrom = (VIDEO_STATUSES as readonly VideoStatus[]).filter((from) =>
canTransition(from, target),
);
const placeholders = allowedFrom.map(() => '?').join(', ');
const result = await c.env.DB.prepare(
'UPDATE videos SET status = ?, playback_hls_url = COALESCE(?, playback_hls_url), updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN (' + placeholders + ')'
)
.bind(target, payload.playbackHlsUrl ?? null, payload.videoId, ...allowedFrom)
.run();
const changes = (result.meta?.changes as number | undefined) ?? 0;
if (changes === 0) {
return c.json({ ok: true, matched: 0, status: target }, 202);
}
return c.json({ ok: true, matched: changes, status: target });| import type { Context, MiddlewareHandler } from 'hono'; | ||
| import { Hono } from 'hono'; | ||
| import { verifyWebhookSignature } from './stream-webhook'; | ||
| import { transitionVideoStatus, type VideoStatus } from './video-status'; |
There was a problem hiding this comment.
Import VIDEO_STATUSES and canTransition to support the combined status transition and metadata update logic.
| import { transitionVideoStatus, type VideoStatus } from './video-status'; | |
| import { VIDEO_STATUSES, canTransition, transitionVideoStatus, type VideoStatus } from './video-status'; |
| if (env.FFMPEG_WORKER_URL) { | ||
| await transitionVideoStatus(env.DB, videoId, 'encoding'); | ||
| await sendToFfmpegWorker(env, videoId, r2Key); | ||
| return; | ||
| } |
There was a problem hiding this comment.
The result of transitionVideoStatus should be checked before dispatching to the FFmpeg worker. If the transition fails (e.g., the video has already moved to a terminal state like ready or failed), we should avoid initiating a new encoding job.
| if (env.FFMPEG_WORKER_URL) { | |
| await transitionVideoStatus(env.DB, videoId, 'encoding'); | |
| await sendToFfmpegWorker(env, videoId, r2Key); | |
| return; | |
| } | |
| if (env.FFMPEG_WORKER_URL) { | |
| const transition = await transitionVideoStatus(env.DB, videoId, 'encoding'); | |
| if (!transition.ok) return; | |
| await sendToFfmpegWorker(env, videoId, r2Key); | |
| return; | |
| } |
There was a problem hiding this comment.
Code Review
This pull request introduces a fallback video encoding path using R2 and an external FFmpeg worker, alongside the existing Cloudflare Stream integration. It includes a new webhook handler to process completion callbacks from the FFmpeg worker and updates the encoding dispatcher to support this secondary path. A significant issue was identified in the webhook handler where database updates for the playback URL and video status are performed non-atomically, potentially leading to race conditions or inconsistent states if webhooks arrive out of order.
| if (target === 'ready' && payload.playbackHlsUrl) { | ||
| // Persist the HLS manifest URL alongside the status flip. | ||
| const result = await c.env.DB.prepare( | ||
| `UPDATE videos | ||
| SET playback_hls_url = ?, | ||
| updated_at = CURRENT_TIMESTAMP | ||
| WHERE id = ?`, | ||
| ) | ||
| .bind(payload.playbackHlsUrl, payload.videoId) | ||
| .run(); | ||
| if ((result.meta?.changes ?? 0) === 0) { | ||
| return c.json({ ok: true, matched: 0, status: target }, 202); | ||
| } | ||
| } | ||
|
|
||
| const transition = await transitionVideoStatus(c.env.DB, payload.videoId, target); | ||
| if (!transition.ok) { | ||
| return c.json({ ok: true, matched: 0, status: target }, 202); | ||
| } | ||
|
|
||
| return c.json({ ok: true, matched: transition.changes, status: target }); |
There was a problem hiding this comment.
The current implementation performs two separate database updates: one for the playback_hls_url and another for the status transition via transitionVideoStatus. This is non-atomic and bypasses the state machine's safety checks for the URL update. Specifically, the first update (lines 70-77) does not verify if the video is in a valid state to receive a manifest URL, which could lead to database inconsistency if a stale or out-of-order webhook arrives (e.g., updating the URL of a video that has already failed or been deleted).
It is recommended to combine these into a single atomic UPDATE query that includes the status check in the WHERE clause, ensuring that the URL is only updated if the status transition is also valid. This follows the pattern already established in handleStreamWebhook.
const allowedFrom = (VIDEO_STATUSES as readonly VideoStatus[]).filter((from) =>
canTransition(from, target),
);
const placeholders = allowedFrom.map(() => '?').join(', ');
const result = await c.env.DB.prepare(
`UPDATE videos
SET status = ?,
playback_hls_url = COALESCE(?, playback_hls_url),
updated_at = CURRENT_TIMESTAMP
WHERE id = ?
AND status IN (${placeholders})`,
)
.bind(target, payload.playbackHlsUrl ?? null, payload.videoId, ...allowedFrom)
.run();
const changes = result.meta?.changes ?? 0;
if (changes === 0) {
return c.json({ ok: true, matched: 0, status: target }, 202);
}
return c.json({ ok: true, matched: changes, status: target });![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8e037341e6
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if (env.FFMPEG_WORKER_URL) { | ||
| await transitionVideoStatus(env.DB, videoId, 'encoding'); | ||
| await sendToFfmpegWorker(env, videoId, r2Key); |
There was a problem hiding this comment.
When FFMPEG_WORKER_URL is set but FFMPEG_WORKER_SECRET is missing, this branch still transitions the row to encoding and calls sendToFfmpegWorker, which throws; the catch path marks the row failed and rethrows, and queue() then calls message.retry() (src/workers/index.ts). In that configuration every encoding message becomes a poison retry loop (flipping failed→encoding→failed) instead of staying queued until config is fixed, which can churn the queue and dead-letter valid jobs.
Useful? React with 👍 / 👎.
| } | ||
| } | ||
|
|
||
| const transition = await transitionVideoStatus(c.env.DB, payload.videoId, target); |
There was a problem hiding this comment.
This transitions FFmpeg-processed videos to ready without ever setting stream_video_id, but the watch client only uses HLS when stream_video_id is present (src/frontend/pages/Watch.tsx) and /api/videos/:id does not expose playback_hls_url. In environments using the new FFmpeg path (STREAM_ENABLED !== 'true'), ready rows for non-browser-native source formats (e.g. MOV/MKV) still fall back to /api/videos/:id/stream and remain unplayable despite successful transcoding.
Useful? React with 👍 / 👎.
When Cloudflare Stream is disabled, dispatch encoding jobs to a containerized FFmpeg worker that writes HLS variants back to R2 and posts to /api/webhooks/ffmpeg with an HMAC-signed completion callback. Adds playback_hls_path column, /api/videos/:id/hls/* serve route, and a derivePlaybackUrl helper that hides the encoder choice from the player.
aloewright
force-pushed
the
conductor/alo-136-r2-ffmpeg-fallback-encoding-path
branch
from
8e03734 to
feed183
Compare
|
ECC bundle files are already tracked in this repository. Skipping generation of another bundle PR. |
Closes ALO-136.
Summary
encoding.ts): whenSTREAM_ENABLED !== 'true', marks the rowencodingand POSTs{videoId, sourceR2Key, outputR2Prefix}toFFMPEG_ENCODER_URL. Outbound is HMAC-signed withFFMPEG_ENCODER_SECRET(x-ffmpeg-signature: time=…,sig1=…). Missing URL is a graceful no-op so a fresh deploy doesn't fail uploads before the encoder Container is wired./api/webhooks/ffmpeg): mirrors the Stream webhook shape — same HMAC verifier, same canonical state-machine guard. Receives{videoId, status, outputR2Prefix, masterPlaylist?}and flips the lifecycle toready/failed, persisting the master playlist's R2 key in a newvideos.playback_hls_pathcolumn./api/videos/:id/hls/*): streams the master playlist + variant playlists +.tssegments straight out of R2. Path traversal is rejected atresolveHlsKey, hidden / DMCA rules match the existing/streamroute, manifests are short-cached and segments are immutable.derivePlaybackUrlhelper resolves Stream URL → fallback URL → null, exposed on/api/videos/:idasplayback_urlso the player stays agnostic of which encoder produced the asset.0019_video_hls_path.sql): adds theplayback_hls_pathcolumn (the existingplayback_hls_urlkeeps holding Stream's videodelivery.net URL).FFMPEG_ENCODER_URL/FFMPEG_ENCODER_SECRETdocumented inwrangler.toml. README updated with the full R2-only flow.Test plan
npm test— 533 tests pass (47 added acrossencoding.test.ts,ffmpeg-webhook.test.ts,hls.test.ts, plus 5 new cases invideos.test.ts)npm run lint(oxlint + AI Gateway guard) cleannpm run type-checkclean🤖 Generated with Claude Code