{"record":{"id":"3b8dbf850fc9a9f9","repo":"vectordotdev/vector","slug":"can-t-run-runner-twice","errorCode":null,"errorMessage":"can't run runner twice","messagePattern":"can't run runner twice","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/topology/builder.rs","lineNumber":1334,"sourceCode":"    }\n\n    async fn send_outputs(&mut self, outputs_buf: &mut TransformOutputsBuf) -> crate::Result<()> {\n        self.timer_tx.try_send_start_wait();\n        let now = Instant::now();\n        outputs_buf.for_each_array_mut(|array| self.latency_recorder.on_send(array, now));\n        self.outputs.send(outputs_buf).await\n    }\n\n    async fn run_inline(mut self) -> TaskResult {\n        // 128 is an arbitrary, smallish constant\n        const INLINE_BATCH_SIZE: usize = 128;\n\n        let mut outputs_buf = self.outputs.new_buf_with_capacity(INLINE_BATCH_SIZE);\n\n        let mut input_rx = self\n            .input_rx\n            .take()\n            .expect(\"can't run runner twice\")\n            .into_stream()\n            .filter(move |events| ready(filter_events_type(events, self.input_type)));\n\n        self.timer_tx.try_send_start_wait();\n        while let Some(events) = input_rx.next().await {\n            self.on_events_received(&events);\n            self.transform.transform_all(events, &mut outputs_buf);\n            self.send_outputs(&mut outputs_buf)\n                .await\n                .map_err(TaskError::wrapped)?;\n        }\n\n        Ok(TaskOutput::Transform)\n    }\n\n    async fn run_concurrently(mut self) -> TaskResult {\n        let input_rx = self\n            .input_rx","sourceCodeStart":1316,"sourceCodeEnd":1352,"githubUrl":"https://github.com/vectordotdev/vector/blob/bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013/src/topology/builder.rs#L1316-L1352","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure each Runner instance is driven exactly once; create a new runner via the builder for every execution.","Do not clone or share the runner across tasks; wrap it in an owned task.","If retrying a failed run, rebuild the topology piece instead of reusing the old runner."],"exampleFix":"// before\nlet runner = build_sync_transform(...);\nrunner.run_inline().await;\nrunner.run_inline().await; // panics: channel already taken\n\n// after\nlet runner = build_sync_transform(...);\nrunner.run_inline().await;\nlet runner2 = build_sync_transform(...); // fresh runner for a second run\nrunner2.run_inline().await;","handlingStrategy":"validation","validationCode":"fn can_run(runner: &Runner) -> bool { runner.has_input() } // expose a debug check; run only once per runner","typeGuard":"fn runner_is_fresh(rx: &Option<Receiver>) -> bool { rx.is_some() }","tryCatchPattern":null,"preventionTips":["Treat Runner as single-use: consume it (self) per run, as the API already does.","Never clone or share runners across tasks.","Rebuild topology pieces instead of retrying a run on the same runner."],"tags":["rust","topology","panic","internal-invariant"],"backgroundTag":"internal-invariant-violation","analyzedSha":"bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013","analyzedAt":"2026-09-16T02:53:35.741Z","contentChangedAt":"2026-09-16T02:53:35.741Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}