moq-gst: moqsink on a bare Element with direct (no-channel) writes

[kixelated]

Summary

A third moqsink rewrite for comparison, the bare gst::Element with direct, no-channel writes (option B from the discussion). The direct-write core was coded independently, but it builds on the same media foundation @arielmol established in #1771 (timeline/running-time policy, codec/caps factory, B-frames, FLUSH/EOS, per-pad error isolation). Ariel's #1771 commits are merged into this branch so his authorship is credited on merge, and he is listed in the element's author metadata.

Each request pad has its own chain function that writes buffers straight into the moq producers from the streaming thread. There is no intermediate channel and no worker task: moq_net's producer writes are synchronous (an in-memory append bounded by group eviction), so the streaming thread never blocks on the network. A thin async task owns only connect and the session lifetime.

Why bare Element instead of GstAggregator (#1870)

GstAggregator synchronizes pads: it will not call aggregate() until every non-EOS pad has a buffer, so a sparse pad batches its siblings into bursts (the ~2s stalls measured on #1870). moqsink wants the opposite, independent per-pad passthrough (each pad is its own moq track, delivered independently). A bare Element gives that for free: each pad's chain runs on its own thread and never waits on another pad. It also stays on the GStreamer 1.14 baseline (no force_live / 1.22).

Why no channel (vs #1771)

#1771 keeps the bare Element but hands buffers to a worker over an unbounded channel. Since GroupProducer::write_frame is synchronous and non-blocking, that handoff is not load-bearing, so this PR drops it: the chain writes the producer directly. The result is fewer moving parts (no DataMsg, no worker loop, no per-pad generations) while keeping pad independence.

Structure

• sink/imp.rs: bare Element. Request pads with chain + event functions; one Mutex<Option<State>> holds the producers + per-pad media, held only briefly per buffer. EOS aggregated across pads. Oversized frames are left to moq-net, whose create_frame rejects them (FrameTooLarge) before allocating the group slot, invalidating just that pad.
• sink/pad.rs: per-pad media (caps -> producer, SEGMENT/running-time policy, frame import) + unit tests.
• sink/session.rs: connect + transport lifetime + observable Status, no media.
• sink/timeline.rs: pure SEGMENT/running-time policy.
• Three read-only diagnostics: connected, moq-version, estimated-send-bitrate.

Public surface / branch targeting

No public library API or wire change. Same element name (moqsink), sink_%u pads, codec caps, and writable properties. No gstreamer-base dependency (it is a bare Element). Internal to the rs/moq-gst plugin, so this targets main.

Testing

• cargo test -p moq-gst: 34 unit + 3 element tests; cargo clippy --all-targets and cargo fmt --check clean.
• Live relay: videotestsrc ! vtenc_h264 ! moqsink.sink_0 plus audiotestsrc ! opusenc ! moqsink.sink_1 to a local moq-relay logged session connected -> two independent pads -> finalized on EOS: ["sink_0", "sink_1", "catalog"] -> EOS.

Relationship to the other PRs

• moq-gst: replace moqsink with a rewritten, unit-tested core #1771 (Ariel): bare Element + unbounded channel + worker. Proven, deployed, 1.16-safe.
• moq-gst: replace moqsink with a GstAggregator-based core #1870 (mine): GstAggregator. Smaller, but synchronizes pads (sparse-pad stall) and force_live is 1.22.
• This PR (B): bare Element + direct writes. Independent pads like moq-gst: replace moqsink with a rewritten, unit-tested core #1771, no channel/worker, 1.14 baseline. Newer than moq-gst: replace moqsink with a rewritten, unit-tested core #1771, so less battle-tested.

(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: abcdded5-649f-47f5-a8a2-3940fb2397a4

📥 Commits

Reviewing files that changed from the base of the PR and between 8a61d46 and df0d25c.

📒 Files selected for processing (1)

• rs/moq-gst/src/sink/imp.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• rs/moq-gst/src/sink/imp.rs

---

Walkthrough

The MoqSink GStreamer element is refactored from an async Tokio task/mpsc-channel design to a synchronous per-pad producer model. Three new submodules are introduced: timeline.rs defines SegmentInfo, classify_segment, and frame_micros for segment continuity and timestamp conversion; session.rs owns transport connection lifecycle, observable Status (connected/version/bitrate), and async task management; pad.rs implements per-pad caps negotiation, PadState transitions, buffer forwarding, and finalization. imp.rs replaces SessionHandle with a synchronous State struct and adds three read-only GObject properties (connected, moq-version, estimated-send-bitrate). Integration tests and documentation are added.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately captures the main architectural change: rewriting moqsink as a bare Element with direct writes instead of channel-based workers.
Description check	✅ Passed	The description comprehensively explains the rationale, design decisions, and implementation structure, clearly relating to all major changes in the changeset.
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/moqsink-direct

---

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: 3

🤖 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 264-276: Extract the EOS aggregation check and posting logic from
the handle_eos method (specifically the all_ended condition check and EOS
message posting) into a new helper method called maybe_post_eos, then call this
helper from both handle_eos and at the end of release_pad (after
state.ended.remove and self.obj().remove_pad calls) to ensure that when a pad is
released, the aggregation condition is re-evaluated and any pending EOS message
is posted if all remaining pads have ended.

In `@rs/moq-gst/src/sink/pad.rs`:
- Around line 44-55: Add a doc comment above the public method `Pad::new()` in
the Pad impl block. Use Rust's triple-slash doc comment syntax to document what
this constructor method does, creating a new Pad instance with default values.
The doc comment should clearly explain the purpose of this initialization method
and appear immediately before the function signature.

In `@rs/moq-gst/src/sink/timeline.rs`:
- Around line 5-11: Add a doc comment for the `rate` field in the `SegmentInfo`
struct. Locate the `rate: f64,` field declaration and add a documentation
comment above it using the same format as the existing doc comments for
`time_format` and `base_nanos` fields. The comment should briefly describe the
purpose or meaning of the rate field to maintain consistency with the public API
documentation standards.

🪄 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: 1e86c1b3-cc2f-4461-b9b9-08ee581718a2

📥 Commits

Reviewing files that changed from the base of the PR and between 8469563 and 21c253a.

📒 Files selected for processing (7)

• doc/bin/gstreamer.md
• 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

[kixelated]

This should be merged now, so you can remove MAX_FRAME_BYTES

[kixelated]

Done in 8c0e920. Removed the constant and the pre-check. moq-net's create_frame now rejects oversized frames with FrameTooLarge before it preallocates the group slot (group.rs:184), so an oversized buffer just invalidates the pad like any other bad bitstream, which is what pad.rs already documented ("an oversized frame, rejected by moq-net"). (Written by Claude)

[kixelated]

IDK if it would be cleaner to have this be Ok(Vec::new) to start with and we only push when ok_mut() (is that even a function...?)

[kixelated]

Done. finalize_all now starts as Ok(Vec::new()) and pushes names while it's still Ok (via result.as_mut()), switching to the first error and leaving the partial order behind. Dropped the separate failure accumulator. One behavioral nuance: it's now first-error-wins rather than catalog-takes-precedence; both are teardown finalize errors so the distinction is cosmetic, and I updated the doc comment to match. (Written by Claude)

[kixelated]

Can we do this before grabbing the lock?

[kixelated]

Done. The map_readable + copy_from_slice now happen before the lock is taken (neither needs shared state), so the per-pad lock is held only for the producer write. Trade-off: a buffer arriving on an already-failed pad now incurs the copy before we discover it's failed, but that's an error path and checking is_failed needs the lock anyway. (Written by Claude)

[kixelated]

This will be occupied 99% of the time right? Maybe don't use entry() to avoid allocating a string every iteration.

[kixelated]

Done. The hot path now does contains_key + get_mut (no allocation) and only builds the owned String key on the rare first-buffer insert. (Written by Claude)

[coderabbitai]

Actionable comments posted: 1

🤖 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 341-349: The buffer copy at line 349 with Bytes::copy_from_slice
happens before moq-net validates the frame size, causing unnecessary allocations
for oversized frames. Expose MAX_FRAME_SIZE from moq-net as public or create a
public preflight validation helper function that moq-gst can import. Add a size
check before the Bytes::copy_from_slice call at line 349 to validate the frame
size against the limit and return an error if it exceeds the maximum allowed
size. Additionally, update the misleading comment at lines 342-343 to accurately
reflect that the allocation happens first and then moq-net rejects it, rather
than claiming rejection happens before allocation.

🪄 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: c130c41f-9756-47f9-884b-848d39a7e095

📥 Commits

Reviewing files that changed from the base of the PR and between 21c253a and 8c0e920.

📒 Files selected for processing (4)

• rs/moq-gst/src/sink/imp.rs
• rs/moq-gst/src/sink/pad.rs
• rs/moq-gst/src/sink/session.rs
• rs/moq-gst/src/sink/timeline.rs

🚧 Files skipped from review as they are similar to previous changes (3)

• rs/moq-gst/src/sink/timeline.rs
• rs/moq-gst/src/sink/session.rs
• rs/moq-gst/src/sink/pad.rs

[coderabbitai]

🩺 Stability & Availability | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify where frame-size validation is exposed and confirm the copy-before-validation path.
set -euo pipefail

# Expect to find the moq-net FrameTooLarge path and whether a reusable public limit/preflight exists.
rg -n -C3 'FrameTooLarge|MAX_FRAME|frame.*too.*large|create_frame|fn create_frame' rs -g '*.rs'

# Expect this to show Bytes::copy_from_slice before Pad::push_buffer.
sed -n '337,376p' rs/moq-gst/src/sink/imp.rs

Repository: moq-dev/moq

Length of output: 14821

---

🏁 Script executed:

# Check if MAX_FRAME_SIZE is publicly exported from moq-net root or if it's crate-private
rg -n 'pub.*MAX_FRAME_SIZE|pub mod|pub use.*MAX_FRAME_SIZE' rs/moq-net/src --type rust | head -20

# Check moq-net's public API surface (lib.rs or mod.rs)
cat rs/moq-net/src/lib.rs 2>/dev/null | head -100

Repository: moq-dev/moq

Length of output: 3693

---

🏁 Script executed:

# Check what's exported from the model module
rg -n 'pub (const|fn|struct|enum)' rs/moq-net/src/model --type rust | grep -i 'frame\|size'

# Check moq-net's model module structure
ast-grep outline rs/moq-net/src/model/mod.rs --view expanded | head -50

# See what moq-gst imports from moq-net
rg -n 'use.*moq_net|use moq_net' rs/moq-gst/src -A2 --type rust | head -30

Repository: moq-dev/moq

Length of output: 3276

---

🏁 Script executed:

# Check moq-net's model/mod.rs to see what's re-exported
cat rs/moq-net/src/model/mod.rs

# Also check if MAX_FRAME_SIZE is used anywhere in moq-gst
rg -n 'MAX_FRAME_SIZE' rs/moq-gst -A2 -B2

Repository: moq-dev/moq

Length of output: 369

---

🏁 Script executed:

# Check if MAX_FRAME_SIZE is exported via pub use in frame.rs
cat rs/moq-net/src/model/frame.rs | head -50

# Check the comments around MAX_FRAME_SIZE to understand the design intent
sed -n '15,35p' rs/moq-net/src/model/frame.rs

# See if moq-gst accesses any GroupProducer methods
rg -n 'GroupProducer\|create_frame\|write_frame' rs/moq-gst/src -B2 -A3

Repository: moq-dev/moq

Length of output: 3197

---

🏁 Script executed:

# Check how moq-gst accesses the frame/group writing
rg -n 'push_buffer|Pad::push|write|producer' rs/moq-gst/src/sink/pad.rs -B2 -A5 | head -60

# See the full pad.rs implementation to understand the frame path
sed -n '1,100p' rs/moq-gst/src/sink/pad.rs

Repository: moq-dev/moq

Length of output: 6474

---

🏁 Script executed:

# Check the Framed implementation in moq_mux to see if it validates frame size
rg -n 'push_buffer|write_frame|create_frame' rs/moq-mux/src/import --type rust -B3 -A3 | head -80

# Also check the Framed implementation signature
ast-grep outline rs/moq-mux/src/import.rs --view expanded 2>/dev/null || echo "File not found, checking directory structure"

# Find the import module
find rs/moq-mux/src -name "import*" -o -name "*framed*" | head -5
</parameter>

Repository: moq-dev/moq

Length of output: 2982

---

Make frame size validation preflight-accessible from moq-net.

The buffer is copied unconditionally to Bytes at line 349 before reaching moq-net's FrameTooLarge validation in GroupProducer::create_frame, so an oversized gst::Buffer allocates on the streaming thread even though moq-net will reject it. MAX_FRAME_SIZE is pub(crate) in moq-net and cannot be imported here for a preflight check. Expose either MAX_FRAME_SIZE as public or a small preflight helper from moq-net so moq-gst can validate before copying.

Additionally, the comment at lines 342–343 is misleading: it claims oversized frames are rejected "before allocating," but the Bytes::copy_from_slice allocation always happens first.

🤖 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 341 - 349, The buffer copy at line
349 with Bytes::copy_from_slice happens before moq-net validates the frame size,
causing unnecessary allocations for oversized frames. Expose MAX_FRAME_SIZE from
moq-net as public or create a public preflight validation helper function that
moq-gst can import. Add a size check before the Bytes::copy_from_slice call at
line 349 to validate the frame size against the limit and return an error if it
exceeds the maximum allowed size. Additionally, update the misleading comment at
lines 342-343 to accurately reflect that the allocation happens first and then
moq-net rejects it, rather than claiming rejection happens before allocation.