fix: codec enforcement, publisher stats, stream deletion, and thread shutdown#26
Conversation
…down - rtmp_bridge.rs: Add stream_id and allowed_codecs fields to ConnState. on_publish() stores them from the DB stream row. on_frame() enforces allowed_codecs list (comma-separated, case-insensitive) and returns false to kick connections sending a disallowed codec. Fix double-lock bug in on_play() where two sequential Mutex::lock() calls on a non-reentrant mutex would deadlock. Add update_publisher_stats() that rate-limits DB writes to 1/second and calculates bitrate from bytes delta. Add stream_id_for_conn() accessor used by the poll loop. - server.rs: Thread shutdown via Arc<AtomicBool> + JoinHandle::join() after axum::serve completes. Share Arc<Mutex<HashSet<String>>> deleted_streams between AppState and the RTMP poll loop. TrackedConn gains stream_id, video_codec, audio_codec fields. Per-poll: pick up detected codecs from Conn fields, call rtmp_bridge.on_frame() for codec enforcement, call update_publisher_stats() for active publishers, kick connections whose stream_id appears in the deleted set. - http.rs: Add deleted_streams Arc<Mutex<HashSet<String>>> to AppState. handle_stream_delete() inserts the stream ID into deleted_streams after a successful DB delete so the RTMP thread evicts live connections. - Cargo.toml: Point librtmp2 dependency at the feature branch that contains the new Conn fields (detected_video_codec, detected_audio_codec, on_connect_cb) and the fixed Transport/TLS implementation. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
|
Warning Review limit reached
Next review available in: 34 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. 📝 WalkthroughWalkthroughAdds shared stream-deletion signaling between HTTP and RTMP, enforces per-connection codec allow-lists, tracks stream and codec state per RTMP connection, rate-limits publisher stats updates, coordinates RTMP thread shutdown, and pins the ChangesStream eviction and codec enforcement
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
- Reorder imports alphabetically (atomic before sync) - Collapse deleted_now chain onto two lines per rustfmt line-length rule - Collapse single-arg log_warn! onto one line Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/server.rs (1)
332-340: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAlways stop and join the RTMP thread on HTTP server errors.
The
?on Line 335 returns before Lines 338-340, so an HTTP serve error skips the new RTMP shutdown path and leaves the background thread running.Proposed shutdown ordering
- axum::serve(http_listener, app) + let http_result = axum::serve(http_listener, app) .with_graceful_shutdown(shutdown_signal()) .await - .map_err(|e| format!("HTTP server error: {e}"))?; + .map_err(|e| format!("HTTP server error: {e}")); crate::log_info!("Shutting down..."); rtmp_stop.store(true, Ordering::Relaxed); let _ = rtmp_thread.join(); crate::log_info!("RTMP thread joined."); + http_result?; Ok(())🤖 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 `@src/server.rs` around lines 332 - 340, Ensure the RTMP shutdown path in the HTTP server startup flow always runs, even when axum::serve in the server entrypoint returns an error. In the code around the serve call and the subsequent shutdown steps, avoid using ? directly on the result before the RTMP stop/join logic; instead, capture the error from axum::serve, run the existing rtmp_stop.store and rtmp_thread.join cleanup in all cases, then return or propagate the HTTP server error afterward. Reference the server startup/shutdown flow in src::server and the shutdown_signal/rtmp_thread handling so the background thread cannot be left running on HTTP failures.
🤖 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 `@src/server.rs`:
- Around line 183-185: The deleted-stream tracking in `src/server.rs` is only
being snapshotted in the poll loop and never cleared, so processed IDs remain
forever and can affect recreated streams. Update the logic around the
`deleted_streams` snapshot and the connection rejection/removal paths to drain
or remove the IDs once they have been handled, using the existing
`deleted_streams` set and the related poll-loop code that checks `deleted_now`
and closes rejected connections. Keep the removal tied to the point where stale
connections are actually dropped so each deletion marker is processed once.
- Around line 170-178: The shutdown path in the RTMP poll loop skips cleanup, so
active entries in tracked remain open in the DB. Update the poll-thread logic in
server.rs to invoke rtmp_bridge.on_close for every still-tracked connection
before breaking when rtmp_stop_clone is set, using the existing tracked
collection and rtmp_bridge handle in the poll loop. Ensure the cleanup runs on
the stop branch before server.stop() exits the loop, and keep the existing
server.poll error handling unchanged.
- Around line 273-286: The codec enforcement in the publishing path only checks
entry.video_codec and skips entry.audio_codec, so disallowed audio can still
pass through. Update the codec-policy block in server::handle_publish (or the
surrounding loop that builds FrameInfo and calls rtmp_bridge.on_frame) to
perform the same allow-list check for audio as for video, using
entry.audio_codec and FrameKind::Audio, and keep the existing
reject_indices/continue flow when on_frame returns false.
---
Outside diff comments:
In `@src/server.rs`:
- Around line 332-340: Ensure the RTMP shutdown path in the HTTP server startup
flow always runs, even when axum::serve in the server entrypoint returns an
error. In the code around the serve call and the subsequent shutdown steps,
avoid using ? directly on the result before the RTMP stop/join logic; instead,
capture the error from axum::serve, run the existing rtmp_stop.store and
rtmp_thread.join cleanup in all cases, then return or propagate the HTTP server
error afterward. Reference the server startup/shutdown flow in src::server and
the shutdown_signal/rtmp_thread handling so the background thread cannot be left
running on HTTP failures.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: cc086a34-52ea-4a9e-aec7-1843b7d0cbb1
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (4)
Cargo.tomlsrc/http.rssrc/rtmp_bridge.rssrc/server.rs
- Capture axum::serve result before stopping/joining the RTMP thread so HTTP server errors still trigger a clean shutdown instead of leaving the background thread running indefinitely. - After the poll loop exits (stop signal or poll error), call on_close for every connection still in the tracked map so the bridge can deactivate publishers and players in the database. - Drain deleted_streams after each eviction cycle: retain only IDs that are still referenced by a live connection, preventing the set from growing without bound across many create/delete cycles. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
![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: c2719884c4
ℹ️ 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".
| crate::log_info!("Stream deleted: {id}"); | ||
| // Signal the RTMP poll loop to evict any active connections for | ||
| // this stream. The set is drained once no connections remain. | ||
| state.deleted_streams.lock().unwrap().insert(id); |
There was a problem hiding this comment.
When an operator deletes a stream and later recreates the same stream id, this insertion leaves a permanent tombstone: the RTMP loop only snapshots deleted_streams and never removes ids from it. As soon as the new publisher/player is accepted and entry.stream_id is set, the deletion check in server.rs still matches the old id and immediately kicks every connection for the recreated stream.
Useful? React with 👍 / 👎.
| } | ||
|
|
||
| // Enforce codec policy via the bridge (which checks allowed_codecs). | ||
| if !entry.video_codec.is_empty() { |
There was a problem hiding this comment.
For streams whose policy excludes an audio codec, or for audio-only publishers, this branch never checks entry.audio_codec even though detected_audio_codec was read above. That means a publisher can use a disallowed audio codec such as AAC/mp4a and remain connected as long as the video codec is allowed, so allowed_codecs is only partially enforced.
Useful? React with 👍 / 👎.
Picks up the transport.rs, client/mod.rs, and session/conn.rs fixes pushed to librtmp2 claude/openrtmp-bugs-risks-amf3n6. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
Adds /// doc lines to DbRtmpBridge::new and http::router, which were the only public functions lacking documentation. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
Fills the remaining gaps in public-API docstring coverage for the RtmpEventHandler trait. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcC3LrUD3aeyyHpa9r7AuE
Summary
allowed_codecsnot enforced:ConnStateinrtmp_bridge.rsnow storesallowed_codecsandstream_idfrom the DB stream row aton_publishtime.on_frame()checks the incoming codec against the comma-separated list (case-insensitive) and returnsfalseto kick the connection if the codec is not permitted. The poll loop inserver.rsreadsdetected_video_codec/detected_audio_codecfromlibrtmp2::session::Connand callson_frame()with the detected codec.on_play()previously lockedself.connstwice sequentially, which deadlocks a non-reentrantMutex. Fixed by using a single lock guard for both the player insertion and thestream_idassignment.update_publisher_stats()toDbRtmpBridge. It calculatesbitrate_kbpsfrom bytes delta over elapsed time, rate-limits DB writes to once per second, and updatesvideo_codec/audio_codec/bytes_inin the publisher row. The RTMP poll loop calls this after each poll iteration for publishing connections.Arc<Mutex<HashSet<String>>> deleted_streamsshared betweenAppStateand the RTMP background thread.handle_stream_delete()inserts the stream ID into the set; the poll loop evicts any connection whosestream_idappears in it. AddedArc<AtomicBool>shutdown signal andJoinHandle::join()so the RTMP thread exits cleanly when the HTTP server shuts down.librtmp2dependency atclaude/openrtmp-bugs-risks-amf3n6branch which addsdetected_video_codec,detected_audio_codec, andon_connect_cbtoConn, and fixes the TLS transport.Test plan
cargo +stable test— all 28 tests passcargo +stable build— no errors/statsendpoint shows non-zerobitrate_kbpsandvideo_codecfor a live publisherDELETE /api/v1/streams/:idwhile a publisher is live → publisher connection dropped within 50 msSIGTERM→ RTMP thread joins cleanly (no zombie thread or fd leak)Generated by Claude Code
Summary by CodeRabbit
librtmp2dependency to use a different git branch.