AudioProxy.RenderCoordinator (audio_proxy v0.4.0)

Copy Markdown View Source

One render per in-flight cache key, shared by everyone who asks for it.

API doc §5: concurrent requests for the same not-yet-cached variant must not each spawn an encoder. This is the single-flight in front of AudioProxy.Ffmpeg.Render — a Registry of unique cache keys, a coordinator process per key, and a DynamicSupervisor holding them.

Subscribing

subscribe/2 either starts the render or attaches to the running one:

{:ok, :miss, render, []}          # this caller started it
{:ok, :coalesced, render, chunks} # it was already running; `chunks` is the catch-up

render is the coordinator's pid and stands where AudioProxy.Ffmpeg.Render's did. It broadcasts the pipeline's own contract, with itself as the handle:

  • {:chunk, render, binary}
  • {:done, render, %{exit_status: 0}}
  • {:error, render, failure}

So a consumer written against the pipeline works here unchanged, except for the two calls that name a handle: Render.ack/2 has no counterpart (see Backlog below) and Render.cancel/1 becomes unsubscribe/1, because a subscriber leaving must not cancel a render other subscribers are still reading.

The slot comes first

A coordinator does not spawn its render until AudioProxy.Semaphore grants it a slot, and releases that slot from terminate/2 — so the cap counts renders rather than requests, and a variant twenty clients are waiting on costs one.

It asks with Semaphore.request/1 rather than the blocking acquire/1, because this process must keep answering joins while it waits: the requests coalescing onto a queued render are exactly the ones that should not each take a slot of their own. So a coordinator has one phase more than the render does — :queued, before :rendering — and a subscriber cannot tell them apart, which is the point. It is waiting for bytes either way.

The slot is asked for in init/1, which is what makes a full queue a start failure rather than a render failure: subscribe/2 answers {:error, {:queue_full, retry_after}} and no coordinator is left registered under that key, so the next request asks the semaphore again instead of joining something that is never going to render. AudioProxy.ErrorJSON already renders that tuple as §5's 429, Retry-After and all.

The start race

Starting and joining are the same call. DynamicSupervisor.start_child/2 either succeeds — this caller is the MISS, and its pid is already in the subscriber list before init/1 returns, so no chunk can be produced before it is listening — or answers {:error, {:already_started, pid}}, which is the COALESCED path and a join.

The join is a call that returns the backlog and registers the subscriber in one callback. That atomicity is the whole seam: a joiner cannot be handed a backlog and then miss the chunk that was broadcast while it was being handed one, and cannot be registered first and receive a chunk it will also find in its backlog.

Losing the race to a coordinator that is stopping — a finished render past its linger, a failed one that has already unregistered — is a caught exit and a retry, which starts a fresh render.

Backlog, and the memory bound

Everything the render produces is retained, in order, so a subscriber that joins at any point receives the whole stream. Retaining it is not a cost the coalescing imposes: the variant-cache slice tees the same bytes to storage.

Because the bytes are held anyway, the pipeline is acknowledged the moment a chunk arrives, and the pipeline's own high-water mark stops being the bound. What bounds this instead is AP_MAX_VARIANT_BYTES: a render whose output passes it fails for every subscriber rather than growing without limit. It is a ceiling on the variant, separate from AP_MAX_SRC_BYTES's ceiling on the source, so a deployment can accept two-hour masters and still bound one render's retention to a preview's worth of bytes. It defaults to the effective AP_MAX_SRC_BYTES, so a deployment that sets neither is bounded exactly where it was before the two were separated.

The breach is only detectable once the response has committed to 200 and begun streaming, so it is a failed request rather than a 413 — the source ceiling stays the only one enforced before a render starts. That is the reason the escalation is written down: the named-pipe (FIFO) pattern in AudioProxy.Ffmpeg.Render's moduledoc spools instead of retaining, and nothing in the contract above changes when it lands.

Raising the ceiling does not buy capacity. It bounds one render while the bill is AP_MAX_CONCURRENCY × backlog, so raising it licenses every slot to reach the larger figure — which turns one killed render into an exhausted container. docs/capacity.md has the arithmetic.

Teardown

Three ways this ends, and they differ in what happens to the key:

  • Done — the completion is broadcast, then the coordinator lingers briefly, still registered, so a request that passed its cache check a moment too late still gets the finished bytes instead of re-rendering.
  • Failure — broadcast to every current subscriber, then the key is unregistered immediately, so the next request retries rather than attaching to a corpse.
  • Last subscriber gone — nobody is listening, so the subprocess is cancelled. unsubscribe/1 returns only once that has happened, which is what makes it usable as a barrier the way Render.cancel/1 is.

One subscriber dying never affects the others: it is removed and the render continues.

The write-back tee is a subscriber

When a variant store is configured and the spec carries :metadata, the coordinator starts an AudioProxy.VariantStore.Tee and registers it like any other subscriber — before the first chunk can exist, so the store receives the whole stream. That one fact is the disconnect policy: with a store, the last client leaving still leaves the tee counted, so the render completes into the store and the next request is a HIT; without one, the subscriber count reaches zero and the render is cancelled as before. A tee that dies is forgotten like any subscriber — and if no client remains either, the render is cancelled, because nothing is left that could profit from it.

It starts with the render, not with the coordinator, which is only visible while a slot is being waited for. A queued coordinator has no tee, so a queued render every client has abandoned stops rather than holding its place to encode for the cache alone — which would spend a slot, under exactly the contention that made it queue, on a render no client wants.

Summary

Types

What a subscriber receives.

How to render, when this subscriber turns out to be the one starting it.

Whether this subscriber started the render or attached to it.

t()

The coordinator, which is also the handle in the messages above.

Functions

The registry and supervisor this module needs, for the application tree.

How many renders are in flight right now.

Subscribes the calling process to the render for cache_key, starting it from spec if it is not already running.

Detaches the calling process from render.

Types

message()

@type message() ::
  {:rendering, t()}
  | {:chunk, t(), binary()}
  | {:done, t(), %{exit_status: 0}}
  | {:error, t(), AudioProxy.Ffmpeg.Render.failure()}

What a subscriber receives.

The pipeline's own three, plus {:rendering, _} — sent only to subscribers that were attached while this coordinator was waiting for a render slot, and meaning "bytes are now possible". A consumer that measures how long the render has been silent needs it, because until then there was no render. One that does not may ignore it.

render_spec()

@type render_spec() :: keyword()

How to render, when this subscriber turns out to be the one starting it.

The options of AudioProxy.Ffmpeg.Render.start_link/1 less :consumer, which is the coordinator itself — plus :metadata (AudioProxy.VariantStore.metadata/0), which never reaches the pipeline: it is what the write-back tee stores alongside the bytes, and without it no tee is started.

status()

@type status() :: :miss | :coalesced

Whether this subscriber started the render or attached to it.

t()

@type t() :: pid()

The coordinator, which is also the handle in the messages above.

Functions

children()

@spec children() :: [Supervisor.child_spec() | {module(), term()}]

The registry and supervisor this module needs, for the application tree.

Start them after AudioProxy.Ffmpeg.RenderSupervisor: a coordinator spawns a render in init/1, and shutdown then tears the coordinators down first.

in_flight()

@spec in_flight() :: non_neg_integer()

How many renders are in flight right now.

One per registered cache key, which is the same thing AP_MAX_CONCURRENCY counts: the twenty requests coalesced onto one encode are one render here. AudioProxy.Metrics samples it per scrape rather than maintaining a gauge from the render events, because a registry entry cannot outlive the process that holds it — an abnormal exit corrects this number for free, where an event-driven counter would be permanently one too high.

Briefly includes a coordinator that has finished and not yet unregistered, which is the same window that lets a late request coalesce onto its backlog.

subscribe(cache_key, spec)

@spec subscribe(String.t(), render_spec()) ::
  {:ok, status(), t(), [binary()]} | {:error, term()}

Subscribes the calling process to the render for cache_key, starting it from spec if it is not already running.

Returns {:ok, status, render, backlog}. backlog is the bytes produced before this subscriber attached, in order, and is always empty for a :miss; a consumer must deliver it before the chunks that follow.

Errors are the pipeline's own — {:error, :ffmpeg_not_found} and friends — plus {:error, {:queue_full, retry_after}} when AudioProxy.Semaphore had neither a slot nor room to wait, and {:error, :coordinator_unavailable} if the start-or-join loop could not settle, which means something is repeatedly killing coordinators.

unsubscribe(render)

@spec unsubscribe(t()) :: :ok

Detaches the calling process from render.

The render continues for whoever is left. If nobody is, it is cancelled and its subprocess killed before this returns — so, like AudioProxy.Ffmpeg.Render.cancel/1, this is a barrier and not a request.

Detaching from a coordinator that has already stopped is :ok: that is the outcome asked for.