StarView.StreamRegistry (star_view v0.7.0)

Copy Markdown View Source

Opt-in per-tab stream ownership with atomic claims and bounded handover.

Add StarView.StreamRegistry to the supervision tree. Each {scope_key, tabId} has one active owner. A new claim retires the previous generation, signals it to stop, and escalates only if that generation has not released within :grace_ms (default 1,000ms).

Put the per-tab signal in the root layout:

<body data-signals:tab-id__ifmissing="crypto.randomUUID()">

A per-document id avoids copied sessionStorage ids making duplicated tabs replace each other's streams. It survives reconnects within the document; a full navigation/reload gets a new id. This is not durable tab-state storage.

Prefer StarView.Stream.open/2 and StarView.Stream.run/2, which release the exact claim on teardown and ignore stale takeover messages on HTTP/1.1 keep-alive processes. Manual streams must call release/1 when they finish. start_stream/2 retains the legacy :replaced exit reason; StarView.Stream opts into generation-tagged replacement messages.

Adapted from dstar (MIT), Copyright (c) 2025 Rico Trevisan.

Summary

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

owner(key)

Returns the active owner and its generation, or :error when absent.

release(conn)

Synchronously releases the exact claim stored in conn. A stale conn cannot release a newer generation. Once this returns, the released generation's escalation cannot kill a reused keep-alive process. Unkeyed conns are a no-op.

replace_and_register(key)

Atomically claims a manual stream key. Returns :ok only when the caller became active owner; coordinator failures return {:error, reason}. Manual callers retain the legacy :replaced exit reason and must call unregister_self/0 before reusing their request process.

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

Starts SSE, atomically claiming {scope_key, tabId} when a usable tab id is present. Malformed signals return halted HTTP 400, oversized signals 413, and an unavailable coordinator 503, before SSE starts. Missing or invalid tab ids retain the unkeyed fallback. Accepts :max_bytes for raw signals.

unregister_self()

Synchronously releases all claims owned by the calling process.

unregister_stream(conn)

Releases a stream's claim and returns the conn for manual pipelines.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

owner(key)

@spec owner(term()) :: {:ok, pid(), reference()} | :error

Returns the active owner and its generation, or :error when absent.

release(conn)

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

Synchronously releases the exact claim stored in conn. A stale conn cannot release a newer generation. Once this returns, the released generation's escalation cannot kill a reused keep-alive process. Unkeyed conns are a no-op.

replace_and_register(key)

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

Atomically claims a manual stream key. Returns :ok only when the caller became active owner; coordinator failures return {:error, reason}. Manual callers retain the legacy :replaced exit reason and must call unregister_self/0 before reusing their request process.

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

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

Starts SSE, atomically claiming {scope_key, tabId} when a usable tab id is present. Malformed signals return halted HTTP 400, oversized signals 413, and an unavailable coordinator 503, before SSE starts. Missing or invalid tab ids retain the unkeyed fallback. Accepts :max_bytes for raw signals.

unregister_self()

@spec unregister_self() :: :ok

Synchronously releases all claims owned by the calling process.

unregister_stream(conn)

Releases a stream's claim and returns the conn for manual pipelines.