{"record":{"id":"aeb2dfa0958af950","repo":"vectordotdev/vector","slug":"concurrent-map-task-cancelled-outside-of-our-contr","errorCode":null,"errorMessage":"concurrent map task cancelled outside of our control","messagePattern":"concurrent map task cancelled outside of our control","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"lib/vector-stream/src/concurrent_map.rs","lineNumber":106,"sourceCode":"        }\n\n        match ready!(this.in_flight.poll_next_unpin(cx)) {\n            // If the stream is done and there is no futures managed by FuturesOrdered,\n            // we must end the stream by returning Poll::Ready(None).\n            None if this.stream.is_done() => Poll::Ready(None),\n            // If there are no in-flight futures managed by FuturesOrdered but the underlying\n            // stream is not done, then we must keep polling that stream.\n            None => Poll::Pending,\n            Some(result) => match result {\n                Ok(item) => Poll::Ready(Some(item)),\n                Err(e) => {\n                    if let Ok(reason) = e.try_into_panic() {\n                        // Resume the panic here on the calling task.\n                        panic::resume_unwind(reason);\n                    } else {\n                        // The task was cancelled, which makes no sense, because _we_ hold the join\n                        // handle. Only sensible thing to do is panic, because this is a bug.\n                        panic!(\"concurrent map task cancelled outside of our control\");\n                    }\n                }\n            },\n        }\n    }\n}\n\n#[cfg(test)]\nmod tests {\n    use futures_util::stream::StreamExt;\n\n    use super::*;\n\n    #[tokio::test]\n    async fn test_concurrent_map_on_empty_stream() {\n        let stream = futures_util::stream::empty::<()>();\n        let limit = Some(NonZeroUsize::new(2).unwrap());\n        // The `as _` is required to construct a `dyn Future`","sourceCodeStart":88,"sourceCodeEnd":124,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/lib/vector-stream/src/concurrent_map.rs#L88-L124","documentation":"ConcurrentMap drives a FuturesOrdered of tokio::spawn'd mapping tasks and polls their JoinHandles. When a JoinHandle yields a JoinError that is not a panic, the mapped task was cancelled. Since the stream itself owns the handle, cancellation 'should' be impossible from outside - it can only occur when the Tokio runtime is shut down (or tasks aborted) while the ConcurrentMap stream is still being polled or still holds in-flight work. The library panics because this represents a driver/runtime lifecycle bug, not a data error.","triggerScenarios":"Dropping or shutting down the Tokio runtime while a ConcurrentMap has spawned tasks in flight, then continuing to poll the stream; also aborting the tasks through some external handle. The JoinError::try_into_panic() path fails (cancellation, not panic) and the else branch fires.","commonSituations":"Test harnesses using #[tokio::test] or block_on that drop the runtime before the pipeline stream completes; graceful-shutdown code that tears down the runtime while sink/source streams still run; nested runtimes or runtime drop ordering bugs in embedding applications.","solutions":["Drive all pipeline streams (including ConcurrentMap) to completion or explicit timeout-cancellation before dropping the Runtime","Own the stream inside a task spawned on the same runtime, so it is dropped when the runtime drops tasks, not polled afterwards","Use Runtime::shutdown_timeout after confirming tasks finished instead of dropping the runtime mid-flight"],"exampleFix":"// before\nlet rt = tokio::runtime::Runtime::new().unwrap();\nlet out: Vec<_> = rt.block_on(stream.collect()); // stream dropped mid-flight elsewhere\n// ...\ndrop(rt); // later polling elsewhere -> cancelled join error\n\n// after\nlet rt = tokio::runtime::Runtime::new().unwrap();\nrt.block_on(async {\n    let out: Vec<_> = stream.collect().await; // fully driven on the runtime\n});","handlingStrategy":"validation","validationCode":"// Ensure the stream fully completes while the runtime is alive:\nrt.block_on(async {\n    use futures_util::StreamExt;\n    while let Some(item) = stream.next().await { /* consume all */ }\n});\n// Only after in-flight tasks finish, tear the runtime down:\nrt.shutdown_timeout(std::time::Duration::from_secs(5));","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never drop or shut down a Tokio runtime while pipeline streams have in-flight spawned tasks","Drive streams to completion (or explicit timeout) on the runtime that spawned their tasks","Keep stream ownership inside tasks on the runtime rather than in runtime-external scopes","In tests, prefer #[tokio::test] with fully-awaited pipelines over hand-built runtimes"],"tags":["rust","tokio","vector","runtime-shutdown","join-error","panic"],"backgroundTag":"tokio-task-cancelled","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T20:17:18.057Z"}