moq-gst: replace moqsink with a GstAggregator-based core

[kixelated]

Summary

Rebuilds the moqsink element on GstAggregator instead of a bare gst::Element with a hand-rolled queue. This is an alternative core to the bounded-channel design in #1771: it keeps that PR's media foundation and replaces the transport seam.

It builds directly on @arielmol's commits from #1771 (the rewritten timeline, codec/caps handling, B-frame support, frame-size cap, EOS aggregation, and tests), so this PR is diffed against main rather than stacked on that branch. The final commit swaps the session core for Aggregator; the 14 commits beneath it are ArielM's and carry his authorship.

Why

The #1771 core put a bounded mpsc channel between the GStreamer streaming threads and the async session, for backpressure. A blocking send on that channel cannot be interrupted by FLUSH_START (which arrives out of band on another thread), so it would wedge a seek/flush/state-change. The fix there was a flush-cancellation apparatus: per-pad watch<bool> gates, an out-of-band FlushSignal, biased cancellable sends (send_or_flush), and per-pad generations to discard stale in-flight messages.

GstAggregator already owns per-pad input queues, FLUSH/EOS/SEGMENT handling, and a single aggregate thread, so none of that is needed. Funneling serialized events and buffers onto one thread also removes the generation bookkeeping and the cross-thread failure map.

A second simplification falls out: GroupProducer::write_frame is synchronous (it appends and evicts in memory, never blocking on the network). So the aggregate thread writes frames straight into the moq producers under a RUNTIME.enter() guard. There is no data channel at all — the async task only connects, holds the session, and reports status. Because the producers are created up front, frames buffered before the connection completes are sent once it does (a "future work" item in #1771, free here).

What changed

• sink/imp.rs — now a GstAggregator subclass. aggregate() drains each pad with pop_buffer() and writes; sink_event rejects unsupported caps; flush() re-anchors timelines; start/stop own the session and producers.
• sink/session.rs — slimmed from ~2035 to ~240 lines: just Status and a connect/lifecycle task.
• sink/pad.rs (new) — per-pad media state (caps -> producer, SEGMENT policy, frame import) extracted from the old Pad/PadSet, plus 19 hermetic unit tests.
• sink/timeline.rs — unchanged.
• sink/mod.rs — @extends gst_base::Aggregator.
• Cargo.toml — adds gstreamer-base (no version-feature bump; stays at the existing 1.14 baseline).

Deleted relative to the #1771 core: the bounded channel + DATA_CHANNEL_BOUND, send_or_flush/SendOutcome, FlushSignal + flush watches + toggle_pad_flush/reset_pad_flushes, per-pad generations, the Status failure map, PadSet, the manual chain/event pad functions, and the request-pad rollback dance.

Public surface / branch targeting

No public library API or wire change. Same element name (moqsink), sink_%u request pads, the seven codec caps, the three writable properties, and the three read-only diagnostics (connected, moq-version, estimated-send-bitrate). Internal to the rs/moq-gst plugin, so this targets main (same as #1771). One cosmetic change: the element now carries an unused src pad (it is SINK-flagged and never pushes).

Testing

• cargo test -p moq-gst: 31 unit + 3 element tests pass, including frame_through_h264_emits_a_frame (a real IDR access unit flows caps -> producer -> published track, hermetically).
• cargo clippy -p moq-gst --all-targets clean; cargo fmt --check clean.
• gst-inspect-1.0 moqsink: registers Sink/Network/MoQ, hierarchy +----GstAggregator, both pad templates GstAggregatorPad, all properties.
• Live relay: videotestsrc ! vtenc_h264 ! h264parse ! moqsink to a local moq-relay logged session connected -> finalized on EOS: ["sink_0", "catalog"] -> posting EOS -> clean session terminated.

Not yet validated (same boundary #1771 drew)

• A subscribe-back byte round-trip against a relay.
• A genuinely stalled sparse pad. Aggregator fires aggregate() when every pad has a buffer; for continuous A/V that is fine, but a stalled non-EOS pad could hold up others. If sparse data tracks matter, set_force_live(true) is the knob (raises the floor to GStreamer 1.22).
• FLUSH/seek on a live pipeline.

(Written by Claude)

[coderabbitai]

Note

Reviews paused

It 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 reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

The moqsink GStreamer element is rewritten from a tokio-channel/async-task architecture to a GstAggregator-based implementation. Three new internal modules are introduced: timeline.rs for pure timeline policy functions (classify_segment, frame_micros) using Result-based decision semantics; session.rs for async MoQ connection lifecycle with observable Status (connected, version, send_bitrate); and pad.rs for per-sink-pad media producer state machines (PadState, codec-specific Framed producer construction, flush/finalize). The aggregator core in imp.rs is rewritten to implement AggregatorImpl, draining buffers synchronously in aggregate(), adding per-buffer size enforcement via write_buffer(), and posting EOS once all sink pads have reached end-of-stream. Three read-only GObject properties (connected, moq-version, estimated-send-bitrate) are added and backed by Session::status(). Frame size enforcement is added to GroupProducer::append_frame in moq-net to reject oversized frames at the producer boundary. Integration tests and updated documentation validate the new behavior.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically describes the main architectural change: replacing moqsink with a GstAggregator-based implementation.
Description check	✅ Passed	The description thoroughly explains the rationale, design changes, and implementation details of replacing the bounded-channel design with a GstAggregator-based approach.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch claude/quirky-ramanujan-71c191

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 4

🧹 Nitpick comments (1)

rs/moq-gst/tests/element.rs (1)

19-68: 🧹 Nitpick | 🔵 Trivial | 🏗️ Heavy lift

Move these tests inline next to the sink implementation per repo test policy.

These are currently integration tests under tests/, but this repo’s Rust guidance requires colocated inline tests (#[cfg(test)] mod tests) in source files. Please move them near the corresponding sink module code.

As per coding guidelines, “Rust tests are integrated within source files” and “Inline Rust tests in source files as #[cfg(test)] mod tests within the same file.”

🤖 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 `@rs/moq-gst/tests/element.rs` around lines 19 - 68, Move the three test
functions request_and_release_sink_pads, missing_url_fails_state_change, and
connect_failure_posts_error_to_bus from the tests/element.rs file to the source
file containing the moqsink implementation. Create a cfg(test) mod tests module
within the source file and place all three test functions there with their
#[test] attributes intact. Remove the now-empty tests/element.rs file after
moving the tests.

Source: Coding guidelines

🤖 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 `@rs/moq-gst/Cargo.toml`:
- Line 23: The gst-base dependency specification on line 23 uses an explicit
version string instead of workspace-managed dependency specs. Replace the
explicit version specification with `{ workspace = true }` for the gst-base
package entry. Apply the same fix to the dependency entry on line 37 that is
also mentioned in the review comment. This ensures all dependencies in the
Cargo.toml file use workspace-pinned versions consistently.

In `@rs/moq-gst/src/sink/mod.rs`:
- Around line 9-10: Add a public doc comment above the glib::wrapper! macro that
defines the pub struct MoqSink. The doc comment should be a doc string (starting
with ///) that provides a clear description of what the MoqSink struct is and
its purpose in the module. This will satisfy the coding guideline requirement
that all exported public symbols must have documentation comments.

In `@rs/moq-gst/src/sink/pad.rs`:
- Around line 149-171: The issue is that when observe_segment rejects a segment
due to a discontinuity in the Active state (line 168 transition to Invalid), the
segment_info is not updated. On the next buffer with the same sticky segment,
the state check at line 156 sets prev to None (since state is Invalid), causing
classify_segment to accept the same rejected segment again. To fix this, update
self.segment_info to the rejected segment info in the SegmentDecision::Reject
branch before transitioning to Invalid state, so that the early return check at
lines 151-153 will catch and reject the same segment on subsequent calls.
Additionally, add a regression test that verifies after an Active to Invalid
transition from a rewinding segment, re-observing that exact same segment keeps
the pad Invalid and continues dropping frames.

In `@rs/moq-gst/src/sink/timeline.rs`:
- Around line 3-23: Add documentation comments to the public types SegmentInfo,
SegmentDecision, and FrameDecision, and to all their fields and enum variants,
since every exported public symbol must have a doc comment per coding
guidelines. Additionally, mark both SegmentDecision and FrameDecision enums with
the #[non_exhaustive] attribute to prevent downstream breakage when new variants
are added in the future, as required by public API guidelines for Rust code.

---

Nitpick comments:
In `@rs/moq-gst/tests/element.rs`:
- Around line 19-68: Move the three test functions
request_and_release_sink_pads, missing_url_fails_state_change, and
connect_failure_posts_error_to_bus from the tests/element.rs file to the source
file containing the moqsink implementation. Create a cfg(test) mod tests module
within the source file and place all three test functions there with their
#[test] attributes intact. Remove the now-empty tests/element.rs file after
moving the tests.

🪄 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: CHILL

Plan: Pro

Run ID: a5207109-a4f7-4fd6-8c13-b028d850611c

📥 Commits

Reviewing files that changed from the base of the PR and between 4c67357 and 77e71da.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (10)

• doc/bin/gstreamer.md
• rs/justfile
• rs/moq-gst/Cargo.toml
• rs/moq-gst/src/lib.rs
• rs/moq-gst/src/sink/imp.rs
• rs/moq-gst/src/sink/mod.rs
• rs/moq-gst/src/sink/pad.rs
• rs/moq-gst/src/sink/session.rs
• rs/moq-gst/src/sink/timeline.rs
• rs/moq-gst/tests/element.rs

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)

rs/moq-gst/src/sink/pad.rs (2)

106-117: ⚠️ Potential issue | 🟠 Major

Reject under-specified Opus caps instead of silently defaulting to stereo/48 kHz.

Lines 108–109 use unwrap_or(2) and unwrap_or(48_000) to mask missing Opus channels and rate caps fields. This advertises incorrect audio configuration for malformed caps. The pad template at imp.rs:256 does not constrain these fields, so require them here or pin them in the template.

🛠️ Proposed fix

-				let channels: i32 = structure.get("channels").unwrap_or(2);
-				let rate: i32 = structure.get("rate").unwrap_or(48_000);
-				let channel_count = u32::try_from(channels)
-					.with_context(|| format!("Opus caps has negative channel count {channels}"))?;
-				let sample_rate =
-					u32::try_from(rate).with_context(|| format!("Opus caps has negative sample rate {rate}"))?;
+				let channels: i32 = structure.get("channels").context("Opus caps missing channels")?;
+				let rate: i32 = structure.get("rate").context("Opus caps missing rate")?;
+				if channels <= 0 {
+					anyhow::bail!("Opus caps has non-positive channel count {channels}");
+				}
+				if rate <= 0 {
+					anyhow::bail!("Opus caps has non-positive sample rate {rate}");
+				}
+				let channel_count = channels as u32;
+				let sample_rate = rate as u32;

Add a test case (e.g., opus_caps_without_channels_fails) to prevent regression.

🤖 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 `@rs/moq-gst/src/sink/pad.rs` around lines 106 - 117, The Opus audio codec
configuration in the "audio/x-opus" match arm is silently defaulting to stereo
(2 channels) and 48 kHz sample rate when the channels and rate fields are
missing from the caps, which can propagate incorrect audio configuration.
Replace the unwrap_or(2) call on the channels extraction and the
unwrap_or(48_000) call on the rate extraction with proper error handling that
rejects under-specified caps instead of accepting them with defaults.
Additionally, add a test case (for example, opus_caps_without_channels_fails) to
verify that Opus caps without required fields are properly rejected and prevent
this regression in the future.

---

140-155: ⚠️ Potential issue | 🟠 Major

Refresh the stored segment when only start changes.

The equality check self.segment_info == Some(info) only compares time_format, rate, and base_nanos, but omits the start field. When a new TIME segment arrives with the same base and rate but a different media start, the early return leaves self.segment pointing to the old segment object. Subsequent calls to frame_timestamp() then invoke to_running_time_full(pts) with a stale start value, mapping the PTS incorrectly.

The fix allows Active pads to refresh self.segment by skipping the early return only when the state is Active, since only Active pads have a cached segment whose fields matter for timestamp mapping.

Proposed fix and regression test

-		// Re-observing the exact segment we last classified is a no-op. observe_segment runs on every
-		// buffer (the segment is sticky), so without this an Invalidated pad would re-anchor on the next
-		// buffer (Invalid -> prev=None -> classify accepts) and silently recover on the same rewound
-		// segment. Recording `info` even on reject is what makes this hold.
-		if self.segment_info == Some(info) {
+		// Re-observing a rejected/non-started segment is a no-op. Active segments still refresh
+		// `self.segment` because `SegmentInfo` omits fields like `start` that affect timestamp mapping.
+		if self.segment_info == Some(info) && self.state != PadState::Active {
 			return;
 		}

 	#[test]
 	fn moved_start_with_advancing_base_stays_continuous() {
 		gst::init().unwrap();
 		let mut pad = Pad::new();
 		pad.observe_segment(time_segment_at(0, 0));
 		assert_eq!(pad.state, PadState::Active);
 		pad.observe_segment(time_segment_at(30_000, 5_000));
 		assert_eq!(pad.state, PadState::Active);
 	}
+
+	#[test]
+	fn moved_start_with_equal_base_refreshes_timestamp_mapping() {
+		gst::init().unwrap();
+		let mut pad = Pad::new();
+		pad.observe_segment(time_segment_at(0, 5_000));
+		pad.observe_segment(time_segment_at(3_000, 5_000));
+		assert_eq!(
+			pad.frame_timestamp(Some(gst::ClockTime::from_mseconds(6_000))),
+			Ok(8_000_000)
+		);
+	}

🤖 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 `@rs/moq-gst/src/sink/pad.rs` around lines 140 - 155, The early return
condition comparing self.segment_info == Some(info) does not account for changes
in the start field of the segment info. When the state is Active and a new
segment arrives with the same base and rate but a different start value, the
early return prevents refreshing self.segment, leaving it pointing to the old
segment with an incorrect start value that breaks timestamp mapping in
frame_timestamp(). Modify the early return condition to allow Active pads to
skip the return and refresh their cached segment, while still returning early
for other pad states or when all fields including start remain unchanged.

🧹 Nitpick comments (2)

rs/moq-gst/src/sink/imp.rs (2)

120-126: 🧹 Nitpick | 🔵 Trivial | ⚡ Quick win

Add the required doc comment for MoqSink.

MoqSink is pub, so it needs a consumer-facing /// comment.

As per coding guidelines, “Every exported public symbol must have a doc comment: each pub Rust item … gets a doc comment.”

📝 Proposed doc comment

+/// Implements the `moqsink` GStreamer element's settings and live aggregator state.
 #[derive(Default)]
 pub struct MoqSink {

🤖 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 `@rs/moq-gst/src/sink/imp.rs` around lines 120 - 126, The `MoqSink` struct is a
public symbol that is missing the required doc comment. Add a consumer-facing
`///` doc comment above the `pub struct MoqSink` definition to document its
purpose and functionality. The doc comment should follow Rust documentation
conventions and provide a clear description of what this struct represents and
its role as a GStreamer sink element.

Source: Coding guidelines

---

63-82: 🧹 Nitpick | 🔵 Trivial | ⚡ Quick win

Skip failed pads before mapping and copying buffers.

Line 67 copies the buffer before Line 81 checks pad.is_failed(), so a pad that is already being dropped still pays the map/copy cost and can still fail the whole element on an unnecessary map error. Move the pad lookup/failure check before map_readable().

♻️ Proposed ordering fix

 	fn write_buffer(&mut self, agg_pad: &gst_base::AggregatorPad, buffer: gst::Buffer) -> Result<(), gst::FlowError> {
 		let caps = agg_pad.current_caps();
 		let segment = agg_pad.segment();
 		let pts = buffer.pts();
-		let map = buffer.map_readable().map_err(|_| {
-			gst::error!(CAT, "failed to map buffer on pad {}", agg_pad.name());
-			gst::FlowError::Error
-		})?;
-		let data = Bytes::copy_from_slice(map.as_slice());
 
 		// Disjoint field borrows: the pad entry borrows `pads` while observe_caps reads broadcast/catalog.
 		let Self {
 			broadcast,
@@
 			..
 		} = self;
-		let pad = pads.entry(agg_pad.name().to_string()).or_insert_with(Pad::new);
+		let pad_name = agg_pad.name();
+		let pad = pads.entry(pad_name.to_string()).or_insert_with(Pad::new);
 		if pad.is_failed() {
 			return Ok(());
 		}
+		let map = buffer.map_readable().map_err(|_| {
+			gst::error!(CAT, "failed to map buffer on pad {}", pad_name);
+			gst::FlowError::Error
+		})?;
+		let data = Bytes::copy_from_slice(map.as_slice());

🤖 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 `@rs/moq-gst/src/sink/imp.rs` around lines 63 - 82, The write_buffer function
maps and copies buffer data before checking if the pad has failed, which wastes
resources on pads that are already failed. Move the pad lookup using
pads.entry(agg_pad.name().to_string()).or_insert_with(Pad::new) and the
subsequent is_failed() check to occur before the buffer.map_readable() call, so
that failed pads can be skipped without incurring the cost of mapping and
copying the buffer data.

🤖 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.

Outside diff comments:
In `@rs/moq-gst/src/sink/pad.rs`:
- Around line 106-117: The Opus audio codec configuration in the "audio/x-opus"
match arm is silently defaulting to stereo (2 channels) and 48 kHz sample rate
when the channels and rate fields are missing from the caps, which can propagate
incorrect audio configuration. Replace the unwrap_or(2) call on the channels
extraction and the unwrap_or(48_000) call on the rate extraction with proper
error handling that rejects under-specified caps instead of accepting them with
defaults. Additionally, add a test case (for example,
opus_caps_without_channels_fails) to verify that Opus caps without required
fields are properly rejected and prevent this regression in the future.
- Around line 140-155: The early return condition comparing self.segment_info ==
Some(info) does not account for changes in the start field of the segment info.
When the state is Active and a new segment arrives with the same base and rate
but a different start value, the early return prevents refreshing self.segment,
leaving it pointing to the old segment with an incorrect start value that breaks
timestamp mapping in frame_timestamp(). Modify the early return condition to
allow Active pads to skip the return and refresh their cached segment, while
still returning early for other pad states or when all fields including start
remain unchanged.

---

Nitpick comments:
In `@rs/moq-gst/src/sink/imp.rs`:
- Around line 120-126: The `MoqSink` struct is a public symbol that is missing
the required doc comment. Add a consumer-facing `///` doc comment above the `pub
struct MoqSink` definition to document its purpose and functionality. The doc
comment should follow Rust documentation conventions and provide a clear
description of what this struct represents and its role as a GStreamer sink
element.
- Around line 63-82: The write_buffer function maps and copies buffer data
before checking if the pad has failed, which wastes resources on pads that are
already failed. Move the pad lookup using
pads.entry(agg_pad.name().to_string()).or_insert_with(Pad::new) and the
subsequent is_failed() check to occur before the buffer.map_readable() call, so
that failed pads can be skipped without incurring the cost of mapping and
copying the buffer data.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 731d90f2-5e0b-488a-a480-04c9943efbcc

📥 Commits

Reviewing files that changed from the base of the PR and between 3cfbbc9 and 585b92f.

📒 Files selected for processing (5)

• rs/moq-gst/src/sink/imp.rs
• rs/moq-gst/src/sink/pad.rs
• rs/moq-gst/src/sink/timeline.rs
• rs/moq-net/src/model/frame.rs
• rs/moq-net/src/model/group.rs

✅ Files skipped from review due to trivial changes (1)

• rs/moq-net/src/model/frame.rs

[arielmol]

A genuinely stalled sparse pad. Aggregator fires aggregate() when every pad has a buffer... set_force_live(true) is the knob (raises the floor to GStreamer 1.22)

I'd gently unbundle from "same boundary." it isn't an unvalidated edge in #1771 it's structurally absent. the Fanout in Element has no aggregate() barrier, each pad's chain runs independently to its own track, so a stalled non-EOS pad never holds up the others. there's nothing to force_live around (no need to upgrade gst to 1.22+).

subscribe-back byte round-trip against a relay & FLUSH/seek on a live pipeline

Indeed on both, what I do have is live end2end from about 15+ of the fleet (x86/aarch64) so transport is exercised end to end. I just haven't asserted byte equality, but doable, ... will do! 😄

[arielmol]

Update on the byte round-trip for #1771 done. I ran a VP8 payload round-trip: videotestsrc(ball) -> vp8enc -> moqsink -> relay -> moqsrc, capturing encoded buffers before moqsink and after moqsrc. The harness writes one file per GStreamer buffer and checks received hashes against a contiguous slice of the published hashes.

150 buffers published, all 150 distinct; 149 received; every received buffer matched the published VP8 payload byte-for-byte and in order. In this run the missing buffer was the tail not draining before teardown, not a payload mismatch.