LangExtract.Runner.Limiter (LangExtract v0.11.0)

Copy Markdown View Source

The runner's shared request budget: an RPM token bucket plus an in-flight cap, with global backoff on retry-after.

Chunk tasks call acquire/2 before each HTTP request and release/1 after it completes. Acquirers are monitored — a killed task (stream halt, crash) releases its slot automatically, so budget can never leak. release_and_pause/2 holds all admission until a deadline: one 429 informs every in-flight chunk instead of N requests independently colliding with the same exhausted window.

Emits [:lang_extract, :limiter, :wait] whenever an acquire had to wait, with the wait duration, the reason that blocked it first (:rpm | :in_flight | :retry_after), and the limiter pid.

Tokens refill lazily from elapsed time — no timer ticks. The clock is injectable (:clock, a zero-arity fun returning milliseconds) so tests run on virtual time.

Internal — no stability guarantees; see the README's "Stability" section. Documented because it explains how the library works, not because it is API.

Summary

Functions

Blocks until a request slot and a token are granted.

Returns a specification to start this module under a supervisor.

Releases the caller's in-flight slot after its request completes.

Releases the caller's slot and pauses all admission for ms milliseconds (a retry-after deadline) in one cast.

Types

option()

@type option() ::
  {:rpm, pos_integer() | :infinity}
  | {:max_in_flight, pos_integer()}
  | {:clock, (-> integer())}
  | {:name, GenServer.name()}

Functions

acquire(limiter, timeout \\ :infinity)

@spec acquire(GenServer.server(), timeout()) :: :ok

Blocks until a request slot and a token are granted.

Returns :ok. The caller is monitored until it calls release/1 (or dies, which releases implicitly).

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

release(limiter)

@spec release(GenServer.server()) :: :ok

Releases the caller's in-flight slot after its request completes.

release_and_pause(limiter, ms)

@spec release_and_pause(GenServer.server(), non_neg_integer()) :: :ok

Releases the caller's slot and pauses all admission for ms milliseconds (a retry-after deadline) in one cast.

Used on 429: separate release then pause casts would let admit_waiting grant queued work in the gap before the pause is visible. One message applies both, then runs admission under the new deadline. Releasing a pid with no slot is a no-op, so the pause half stands alone for callers that hold nothing.

Repeated pauses extend to the furthest deadline; they never shorten it. The deadline is honored verbatim: a long quota-reset retry-after stalls admission until it expires (callers cap synthesized backoff at the source — see LangExtract.Runner.Request).

start_link(opts)

@spec start_link([option()]) :: GenServer.on_start()