Finalize fetch relay coverage and error handling

dmorn · dmorn · commit cdad3ceeb270 · 2026-04-15T23:23:10.000+02:00
Closes #11
diff --git a/README.md b/README.md
@@ -44,6 +44,7 @@ Today `moqx` supports a single client-side path:
 - broadcasts, tracks, and frame delivery
 - live subscription via SUBSCRIBE with `FilterType::LatestObject`
 - raw fetch for retrieving track objects by range (subscriber sessions only)
+  - on the current moqtail relay, standalone fetch succeeds only for objects the relay already has in cache; upstream relay-to-publisher standalone fetch is not implemented yet
 - optional helper-layer catalog publication via `MOQX.Helpers.publish_catalog/2`
   and `MOQX.Helpers.update_catalog/2`
 - optional helper-layer catalog retrieval via `MOQX.Helpers.fetch_catalog/2`
@@ -315,12 +316,12 @@ mix moqx.inspect https://draft-14.cloudflare.mediaoverquic.com --namespace bbb -
 The task will:
 
 1. connect as a subscriber,
-2. load catalog via fetch (with live-subscribe fallback when fetch has no objects),
+2. load catalog via fetch (with live-subscribe fallback when fetch has no objects or the relay has not cached the track yet),
 3. optionally apply a known relay preset (`--preset`) or choose one interactively (`--choose-relay`),
 4. try `"catalog"` and then `".catalog"` unless `--catalog-track` is set,
 5. optionally skip fetch entirely with `--no-fetch` and go straight to live subscribe,
 6. prompt you to choose a track (or use `--track <name>`),
-5. subscribe and print live stats each interval:
+7. subscribe and print live stats each interval:
    - PRFT latency (or `n/a` if unavailable),
    - bandwidth (`B/s` and `kbps`),
    - groups/sec,
@@ -364,6 +365,14 @@ Fetch retrieves raw track objects by range from a subscriber session.
 `fetch/4` returns `{:ok, ref}` immediately, then delivers messages to the
 caller's mailbox correlated by `ref`.
 
+Important moqtail relay note: the current relay only serves standalone fetches
+from its local track cache. In practice that means fetch works end-to-end for
+objects the relay has already seen (for example after live delivery to a
+subscriber), but it does not yet forward standalone fetch upstream to a
+publisher on cache miss. On such a cache miss, `moqx` surfaces the relay reply
+as a typed `{:moqx_request_error, %MOQX.RequestError{op: :fetch, ...}}` rather
+than hanging silently.
+
 The fetch message contract is:
 
 - `{:moqx_fetch_ok, %MOQX.FetchOk{ref, namespace, track_name}}`
diff --git a/native/moqx_native/src/lib.rs b/native/moqx_native/src/lib.rs
@@ -374,7 +374,9 @@ impl SessionInner {
         track_name: String,
     ) -> (Option<u64>, bool) {
         let mut guard = self.track_lifecycle.lock().unwrap();
-        let entry = guard.entry(key.to_string()).or_insert_with(TrackLifecycle::pending);
+        let entry = guard
+            .entry(key.to_string())
+            .or_insert_with(TrackLifecycle::pending);
         entry.created = true;
         entry.closed = false;
 
@@ -401,7 +403,9 @@ impl SessionInner {
 
     fn activate_track(&self, key: &str, track_alias: u64) {
         let mut guard = self.track_lifecycle.lock().unwrap();
-        let entry = guard.entry(key.to_string()).or_insert_with(TrackLifecycle::pending);
+        let entry = guard
+            .entry(key.to_string())
+            .or_insert_with(TrackLifecycle::pending);
         if entry.closed {
             return;
         }
@@ -435,7 +439,9 @@ impl SessionInner {
 
     fn close_track(&self, key: &str) -> Vec<TrackNotifier> {
         let mut guard = self.track_lifecycle.lock().unwrap();
-        let entry = guard.entry(key.to_string()).or_insert_with(TrackLifecycle::created);
+        let entry = guard
+            .entry(key.to_string())
+            .or_insert_with(TrackLifecycle::created);
 
         if entry.closed {
             return Vec::new();
@@ -919,26 +925,128 @@ async fn control_loop(
     inner: Arc<SessionInner>,
     shutdown_rx: &mut watch::Receiver<bool>,
 ) {
-    loop {
+    let exit_reason = loop {
         tokio::select! {
-            _ = shutdown_rx.changed() => break,
+            _ = shutdown_rx.changed() => break None,
             msg = control_rx.recv() => {
                 match msg {
                     Some(msg) => {
-                        if handler.send(&msg).await.is_err() {
-                            break;
+                        if let Err(err) = handler.send(&msg).await {
+                            break Some(format!("control stream send failed: {err:?}"));
                         }
                     }
-                    None => break,
+                    None => break None,
                 }
             }
             result = handler.next_message() => {
                 match result {
                     Ok(msg) => dispatch_control_response(msg, &inner),
-                    Err(_) => break,
+                    Err(err) => break Some(format!("control stream receive failed: {err:?}")),
                 }
             }
         }
+    };
+
+    if let Some(reason) = exit_reason {
+        fail_pending_control_ops(&inner, &reason);
+    }
+}
+
+fn fail_pending_control_ops(inner: &Arc<SessionInner>, reason: &str) {
+    let pending_publishes = {
+        let mut guard = inner.pending_publishes.lock().unwrap();
+        std::mem::take(&mut *guard)
+    };
+
+    for (_, pending) in pending_publishes {
+        let mut msg_env = OwnedEnv::new();
+        let pid = pending.caller_pid;
+        pending.ref_env.run(|ref_env| {
+            let publish_ref = pending.publish_ref_term.load(ref_env);
+            let _ = msg_env.send_and_clear(&pid, |env| {
+                let nil = rustler::types::atom::nil().to_term(env);
+                let payload = TransportErrorOut {
+                    op: atoms::publish(),
+                    message: reason.to_string(),
+                    kind: Some(atoms::runtime()),
+                    r#ref: publish_ref.in_env(env),
+                    handle: nil,
+                };
+                (atoms::moqx_transport_error(), payload).encode(env)
+            });
+        });
+    }
+
+    let pending_subscribes = {
+        let mut guard = inner.pending_subscribes.lock().unwrap();
+        std::mem::take(&mut *guard)
+    };
+
+    for (_, pending) in pending_subscribes {
+        let mut msg_env = OwnedEnv::new();
+        let pid = pending.caller_pid;
+        pending.term_env.run(|term_env| {
+            let sub_ref = pending.subscription_ref_term.load(term_env);
+            let _ = msg_env.send_and_clear(&pid, |env| {
+                let nil = rustler::types::atom::nil().to_term(env);
+                let payload = TransportErrorOut {
+                    op: atoms::subscribe(),
+                    message: reason.to_string(),
+                    kind: Some(atoms::runtime()),
+                    r#ref: nil,
+                    handle: sub_ref.in_env(env),
+                };
+                (atoms::moqx_transport_error(), payload).encode(env)
+            });
+        });
+    }
+
+    let pending_fetches = {
+        let mut guard = inner.pending_fetches.lock().unwrap();
+        std::mem::take(&mut *guard)
+    };
+
+    for (_, pending) in pending_fetches {
+        let mut msg_env = OwnedEnv::new();
+        let pid = pending.caller_pid;
+        pending.ref_env.run(|ref_env| {
+            let fetch_ref = pending.ref_term.load(ref_env);
+            let _ = msg_env.send_and_clear(&pid, |env| {
+                let nil = rustler::types::atom::nil().to_term(env);
+                let payload = TransportErrorOut {
+                    op: atoms::fetch(),
+                    message: reason.to_string(),
+                    kind: Some(atoms::runtime()),
+                    r#ref: fetch_ref.in_env(env),
+                    handle: nil,
+                };
+                (atoms::moqx_transport_error(), payload).encode(env)
+            });
+        });
+    }
+
+    let active_subscriptions = {
+        let mut guard = inner.active_subscriptions.lock().unwrap();
+        std::mem::take(&mut *guard)
+    };
+
+    for (_, active) in active_subscriptions {
+        let mut msg_env = OwnedEnv::new();
+        let pid = active.caller_pid;
+        active.ref_env.run(|ref_env| {
+            let sub_ref = active.subscription_ref_term.load(ref_env);
+            let _ = msg_env.send_and_clear(&pid, |env| {
+                let nil = rustler::types::atom::nil().to_term(env);
+                let payload = TransportErrorOut {
+                    op: atoms::subscribe(),
+                    message: reason.to_string(),
+                    kind: Some(atoms::runtime()),
+                    r#ref: nil,
+                    handle: sub_ref.in_env(env),
+                };
+                (atoms::moqx_transport_error(), payload).encode(env)
+            });
+        });
     }
 }
 
@@ -1424,8 +1532,13 @@ async fn handle_subgroup_stream(
     // Control and data travel on independent streams. A subgroup stream may arrive
     // just before the control loop processes the matching SubscribeOk(track_alias).
     // Wait briefly for local activation before deciding this alias is unknown.
-    let has_subscription =
-        wait_for_active_subscription(inner, track_alias, Duration::from_millis(2_000), Duration::from_millis(5)).await;
+    let has_subscription = wait_for_active_subscription(
+        inner,
+        track_alias,
+        Duration::from_millis(2_000),
+        Duration::from_millis(5),
+    )
+    .await;
 
     if !has_subscription {
         return Ok(());
diff --git a/test/moqx_integration_test.exs b/test/moqx_integration_test.exs
@@ -125,6 +125,62 @@ defmodule MOQXIntegrationTest do
     end
   end
 
+  defp await_fetch_ok!(fetch_ref, namespace, track_name, timeout \\ @timeout) do
+    receive do
+      {:moqx_fetch_ok,
+       %MOQX.FetchOk{ref: ^fetch_ref, namespace: ^namespace, track_name: ^track_name}} ->
+        :ok
+
+      {:moqx_request_error, %MOQX.RequestError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch failed: #{inspect(reason)}")
+
+      {:moqx_transport_error, %MOQX.TransportError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch transport failed: #{inspect(reason)}")
+    after
+      timeout -> flunk("fetch ok timeout for #{namespace}/#{track_name}")
+    end
+  end
+
+  defp await_fetch_object!(fetch_ref, expected_payload, timeout \\ @timeout) do
+    receive do
+      {:moqx_fetch_object,
+       %MOQX.FetchObject{
+         ref: ^fetch_ref,
+         group_id: group_id,
+         object_id: object_id,
+         payload: payload
+       }} ->
+        if payload == expected_payload do
+          {group_id, object_id, payload}
+        else
+          await_fetch_object!(fetch_ref, expected_payload, timeout)
+        end
+
+      {:moqx_request_error, %MOQX.RequestError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch failed: #{inspect(reason)}")
+
+      {:moqx_transport_error, %MOQX.TransportError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch transport failed: #{inspect(reason)}")
+    after
+      timeout -> flunk("fetch object timeout waiting for #{inspect(expected_payload)}")
+    end
+  end
+
+  defp await_fetch_done!(fetch_ref, timeout \\ @timeout) do
+    receive do
+      {:moqx_fetch_done, %MOQX.FetchDone{ref: ^fetch_ref}} ->
+        :ok
+
+      {:moqx_request_error, %MOQX.RequestError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch failed: #{inspect(reason)}")
+
+      {:moqx_transport_error, %MOQX.TransportError{op: :fetch, ref: ^fetch_ref, message: reason}} ->
+        flunk("fetch transport failed: #{inspect(reason)}")
+    after
+      timeout -> flunk("fetch done timeout")
+    end
+  end
+
   describe "integration relay: connect" do
     @tag :integration
     test "subscriber connects and reports draft-14 version" do
@@ -358,6 +414,74 @@ defmodule MOQXIntegrationTest do
       end
     end
 
+    @tag :integration
+    test "fetch returns cached object bytes end-to-end on the relay" do
+      publisher = connect_publisher!()
+      subscriber = connect_subscriber!()
+
+      try do
+        ns = "moqx-e2e-fetch-#{System.system_time(:millisecond)}"
+        track_name = "catalog"
+
+        payload =
+          ~s({"version":1,"supportsDeltaUpdates":false,"tracks":[{"name":"demo","role":"video"}]})
+
+        broadcast = publish_and_await_broadcast!(publisher, ns)
+        {:ok, track} = MOQX.create_track(broadcast, track_name)
+
+        handle = subscribe_and_await_handle!(subscriber, ns, track_name)
+        await_track_active!(track, ns, track_name)
+
+        :ok = MOQX.write_frame(track, payload)
+
+        {live_group_id, live_payload} = await_matching_payload_frame!(payload)
+        assert live_group_id == 0
+        assert live_payload == payload
+
+        :ok = MOQX.unsubscribe(handle)
+        assert_receive {:moqx_publish_done, %MOQX.PublishDone{handle: ^handle}}, @timeout
+
+        {:ok, fetch_ref} = MOQX.fetch(subscriber, ns, track_name, start: {0, 0}, end: {0, 1})
+
+        :ok = await_fetch_ok!(fetch_ref, ns, track_name)
+        {group_id, object_id, fetched_payload} = await_fetch_object!(fetch_ref, payload)
+        :ok = await_fetch_done!(fetch_ref)
+
+        assert group_id == 0
+        assert object_id == 0
+        assert fetched_payload == payload
+
+        :ok = MOQX.finish_track(track)
+      after
+        :ok = MOQX.close(subscriber)
+        :ok = MOQX.close(publisher)
+      end
+    end
+
+    @tag :integration
+    test "fetch cache miss returns typed request error on current moqtail relay" do
+      publisher = connect_publisher!()
+      subscriber = connect_subscriber!()
+
+      try do
+        ns = "moqx-e2e-fetch-miss-#{System.system_time(:millisecond)}"
+        track_name = "catalog"
+
+        _broadcast = publish_and_await_broadcast!(publisher, ns)
+
+        {:ok, fetch_ref} = MOQX.fetch(subscriber, ns, track_name, start: {0, 0}, end: {0, 1})
+
+        assert_receive {:moqx_request_error,
+                        %MOQX.RequestError{op: :fetch, ref: ^fetch_ref, message: reason}},
+                       @timeout
+
+        assert reason =~ "TrackDoesNotExist"
+      after
+        :ok = MOQX.close(subscriber)
+        :ok = MOQX.close(publisher)
+      end
+    end
+
     @tag :integration
     test "unsubscribe/1 stops frame delivery and emits moqx_publish_done" do
       publisher = connect_publisher!()
