What causes "concurrent map task cancelled outside of our control" in vectordotdev/vector?

Drive all pipeline streams (including ConcurrentMap) to completion or explicit timeout-cancellation before dropping the Runtime

vectordotdev/vector · critical

concurrent map task cancelled outside of our control

Error message

concurrent map task cancelled outside of our control

What it means

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.

Source

Thrown at lib/vector-stream/src/concurrent_map.rs:106

        }

        match ready!(this.in_flight.poll_next_unpin(cx)) {
            // If the stream is done and there is no futures managed by FuturesOrdered,
            // we must end the stream by returning Poll::Ready(None).
            None if this.stream.is_done() => Poll::Ready(None),
            // If there are no in-flight futures managed by FuturesOrdered but the underlying
            // stream is not done, then we must keep polling that stream.
            None => Poll::Pending,
            Some(result) => match result {
                Ok(item) => Poll::Ready(Some(item)),
                Err(e) => {
                    if let Ok(reason) = e.try_into_panic() {
                        // Resume the panic here on the calling task.
                        panic::resume_unwind(reason);
                    } else {
                        // The task was cancelled, which makes no sense, because _we_ hold the join
                        // handle. Only sensible thing to do is panic, because this is a bug.
                        panic!("concurrent map task cancelled outside of our control");
                    }
                }
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use futures_util::stream::StreamExt;

    use super::*;

    #[tokio::test]
    async fn test_concurrent_map_on_empty_stream() {
        let stream = futures_util::stream::empty::<()>();
        let limit = Some(NonZeroUsize::new(2).unwrap());
        // The `as _` is required to construct a `dyn Future`

View on GitHub (pinned to 3708c39b12)

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

Example fix

// before
let rt = tokio::runtime::Runtime::new().unwrap();
let out: Vec<_> = rt.block_on(stream.collect()); // stream dropped mid-flight elsewhere
// ...
drop(rt); // later polling elsewhere -> cancelled join error

// after
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
    let out: Vec<_> = stream.collect().await; // fully driven on the runtime
});

Defensive patterns

Strategy: validation

Validate before calling

// Ensure the stream fully completes while the runtime is alive:
rt.block_on(async {
    use futures_util::StreamExt;
    while let Some(item) = stream.next().await { /* consume all */ }
});
// Only after in-flight tasks finish, tear the runtime down:
rt.shutdown_timeout(std::time::Duration::from_secs(5));

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of vectordotdev/vector@3708c39b12 (2026-08-20). Data as JSON: /api/errors/aeb2dfa0958af950. Report an issue: GitHub.