[next] [5] Rework stream architecture with lifecycle state machine and latency optimizations#593
Merged
patrickelectric merged 35 commits intoApr 16, 2026
Conversation
This was referenced Mar 26, 2026
joaoantoniocardoso
force-pushed
the
pr/stream-architecture-overhaul
branch
2 times, most recently
from
ea151a8 to
fa8166b
Compare
joaoantoniocardoso
changed the title
joaoantoniocardoso
force-pushed
the
pr/stream-architecture-overhaul
branch
8 times, most recently
from
16d6a50 to
e8aed54
Compare
Updates vergen-gix from 1.0.9 to 9.1.0 (major version bump). The new version reorganizes sub-crate versioning (vergen, vergen-lib now 9.1.0).
Adds gstreamer-rtp 0.25 bindings. Required by the new playout-delay RTP header extension probe in the WebRTC sink.
Adds proptest 1.x for property-based testing. Used by the lifecycle state machine tests.
joaoantoniocardoso
force-pushed
the
pr/stream-architecture-overhaul
branch
3 times, most recently
from
5b0df49 to
f2edbf4
Compare
Removes redundant `&camera` borrows where `camera` is already a reference. Pure clippy-style cleanup with no behavioral change.
Removes `.map(|zid| zid)` identity closure. No behavioral change.
Removes the wrap_fn debug middleware and the actix_web::middleware::Logger layer. TracingLogger already covers request logging; the removed layers added duplicate noise and imported actix_service::Service unnecessarily.
Adds #[serial] to two tests that were missing it, preventing race conditions with global state in parallel test execution.
joaoantoniocardoso
force-pushed
the
pr/stream-architecture-overhaul
branch
4 times, most recently
from
7a59bc5 to
21070bd
Compare
Adds four new CLI flags:
--enable-dot: gates the /dot WebSocket endpoint
--rtsp-port PORT (default 8554): configurable RTSP server port
--disable-onvif: skips ONVIF camera discovery
--enable-realtime-threads: enables SCHED_RR for GStreamer threads
(requires CAP_SYS_NICE)
Also fixes Args::parse to use parse_from(["mavlink-camera-manager"]) in
cfg(test) so unit tests don't consume process argv.
Adds two functions:
lower_thread_priority(): sets nice 10 so GStreamer RT threads are
preferred by the OS scheduler.
lower_to_background_priority(): sets SCHED_OTHER + nice 19 for
background tasks (e.g. thumbnail pipelines).
Both are no-ops on non-Linux targets.
Member
|
can you squash the fixup commits ? |
Introduces a lock-free atomic state machine with four phases (Idle,
Waking, Running, Draining) and a consumer reference count packed into
a single AtomicU32.
Key API:
add_consumer(notify): Idle->Waking (wakes pipeline), or increments
count. Draining->Running on reconnect.
remove_consumer(is_lazy): decrements count. Running->Draining when
last lazy consumer leaves.
transition_to_running() / transition_to_idle(): phase transitions
driven by the stream watcher.
handle_pipeline_error(): exponential backoff (1s..60s) for pipeline
creation failures.
The state machine enables lazy-idle behavior: streams start their
pipeline only when a consumer (WebRTC session, RTSP client, thumbnail
request) connects, and tear down back to Idle when all consumers leave
after a grace period.
Includes comprehensive property-based (proptest) and unit tests.
Extends try_set_property to resolve enum-typed properties from string nicks (e.g. "downstream") via EnumClass::to_value_by_nick, in addition to the existing i32 path. This allows callers to set enum properties using human-readable nick strings without manual conversion.
Adds three public helpers for runtime GStreamer pipeline surgery:
excise_single_element(): surgically unlinks a passthrough element from
its chain and relinks upstream/downstream neighbours, then cleans up
on a background thread. Used by WebRTC send-path optimization and
proxysrc queue removal.
excise_proxysrc_queue(): one-shot BLOCK_DOWNSTREAM probe that excises
the internal queue inside proxysrc, removing one frame of intermediate
buffering latency. Used by UDP and Zenoh sinks.
bypass_jitterbuffer(): hooks deep-element-added to detect and excise
rtpjitterbuffer elements created by rtspsrc's internal rtpbin after a
5-second delay. Eliminates receive-side buffering for relay pipelines
where retransmission is not needed. Also forces do-retransmission=false
on any rtspsrc elements. Used by ONVIF pipelines.
wait_for_element_state(): generic polling wait for any GStreamer
element (not just pipelines) to reach a target state, using
current_state() checks. Used for reliable webrtcbin Null teardown
where state(timeout) can return prematurely on some GStreamer
versions.
Adds dump_bin_elements() which logs every element in a GStreamer bin with its factory name, interesting properties (leaky, latency, sync, etc.), pad caps, and peer connections. Useful for diagnosing pipeline topology at runtime. Gated behind --enable-dot in callers.
…ding detection Extracts teardown_probe_pipeline() which uses spawn_blocking + 5s timeout to prevent rtspsrc from blocking the async runtime during set_state(Null). Reduces encoding detection timeout from 15s to 3s for faster stream configuration probing. Fixes get_capture_configuration_using_encoding() to include encoding-name in the caps filter and tolerate slow set_state(Playing) transitions. Disables do-retransmission on rtspsrc in probe pipelines and adds async=false to probe fakesinks to avoid state-change deadlocks.
…tory
Replaces the shmsrc-based RTSP media pipeline with an appsrc-based
bridge. The factory now receives encoded video frames directly from the
source pipeline's video tee via an AppSrc, eliminating the shared-memory
socket and the depay/repay round-trip.
Changes:
- Uses cli::manager::rtsp_server_port() instead of hardcoded 8554
- Adds has_factory() for checking existing mounts during lazy resume
- Adds port() accessor for dynamic port retrieval
- Caps are now derived from the first sample rather than pre-set,
allowing the appsrc to adapt to runtime encoding changes
- Adds force_rtsp_transport_sinks_sync_false() on media state change
to disable sync on RTSP transport sinks for zero-latency delivery
- Adds media_configure callback that wires the AppSrc and installs
an unprepared handler for client disconnect tracking via
RtspFlowHandle
- Factory launch strings now use "appsrc name=source" followed by
the appropriate payloader, matching video/x-h264, video/x-h265,
video/x-raw, and image/jpeg caps
- Handles server.attach() failure gracefully by logging and retrying
instead of panicking
…ycle integration
Major rewrite of the RTSP sink to work with the new appsrc-based RTSP
media factory and the lifecycle state machine.
Changes:
- Replaces shmsink with an appsink that pushes video frames to the
RTSP factory's AppSrc via shared Arc<Mutex<Option<AppSrc>>>
- Clones appsrc out of the lock before pushing samples to minimize
lock contention
- Derives caps from incoming samples with needs_caps_update() check,
updating the appsrc caps only when they change
- Adds RtspFlowHandle with valve-based data flow control and consumer
counting integrated with the lifecycle state machine. The valve
blocks data when no RTSP clients are connected.
- Adds RtspSinkPersistent struct to carry shared state (appsrc,
pts_offset, flow_handle) across pipeline recreations during lazy
resume, so the RTSP factory and connected clients survive pipeline
teardown/recreation cycles.
- PTS offset tracking ensures continuous timestamps across pipeline
wake cycles, preventing timestamp discontinuities that would cause
RTSP clients to stall.
…ppsink teardown Removes rtp_queue_max_time_ns() and FALLBACK_RTP_QUEUE_TIME_NS since queue timing is now handled per-sink or eliminated entirely. Updates create_rtsp_sink() to accept lifecycle and persistent state parameters for the new appsrc-bridge RTSP sink. Reworks unlink_sink_from_tee(): scopes the BLOCK_DOWNSTREAM probe tightly to just the pad release (not the full teardown), then sends EOS directly to webrtcbin elements while still in the pipeline so ICE/DTLS can flush cleanly. Waits for webrtcbin to truly reach Null via wait_for_element_state() before pipeline removal, preventing thread leaks caused by premature removal. Uses a temp pipeline only for non-webrtcbin elements. Adds force_sync_false_on_element() and force_sync_false_on_element_tree() helpers to disable sync on all pipeline sinks, called during link_sink_to_tee for consistent zero-latency behavior.
…ueue excision Removes the explicit queue element between the tee and proxysink. Buffering is now handled by the proxysrc internal queue with minimal settings (max-size-buffers=1, no time/byte limits). Installs a one-shot pad probe on the proxysrc queue's src pad that excises the queue after the first buffer flows through, eliminating one frame of intermediate buffering latency. The proxysink is now linked directly to the tee.
… queue excision Same pattern as the UDP sink: removes the explicit queue element, configures the proxysrc internal queue to minimal settings, and installs an excision probe to remove it after the first buffer flows. Also adds async=false, enable-last-sample=false, qos=false, and leaky-type=downstream to the appsink for lower end-to-end latency.
…down
Major improvements to the WebRTC sink for lower latency and more
reliable session lifecycle:
Teardown:
- Adds Drop impl that releases the webrtcbin request pad, sets
webrtcbin to Null, and polls wait_for_element_state() to ensure
internal threads fully exit before the element is freed.
- Disables UPnP/IGD on the underlying NiceAgent (upnp=false) to
prevent gupnp-igd SSDP discovery threads from leaking.
- EOS is now a no-op; unlink_sink_from_tee sends EOS directly and
the Drop handler handles final cleanup. The previous post_message
approach caused the bus watcher to kill the entire pipeline.
Negotiation:
- Sets codec-preferences on the transceiver from upstream caps so
on-negotiation-needed fires without waiting for buffer caps.
- Installs a pre-excision BLOCK probe on the queue src pad to prevent
buffers from reaching RED/FEC/RTX encoders before they are excised.
- Retains H265 sprop-vps/sps/pps in SDP fmtp alongside H264
sprop-parameter-sets for correct codec initialization.
Latency:
- Adds playout-delay RTP header extension (min=0, max=0) via pad
probe and SDP extmap, telling browsers to render immediately.
- Strips FEC/RED payload types from the SDP offer to avoid
unnecessary packet processing.
- On Connected: excises rtpulpfecenc, rtpredenc, rtprtxsend, disables
sync on internal sinks, excises the queue, and sends ForceKeyUnit
upstream for an immediate fresh keyframe.
- Disables RTP storage (was 1s) since retransmission is not used.
Reliability:
- Enables ICE keepalive-conncheck for faster peer-loss detection.
- Reworks FailSafeKiller to use recv_timeout(10s) instead of
sleep(9s) + recv_timeout(1s), so it exits immediately when the
session is torn down or the peer connects.
- Preserves original H264 profile level from source caps instead of
hardcoding constrained baseline.
Adapts the pipeline module for the new appsrc-based RTSP bridge and
the lifecycle state machine:
- Routes RTSP sink through video_tee instead of rtp_tee, since the
appsrc bridge receives encoded video frames, not RTP packets.
- Reuses existing RTSP factory during lazy-resume recreation via
RTSPServer::has_factory(), so connected RTSP clients survive
pipeline teardown/recreation cycles.
- Gets caps from video_tee or capsfilter instead of rtp_tee for
RTSP factory creation.
- Passes new appsrc/pts_offset/flow_handle arguments to
RTSPServer::add_pipeline().
- On sink removal: preserves RTSP factory if
should_preserve_factory() instead of always stopping it.
- Removes the old "set pipeline to Null when no consumers" logic,
which is now handled by the lifecycle state machine.
…ypass and source-info Changes rtspsrc parameters: buffer-mode=none, do-retransmission=false, udp-buffer-size=2621440 for lower latency and no retransmission overhead. Adds a RawRtpTee that tees raw RTP before depay, then re-depays, parses (h264parse/h265parse with config-interval=-1 for header re-injection), and re-pays for the rtp_tee branch. This preserves source-info metadata (which is lost by depay) for RTP header extensions that need original source information. Adds source-info=true to depay/pay elements and perfect-rtptime=false to payloaders so timestamps are derived from the pipeline clock rather than RTP timestamps. Removes the hardcoded H265 profile=main caps constraint, allowing any profile from the source. Calls bypass_jitterbuffer() on the constructed pipeline to excise rtpjitterbuffer elements after a delay.
…ers and encoding detection Same rtspsrc parameter changes as ONVIF: buffer-mode=none, do-retransmission=false, udp-buffer-size=2621440. Adds encoding-name to RTP source caps for explicit codec negotiation and removes hardcoded profile=main from the H265 capsfilter to avoid rejecting non-main-profile streams. Adds RawRtpTee for source-info preservation across depay/repay with parse elements (config-interval=-1) for header re-injection, adds source-info=true to depay/pay elements, and perfect-rtptime=false to payloaders.
…tartup Simplifies start-command wait: removes tokio::select! with the finish channel. During initial pipeline creation, add_sink may set the pipeline to Playing before the start command arrives; residual bus messages (e.g. EOS from a previous StreamState teardown) must not kill the runner before it is ready. Increases max_lost_ticks from 30 to 150, making the stuck-position detector more tolerant of transient stalls (e.g. during rtspsrc reconnection). Skips position tracking when pipeline is not in Playing state, resetting lost_ticks to avoid false positives during state transitions. EOS filtering: ignores EOS messages from child elements (e.g. webrtcbin during dynamic sink removal). Only pipeline-level (aggregated) EOS triggers runner shutdown. Error filtering: ignores not-linked errors from rtspsrc (expected for unconsumed audio streams) and errors from webrtcbin internals (expected during normal WebRTC disconnection). When --enable-dot is active: dumps pipeline elements on PLAYING state change for runtime topology inspection.
Integrates the lifecycle state machine into the Stream struct and
rewrites the watcher loop from a simple "restart when dead" approach
to a lifecycle-driven state machine.
New Stream fields: lifecycle, notify, thumbnail_cooldown, mavlink_camera,
active_webrtc_sessions. Moves mavlink_camera from StreamState to Stream
so it persists across idle/wake cycles.
Watcher state machine phases:
Idle: drops pipeline state, waits for notify from add_consumer.
Waking: tears down old pipeline (with 10s timeout), re-probes source
configuration, creates new StreamState, extracts persistent RTSP
handles, creates MavlinkCamera on first successful wake, transitions
to Running.
Running: monitors pipeline health via PipelineRunner, handles pipeline
errors with exponential backoff.
Draining: starts 5s idle grace timer; transitions to Idle if no
consumers reconnect within the grace period.
StreamState::try_new now accepts lifecycle, notify, and persistent RTSP
state parameters.
On Stream construction: adds an initial consumer then immediately
removes it for lazy streams (drains to Idle), or keeps it for non-lazy
streams (disable_lazy=true stays Running permanently).
Reworks stream manager methods for lifecycle-aware operation: add_session(): adds a lifecycle consumer first, then waits for the pipeline to reach Playing with position-advance verification. Handles Idle (full recreation), Waking (mid-recreation), and Running uniformly. Includes a 3s live-source grace period for streams where query_position doesn't advance (e.g. rtspsrc). On failure, removes the consumer to avoid leaked references. remove_session(): tracks sessions in active_webrtc_sessions HashSet. Double-removal (e.g. from both signalling cleanup and DTLS close callback) is now idempotent -- the second call skips the lifecycle consumer decrement. get_jpeg_thumbnail_from_source(): lifecycle-aware thumbnail with cooldown. Adds a consumer to wake the pipeline, waits for it to be Playing with advancing position, serves the thumbnail, then keeps the consumer alive for a 15s cooldown period so repeated thumbnail requests don't cycle the pipeline through Idle/Waking. Stream status now uses lifecycle.stream_status() for the state field and reads mavlink from the persistent Stream.mavlink_camera.
Tracks active sessions per WebSocket connection in Arc<Mutex<Vec<BindAnswer>>> so orphaned sessions can be cleaned up when the WebSocket disconnects unexpectedly (e.g. browser navigated away without sending EndSession). Sends EndSession notification to the client before server-side session removal so the frontend can react to server-initiated teardowns. Reduces sender-task timeout from 30s to 5s for faster disconnect detection. Uses tokio::select! to abort the counterpart task immediately when one exits, instead of waiting for both via tokio::join!. Signalling now lists streams in Idle and Running states (not just Running), so lazy-idle streams are visible and can be connected to.
Calls lower_thread_priority() on TURN server worker threads so GStreamer pipeline threads running at SCHED_RR realtime are always preferred by the OS scheduler.
BlueROV: uses RTSPServer::port() instead of hardcoded 8554 so the RTSP endpoint respects the --rtsp-port CLI flag. mod.rs: gates mod test behind #[cfg(feature = "webrtc-test")] so it only compiles when the test feature is active. test.rs: sets disable_lazy: true in the extended configuration so the WebRTC test stream stays alive permanently, bypassing the lazy idle behavior that would shut down the pipeline when no consumers are connected.
The /dot WebSocket endpoint now returns 404 unless the --enable-dot CLI flag is set. Prevents unnecessary GStreamer DOT graph generation in production deployments.
Removes the single-version WebRTC leak test workflow. Thread leak detection is now part of the integration test suite and the new multi-GStreamer compatibility CI covers the same ground across multiple versions.
joaoantoniocardoso
force-pushed
the
pr/stream-architecture-overhaul
branch
from
e7d3b47 to
cdf20c4
Compare
This was referenced May 21, 2026
Merged
Closed
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
2 participants
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.
Summary
Major rework of the stream pipeline architecture, introducing a lifecycle state machine for lazy pipeline management, WebRTC latency optimizations, and supporting infrastructure.
Review Progression
This PR is organized into 35 commits across 10 phases. Each phase builds on the previous one and can be reviewed as a unit.
Phase 1: Dependencies and tooling (commits 1-4)
Cargo-only changes. Adds
gst-rtp(RTP header extension support),proptest(lifecycle/property tests), and bumpsvergen-gix. Lock file updated last.Phase 2: Small independent cleanups (commits 5-8)
Four self-contained fixes across unrelated modules (mavlink, zenoh, server, settings). No streaming behavior changes. Quick to review.
Phase 3: CLI, helpers, and runtime setup (commits 9-11)
Adds the new CLI/runtime prerequisites:
--enable-dot,--rtsp-port,--disable-onvif,--enable-realtime-threads, and the feature-gated--enable-webrtc-task-testcompatibility hook, plus Linux thread-priority helpers and an explicit Tokio runtime builder replacing#[tokio::main].Phase 4: Stream types and lifecycle foundation (commits 12-13)
The core addition: a lock-free atomic lifecycle state machine with four phases (
Idle,Waking,Running,Draining) and consumer reference counting.StreamStatusStateis added to the stream types module.Phase 5: GStreamer utility additions (commits 14-17)
New helpers that later sink and pipeline commits depend on: enum-nick property setting, element excision, proxysrc queue removal, jitterbuffer bypass, pipeline debug dumping, and hardened probe-pipeline teardown.
Phase 6: RTSP server and sink rework (commits 18-19)
Replaces the
shmsrc/shmsinkshared-memory RTSP bridge with anappsrc/appsinkin-process bridge. The RTSP server now receives encoded frames directly from the video tee. The RTSP sink gains valve-based flow control and PTS offset tracking across wake cycles.Phase 7: Sink adaptations (commits 20-23)
Adapts all sink types to the new architecture:
sink/mod.rs: tighter unlink/EOS handlingudp_sink/zenoh_sink: RTP queue removal + proxysrc queue excision for lower latencywebrtc_sink: playout-delay extension, FEC/RED/RTX excision on connect, hardened teardown inDropPhase 8: Pipeline module rework (commits 24-27)
Routes RTSP through
video_teeinstead ofrtp_tee, addsRawRtpTeefor source-info preservation in ONVIF/redirect pipelines, hardens the bus watcher, and enables pipeline debug dumps behind--enable-dot.Phase 9: Stream core and manager integration (commits 28-29)
Integrates the lifecycle state machine into
Streamand rewrites the watcher loop from a simple restart loop to a four-phase lifecycle controller. The stream manager gets lifecycle-aware session handling and cooldown-based thumbnail behavior.Phase 10: WebRTC and rollout polish (commits 30-35)
Final integration work: WebRTC signalling/session cleanup, lower-priority TURN threads, BlueROV/test configuration updates,
/dotendpoint gating, removal of the superseded WebRTC leak workflow, and the final module-level polish commit.Test plan