vectordotdev/vector · error

can't run runner twice

Error message

can't run runner twice

What it means

This panic comes from `Option::take().expect(...)` on the runner's `input_rx` channel in `run_inline` (src/topology/builder.rs:1334). The runner consumes its input receiver exactly once; a second call means the same runner was driven again after it already started. The library treats that as an internal invariant violation rather than a recoverable error.

Solutions

1. Ensure each Runner instance is driven exactly once; create a new runner via the builder for every execution.
2. Do not clone or share the runner across tasks; wrap it in an owned task.
3. If retrying a failed run, rebuild the topology piece instead of reusing the old runner.

Example fix

// before
let runner = build_sync_transform(...);
runner.run_inline().await;
runner.run_inline().await; // panics: channel already taken

// after
let runner = build_sync_transform(...);
runner.run_inline().await;
let runner2 = build_sync_transform(...); // fresh runner for a second run
runner2.run_inline().await;

fn can_run(runner: &Runner) -> bool { runner.has_input() } // expose a debug check; run only once per runner

fn runner_is_fresh(rx: &Option<Receiver>) -> bool { rx.is_some() }

When it happens

Trigger: Calling `run_inline` twice on the same runner instance, or calling both `run_inline` and `run_concurrently` on the same runner, typically when build_sync_transform wires the runner into the topology more than once.

Common situations: Custom topology construction code (e.g. in tests or external tooling that reuses vector's builder) that retries or re-runs a transform runner after the first invocation already took the input channel; cloning/sharing a runner across two tasks.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• join error or bad poll
• Pausing unknown sink from fanout
• Replacing unknown sink from fanout
• source output misconfigured
• source output misconfigured - output for port

AI-assisted analysis of vectordotdev/vector@bdb87aeaa4 (2026-09-16). Data as JSON: /api/errors/3b8dbf850fc9a9f9. Report an issue: GitHub.

Appendix: source

Thrown at src/topology/builder.rs:1334

    }

    async fn send_outputs(&mut self, outputs_buf: &mut TransformOutputsBuf) -> crate::Result<()> {
        self.timer_tx.try_send_start_wait();
        let now = Instant::now();
        outputs_buf.for_each_array_mut(|array| self.latency_recorder.on_send(array, now));
        self.outputs.send(outputs_buf).await
    }

    async fn run_inline(mut self) -> TaskResult {
        // 128 is an arbitrary, smallish constant
        const INLINE_BATCH_SIZE: usize = 128;

        let mut outputs_buf = self.outputs.new_buf_with_capacity(INLINE_BATCH_SIZE);

        let mut input_rx = self
            .input_rx
            .take()
            .expect("can't run runner twice")
            .into_stream()
            .filter(move |events| ready(filter_events_type(events, self.input_type)));

        self.timer_tx.try_send_start_wait();
        while let Some(events) = input_rx.next().await {
            self.on_events_received(&events);
            self.transform.transform_all(events, &mut outputs_buf);
            self.send_outputs(&mut outputs_buf)
                .await
                .map_err(TaskError::wrapped)?;
        }

        Ok(TaskOutput::Transform)
    }

    async fn run_concurrently(mut self) -> TaskResult {
        let input_rx = self
            .input_rx

View on GitHub (pinned to bdb87aeaa4)