Function stream_blocking_iter 

pub fn stream_blocking_iter<Init, Iter, Item, Error>(
    chunk_size: 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>> + Send + 'static,
    Item: Send + 'static,
    Error: Send + 'static,

Bridge a blocking, fallible iterator into a bounded Stream, decoding on Tokio’s blocking pool.

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. The decode runs in chunks of chunk_size items, each on a tokio::task::spawn_blocking task that decodes the chunk, returns it, and gives its thread back. When a chunk arrives the stream starts the next one, so decoding overlaps consumption, and it starts no further chunk until the consumer has drained the one it holds. That is the back-pressure: the decoder runs at most two chunks, 2 × chunk_size items, ahead of whatever polls this stream, whatever the dataset’s size. init runs in the first chunk’s task, so opening the source is off the runtime’s workers too, as it usually needs to be.

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 two chunks ahead of its own consumer”, not an end-to-end memory bound for whatever pipeline it is embedded in.

The bound is also per stream. A merge_time_sorted of N of these polls every input until each has delivered a chunk, so it holds up to 2 × N × chunk_size decoded items. For a merge of thousands of inputs, lower chunk_size.

No thread is held while waiting, so any number of these can be merged

A blocking thread is held only while a chunk is being decoded, never while the stream waits for its consumer. Building a stream starts nothing: the first chunk is started on first poll. Any number of these can therefore be driven together, for example by merge_time_sorted, which cannot emit until every input has buffered an event. Past Tokio’s max_blocking_threads (512 by default), chunks queue for a thread and every one of them still finishes, so the merge is slower but always progresses.

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 ends the stream with None, after a warn!. This one’s handle is never exposed, so nothing aborts the task, and the only way to reach that is runtime shutdown cancelling a chunk still queued for a thread. The consumer is normally being torn down for the same reason; the warn! says the decode was cut short in case it is not.

Cancellation

Dropping the returned stream starts no further chunk. The chunk already started is not cancelled, whether it is being decoded or still waiting for a thread: it runs to its end, at most chunk_size items, and its result is discarded with the iterator. Letting it run keeps the iterator, and whatever its Drop does, on the blocking pool.

Panics

Panics if chunk_size is zero (a chunk must deliver at least one item), and, when first polled, outside a Tokio runtime, like any tokio::task::spawn_blocking caller. chunk_size bounds the items per chunk and is not allocated up front, so a large value such as usize::MAX is safe.

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 decoded before the panic, including those earlier in the same chunk, 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.

The iterator must be Send + 'static

The iterator moves from one chunk’s task to the next, so it has to cross threads. A decoder with !Send internals is not accepted. Holding one thread for the whole decode would allow it, and is exactly what made a merge of more inputs than the pool has threads deadlock.

Examples

let _stream = stream_blocking_iter(DEFAULT_BLOCKING_CHUNK_SIZE, decode);