Method run_concurrent_inner

lambda-runtime/src/runtime.rs:201–277

Concurrent processing using N independent long-poll loops (for Lambda managed-concurrency).

(
        service: S,
        config: Arc<Config>,
        client: Arc<ApiClient>,
        concurrency_limit: u32,
    )

    /// Concurrent processing using N independent long-poll loops (for Lambda managed-concurrency).
    async fn run_concurrent_inner(
        service: S,
        config: Arc<Config>,
        client: Arc<ApiClient>,
        concurrency_limit: u32,
    ) -> Result<(), BoxError> {
        let limit = concurrency_limit as usize;

        // Use FuturesUnordered so we can observe worker exits as they happen,
        // rather than waiting for all workers to finish (join_all).
        let mut workers: FuturesUnordered<tokio::task::JoinHandle<(tokio::task::Id, Result<(), BoxError>)>> =
            FuturesUnordered::new();
        let spawn_worker = |service: S, config: Arc<Config>, client: Arc<ApiClient>| {
            tokio::spawn(async move {
                let task_id = tokio::task::id();
                let result = concurrent_worker_loop(service, config, client).await;
                (task_id, result)
            })
        };
        // Spawn one worker per concurrency slot; the last uses the owned service to avoid an extra clone.
        for _ in 1..limit {
            workers.push(spawn_worker(service.clone(), config.clone(), client.clone()));
        }
        workers.push(spawn_worker(service, config, client));

        // Track worker exits across tasks. A single worker failing should not
        // terminate the whole runtime (LMI keeps running with the remaining
        // healthy workers). We only return an error once there are no workers
        // left (i.e., we cannot keep at least 1 worker alive).
        //
        // Note: Handler errors (Err returned from user code) do NOT trigger this.
        // They are reported to Lambda via /invocation/{id}/error and the worker
        // continues. This only captures unrecoverable runtime failures like
        // API client failures, runtime panics, etc.
        let mut errors: Vec<WorkerError> = Vec::new();
        let mut remaining_workers = limit;
        while let Some(result) = futures::StreamExt::next(&mut workers).await {
            remaining_workers = remaining_workers.saturating_sub(1);
            match result {
                Ok((task_id, Ok(()))) => {
                    // `concurrent_worker_loop` runs indefinitely, so an Ok return indicates
                    // an unexpected worker exit; we still decrement because the task is gone.
                    error!(
                        task_id = %task_id,
                        remaining_workers,
                        "Concurrent worker exited cleanly (unexpected - loop should run forever)"
                    );
                    errors.push(WorkerError::CleanExit(task_id));
                }
                Ok((task_id, Err(err))) => {
                    error!(
                        task_id = %task_id,
                        error = %err,
                        remaining_workers,
                        "Concurrent worker exited with error"
                    );
                    errors.push(WorkerError::Failure(task_id, err));
                }