One ffprobe per in-flight source, shared by everyone who asks about it.
The single-flight in front of AudioProxy.Ffprobe.probe/2, in the shape
AudioProxy.RenderCoordinator already uses for renders: a Registry of
unique identities, a coordinator process per identity, a DynamicSupervisor
holding them, and a broadcast to everyone waiting.
Without it, N concurrent requests for one variant spawn N probes and exactly one render — the registry that exists to stop duplicate work sits behind the gate that now does some. Worse for the case the audio-only policy is about: N concurrent requests for a refused source spawn N probes and no render at all, so the render queue, which is what sheds load, never sees them.
The identity is the source, not the cache key
This is the one place this module departs from the render coordinator, and it
is deliberate. A render is identified by its cache key, because the bytes it
produces depend on every option in the URL. A probe reads container headers,
and what it finds depends on the source and nothing else — f:mp3/br:128
and f:opus of one file ask ffprobe the identical question about the
identical bytes.
So the key is AudioProxy.Source.canonical/1, which buys two things a
cache-key identity would not:
- concurrent requests for different variants of one source share a probe, which is the common shape when a client is fetching several renditions;
/infoshares the mechanism with the render gate, as it must, since/infohas no variant and therefore no cache key to be identified by.
The staleness window is unchanged either way: it is one probe's lifetime, and the render that follows reads the same bytes the probe did.
Asking
probe/3 either starts the probe or attaches to the running one, and returns
what AudioProxy.Ffprobe.probe/2 would have returned — {:ok, decoded_json}
or {:error, reason} in that module's own vocabulary — plus one reason of its
own, {:error, {:queue_full, retry_after}}, when AudioProxy.ProbeLimiter
had no slot. AudioProxy.ErrorJSON already renders that tuple as §5's 429.
A caller cannot tell whether it started the probe or joined one, and has no reason to: it is waiting for a verdict either way.
The slot comes first, and only a spawn takes one
A coordinator takes an AP_MAX_PROBE_CONCURRENCY slot in init/1, before it
spawns anything, and releases it the moment the verdict arrives — before the
linger below, because what the linger holds is a decoded map and a map costs
no CPU.
Taking the slot in the coordinator rather than in the caller is what makes
coalescing and the ceiling compose: requests that coalesce onto a running
probe do not consume the ceiling, so the bound counts ffprobe processes
rather than requests, exactly as the render semaphore counts encoders rather
than subscribers.
The start race
Starting and joining are the same call, as in AudioProxy.RenderCoordinator:
DynamicSupervisor.start_child/2 either succeeds — this caller spawned the
probe, and its pid was in the waiter list before init/1 returned, so no
verdict can be broadcast before it is listening — or answers
{:error, {:already_started, pid}}, which is a join. Losing the race to a
coordinator that is stopping is a caught exit and a retry, which starts a
fresh probe.
The probe runs beside the coordinator, not inside it
Ffprobe.probe/2 blocks: it spawns the subprocess with itself as consumer and
collects chunks until the output ends. A GenServer cannot do that and still
answer the joins that are coalescing onto it — the same reason
AudioProxy.RenderCoordinator asks the semaphore with request/1 rather than
acquire/1.
So the coordinator spawns a monitored runner process which calls
Ffprobe.probe/2 and sends the result back. The runner owns the subprocess,
which means the kill discipline and AP_PROBE_TIMEOUT are unchanged and still
live in one place. A runner that dies without reporting is a DOWN, and every
waiter is told :probe_failed — a probe that produced no verdict is not a
verdict.
Teardown
Two ways this ends, and they differ in what happens to the identity:
- A verdict — broadcast, then the coordinator lingers briefly, still registered, so a request arriving a moment later is answered from it without a second spawn. The slot is already back.
- A failure — broadcast to every waiter, then the identity is unregistered immediately, so the next request probes again rather than attaching to a corpse. Same discipline the render coordinator uses, and the same reason: a failed probe is not a cached "no".
Waiters are deliberately not monitored, which is where this diverges from
the render coordinator's lifecycle. There, the last subscriber leaving must
cancel a subprocess that could otherwise run for minutes; here the runner is
bounded by AP_PROBE_TIMEOUT and finishes in tens of milliseconds, so a
monitor per waiter would buy a lifecycle to maintain and at most a few
milliseconds of a slot. Sending a verdict to a pid that has since exited is
harmless.
Summary
Types
Why a probe produced no verdict. AudioProxy.Ffprobe's, plus a full pool.
Functions
The registry and supervisor this module needs, for the application tree.
Probes input, or joins the probe already running for identity.
Types
@type error_reason() :: AudioProxy.Ffprobe.error_reason() | {:queue_full, pos_integer()}
Why a probe produced no verdict. AudioProxy.Ffprobe's, plus a full pool.
Functions
@spec children() :: [Supervisor.child_spec() | {module(), term()}]
The registry and supervisor this module needs, for the application tree.
Start them after AudioProxy.ProbeLimiter: a coordinator takes a slot in
init/1 and releases one on its way out, so the thing it releases into has to
outlive it, which reverse-order shutdown gives for free.
@spec probe(String.t(), String.t(), keyword()) :: {:ok, map()} | {:error, error_reason()}
Probes input, or joins the probe already running for identity.
identity is what two callers must agree on to share a probe —
AudioProxy.Source.canonical/1, per The identity is the source above.
input and opts are AudioProxy.Ffprobe.probe/2's, and are used only by
the caller that turns out to be starting the probe: a joiner's are ignored,
because the probe it is joining has already been spawned. That matters in
exactly one place — a test whose two callers pass different :executable
stand-ins for one source get whichever spawned first — and nowhere in
production, where both are derived from the resolved source.