MoqSrc/MoqSink: refactor to async session tasks with channel-based pad forwarding

[SteveMcFarlin]

MoqSink Summary

• State management: Replace the block_on-based setup() with start_session()/stop_session() and a Tokio-spawned run_session, so GLib state changes no longer block while establishing MoQ sessions. All session state now lives inside the async task’s RuntimeState.
• Pad handling: Keep dynamic pad creation, but forward CAPS/buffers through a bounded control channel (forward_event/forward_buffer). Decoder instantiation (handle_caps) and timestamp math remain identical; actual decoding/publishing now happens off the GLib thread.
• Network setup: Rebuild the publish pipeline asynchronously—create OriginProducer, BroadcastProducer, catalog via moq_mux::CatalogProducer::new, publish the broadcast, and connect with client.with_publish(origin.consume()). The live Session is retained inside RuntimeState.
• Flow control: blocking_send provides clearer backpressure semantics (eventually yields FLOW_FLUSHING), though upstream queues are still recommended for bursty inputs.

Potential regressions / tests

1. RuntimeState owns all session/broadcast state now—verify nothing else expects synchronous access.
2. CAPS/buffer ordering per pad is preserved, but cross-pad ordering depends on message interleaving; test pad add/remove under load.
3. Only CAPS events are forwarded explicitly; add control messages if flush/segment handling is needed.
4. Errors are logged and the channel closes; integration tests should cover handshake failure, decoder errors, and shutdown timing to ensure pipelines react gracefully.

MoqSrc Summary

• Element Type – imp.rs now subclasses gst::Element instead of gst::Bin, so the source behaves like a leaf element and owns all pad logic itself.
• State Management – change_state no longer blocks on setup(). Ready→Paused just spins up a Tokio‑backed SessionController, while Paused→Ready signals it to stop
• Session Lifecycle – Catalog discovery, pad creation, and buffer forwarding now flow through a control_tx/control_rx channel plus run_session/spawn_track_pump, so all networking happens off the GLib thread.
• Pad Handling – Pads are created on demand via ControlMessage::CreatePad, and PadMessage routing through a GLib helper keeps buffer/event pushes serialized on the main loop
• Codec Support – video_caps now emits caps for H.264, H.265 (hev1/hvc1/byte-stream), and AV1, bringing the source in line with the sink’s decoder stack.
• Cleanup – stop_session() drops the control task, sends PadMessage::Drop to every pad, and clears the channel sender so downstream sees a flush.
• Testing – Added a unit test covering the GLib forwarder to ensure channel messages stay in order.
• Runtime Usage – Still uses a dedicated Tokio runtime, but with LazyLock and all async work centralized in the session task instead of ad-hoc tokio::spawn calls inside setup().

Potential regressions

• Channel backpressure vs. GLib timing – Pads now use mpsc::UnboundedSender into the main-context forwarder. If the forwarder stalls (e.g., GLib thread busy) those queues can grow without bound; previously the direct pad.push would backpressure immediately.
• Per-pad ordering – Catalog pads are now independent tasks sending through PadMessage channels. Ordering is preserved per pad, but cross-pad sequencing may differ from the old implementation that happened to poll tracks sequentially inside setup().
• Shutdown timing – stop_session sends PadMessage::Drop and aborts the GLib task, but downstream might see EOS/flush later than before because we now wait for the async pumps to notice shutdown.
• Codec negotiation – H.265/AV1 caps mirror the sink’s assumptions (e.g., choosing hev1 vs hvc1 based on description), but if downstream expects different stream-format strings or fields, negotiation could fail.
• Error propagation – Session failures are only surfaced via ControlMessage::ReportError → gst::element_error!. There’s still no recovery path; repeated errors may leave the element in a flushing state without attempting reconnection, same as before but now more asynchronous.

[kixelated]

nits

[kixelated]

Probably not needed. Quinn itself doesn't scale past a single thread.

[kixelated]

Switch to an unbounded sender?

[kixelated]

IMO return an error

-> anyhow::Result<()> {

So we only need to gst::error! in one place.

[kixelated]

IMO you should remove a level of indentation by inlining the spawn:

let handle = RUNTIME.spawn(run_track_pump);

fn run_track_pump(
	mut track: hang::container::OrderedConsumer,
	descriptor: TrackDescriptor,
	pad_endpoint: PadEndpoint,
	mut shutdown: watch::Receiver<bool>,
) {

[kixelated]

What is this? Should we be using this runtime instead..?

[kixelated]

idk if this test is needed. mpsc works