Integrate relay docker service tests and finalize subscribe timeout typing

Author: dmorn

.github/workflows/ci.yml

@@ -62,13 +62,6 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
-      - name: Check out pinned moq-dev relay workspace
-        uses: actions/checkout@v6
-        with:
-          repository: moq-dev/moq
-          ref: aae43f824b9b8d93b68dd6a765f8bec691f60c0b
-          path: .moq-dev
-
       - uses: erlef/setup-beam@v1
         with:
           elixir-version: '1.19.0'
@@ -94,16 +87,27 @@ jobs:
             ~/.cargo/registry
             ~/.cargo/git
             target
-            .moq-dev/target
-          key: ${{ runner.os }}-cargo-${{ hashFiles('mix.lock', 'native/moqx_native/Cargo.lock', '.moq-dev/Cargo.lock') }}
+          key: ${{ runner.os }}-cargo-${{ hashFiles('mix.lock', 'native/moqx_native/Cargo.lock') }}
           restore-keys: |
             ${{ runner.os }}-cargo-
 
       - name: Install Elixir dependencies
         run: mix deps.get
 
-      - name: Build relay binary for integration tests
-        run: cargo build --manifest-path .moq-dev/Cargo.toml --release -p moq-relay
+      - name: Generate integration relay certificates
+        run: scripts/generate_integration_certs.sh .tmp/integration-certs
+
+      - name: Start pinned relay image
+        env:
+          MOQX_RELAY_IMAGE: ghcr.io/moqtail/relay:sha-190e502@sha256:36c929b71140a83158da383721f1d59f199a9f643ab5d033910258f5aa2903ee
+        run: docker compose -f docker-compose.integration.yml up -d relay
 
       - name: Run relay-backed integration tests
+        env:
+          MOQX_EXTERNAL_RELAY_URL: https://localhost:4433
+          MOQX_RELAY_CACERTFILE: .tmp/integration-certs/ca.pem
         run: mix test.integration
+
+      - name: Stop relay image
+        if: always()
+        run: docker compose -f docker-compose.integration.yml down --remove-orphans

.gitignore

@@ -27,3 +27,4 @@ moqx-*.tar
 /target/
 priv/native
 .moq-dev/
+.tmp

AGENTS.md

@@ -1,2 +1,4 @@
 - before committing, ensure you `mix format`, run the tests, run `mix credo` (elixir's linter) and refactor accordingly, ensure documentation stays consistent
 - if you find stale documentation, point it out: we either have to update or remove it
+- this project is based on moqtail. Before making any design decisions, ensure you check out what has been chosen there
+- this project is based on a series of MOQT IETF documents (linked in README), which should always be consulted before making decisions

CHANGELOG.md

@@ -46,6 +46,15 @@ All notable changes to `moqx` will be documented in this file.
   `await_catalog/2`).
 - Mix tasks (`moqx.e2e.pubsub`, `moqx.moqtail.demo`) and integration helpers now
   follow the typed async contract.
+- Integration tests now assume a relay endpoint provided by environment
+  (`MOQX_EXTERNAL_RELAY_URL`) and trusted CA path (`MOQX_RELAY_CACERTFILE`),
+  with Docker-based local/CI orchestration as the primary deterministic path.
+- Removed `:public_relay_live` integration tests; public relay interop checks now
+  live in manual mix tasks.
+- Subscribe availability timeout is now exposed as
+  `rendezvous_timeout_ms` (`delivery_timeout_ms` remains as a deprecated alias)
+  and subscribe request rejections now carry typed `RequestError.code` values
+  (for example `:track_does_not_exist` / `:timeout`).
 
 ### Documentation
 

README.md

@@ -227,7 +227,9 @@ Subscriptions are asynchronous and correlated by subscription handle.
 
 `subscribe/4` options:
 
-- `delivery_timeout_ms` -- MOQT DELIVERY TIMEOUT parameter (`0x02`) in milliseconds.
+- `rendezvous_timeout_ms` -- how long the relay may wait for publisher availability
+  before rejecting subscribe (encoded as MOQT DELIVERY TIMEOUT parameter `0x02`).
+- `delivery_timeout_ms` -- deprecated alias for `rendezvous_timeout_ms`.
 
 The subscription message contract is:
 
@@ -240,7 +242,7 @@ The subscription message contract is:
 - `{:moqx_transport_error, %MOQX.TransportError{op: :subscribe, handle, ...}}`
 
 ```elixir
-{:ok, handle} = MOQX.subscribe(subscriber, "moqtail", "catalog", delivery_timeout_ms: 1_500)
+{:ok, handle} = MOQX.subscribe(subscriber, "moqtail", "catalog", rendezvous_timeout_ms: 1_500)
 
 receive do
   {:moqx_subscribe_ok, %MOQX.SubscribeOk{handle: ^handle}} -> :ok
@@ -487,13 +489,29 @@ For an explicit split between fast checks and integration coverage:
 
 ```bash
 mix ci
-mix test.integration
+scripts/test_integration.sh
 ```
 
 - `mix ci` runs formatting, Credo, and non-integration tests
-- `mix test.integration` runs the deterministic integration suite (excludes `:public_relay_live` tests)
-- run live public-relay coverage explicitly with:
-  `mix test --include integration --include public_relay_live`
+- `scripts/test_integration.sh` starts a pinned relay Docker image with local
+  TLS certs and runs `mix test.integration`
+
+You can override relay version independently from the locally compiled moqtail
+library by setting `MOQX_RELAY_IMAGE`, for example:
+
+```bash
+MOQX_RELAY_IMAGE=ghcr.io/moqtail/relay:sha-190e502 scripts/test_integration.sh
+```
+
+Set a digest-pinned reference for strict reproducibility:
+
+```bash
+MOQX_RELAY_IMAGE='ghcr.io/moqtail/relay:sha-190e502@sha256:36c929b71140a83158da383721f1d59f199a9f643ab5d033910258f5aa2903ee' scripts/test_integration.sh
+```
+
+`mix test.integration` can still be run directly if you provide a relay URL and
+trusted CA path via environment (`MOQX_EXTERNAL_RELAY_URL`,
+`MOQX_RELAY_CACERTFILE`).
 
 ### Local relay TLS
 

docker-compose.integration.yml

@@ -0,0 +1,18 @@
+services:
+  relay:
+    image: ${MOQX_RELAY_IMAGE:-ghcr.io/moqtail/relay:sha-190e502@sha256:36c929b71140a83158da383721f1d59f199a9f643ab5d033910258f5aa2903ee}
+    platform: linux/amd64
+    command:
+      - --host
+      - 0.0.0.0
+      - --port
+      - "4433"
+      - --cert-file
+      - /certs/cert.pem
+      - --key-file
+      - /certs/key.pem
+    ports:
+      - "4433:4433/udp"
+    volumes:
+      - ./.tmp/integration-certs/cert.pem:/certs/cert.pem:ro
+      - ./.tmp/integration-certs/key.pem:/certs/key.pem:ro

lib/moqx.ex

@@ -163,7 +163,8 @@ defmodule MOQX do
 
   @typedoc "Subscribe options accepted by `subscribe/4` and `subscribe_track/4`."
   @type subscribe_opt ::
-          {:delivery_timeout_ms, non_neg_integer()}
+          {:rendezvous_timeout_ms, non_neg_integer()}
+          | {:delivery_timeout_ms, non_neg_integer()}
           | {:init_data, binary()}
           | {:track_meta, map()}
           | {:track, Track.t()}
@@ -693,7 +694,9 @@ defmodule MOQX do
 
   Supported options:
 
-  - `:delivery_timeout_ms` -- MOQT DELIVERY TIMEOUT (parameter type `0x02`) in milliseconds
+  - `:rendezvous_timeout_ms` -- how long the relay may wait for publisher availability
+    before rejecting the request (encoded as MOQT DELIVERY TIMEOUT parameter `0x02`)
+  - `:delivery_timeout_ms` -- deprecated alias for `:rendezvous_timeout_ms`
   - `:init_data` -- binary init segment/configuration to surface in `:moqx_track_init`
   - `:track_meta` -- map surfaced in `:moqx_track_init`
   - `:track` -- `%MOQX.Catalog.Track{}` convenience; fills `:init_data` and `:track_meta`
@@ -715,8 +718,7 @@ defmodule MOQX do
           {:ok, subscription_handle()} | {:error, MOQX.RequestError.t()}
   def subscribe(session, broadcast_path, track_name, opts)
       when is_binary(broadcast_path) and is_binary(track_name) and is_list(opts) do
-    delivery_timeout_ms =
-      opts |> Keyword.get(:delivery_timeout_ms) |> normalize_delivery_timeout_ms!()
+    rendezvous_timeout_ms = normalize_rendezvous_timeout_ms!(opts)
 
     {init_data, track_meta} = normalize_subscribe_track_payload!(opts)
 
@@ -726,7 +728,7 @@ defmodule MOQX do
            session,
            broadcast_path,
            track_name,
-           delivery_timeout_ms,
+           rendezvous_timeout_ms,
            init_data,
            track_meta
          ) do
@@ -790,7 +792,7 @@ defmodule MOQX do
   end
 
   defp validate_subscribe_opts_keys!(opts) do
-    allowed_keys = [:delivery_timeout_ms, :init_data, :track_meta, :track]
+    allowed_keys = [:rendezvous_timeout_ms, :delivery_timeout_ms, :init_data, :track_meta, :track]
 
     case Keyword.keys(opts) -- allowed_keys do
       [] -> :ok
@@ -843,15 +845,33 @@ defmodule MOQX do
           "expected :track_meta to be a map, got: #{inspect(track_meta)}"
   end
 
-  defp normalize_delivery_timeout_ms!(nil), do: nil
+  defp normalize_rendezvous_timeout_ms!(opts) do
+    rendezvous_timeout_ms = Keyword.get(opts, :rendezvous_timeout_ms)
+    delivery_timeout_ms = Keyword.get(opts, :delivery_timeout_ms)
 
-  defp normalize_delivery_timeout_ms!(delivery_timeout_ms)
-       when is_integer(delivery_timeout_ms) and delivery_timeout_ms >= 0,
-       do: delivery_timeout_ms
+    case {rendezvous_timeout_ms, delivery_timeout_ms} do
+      {nil, nil} ->
+        nil
 
-  defp normalize_delivery_timeout_ms!(delivery_timeout_ms) do
+      {timeout_ms, nil} ->
+        normalize_timeout_ms!(timeout_ms, :rendezvous_timeout_ms)
+
+      {nil, timeout_ms} ->
+        normalize_timeout_ms!(timeout_ms, :delivery_timeout_ms)
+
+      {_rendezvous, _delivery} ->
+        raise ArgumentError,
+              "subscribe/4 accepts only one of :rendezvous_timeout_ms or :delivery_timeout_ms"
+    end
+  end
+
+  defp normalize_timeout_ms!(timeout_ms, _key)
+       when is_integer(timeout_ms) and timeout_ms >= 0,
+       do: timeout_ms
+
+  defp normalize_timeout_ms!(timeout_ms, key) do
     raise ArgumentError,
-          "expected :delivery_timeout_ms to be a non-negative integer, got: #{inspect(delivery_timeout_ms)}"
+          "expected #{inspect(key)} to be a non-negative integer, got: #{inspect(timeout_ms)}"
   end
 
   # ---------------------------------------------------------------------------

lib/moqx/native.ex

@@ -41,7 +41,7 @@ defmodule MOQX.Native do
         _session,
         _broadcast_path,
         _track_name,
-        _delivery_timeout_ms,
+        _rendezvous_timeout_ms,
         _init_data,
         _track_meta
       ),

mix.exs

@@ -68,7 +68,7 @@ defmodule MOQX.MixProject do
   defp aliases do
     [
       ci: ["format --check-formatted", "credo --strict", "test --exclude integration"],
-      "test.integration": ["test --only integration --exclude public_relay_live"]
+      "test.integration": ["test --only integration"]
     ]
   end
 end

native/moqx_native/src/lib.rs

@@ -16,7 +16,7 @@ use moqtail::model::common::pair::KeyValuePair;
 use moqtail::model::common::tuple::{Tuple, TupleField};
 use moqtail::model::control::client_setup::ClientSetup;
 use moqtail::model::control::constant::{
-    GroupOrder as MoqGroupOrder, PublishDoneStatusCode, DRAFT_14,
+    GroupOrder as MoqGroupOrder, PublishDoneStatusCode, SubscribeErrorCode, DRAFT_14,
 };
 use moqtail::model::control::control_message::ControlMessage;
 use moqtail::model::control::fetch::{Fetch as MoqFetch, StandAloneFetchProps};
@@ -73,6 +73,14 @@ mod atoms {
         does_not_exist,
         end_of_group,
         end_of_track,
+        internal_error,
+        unauthorized,
+        timeout,
+        not_supported,
+        track_does_not_exist,
+        invalid_range,
+        malformed_auth_token,
+        expired_auth_token,
     }
 }
 
@@ -229,6 +237,19 @@ fn publish_done_status_to_atom(status: PublishDoneStatusCode) -> Atom {
     }
 }
 
+fn subscribe_error_code_to_atom(code: SubscribeErrorCode) -> Atom {
+    match code {
+        SubscribeErrorCode::InternalError => atoms::internal_error(),
+        SubscribeErrorCode::Unauthorized => atoms::unauthorized(),
+        SubscribeErrorCode::Timeout => atoms::timeout(),
+        SubscribeErrorCode::NotSupported => atoms::not_supported(),
+        SubscribeErrorCode::TrackDoesNotExist => atoms::track_does_not_exist(),
+        SubscribeErrorCode::InvalidRange => atoms::invalid_range(),
+        SubscribeErrorCode::MalformedAuthToken => atoms::malformed_auth_token(),
+        SubscribeErrorCode::ExpiredAuthToken => atoms::expired_auth_token(),
+    }
+}
+
 fn runtime() -> &'static Runtime {
     static RT: OnceLock<Runtime> = OnceLock::new();
     RT.get_or_init(|| {
@@ -1003,7 +1024,7 @@ fn dispatch_control_response(msg: ControlMessage, inner: &Arc<SessionInner>) {
                         let payload = RequestErrorOut {
                             op: atoms::subscribe(),
                             message: reason.clone(),
-                            code: None,
+                            code: Some(subscribe_error_code_to_atom(err.error_code)),
                             r#ref: nil,
                             handle: sub_ref.in_env(env),
                         };
@@ -2470,7 +2491,7 @@ fn subscribe<'a>(
     session: ResourceArc<SessionRes>,
     broadcast_path: String,
     track_name: String,
-    delivery_timeout_ms: Option<u64>,
+    rendezvous_timeout_ms: Option<u64>,
     init_data: Option<Binary>,
     track_meta: rustler::Term,
 ) -> rustler::NifResult<rustler::Term<'a>> {
@@ -2484,11 +2505,11 @@ fn subscribe<'a>(
     let inner = session.inner.clone();
     let namespace = normalize_path(&session.root, &broadcast_path);
 
-    let subscribe_parameters = match delivery_timeout_ms {
+    let subscribe_parameters = match rendezvous_timeout_ms {
         Some(timeout_ms) => {
             vec![KeyValuePair::try_new_varint(0x02, timeout_ms).map_err(|e| {
                 rustler::Error::Term(Box::new(format!(
-                    "invalid delivery_timeout_ms parameter: {:?}",
+                    "invalid rendezvous timeout parameter: {:?}",
                     e
                 )))
             })?]