Dstar.Utility.StreamRegistry (dstar v0.3.0)

Copy Markdown View Source

Opt-in per-tab stream deduplication with linearizable claims.

The registry is a single coordinator process. It serializes the small, non-blocking state transition for every claim, while handover remains asynchronous and bounded per key. A successful claim means the caller owns the key at that linearization point; later contenders replace it in the same total order.

Problem

With full-page navigation, SSE stream processes do not learn that the client disconnected until their next write. Stale processes keep subscriptions, perform wasted work, and can exhaust HTTP/1.1's per-origin connection limit.

Setup

Add the registry to your application's supervision tree:

children = [
  Dstar.Utility.StreamRegistry,
  # ...
]

Then put a per-tab id in your root layout:

<body data-signals:tab-id="sessionStorage.getItem('_ds_tab') || (() => { const id = crypto.randomUUID(); sessionStorage.setItem('_ds_tab', id); return id; })()">

sessionStorage is per-tab. Use data-signals:tab-id, not tabId (HTML lowercases attribute names and Datastar camelizes the hyphenated form), and do not prefix the signal with _ because Datastar keeps underscore-prefixed signals client-side.

Claim and handover contract

start_stream/2,3 reads the validated tabId, atomically claims {scope_key, tab_id}, and only then starts SSE. Missing or invalid tabId intentionally falls back to an ordinary, non-deduplicated stream for rollout compatibility. A valid keyed request whose claim fails instead returns a halted, non-SSE 503 response — it never starts an untracked stream.

On replacement, the old holder receives a generation-tagged exit signal. Dstar.Stream.run/2 (which Dstar.Page streams run on) recognizes only the generation stored on its conn, synchronously releases it, and then performs graceful application teardown. A trapping holder that does not release is killed after the grace period. Release and escalation are serialized by this coordinator, so a release that has returned cannot be followed by a stale kill against a keep-alive process now serving unrelated work. Stale replacement messages from an older generation cannot stop a newer stream in the same process.

The coordinator monitors every claim, so process death also cleans it up. It remains the package's only opt-in process.

Summary

Types

claim()
key()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

release(conn)

Synchronously releases the exact stream claim stored on conn.

replace_and_register(key)

Replaces the active owner of key and atomically claims it for the caller.

start_stream(conn, scope_key, opts \\ [])

Starts an SSE stream with per-tab deduplication.

tab_id(conn)

Returns a fetched signal map's (or already-fetched conn's) usable tabId.

unregister_self()

Releases every active or handing-over claim owned by the calling process.

Types

claim()

@type claim() :: reference()

key()

@type key() :: term()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

release(conn)

@spec release(Plug.Conn.t()) :: :ok

Synchronously releases the exact stream claim stored on conn.

This is the preferred teardown operation for keyed streams. A stale conn can release only its own generation and can never clear a newer claim held by a reused keep-alive process. Unkeyed conns are a no-op.

replace_and_register(key)

@spec replace_and_register(key()) :: :ok | {:error, term()}

Replaces the active owner of key and atomically claims it for the caller.

Returns :ok only after the coordinator has made the caller the active owner at a linearization point. Concurrent contenders are ordered by the coordinator; each may succeed and then be replaced by a later claim. The displaced generation remains tracked through its bounded handover until it releases or is escalated.

The holder's internal exit signal is {:EXIT, registry_pid, {:replaced, claim_generation}}. Dstar.Stream (and so Dstar.Page) validates the generation and presents the legacy {:EXIT, registry_pid, :replaced} shape to application callbacks. Controllers should use Dstar.Stream; hand-rolled loops should use start_stream/2,3 and release/1 rather than consuming this internal signal directly.

start_stream(conn, scope_key, opts \\ [])

@spec start_stream(Plug.Conn.t(), term(), keyword()) :: Plug.Conn.t()

Starts an SSE stream with per-tab deduplication.

Signals are fetched through Dstar.Signals.fetch/2. Malformed/non-object input receives 400 and oversized input receives 413 before any claim or SSE start. Pass max_bytes: n as the third argument to change that input limit.

With a usable tabId, the claim is fail-closed: success stores its ownership generation in conn.private and starts SSE; coordinator failure returns a halted plain-text 503 conn without calling Dstar.start/1. With no usable tabId, the documented non-deduplicated fallback still starts ordinary SSE.

tab_id(conn)

@spec tab_id(Plug.Conn.t() | map()) :: String.t() | nil

Returns a fetched signal map's (or already-fetched conn's) usable tabId.

A usable id is a non-whitespace binary of 1..64 bytes. The exact client-supplied value becomes part of the key; no normalization occurs. This helper does not own raw bodies — call start_stream/2,3 or Dstar.Signals.fetch/2 first when params are unfetched.

unregister_self()

@spec unregister_self() :: :ok

Releases every active or handing-over claim owned by the calling process.

The call is synchronous. Once it returns, no escalation for a released generation can kill that process. This remains as a compatibility helper for code that can hold multiple manual claims; new hand-rolled loops should call release/1 with their stream conn. It is a no-op when the opt-in coordinator is not running.