Complete MOQT draft-16 subscriber lifecycle

Author: dmorn

README.md

@@ -106,9 +106,41 @@ end
 ```
 
 This path negotiates ALPN `moqt-16`, sends native-QUIC `PATH` and `AUTHORITY`
-setup parameters, and decodes draft-16 subgroup objects. The catalog payload is
-currently delivered as `ObjectReceived`; CMSF catalog parsing is tracked
-separately.
+setup parameters, and decodes draft-16 subgroup streams and object datagrams.
+Objects preserve extension headers and end-of-group metadata. The catalog
+payload is currently delivered as `ObjectReceived`; CMSF catalog parsing is
+tracked separately.
+
+Draft-16 also accepts the complete protocol-neutral filter model:
+
+```elixir
+filter = %MOQX.SubscriptionFilter{
+  type: :absolute_range,
+  start_location: {12, 4},
+  end_group: 20
+}
+
+{:ok, subscription} =
+  MOQX.subscribe(client, track,
+    filter: filter,
+    priority: 127,
+    group_order: :ascending,
+    delivery_timeout: 5_000
+  )
+
+:ok =
+  MOQX.update_subscription(client, subscription,
+    start: :next_group,
+    priority: 64
+  )
+```
+
+The relative `:start` policies remain the portable API shared with Cloudflare.
+Absolute start/range filters, request updates, datagrams, and accepted
+subscription parameters are currently implemented by `:draft_16`. Update
+success and rejection arrive as `SubscriptionUpdated` and
+`SubscriptionUpdateFailed`; an update rejection leaves the subscription
+active.
 
 Catalog tracks can be subscribed directly. Delivered objects retain their
 subscription, group, subgroup, object, and priority coordinates:
@@ -156,6 +188,10 @@ ffmpeg -v error -f h264 -i /tmp/cloudflare-bbb.h264 -f null -
 only after every stream advertised by `PUBLISH_DONE` has been processed or the
 subscription's `:delivery_timeout` has elapsed.
 
+Objects are emitted in transport arrival order. No global ordering across
+independent subgroup streams is claimed until
+[issue #29](https://github.com/dmorn/moqx/issues/29) records that policy.
+
 All application-facing output uses typed `MOQX.Event.*` structs inside the
 stable `{:moqx, client, event}` envelope. By default events go to the process
 that calls `MOQX.connect/2`; shared connection owners can choose a router:

docs/adr/0010-compose-versioned-wire-packages-into-explicit-protocol-implementations.md

@@ -22,17 +22,18 @@ explicit protocol-selection decision.
 
 The second active implementation is standard MOQT draft-16, selected as
 `:draft_16` and implemented by `MOQX.Protocol.Draft16`. It coexists with
-Cloudflare draft-14 rather than replacing it. The first landed slice covers
-native QUIC ALPN `moqt-16`, draft-16 setup with `PATH` and `AUTHORITY`,
-subscription using relative start policies, subscription acceptance/error, and
-subgroup object delivery through the existing typed public API.
+Cloudflare draft-14 rather than replacing it. Its subscriber path covers native
+QUIC ALPN `moqt-16`, strict draft-16 setup and request credit, all four
+protocol-neutral subscription filters, request updates, subscription
+acceptance/error/completion, subgroup streams, object datagrams, extension
+preservation, and stream draining through the existing typed public API.
 
 The normative wire reference is
 `draft-ietf-moq-transport-16`. Interoperability behavior is checked against
 Moqtail's `draft-16` branch pinned at
 `c2ff7253479c6a0d7c8282a1cad289d591ebc302` and its public
-`relay.moqtail.dev` endpoint. CMSF catalog decoding and draft-16 publication
-remain separate incremental work.
+`relay.moqtail.dev` endpoint. CMSF catalog decoding, cross-stream ordering
+policy, and draft-16 publication remain separate incremental work.
 
 ## Context
 
@@ -150,9 +151,10 @@ Implementation namespaces follow this shape:
 - a future MOQ Lite implementation under `MOQX.Protocol.MOQLite04`.
 
 Cloudflare draft-14 has subscriber and publisher support. Standard draft-16
-currently has its first subscriber slice. Concrete implementations are
-independent; version and deployment differences must not accumulate as
-conditionals in one global state machine.
+has a complete subscriber lifecycle, subject to the separate cross-stream
+ordering decision. Concrete implementations are independent; version and
+deployment differences must not accumulate as conditionals in one global state
+machine.
 
 ### Versioned wire packages
 
@@ -270,8 +272,8 @@ namespace publication. Secret lookup, issuance, permission, and rotation stay
 outside the library. Secret values and sensitive encoded actions have redacted
 inspection and are unwrapped only at the driver's transport-send boundary.
 
-Datagram delivery and other protocol implementations remain incremental work
-behind the same boundaries.
+Other protocol implementations remain incremental work behind the same
+boundaries.
 
 ## Consequences
 

lib/moqx.ex

@@ -22,8 +22,21 @@ defmodule MOQX do
   @typedoc "Option accepted by `subscribe/3`."
   @type subscription_option ::
           {:start, subscription_start()}
+          | {:filter, MOQX.SubscriptionFilter.t()}
           | {:priority, 0..255}
-          | {:delivery_timeout, non_neg_integer()}
+          | {:group_order, :ascending | :descending}
+          | {:delivery_timeout, pos_integer()}
+          | {:parameters, [MOQX.SubscriptionParameter.t()]}
+
+  @typedoc "Option accepted by `update_subscription/3`."
+  @type subscription_update_option ::
+          {:start, subscription_start()}
+          | {:filter, MOQX.SubscriptionFilter.t()}
+          | {:priority, 0..255}
+          | {:delivery_timeout, pos_integer()}
+          | {:forward, boolean()}
+          | {:new_group, non_neg_integer()}
+          | {:parameters, [MOQX.SubscriptionParameter.t()]}
 
   @doc "Returns the default native QUIC transport implementation."
   @spec transport() :: module()
@@ -63,6 +76,17 @@ defmodule MOQX do
     ConnectionDriver.subscribe(client, track, options)
   end
 
+  @doc "Updates an active subscription's draft-neutral filter and delivery parameters."
+  @spec update_subscription(
+          MOQX.Client.t(),
+          MOQX.Subscription.t(),
+          [subscription_update_option()]
+        ) ::
+          :ok | {:error, term()}
+  def update_subscription(client, subscription, options) do
+    ConnectionDriver.update_subscription(client, subscription, options)
+  end
+
   @doc "Ends an active subscription and sends the selected protocol's unsubscribe message."
   @spec unsubscribe(MOQX.Client.t(), MOQX.Subscription.t()) :: :ok | {:error, term()}
   def unsubscribe(client, subscription) do

lib/moqx/event.ex

@@ -3,6 +3,8 @@ defmodule MOQX.Event do
 
   @type t ::
           MOQX.Event.SubscriptionAccepted.t()
+          | MOQX.Event.SubscriptionUpdated.t()
+          | MOQX.Event.SubscriptionUpdateFailed.t()
           | MOQX.Event.SubscriptionFailed.t()
           | MOQX.Event.SubscriptionDone.t()
           | MOQX.Event.ObjectReceived.t()
@@ -19,6 +21,24 @@ defmodule MOQX.Event do
           | MOQX.Event.ProtocolFailed.t()
 end
 
+defmodule MOQX.Event.SubscriptionUpdated do
+  @moduledoc "The relay accepted a subscription parameter update."
+  @enforce_keys [:subscription]
+  defstruct [:subscription, parameters: []]
+
+  @type t :: %__MODULE__{
+          subscription: MOQX.Subscription.t(),
+          parameters: [MOQX.SubscriptionParameter.t()]
+        }
+end
+
+defmodule MOQX.Event.SubscriptionUpdateFailed do
+  @moduledoc "The relay rejected a subscription update while leaving the subscription active."
+  @enforce_keys [:subscription, :error]
+  defstruct [:subscription, :error]
+  @type t :: %__MODULE__{subscription: MOQX.Subscription.t(), error: MOQX.ProtocolError.t()}
+end
+
 defmodule MOQX.Event.PublicationSubscriptionRequested do
   @moduledoc "An inbound publisher subscription is waiting for an application decision."
   @enforce_keys [:request]
@@ -41,8 +61,13 @@ end
 defmodule MOQX.Event.SubscriptionAccepted do
   @moduledoc "The relay accepted a subscription."
   @enforce_keys [:subscription]
-  defstruct [:subscription]
-  @type t :: %__MODULE__{subscription: MOQX.Subscription.t()}
+  defstruct [:subscription, parameters: [], track_extensions: []]
+
+  @type t :: %__MODULE__{
+          subscription: MOQX.Subscription.t(),
+          parameters: [MOQX.SubscriptionParameter.t()],
+          track_extensions: [MOQX.Extension.t()]
+        }
 end
 
 defmodule MOQX.Event.SubscriptionFailed do

lib/moqx/extension.ex

@@ -0,0 +1,12 @@
+defmodule MOQX.Extension do
+  @moduledoc "Protocol-specific extension information retained losslessly at a public boundary."
+
+  @enforce_keys [:protocol, :identifier, :value]
+  defstruct [:protocol, :identifier, :value]
+
+  @type t :: %__MODULE__{
+          protocol: atom(),
+          identifier: non_neg_integer(),
+          value: non_neg_integer() | binary()
+        }
+end

lib/moqx/object.ex

@@ -15,6 +15,8 @@ defmodule MOQX.Object do
     :object_id,
     :publisher_priority,
     :status,
+    :extensions,
+    :end_of_group?,
     :payload
   ]
 
@@ -25,6 +27,8 @@ defmodule MOQX.Object do
           object_id: non_neg_integer(),
           publisher_priority: 0..255 | nil,
           status: :object_does_not_exist | :end_of_group | :end_of_track | nil,
+          extensions: [MOQX.Extension.t()] | nil,
+          end_of_group?: boolean() | nil,
           payload: binary()
         }
 end

lib/moqx/operation.ex

@@ -7,6 +7,7 @@ defmodule MOQX.Operation do
 
   @type t ::
           Subscribe.t()
+          | UpdateSubscription.t()
           | Unsubscribe.t()
           | Publish.t()
           | AddTrack.t()
@@ -34,6 +35,15 @@ defmodule MOQX.Operation do
     @type t :: %__MODULE__{subscription: term()}
   end
 
+  defmodule UpdateSubscription do
+    @moduledoc "Updates the parameters or object boundary of an active subscription."
+
+    @enforce_keys [:subscription]
+    defstruct [:subscription, options: []]
+
+    @type t :: %__MODULE__{subscription: MOQX.Subscription.t(), options: keyword()}
+  end
+
   defmodule Publish do
     @moduledoc "Advertises one application-level track namespace."