Multiple Sinks & Independent Completion #10

New issue

Open

opened 2026-07-23 14:46:10 +00:00 by erik · 0 comments

erik commented

2026-07-23 14:46:10 +00:00

Owner

Currently, any stage completing triggers downstream drain. With multiple sinks (e.g. write to both OpenSearch and a file), one finishing early would starve the other.

Proposed Approach

Tag stages with their role: Source, Transform, Sink (already implicit in config via inputs and whether the stage has downstream consumers — make it explicit).
Shutdown rule: The pipeline shuts down when all sources have completed OR when a shutdown signal arrives. Sinks complete independently when their input channel drains.
Fan-out to multiple sinks: When a transform feeds multiple sinks, each sink gets its own bounded channel. The transform sends to all of them. A sink completing and dropping its receiver doesn't affect the other sinks — the transform just stops sending to that edge.

Implementation sketch:

// In the pipeline engine, track sink handles separately
let sink_handles: Vec<JoinHandle<()>> = ...;

// Wait for all sources to finish (or shutdown signal)
source_barrier.wait().await;

// Drop source output senders → transforms drain → sinks drain
// Wait for all sinks
futures::future::join_all(sink_handles).await;

Per-sink completion callback: Add on_complete() to the Plugin trait (default no-op) so sinks can report final stats without triggering pipeline shutdown.

Edge Case: Sink Failure

If a sink errors out, it should log and drop its receiver. Other sinks continue. An optional on_error policy per sink (abort-pipeline | continue | retry) could be configured.

Currently, any stage completing triggers downstream drain. With multiple sinks (e.g. write to both OpenSearch and a file), one finishing early would starve the other. ### Proposed Approach 1. **Tag stages with their role**: `Source`, `Transform`, `Sink` (already implicit in config via `inputs` and whether the stage has downstream consumers — make it explicit). 2. **Shutdown rule**: The pipeline shuts down when **all sources** have completed OR when a shutdown signal arrives. Sinks complete independently when their input channel drains. 3. **Fan-out to multiple sinks**: When a transform feeds multiple sinks, each sink gets its own bounded channel. The transform sends to all of them. A sink completing and dropping its receiver doesn't affect the other sinks — the transform just stops sending to that edge. 4. **Implementation sketch:** ```rust // In the pipeline engine, track sink handles separately let sink_handles: Vec<JoinHandle<()>> = ...; // Wait for all sources to finish (or shutdown signal) source_barrier.wait().await; // Drop source output senders → transforms drain → sinks drain // Wait for all sinks futures::future::join_all(sink_handles).await; ``` 5. **Per-sink completion callback**: Add `on_complete()` to the Plugin trait (default no-op) so sinks can report final stats without triggering pipeline shutdown. ### Edge Case: Sink Failure If a sink errors out, it should log and drop its receiver. Other sinks continue. An optional `on_error` policy per sink (`abort-pipeline` | `continue` | `retry`) could be configured.

erik self-assigned this

2026-07-23 14:46:10 +00:00