feat(stream): add pull-based pipeline runtime [STREAM-1668]

[tryangul]

What

New processing/stream/ module — a pull-based pipeline runtime using async streams. Fully additive, no existing code changed.

How

(Sorry for the wall of text)

Core abstractions:

• Stage trait — single process() method to encapsulate a 'step' in the chain. Returns StageResult which encapsulates the outcome of processing a message. Open so consumers can define ad-hoc stages, but we could easily provide "canned" stages for the common usecases.
• PipelineExt combinators — compose the stages together top to bottom via .apply(), .apply_concurrent(), .on_next(), .on_reject(), .commit().
• PipelineRunner — core harness that handles rebalance / restart loop.
• Handlers — side effects without transformations: NextHandler (side-effects on success), RejectionHandler (DLQ/logging). Canned: KafkaProducerHandler, DlqHandler (preserves original headers), LogHandler.

Why

Addresses the pain points in STREAM-1668 via a new runtime rather than rearchitecting the existing processor:

• Backpressure — natural via async streams, no MessageRejected spin loop
• DLQ — Reject variant flows through the pipeline to handlers
• Profiling — each stage is a separate async call
• Concurrency — .apply_concurrent(stage, N) for I/O-bound stages
• Rebalance — PipelineRunner handles the restart loop, caller just defines the pipeline shape

Notes

• Working example: examples/transform_and_produce_pull.rs
• 3 tests: rebalance restart, shutdown exit, natural completion
• Fatal Kafka errors → StageResult::Fail (recoverable errors skipped)

Next steps

• Streams PR incoming — PullBasedAdapter that maps the Python DSL to pull-based stages (tested e2e with real protobuf + parquet)
• Deploy to SBC items_span pipeline to benchmark against the push model

STREAM-1668

[linear-code]

STREAM-1668

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 7db0dbb. Configure here.}

[sentry]

Bug: A shared Notify instance for rebalances can cause a race condition. If rebalances occur rapidly, a new stream may immediately exit, leading to a processing loop without consuming messages.
_{Severity: MEDIUM}

Suggested Fix

To prevent the stale notification permit from being carried over, create a new Notify instance for each new stream created by the source.stream() method. This ensures that each stream instance has its own isolated rebalance signal, avoiding the race condition.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: rust-arroyo/src/processing/stream/source.rs#L159

Potential issue: The `KafkaSource` uses a single shared `Arc<Notify>` instance
(`revoke`) to signal rebalances across stream restarts. A race condition exists where a
rebalance can occur after the old stream has terminated but before the new stream has
registered its `revoke.notified()` future. If rebalances happen in rapid succession, a
permit from a second rebalance can be stored and then immediately consumed by the new
stream. This causes the stream to exit with `PipelineExit::Rebalance` without processing
any messages, potentially leading to a tight loop that stalls message consumption.

Also affects:

• rust-arroyo/src/processing/stream/source.rs:101~101
• rust-arroyo/src/processing/stream/source.rs:128~128

[tryangul]

I think this winds up being a noop because the window is small and we rebuild everything anyway.