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-uprender 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/2 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/1returns only once that has happened, which is what makes it usable as a barrier the wayRender.cancel/1is.
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.
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
@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.
@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.
@type status() :: :miss | :coalesced
Whether this subscriber started the render or attached to it.
@type t() :: pid()
The coordinator, which is also the handle in the messages above.
Functions
@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.
@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.
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.
@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.