Add ab-av1 worker websocket and shared CRF claiming#19
Open
mjc wants to merge 70 commits into
Open
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThis PR adds a websocket-based worker flow for CRF-search, including token authentication, session tracking, atomic work claiming, transfer streaming, dashboard and workers UI updates, diagnostics output, and supporting config/documentation. ChangesWorker Websocket Feature
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Worker
participant WorkerSocket
participant WorkerChannel
participant WorkerSessions
participant Media
Worker->>WorkerSocket: connect(token)
WorkerSocket-->>Worker: assign worker_id
Worker->>WorkerChannel: join("workers:crf_search")
Worker->>WorkerChannel: announce(payload)
WorkerChannel->>WorkerSessions: register(attrs)
WorkerChannel-->>Worker: accepted(protocol_version)
Worker->>WorkerChannel: pull_work
WorkerChannel->>Media: claim_next_video_for_crf_search
Media-->>WorkerChannel: video or nil
WorkerChannel->>WorkerSessions: assign_video
WorkerChannel-->>Worker: work_assigned or no_work
Worker->>WorkerChannel: crf_search_result / crf_search_completed
WorkerChannel->>WorkerSessions: clear_video
WorkerChannel-->>Worker: event_ack
Poem
🚥 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 |
mjc
commented
Jul 7, 2026
mjc
commented
Jul 7, 2026
mjc
commented
Jul 7, 2026
mjc
commented
Jul 7, 2026
mjc
commented
Jul 7, 2026
mjc
commented
Jul 7, 2026
There was a problem hiding this comment.
Pull request overview
Adds a Phoenix WebSocket/Channel surface for remote ab-av1 CRF-search workers, and unifies CRF job dispatching behind an atomic “claim” so local Broadway and remote workers don’t double-claim the same analyzed video.
Changes:
- Introduces authenticated worker WebSocket endpoint (
/workers/socket) with protocol validation, session tracking, heartbeat, and job assignment replies. - Adds an atomic DB claim path for CRF-search work and switches the local CRF Broadway producer to use it.
- Extends diagnostics + dashboard UI/docs/tests to surface worker connection info and validate protocol/session behavior.
Reviewed changes
Copilot reviewed 25 out of 26 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| test/support/channel_case.ex | Adds a ChannelCase helper for channel tests with DB sandbox setup. |
| test/reencodarr/broadway_producers_test.exs | Adds tests asserting atomic claim transitions and prevents redispatch. |
| test/reencodarr/ab_av1/worker_sessions_test.exs | Adds tests for worker session lifecycle, expiry, and diagnostics output. |
| test/reencodarr/ab_av1/worker_protocol_test.exs | Adds tests for building job_assigned payloads. |
| test/reencodarr_web/live/dashboard_v2_live_test.exs | Tests dashboard display of worker WebSocket token fingerprint and URL. |
| test/reencodarr_web/channels/worker_channel_test.exs | End-to-end channel tests for auth, announce, work claiming, heartbeat, and disconnect requeue. |
| lib/reencodarr/media/video_queries.ex | Implements atomic “claim next CRF-search video” query + state transition. |
| lib/reencodarr/media.ex | Exposes claim_next_video_for_crf_search/0 from the Media context. |
| lib/reencodarr/diagnostics.ex | Adds worker session listing to diagnostics output. |
| lib/reencodarr/crf_searcher/broadway/producer.ex | Switches Broadway dispatch to atomic claim to avoid double-claiming. |
| lib/reencodarr/application.ex | Starts the WorkerSessions GenServer under the main supervision tree. |
| lib/reencodarr/ab_av1/worker_sessions.ex | New GenServer/ETS-backed worker session registry with expiry/touch/assignment. |
| lib/reencodarr/ab_av1/worker_protocol.ex | New helpers for protocol validation and server reply payloads. |
| lib/reencodarr_web/live/dashboard_live.ex | Displays worker socket URL and token fingerprint (without exposing the token). |
| lib/reencodarr_web/endpoint.ex | Mounts the worker socket at /workers/socket. |
| lib/reencodarr_web/channels/worker_socket.ex | Implements token-authenticated socket connect and worker socket ID. |
| lib/reencodarr_web/channels/worker_channel.ex | Implements worker channel join/announce/heartbeat/work request/disconnect behavior. |
| docs/ab_av1_worker_websocket.md | Documents initial worker WebSocket contract and payload shapes. |
| config/runtime.exs | Adds runtime worker-token config (env var or derived fallback). |
| .serena/memories/tech_stack.md | Adds project “memory” notes (tech stack). |
| .serena/memories/task_completion.md | Adds project “memory” notes (finish gates). |
| .serena/memories/suggested_commands.md | Adds project “memory” notes (common commands). |
| .serena/memories/memory_maintenance.md | Adds project “memory” notes (maintenance guidance). |
| .serena/memories/core.md | Adds project “memory” notes (repo structure/entry point). |
| .serena/memories/conventions.md | Adds project “memory” notes (key conventions). |
| .gitignore | Ignores Serena cache directory. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Owner
Author
|
needs more type specs |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🤖 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 `@config/runtime.exs`:
- Around line 23-32: The worker token derivation in runtime config uses a raw
SHA-256 hash of secret_key_base concatenated with a label, which should be
replaced with an HMAC-based derivation. Update the worker_token logic in the
runtime config to use :crypto.mac(:hmac, :sha256, secret_key_base,
"ab-av1-worker-token") instead of :crypto.hash/2 with string concatenation,
while keeping the existing Base.url_encode64 formatting and the surrounding
System.get_env("REENCODARR_WORKER_TOKEN") / SECRET_KEY_BASE fallback behavior.
In `@docs/ab_av1_worker_websocket.md`:
- Around line 28-32: The websocket docs example reply is missing the protocol
version field, so update the expected payload in the documentation to match what
WorkerProtocol.accepted/1 actually returns. Locate the “Expected reply payload”
example in the ab_av1 worker websocket doc and change it to include
protocol_version alongside accepted, keeping the documented contract aligned
with the server response.
In `@lib/reencodarr_web/channels/worker_channel.ex`:
- Around line 55-67: Add a catch-all handle_in/3 clause in WorkerChannel to
handle unexpected worker events instead of letting FunctionClauseError crash the
channel process. Keep the existing handle_in("heartbeat", ...),
handle_in("request_work", ...), and handle_in("pull_work", ...) clauses, then
add a fallback that matches any other event and replies with a structured
WorkerProtocol.error-style response so unknown messages are safely rejected
without dropping the socket.
In `@lib/reencodarr_web/channels/worker_socket.ex`:
- Around line 13-18: The token check in worker_socket’s connect/3 uses a normal
pattern match on the fetched :worker_token, which can leak timing information.
Update the authentication branch in connect/3 to compare the provided token
against the stored token with Plug.Crypto.secure_compare/2, and keep the
assign(socket, :worker_id, worker_id()) path only when that constant-time
comparison succeeds.
In `@lib/reencodarr/ab_av1/worker_sessions.ex`:
- Around line 98-137: The `handle_call/3` clauses for `{:touch, ...}`,
`{:assign_video, ...}`, and `{:clear_video, ...}` in `worker_sessions` duplicate
the same lookup/update/reply flow. Extract that shared logic into a private
helper such as `update_session_reply/3` (or similar) that takes
`server_worker_id`, `state`, and an update function, and have each clause
delegate to it while preserving the existing `lookup_session/1`,
`put_session/1`, and `{:error, :unknown_worker_session}` behavior.
- Around line 71-96: The `handle_call({:register, attrs}, _from, state)` path is
crashing on missing keys because it and `build_session/2` rely on
`Map.fetch!/2`, which can take down the GenServer and destroy the private
ETS-backed sessions. Add upfront validation for all required fields in
`register/1` or make `build_session/2` return a safe error instead of raising,
and ensure malformed attrs reply with an error rather than crashing. Use the
`handle_call({:register, ...})` and `build_session/2` symbols to locate the
unsafe fetches and guard them consistently.
In `@lib/reencodarr/media/video_queries.ex`:
- Around line 191-202: The initial candidate lookup in
claim_next_video_for_crf_search_in_tx/1 only fetches one Video id, which leaves
claim_next_video_for_crf_search_in_tx/2 with no rest list to retry when the
guarded update loses contention. Change the query to select a small ordered
batch of analyzed Video ids instead of a single row, so the recursive fallback
in claim_next_video_for_crf_search_in_tx/2 can continue to the next-best
candidate when the first claim fails.
In `@test/reencodarr_web/live/dashboard_v2_live_test.exs`:
- Around line 85-95: The dashboard_v2_live_test test mutates process-wide
Application config while the test module is async, which can leak the temporary
worker token into concurrent tests. Make this test module or the specific "shows
the worker websocket token when configured" test non-async, or refactor the
token lookup used by the LiveView so it can be injected/isolated instead of
relying on Application.get_env/put_env during the test. Keep the restore logic
in on_exit, but ensure no other async test can observe the transient
worker_token value.
In `@test/reencodarr/ab_av1/worker_protocol_test.exs`:
- Around line 1-27: Add unit test coverage in WorkerProtocolTest for the
untested WorkerProtocol.parse_announcement/1 and WorkerProtocol.error/1 paths.
Add cases that exercise invalid/untrusted announcement payloads and verify
parse_announcement/1 handles them as expected, including the missing catch-all
behavior noted in worker_protocol.ex, and add assertions that error/1 maps
worker failures to the correct protocol payload. Use the existing
work_assigned/2 test as a pattern and keep the new tests focused on the public
protocol API used by worker_channel.ex.
In `@test/reencodarr/ab_av1/worker_sessions_test.exs`:
- Around line 12-85: The same worker session attrs map is duplicated across all
four tests, so extract it into a shared helper/fixture and reuse it. Update the
tests in WorkerSessionsTest to call a local helper like worker_session_attrs/1,
or move the shared data into test/support/fixtures.ex if that is the standard
test fixture location, so future changes to the session fields happen in one
place.
- Around line 1-86: Add tests in WorkerSessionsTest to cover the untested error
branches: verify WorkerSessions.register/1 rejects a second registration with
the same server_worker_id or client_worker_id and returns {:error,
:duplicate_worker_id}, and add separate assertions that touch/1, assign_video/2,
and clear_video/1 each return {:error, :unknown_worker_session} when called with
a non-existent worker session identifier. Use the existing WorkerSessions
setup/reset pattern and the same register/list helpers to keep the new tests
consistent with the current module.
In `@test/reencodarr/broadway_producers_test.exs`:
- Around line 119-146: Add a test for the CRF claim path that asserts the same
`:video_mutated` PubSub broadcast behavior already covered for the analyzer
claim flow. Extend `test/reencodarr/broadway_producers_test.exs` near the
existing `CrfProducer.handle_demand/2` tests by invoking
`claim_next_video_for_crf_search` indirectly through `CrfProducer`, then verify
the broadcast payload alongside the state transition. Use
`VideoQueries.claim_next_video_for_crf_search/1` and the existing broadcast
assertion helpers/pattern from the analyzer coverage as the reference for
parity.
🪄 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: ASSERTIVE
Plan: Pro Plus
Run ID: 7cd47dea-01fd-4913-821a-94612a8358b9
📒 Files selected for processing (26)
.gitignore.serena/memories/conventions.md.serena/memories/core.md.serena/memories/memory_maintenance.md.serena/memories/suggested_commands.md.serena/memories/task_completion.md.serena/memories/tech_stack.mdconfig/runtime.exsdocs/ab_av1_worker_websocket.mdlib/reencodarr/ab_av1/worker_protocol.exlib/reencodarr/ab_av1/worker_sessions.exlib/reencodarr/application.exlib/reencodarr/crf_searcher/broadway/producer.exlib/reencodarr/diagnostics.exlib/reencodarr/media.exlib/reencodarr/media/video_queries.exlib/reencodarr_web/channels/worker_channel.exlib/reencodarr_web/channels/worker_socket.exlib/reencodarr_web/endpoint.exlib/reencodarr_web/live/dashboard_live.extest/reencodarr/ab_av1/worker_protocol_test.exstest/reencodarr/ab_av1/worker_sessions_test.exstest/reencodarr/broadway_producers_test.exstest/reencodarr_web/channels/worker_channel_test.exstest/reencodarr_web/live/dashboard_v2_live_test.exstest/support/channel_case.ex
mjc
force-pushed
the
mjc/reenc-133-worker-websocket
branch
from
ca5ec4b to
882c916
Compare
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
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.
Adds the ab-av1 worker websocket path on the Phoenix endpoint with token auth, protocol validation, session tracking, and job assignment replies.
Also switches local CRF Broadway dispatch to the same atomic claim path so remote websocket workers and local Broadway cannot claim the same analyzed video twice.
Verification:
Summary by CodeRabbit