moq-gst: replace moqsink with a rewritten, unit-tested core

[arielmol]

Summary

Replaces the moqsink element with a rewritten core that separates the timeline,
per-pad state, and session lifecycle into independently testable pieces. The old sink
interleaved connection, pads, timing, finalization, and errors, with no dedicated unit
tests and ~16% line coverage. The new sink module reaches ~83% line coverage
(timeline 100%, session 88%); the moq-gst crate's suite is 65 unit + 3 element tests.

It is a compatible superset of the old element: same element name, same sink_%u
request pads, the same seven codec caps (H.264/H.265 byte-stream/au, AV1, VP8, VP9,
AAC raw, Opus), and the same three writable properties (url, broadcast,
tls-disable-verify). It adds three read-only diagnostics: connected, moq-version,
estimated-send-bitrate.

New behavior: shared running-time continuity (no per-pad anchoring), explicit pad
membership with per-pad generations, per-pad error isolation, B-frame support
(decode-order frames, no PTS monotonicity filter), a frame-size cap, and FLUSH handling
(timeline discontinuity + cancellation of blocked sends).

History note: the branch builds the core as a parallel moqsinkspike element, then
promotes it to moqsink (rename + remove the old), then applies post-review polish.
Squash-merge is fine.

Behavior differences vs the old moqsink (intentional, not regressions)

• Stricter caps. AAC requires mpegversion=4 + stream-format=raw; H.264/H.265
  require stream-format=byte-stream + alignment=au; a non-unit-rate SEGMENT
  invalidates the timeline. The old sink validated none of this in code (only the pad
  template), so it accepted inputs the rewrite now rejects per-pad. The seven codec
  types are unchanged.
• Absolute running-time origin. Timestamps use shared running time (for A/V
  alignment) instead of the old per-pad reference_pts rebasing, so a late-joining live
  pipeline emits a large first timestamp rather than ~0.

Testing

• Unit + lint: 65 unit + 3 element tests (crate-wide), cargo clippy -D warnings,
  cargo fmt.
• Live e2e: a fleet of senders (x86_64 + aarch64) publishing to a real relay;
  subscribing back confirmed decoded frames with no errors.
• Transport failure: connect-failure (hermetic element test); mid-stream relay death
  (SIGKILL the local relay → session error on the bus + non-zero exit ~35 s after the
  kill, consistent with QUIC's ~30 s idle timeout for a silent peer; a graceful close is
  instant; this latency is shared with the old sink).
• Memory-leak, local soak (dev box, 35 min): x264 -> moqsink in isolation held RSS
  flat (24 consecutive identical samples) with zero leaked GstMiniObjects at teardown.
• Memory-leak, overnight bisection (target node, about 12-14 h): msdk -> moqsink vs
  msdk -> fakesink (differ only in the sink) showed identical flat RSS slopes
  ( about +0.07 KB/min, R^2~=0.1), so moqsink adds no measurable leak over the encoder.

Notes

• Public API: no change to any pub item in library crates; internal to rs/moq-gst
  (a plugin), plus three property rows in doc/bin/gstreamer.md.
• Targets main (no wire or public library API change).
• Known/accepted: moq_net::Session::closed() always returns Err, so a clean remote
  close surfaces as a bus ERROR rather than EOS.
• Future work (out of scope): draining to the local broadcast during connect, non-VCL
  NAL accumulation in moq-mux::import, Framed::discontinuity() for an AU split across
  buffers, stats/observability, a latency benchmark harness.

(Co-Written by Claude)

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 104ac536-2dc0-451a-ba16-4a515d2f43f1

📥 Commits

Reviewing files that changed from the base of the PR and between a858308 and 50c6f5d.

📒 Files selected for processing (1)

• rs/moq-gst/Cargo.toml

---

Walkthrough

The moq-gst sink element is refactored from a monolithic in-file async implementation into a GObject shell (imp.rs) that delegates all networking and publishing to a new async session.rs worker module. Pure segment/frame timestamp classification is extracted into a new stateless timeline.rs module. MoqSink gains per-pad generation counters, generation-keyed flush watches, and a shared Status handle; these are used to implement safe flush sequencing and to surface three new read-only GObject monitoring properties: connected, moq-version, and estimated-send-bitrate. The plugin license metadata is corrected to "MIT/X11", the gst crate is moved to [dev-dependencies], CI is updated to run moq-gst tests explicitly, and hermetic element-boundary integration tests are added.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically describes the main change: rewriting the moqsink element with an improved, unit-tested core implementation.
Description check	✅ Passed	The description comprehensively documents the rewrite objectives, behavioral changes, testing approach, and known limitations, directly relating to the changeset.
Docstring Coverage	✅ Passed	Docstring coverage is 95.52% 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

---

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 and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (6)

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

12-23: ⚡ Quick win

Mark public decision enums as #[non_exhaustive] for additive evolution.

These public enums are likely to grow (new reject/drop reasons or decision forms), so callers should not be forced into exhaustive matching now.

Proposed compatibility patch

 #[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
 pub enum SegmentDecision {
 	Accept,
 	Reject(&'static str),
 }
 
 #[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
 pub enum FrameDecision {
 	/// Emit at this MoQ timestamp (micros).
 	Emit(u64),
 	Drop(&'static str),
 }

As per coding guidelines, "Add #[non_exhaustive] to public enums that may gain variants in the future".

🤖 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/timeline.rs` around lines 12 - 23, Add the
#[non_exhaustive] attribute above the SegmentDecision enum and the FrameDecision
enum definitions. This will allow these public enums to gain new variants in the
future without breaking existing code that relies on exhaustive pattern
matching. Place the attribute on the line immediately before each #[derive]
attribute for both enums.

Source: Coding guidelines

---

4-23: ⚡ Quick win

Add Rustdoc on exported symbols and all public fields.

SegmentInfo, SegmentDecision, and FrameDecision are public but missing item-level /// docs, and SegmentInfo::rate is undocumented.

Proposed doc-comment patch

 #[derive(Debug, Clone, Copy, PartialEq)]
+/// Normalized SEGMENT metadata used by timeline classification.
 pub struct SegmentInfo {
 	/// Only TIME segments map to a media timeline (not BYTES/DEFAULT).
 	pub time_format: bool,
+	/// Playback rate for the segment; only 1.0 is currently accepted.
 	pub rate: f64,
 	/// Running-time anchor of the segment; continuity is judged on this, not on `start`.
 	pub base_nanos: u64,
 }
 
 #[derive(Debug, PartialEq, Eq)]
+/// Decision for whether an incoming SEGMENT keeps timeline continuity.
 pub enum SegmentDecision {
 	Accept,
 	Reject(&'static str),
 }
 
 #[derive(Debug, PartialEq, Eq)]
+/// Decision for whether a frame should be emitted or dropped.
 pub enum FrameDecision {
 	/// Emit at this MoQ timestamp (micros).
 	Emit(u64),
 	Drop(&'static str),
 }

As per coding guidelines, "Document every exported Rust symbol with doc comments (///), including every pub item and module-level docs (//! block at module root)".

🤖 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/timeline.rs` around lines 4 - 23, Add Rustdoc comments
(using `///`) to all public items and fields that are currently missing
documentation. Specifically, add a doc comment to the SegmentInfo struct itself,
the rate field within SegmentInfo, the SegmentDecision enum, the Accept and
Reject variants of SegmentDecision, the FrameDecision enum, and the Drop variant
of FrameDecision. Each doc comment should clearly describe the purpose and usage
of the item or field it documents, following the existing documentation style
used for other fields like time_format and base_nanos in SegmentInfo.

Source: Coding guidelines

rs/moq-gst/src/sink/session.rs (4)

25-26: 💤 Low value

Add doc comment for the public debug category.

The coding guideline requires doc comments on every exported symbol. CAT is pub static but lacks a /// doc comment.

+/// GStreamer debug category for the MoQ sink element.
 pub static CAT: LazyLock<gst::DebugCategory> =
 	LazyLock::new(|| gst::DebugCategory::new("moq-sink", gst::DebugColorFlags::empty(), Some("MoQ Sink Element")));

🤖 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/session.rs` around lines 25 - 26, Add a doc comment to
the public static variable CAT to document its purpose. Place a /// comment
directly above the CAT declaration that describes what this debug category is
used for in the MoQ Sink element, such as "The debug category for the MoQ Sink
element" or similar description that explains its role in the module.

Source: Coding guidelines

---

138-167: ⚡ Quick win

Add #[non_exhaustive] to the public DataMsg enum.

Per coding guidelines, public enums that may gain variants in the future should be marked #[non_exhaustive]. DataMsg could plausibly add new event types (e.g., Gap, StreamStart, CustomEvent).

+/// Messages sent from the GObject shell to the session worker over the bounded data channel.
+#[non_exhaustive]
 pub enum DataMsg {
 	AddPad {
 		pad: String,

🤖 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/session.rs` around lines 138 - 167, The public DataMsg
enum is missing the #[non_exhaustive] attribute, which should be added per
coding guidelines to indicate that new variants may be introduced in the future.
Add the #[non_exhaustive] macro attribute directly above the pub enum DataMsg
definition to allow the enum to be extended without breaking existing code that
performs exhaustive pattern matching.

Source: Coding guidelines

---

48-51: 💤 Low value

Add doc comment for the public Status type.

Status is a pub struct but lacks a /// doc comment. The internal comment on lines 31-34 explains the design well but should be converted to a doc comment for API documentation.

+/// Shared observable state, read by the element's getters without touching the worker task.
+///
+/// One `Mutex` covers the session generation plus every gated field, so "check the live
+/// generation, then write" is one indivisible step.
 #[derive(Default)]
 pub struct Status {
 	inner: Mutex<StatusInner>,
 }

🤖 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/session.rs` around lines 48 - 51, The public struct
`Status` is missing a doc comment which should be added using the `///`
documentation syntax. Add a doc comment above the `Status` struct definition
that explains its purpose and design. The explanation should be derived from the
internal comments found in the StatusInner section (lines 31-34) and should
provide clear API documentation for users of this public type. Convert the
relevant design details from those internal comments into a proper public
documentation comment for the `Status` struct.

Source: Coding guidelines

---

890-894: Log timestamp overflow instead of silently ignoring it.

Line 892 uses .ok() to silently convert a potential TimeOverflow error to None, allowing frames to be emitted without timestamp information. While overflow is unlikely for typical media durations, this pattern masks errors without logging or documenting why it's safe to ignore.

Consider logging the overflow if it occurs:

 match decision {
 	FrameDecision::Emit(micros) => {
-		let ts = hang::container::Timestamp::from_micros(micros).ok();
+		let ts = match hang::container::Timestamp::from_micros(micros) {
+			Ok(ts) => Some(ts),
+			Err(e) => {
+				gst::warning!(CAT, "timestamp conversion failed for {micros} µs: {e}");
+				None
+			}
+		};
 		framed.decode_frame(&mut data, ts).map_err(|err| anyhow::anyhow!(err))
 	}

🤖 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/session.rs` around lines 890 - 894, In the
FrameDecision::Emit handler, the call to Timestamp::from_micros(micros).ok()
silently converts any TimeOverflow error to None, which masks potential
timestamp conversion failures. Instead of using .ok() to suppress the error,
check the result of Timestamp::from_micros(micros) and log a warning or error
message when overflow occurs before proceeding with the frame decode. This
ensures that timestamp overflow issues are visible in logs for debugging
purposes while still allowing the frame to be processed if needed.

🤖 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/src/sink/imp.rs`:
- Around line 271-274: The blocking_send calls for DataMsg::AddPad and
DataMsg::DropPad control messages lack cancellation support, which can cause
indefinite blocking if the bounded data channel fills or the session connection
stalls. Replace the uncancellable blocking_send approach used in the
sender.blocking_send call with a cancellable alternative that matches the data
path's pattern of send_or_flush with per-pad flush cancellation. Ensure both the
AddPad message (around line 273) and the DropPad message (around line 292) are
updated to use the same cancellation mechanism to prevent these control-path
operations from blocking indefinitely under channel pressure.

In `@rs/moq-gst/tests/element.rs`:
- Around line 47-55: In the connect_failure_posts_error_to_bus test function,
replace the URL property value that uses the nonexistent.invalid hostname with a
loopback address and closed port (such as a localhost IP with port 1) to ensure
immediate connection refusal instead of relying on DNS timeout. This will make
the test deterministic by forcing a connection refused error that is not
dependent on DNS resolution timing, while still exercising the same bus error
handling path.

---

Nitpick comments:
In `@rs/moq-gst/src/sink/session.rs`:
- Around line 25-26: Add a doc comment to the public static variable CAT to
document its purpose. Place a /// comment directly above the CAT declaration
that describes what this debug category is used for in the MoQ Sink element,
such as "The debug category for the MoQ Sink element" or similar description
that explains its role in the module.
- Around line 138-167: The public DataMsg enum is missing the #[non_exhaustive]
attribute, which should be added per coding guidelines to indicate that new
variants may be introduced in the future. Add the #[non_exhaustive] macro
attribute directly above the pub enum DataMsg definition to allow the enum to be
extended without breaking existing code that performs exhaustive pattern
matching.
- Around line 48-51: The public struct `Status` is missing a doc comment which
should be added using the `///` documentation syntax. Add a doc comment above
the `Status` struct definition that explains its purpose and design. The
explanation should be derived from the internal comments found in the
StatusInner section (lines 31-34) and should provide clear API documentation for
users of this public type. Convert the relevant design details from those
internal comments into a proper public documentation comment for the `Status`
struct.
- Around line 890-894: In the FrameDecision::Emit handler, the call to
Timestamp::from_micros(micros).ok() silently converts any TimeOverflow error to
None, which masks potential timestamp conversion failures. Instead of using
.ok() to suppress the error, check the result of Timestamp::from_micros(micros)
and log a warning or error message when overflow occurs before proceeding with
the frame decode. This ensures that timestamp overflow issues are visible in
logs for debugging purposes while still allowing the frame to be processed if
needed.

In `@rs/moq-gst/src/sink/timeline.rs`:
- Around line 12-23: Add the #[non_exhaustive] attribute above the
SegmentDecision enum and the FrameDecision enum definitions. This will allow
these public enums to gain new variants in the future without breaking existing
code that relies on exhaustive pattern matching. Place the attribute on the line
immediately before each #[derive] attribute for both enums.
- Around line 4-23: Add Rustdoc comments (using `///`) to all public items and
fields that are currently missing documentation. Specifically, add a doc comment
to the SegmentInfo struct itself, the rate field within SegmentInfo, the
SegmentDecision enum, the Accept and Reject variants of SegmentDecision, the
FrameDecision enum, and the Drop variant of FrameDecision. Each doc comment
should clearly describe the purpose and usage of the item or field it documents,
following the existing documentation style used for other fields like
time_format and base_nanos in SegmentInfo.

🪄 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: 16ea640f-7e9d-412f-811d-8bc53d8f0e7c

📥 Commits

Reviewing files that changed from the base of the PR and between 2788d79 and a858308.

📒 Files selected for processing (9)

• 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/session.rs
• rs/moq-gst/src/sink/timeline.rs
• rs/moq-gst/tests/element.rs

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify control-path blocking sends and whether DataMsg uses bounded channels.
# Expected: show blocking_send call sites and channel construction/capacity in session worker plumbing.

rg -nP --type=rust -C3 '\bblocking_send\s*\(' rs/moq-gst/src/sink/imp.rs rs/moq-gst/src/sink/session.rs
rg -nP --type=rust -C3 '\bmpsc::channel\s*\(' rs/moq-gst/src/sink/session.rs
rg -nP --type=rust -C3 '\bDataMsg\b|AddPad|DropPad' rs/moq-gst/src/sink/session.rs

Repository: moq-dev/moq

Length of output: 9076

---

Control-path blocking_send for AddPad and DropPad lacks cancellation.

Both DataMsg::AddPad (line 273) and DataMsg::DropPad (line 292) use uncancellable blocking_send on the bounded data channel. Unlike the data path (which uses send_or_flush with per-pad flush cancellation), these control messages can block pad add/remove indefinitely if the channel fills or the session connection stalls. The code at lines 289–291 documents this as a known limitation marked "out of scope for FLUSH"—but it remains a potential stability issue under pressure.

🤖 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 271 - 274, The blocking_send calls
for DataMsg::AddPad and DataMsg::DropPad control messages lack cancellation
support, which can cause indefinite blocking if the bounded data channel fills
or the session connection stalls. Replace the uncancellable blocking_send
approach used in the sender.blocking_send call with a cancellable alternative
that matches the data path's pattern of send_or_flush with per-pad flush
cancellation. Ensure both the AddPad message (around line 273) and the DropPad
message (around line 292) are updated to use the same cancellation mechanism to
prevent these control-path operations from blocking indefinitely under channel
pressure.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Make the connect-failure test deterministic without DNS.

This test depends on DNS failure timing for .invalid, which can be flaky in CI/network-restricted environments. Prefer a loopback closed port to force immediate connect failure while exercising the same bus-error path.

Suggested change

-		.property("url", "https://nonexistent.invalid:443")
+		.property("url", "https://127.0.0.1:1")

Also applies to: 60-63

🤖 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 47 - 55, In the
connect_failure_posts_error_to_bus test function, replace the URL property value
that uses the nonexistent.invalid hostname with a loopback address and closed
port (such as a localhost IP with port 1) to ensure immediate connection refusal
instead of relying on DNS timeout. This will make the test deterministic by
forcing a connection refused error that is not dependent on DNS resolution
timing, while still exercising the same bus error handling path.

[arielmol]

To the holy rabbit:

• The blocking_send on AddPad/DropPad is a known limitation, already called out in the comment and marked as future work.

• For the connect-failure test I'm keeping .invalid. The 127.0.0.1:1 suggestion assumes TCP refused semantics, but this is QUIC over UDP, so a closed port won't refuse fast, it'd just sit until the handshake timeout and get flakier. .invalid is a reserved TLD that never resolves, so it fails immediately.

• The .ok() on the timestamp overflow is fine, it only triggers on ABSURD durations and decode_frame already takes an optional timestamp. Can add a warning log if you'd rather see it.

For CI fails, will fix it asap.