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 has a dedicated unit suite (timeline,
segment classification, session lifecycle, flush, per-pad generations); the moq-gst
crate's suite is 59 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 re-anchor: drop to NoSegment, recover on the next SEGMENT).

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.

Session handoff (non-blocking)

The session handoff is a non-blocking unbounded channel: chain accepts each buffer
toward the session worker and returns, never blocking the streaming thread. On the
connected steady path the worker writes into the MoQ producers, whose cache owns
retention and eviction, so loss under congestion is governed by MoQ cache/retention
semantics rather than by stalling the encoder. moqsink is therefore not a GStreamer
backpressure boundary; pacing a non-live/file source belongs upstream (for example
identity sync=true).

This replaces an earlier bounded-channel design, where a send blocked on a full channel
could not be interrupted by an out-of-band FLUSH_START and so could wedge a
seek/teardown. With a non-blocking handoff there is no blocked send, so the
flush-cancellation apparatus is gone: the bounded channel, send_or_flush/SendOutcome,
and the per-pad flush watch gates that cut a blocked send (plus the AddPad/DropPad
blocking sends). FlushSignal and the per-pad generations stay; FlushSignal now carries
only the timeline re-anchor. FLUSH re-anchors a pad's timeline (drop to NoSegment, recover
on the next SEGMENT), not an anti-deadlock cut.

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: 59 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): building the session producers up front and writing
  directly (removing the data channel and its connect-window residual buffer), 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]

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 and frame timestamp classification is extracted into a new stateless timeline.rs module. MoqSink gains per-pad generation counters for flush sequencing, a shared Status handle for observable session state, and 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 PR title accurately and concisely describes the primary change: rewriting the moqsink element with improved unit test coverage and modular architecture.
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.
Description check	✅ Passed	The pull request description clearly relates to the changeset, describing a comprehensive rewrite of the moqsink GStreamer element with detailed context on objectives, behavior changes, testing, and known limitations.

_{✏️ 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 | 🟡 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.

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