Function stream_blocking_iter 

pub fn stream_blocking_iter<Init, Iter, Item, Error>(
    capacity: usize,
    init: Init,
) -> impl Stream<Item = Result<Item, Error>> + Send + Unpin + 'static
where
    Init: FnOnce() -> Result<Iter, Error> + Send + 'static,
    Iter: Iterator<Item = Result<Item, Error>>,
    Item: Send + 'static,
    Error: Send + 'static,

Bridge a blocking, fallible iterator into a bounded Stream, decoding on a blocking thread.

Historical data usually arrives as something that blocks: a Parquet artifact, a compressed archive, a CSV on a slow disk. Two things then go wrong if such an iterator is simply wrapped in futures::stream::iter:

1. It stalls a runtime worker for the whole decode. Nothing else scheduled on that worker — the engine, another instrument’s decode, a heartbeat — makes progress until the file ends.
2. It never yields Pending, so nothing downstream can pace it. A consumer that forwards into an unbounded channel (the engine feed is one) therefore takes the entire artifact into memory before processing the first event, which is the opposite of the streaming the caller asked for. Laziness alone does not bound memory; the producer has to be made to wait.

This helper fixes both: init runs on a tokio::task::spawn_blocking thread — so opening the source is blocking too, as it usually is — and each item is handed over a channel of capacity items. When the channel is full the blocking thread parks, which is the back-pressure: the decoder runs at most capacity items ahead of whatever polls this stream, whatever the dataset’s size, and decoding overlaps consumption instead of preceding it.

That bound is local to this hand-off. It constrains nothing further downstream: a caller that forwards this stream into another unbounded queue — a backtest engine feed is exactly that shape — reintroduces unbounded growth there whenever the far end is the slower side. What this helper guarantees is “the decoder will not get more than capacity items ahead of its own consumer”, not an end-to-end memory bound for whatever pipeline it is embedded in.

Failure model

The item type is Result, because a source that reads incrementally can fail after it opened successfully. An init that fails yields exactly one Err and ends the stream, so a caller handles open and mid-stream failures on one path.

A panic is not an Err and is not reported as one: Error is an unconstrained generic, so there is no value this helper could construct to represent one. It is re-raised on the consumer instead — see # Panics. What it must never do is end the stream quietly, which would leave a truncated decode indistinguishable from a source that finished, and drive a backtest to a normal-looking summary over partial data.

A tokio::task::JoinError that is not a panic does end the stream as a clean None. A spawn_blocking task cannot be aborted and this one’s handle is never exposed, so the only way to reach that is runtime shutdown — where the consumer is being torn down for the same reason and has nothing left to do with the news.

Cancellation

Dropping the returned stream drops the receiver, the next hand-off fails, and the blocking task returns at its next item boundary. A blocking task cannot be pre-empted, so an iterator that blocks for a long time between items keeps its thread until it comes back — bounded by the iterator’s own step, not by the dataset.

Panics

Panics if capacity is zero (a zero-capacity channel cannot deliver), and requires a Tokio runtime with the rt feature, like any tokio::task::spawn_blocking caller.

A panic in init or in the iterator is re-raised on the task that polls this stream, with the original payload, at the point the stream would otherwise have ended. This mirrors futures::stream::iter over the same iterator, where the panic surfaces on the poller directly: moving the decode to another thread changes where the work happens, not whether the caller hears about it. Items already handed over are still yielded first, so the panic arrives after the last successfully decoded item rather than discarding the batch.

Polling past the end

None is terminal and repeatable: polling again yields None rather than panicking, so a consumer that does not track termination itself needs no .fuse(). The returned stream is also Unpin, so awaiting next() needs no pinning.

Both are part of the contract, not accidents of the current body — a rewrite that broke either (an async_stream generator breaks both) would break callers. Unpin is therefore in the return type, so such a rewrite fails to compile here rather than at some downstream call site; the terminal None cannot be spelled in a signature and is pinned by test instead. FusedStream is deliberately not implemented: the terminal None above already gives a consumer what fusing would, the intended consumer merge_time_sorted fuses its inputs itself, and widening the return type later is an additive change. That is a design judgement rather than an observation — this helper has no in-tree production callers, so there is no consumer set to appeal to.

⚠️ One blocking thread per call, held for the whole decode

tokio::task::spawn_blocking fires at call time, not at first poll, so building N of these starts N blocking tasks immediately, before anything is polled. Each holds its thread for the whole decode: blocking_send parks the thread while the channel is full rather than returning it to the pool, which is precisely the back-pressure described above.

Tokio’s blocking pool defaults to max_blocking_threads: 512. Past that, further tasks are queued and get no thread until a running one returns — and a decode only returns when its consumer drains it.

That combination deadlocks under a k-way merge. merge_time_sorted collects its inputs eagerly and cannot emit until every input has buffered an event or ended (see its # Pace section). With more than 512 inputs, the queued tasks never run, so their streams stay Pending, so the merge stays Pending, so none of the running decodes is drained and none releases its thread. Nothing times out and nothing logs: the backtest simply stops.

A 512-instrument universe is not exotic, so a caller merging more inputs than the pool has threads must either raise max_blocking_threads above the input count, or merge in batches whose size it does not exceed.

Examples

let _stream = stream_blocking_iter(DEFAULT_BLOCKING_CHANNEL_CAPACITY, decode);