Bounded delivery of supervised chunk results to a lazy caller-side stream.
The consumer's demand drives admission: at most :buffer tasks are
outstanding, so the caller's mailbox never holds more than :buffer
undelivered results (plus their :DOWN notices). A slow consumer
throttles admission instead of growing a mailbox — the normative
bounded-delivery constraint from the production-pipeline design.
Task crashes become per-chunk events via the monitor :DOWN reason;
halting the stream early kills all outstanding tasks.
Why hand-rolled
Without drain, this module should collapse into
Task.Supervisor.async_stream_nolink(..., max_concurrency: buffer, ordered: false, zip_input_on_exit: true) — the bounded/unordered/
crash-to-event core duplicates it. Drain is what it can't express:
reporting never-started chunks requires detecting the dead supervisor
at admission time. Layering that detection on top was also considered
and rejected: a Stream.transform can't catch the supervisor-down
exit (it erupts between elements, inside the inner enumerable's
reduction), so the wrapper needs continuation-driving via
Enumerable.reduce plus OTP exit-shape matching plus reconstructing
the never-started set by diffing observed byte ranges — more fragile
than this explicit loop, and dependent on uncontracted stdlib event
ordering. If drain semantics are ever dropped, delete this module in
favor of async_stream_nolink.
Drain
When the runner shuts down mid-stream (deploy, supervisor restart),
chunk tasks trap exits, so the supervisor's shutdown signal gives each
in-flight request up to :shutdown milliseconds to finish naturally —
completed results still reach the consumer. Chunks never started are
emitted as {:error, %ChunkError{reason: :drained}}, detected when
admission hits the dead task supervisor. Consumers already handle
per-chunk errors, so drain adds no new consumer code paths.
Internal — no stability guarantees; see the README's "Stability" section. Documented because it explains how the library works, not because it is API.
Summary
Types
@type event() :: {:ok, LangExtract.ChunkResult.t()} | {:error, LangExtract.ChunkError.t()}
Functions
@spec stream_events( Supervisor.supervisor(), [LangExtract.Chunker.Chunk.t()], (LangExtract.Chunker.Chunk.t() -> {:ok, LangExtract.ChunkResult.t()} | {:error, LangExtract.ChunkError.t()}), keyword() ) :: Enumerable.t()