Multiple Sources & Coordinated Shutdown #11

New issue

Open

opened 2026-07-23 14:47:40 +00:00 by erik · 0 comments

erik commented

2026-07-23 14:47:40 +00:00

Owner

The real problem is shutdown semantics. Currently the pipeline has a single global shutdown signal (Ctrl+C). There's no concept of "all sources are done, begin draining." Consider a pipeline with two sources — a file source and a Kafka source:

The file source finishes reading → its task exits, its broadcast sender drops
The Kafka source keeps running indefinitely
The pipeline never shuts down unless the user hits Ctrl+C

What we want: the pipeline drains and shuts down when all sources have completed, without requiring an external signal. A single source finishing should not trigger shutdown.

Proposed Approach

Classify stages at build time. A source is any stage with no inputs. Tag them:

struct Stage {
    name: String,
    inputs: Vec<String>,
    plugin: Box<dyn Plugin>,
    role: StageRole,  // Source, Transform, Sink
}

enum StageRole { Source, Transform, Sink }

Source = no inputs. Sink = no downstream consumers. Transform = everything else.

Source completion barrier. Collect source task handles separately:

let mut source_handles: Vec<JoinHandle<()>> = Vec::new();
let mut other_handles: Vec<JoinHandle<()>> = Vec::new();

// ... spawn tasks, sort handles by role ...

// Wait for ALL sources to finish (natural EOF, not Ctrl+C)
futures::future::join_all(source_handles).await;

// All sources done → trigger graceful shutdown
// This causes transforms to drain their input channels and complete,
// which cascades to sinks.
// Optionally send shutdown signal so any long-lived transforms also exit.
let _ = shutdown_tx.send(true);

// Wait for remaining stages to drain
futures::future::join_all(other_handles).await;

Per-source status reporting. Sources should be able to report why they stopped:

enum SourceCompletion {
    Eof,              // natural end of input
    Error(String),    // fatal error, couldn't continue
    Shutdown,         // responded to shutdown signal
}

This feeds into observability (section 3) — the dashboard can show which sources are still active and why each one stopped.

Interaction with Ctrl+C. The shutdown signal and source-completion barrier work together:
- Normal flow: sources finish naturally → barrier resolves → pipeline drains
- Ctrl+C: shutdown signal fires → sources break their select! loops → barrier resolves → pipeline drains
- Partial: some sources finish, user hits Ctrl+C for the rest → same drain

Edge Cases

Infinite sources (e.g. Kafka, scheduler with no end): These never complete on their own. The pipeline stays alive until Ctrl+C. This is correct behavior — the barrier just waits for all sources, so one infinite source keeps the pipeline running.
Source restarts: A source that errors could optionally restart (with backoff) rather than counting as "completed." Config: on_error: restart | complete | abort-pipeline.
Dynamic source addition: Out of scope for now, but the barrier design doesn't preclude it — you'd add the new handle to the barrier set.

Migration Path

This is a pipeline-engine-only change. No Plugin trait modifications needed. Sources already exit their start() method when done — we just need to track their handles separately and add the barrier logic in main().

Source plugins are the ones that signal "I'm done" — when a source completes, its broadcast sender drops, which cascades downstream. With the current fan-in design (multiple upstream broadcast receivers merged into one mpsc), a single source completing just closes one forwarder task — the mpsc stays open as long as at least one forwarder is alive. So the data path *accidentally* works correctly for multiple sources today. The real problem is **shutdown semantics**. Currently the pipeline has a single global shutdown signal (Ctrl+C). There's no concept of "all sources are done, begin draining." Consider a pipeline with two sources — a file source and a Kafka source: - The file source finishes reading → its task exits, its broadcast sender drops - The Kafka source keeps running indefinitely - The pipeline never shuts down unless the user hits Ctrl+C What we want: **the pipeline drains and shuts down when all sources have completed**, without requiring an external signal. A single source finishing should not trigger shutdown. ### Proposed Approach 1. **Classify stages at build time.** A source is any stage with no `inputs`. Tag them: ```rust struct Stage { name: String, inputs: Vec<String>, plugin: Box<dyn Plugin>, role: StageRole, // Source, Transform, Sink } enum StageRole { Source, Transform, Sink } ``` `Source` = no inputs. `Sink` = no downstream consumers. `Transform` = everything else. 2. **Source completion barrier.** Collect source task handles separately: ```rust let mut source_handles: Vec<JoinHandle<()>> = Vec::new(); let mut other_handles: Vec<JoinHandle<()>> = Vec::new(); // ... spawn tasks, sort handles by role ... // Wait for ALL sources to finish (natural EOF, not Ctrl+C) futures::future::join_all(source_handles).await; // All sources done → trigger graceful shutdown // This causes transforms to drain their input channels and complete, // which cascades to sinks. // Optionally send shutdown signal so any long-lived transforms also exit. let _ = shutdown_tx.send(true); // Wait for remaining stages to drain futures::future::join_all(other_handles).await; ``` 3. **Per-source status reporting.** Sources should be able to report why they stopped: ```rust enum SourceCompletion { Eof, // natural end of input Error(String), // fatal error, couldn't continue Shutdown, // responded to shutdown signal } ``` This feeds into observability (section 3) — the dashboard can show which sources are still active and why each one stopped. 4. **Interaction with Ctrl+C.** The shutdown signal and source-completion barrier work together: - Normal flow: sources finish naturally → barrier resolves → pipeline drains - Ctrl+C: shutdown signal fires → sources break their `select!` loops → barrier resolves → pipeline drains - Partial: some sources finish, user hits Ctrl+C for the rest → same drain ### Edge Cases - **Infinite sources** (e.g. Kafka, scheduler with no end): These never complete on their own. The pipeline stays alive until Ctrl+C. This is correct behavior — the barrier just waits for *all* sources, so one infinite source keeps the pipeline running. - **Source restarts**: A source that errors could optionally restart (with backoff) rather than counting as "completed." Config: `on_error: restart | complete | abort-pipeline`. - **Dynamic source addition**: Out of scope for now, but the barrier design doesn't preclude it — you'd add the new handle to the barrier set. ### Migration Path This is a pipeline-engine-only change. No Plugin trait modifications needed. Sources already exit their `start()` method when done — we just need to track their handles separately and add the barrier logic in `main()`.

erik self-assigned this

2026-07-23 14:47:40 +00:00