anodizer_core/

parallel.rs

//! Shared bounded-parallelism helper used by stages that run one subprocess
//! per sub-config (makeself, nfpm, snapcraft, flatpak, upx, …).
//!
//! The stages share the same Step 1 / Step 2 / Step 3 shape:
//!
//! 1. **Step 1** (serial, `&mut ctx`): render templates, stage files,
//!    collect a `Vec<Job>` of fully-owned work units.
//! 2. **Step 2** (parallel, bounded by `ctx.options.parallelism`): run one
//!    subprocess per job in `std::thread::scope`.
//! 3. **Step 3** (serial, `&mut ctx`): register the returned artifacts.
//!
//! Before this helper every stage hand-rolled the Step 2 loop —
//! `for chunk in jobs.chunks(n) { thread::scope(|s| …) }` with its own
//! join-unwrap-or-panic handling. The pattern is now shared here so new
//! parallelized stages just write `run_job`.
//!
//! Semantics match the previous hand-rolled loops exactly:
//!
//! - **Bounded concurrency**: at most `parallelism` workers run at once,
//!   enforced by chunking the job list and scoping threads per-chunk.
//! - **Fail-fast within a chunk**: if any worker in a chunk fails, the whole
//!   chunk still runs to completion (threads are already spawned), but the
//!   caller receives the first error and processes no further chunks. The
//!   completed siblings' work is still accounted for: additional failures
//!   in the batch are logged, and a warn summarizes partial progress.
//! - **Panic-safe**: a worker panic becomes an `anyhow::Error` annotated
//!   with `stage_name`, so a panicked thread doesn't leave the pool
//!   deadlocked or drop all other results on the floor.
//! - **Order-preserving**: results are collected in job-submission order, so
//!   downstream artifact registration remains deterministic.

use anyhow::{Result, anyhow};

use crate::log::StageLogger;
use std::sync::{Mutex, MutexGuard};

/// Acquire a `Mutex` guard, recovering from poison rather than panicking.
///
/// A poisoned lock means a sibling worker thread panicked while holding
/// the guard. For the data shapes this helper is used on (counters,
/// `Vec` accumulators), the inner state has no invariant a panic could
/// have broken — the worst case is one partial write missing. Panicking
/// the current worker too would abandon its already-completed network
/// call without updating the count, silently inflating the operator's
/// `failed` bucket.
pub fn lock_recover<'a, T>(m: &'a Mutex<T>, log: &StageLogger, label: &str) -> MutexGuard<'a, T> {
    match m.lock() {
        Ok(g) => g,
        Err(poisoned) => {
            log.warn(&format!(
                "{label} mutex poisoned by sibling thread panic; recovering state"
            ));
            poisoned.into_inner()
        }
    }
}

/// Translate a `thread::JoinHandle::join` result's panic payload into
/// an `anyhow::Error` tagged with `label`. The two common panic
/// payload shapes (`&'static str` / `String`) are downcast so the
/// surfaced message is readable rather than the opaque `Any`
/// placeholder.
///
/// Accepts `Result<T, Box<dyn Any + Send>>` rather than the handle
/// itself so a single helper covers both [`std::thread::JoinHandle`]
/// and [`std::thread::ScopedJoinHandle`] — both expose `.join()`
/// returning the same `Result` shape.
///
/// Use when the worker returns `T` and the caller wants `Result<T>`
/// so a panic doesn't propagate as a silently-lost result. For
/// workers that already return `Result<T, anyhow::Error>`, prefer
/// [`run_parallel_chunks`] which bakes this in.
pub fn join_panic_to_err<T>(join_result: std::thread::Result<T>, label: &str) -> Result<T> {
    join_result.map_err(|panic_payload| {
        let msg = if let Some(s) = panic_payload.downcast_ref::<&'static str>() {
            (*s).to_string()
        } else if let Some(s) = panic_payload.downcast_ref::<String>() {
            s.clone()
        } else {
            format!("{:?}", panic_payload)
        };
        anyhow!("{label} worker thread panicked: {msg}")
    })
}

/// Run `run_job` across `jobs` with bounded parallelism. Returns the
/// per-job results in submission order.
///
/// `stage_name` is embedded in the panic error message so a crash in one
/// stage is attributable at a glance (`"nfpm worker thread panicked"` vs
/// `"snapcraft worker thread panicked"`).
///
/// `parallelism` is clamped to `>= 1` internally, so callers can pass
/// `ctx.options.parallelism` without pre-clamping.
///
/// On failure the FIRST error is returned and no further chunks run, but
/// the failed chunk's completed siblings are never silently discarded:
/// every additional failure in the chunk is logged as a warning (only the
/// first error propagates), and a warn summarizes the partial progress —
/// how many jobs in the batch succeeded before the failure and how many
/// later jobs were never started.
pub fn run_parallel_chunks<J, T, F>(
    jobs: &[J],
    parallelism: usize,
    stage_name: &'static str,
    log: &StageLogger,
    run_job: F,
) -> Result<Vec<T>>
where
    J: Sync,
    T: Send,
    F: Fn(&J) -> Result<T> + Sync,
{
    let parallelism = parallelism.max(1);
    let mut results: Vec<T> = Vec::with_capacity(jobs.len());

    // A worker thread starts outside every retry scope, so the caller's label
    // is read here and re-entered inside each worker; without it the stage's
    // parallel backoff would be filed as unattributed.
    let retry_scope = crate::retry::current_scope();
    for chunk in jobs.chunks(parallelism) {
        let chunk_results: Vec<Result<T>> = std::thread::scope(|s| {
            let handles: Vec<_> = chunk
                .iter()
                .map(|job| {
                    s.spawn(|| {
                        let _scope = crate::retry::RetryScope::inherit(retry_scope.clone());
                        run_job(job)
                    })
                })
                .collect();
            handles
                .into_iter()
                .map(|h| {
                    h.join()
                        .unwrap_or_else(|_| Err(anyhow!("{} worker thread panicked", stage_name)))
                })
                .collect()
        });

        // The whole chunk already ran to completion (its threads were
        // spawned together), so account for EVERY result before propagating:
        // a bare `push(r?)` would silently drop the completed siblings'
        // work and any second/third failure in the same batch.
        let mut first_err: Option<anyhow::Error> = None;
        let mut chunk_ok = 0usize;
        let chunk_len = chunk.len();
        for r in chunk_results {
            match r {
                Ok(t) => {
                    chunk_ok += 1;
                    results.push(t);
                }
                Err(e) => {
                    if first_err.is_none() {
                        first_err = Some(e);
                    } else {
                        // Warn with the root cause only: the full anyhow chain
                        // can embed unredacted subprocess/HTTP detail (upload
                        // URLs, response bodies) that the propagated error gets
                        // caller-side redaction for but this path would not.
                        let root = e.root_cause().to_string();
                        let first = root.lines().next().unwrap_or("");
                        let mut line: String = first.chars().take(200).collect();
                        if first.chars().count() > 200 {
                            line.push('…');
                        }
                        log.warn(&format!(
                            "{stage_name}: additional failure in the same batch \
                             (only the first is propagated): {line}"
                        ));
                        log.verbose(&format!("{stage_name}: additional failure detail: {e:#}"));
                    }
                }
            }
        }
        if let Some(err) = first_err {
            let not_started = jobs.len() - results.len() - (chunk_len - chunk_ok);
            log.warn(&format!(
                "{stage_name}: {chunk_ok} of {chunk_len} item(s) in this batch succeeded \
                 before the failure ({completed} of {total} total completed, \
                 {not_started} never started)",
                completed = results.len(),
                total = jobs.len(),
            ));
            return Err(err);
        }
    }
    Ok(results)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_helpers::test_logger;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn preserves_submission_order() {
        // Even with multi-threaded execution, the returned Vec must mirror
        // the input slice order so downstream artifact registration is
        // deterministic across runs.
        let jobs: Vec<u32> = (0..20).collect();
        let out =
            run_parallel_chunks(&jobs, 4, "test", test_logger(), |job| Ok(*job * 10)).unwrap();
        assert_eq!(out, (0..20).map(|i| i * 10).collect::<Vec<_>>());
    }

    #[test]
    fn bounded_concurrency() {
        // With parallelism=2 across 10 jobs, no more than 2 workers should
        // be in-flight at once. Observed via an AtomicUsize peak
        // counter that each worker increments on entry and decrements on
        // exit, with a small sleep to force overlap.
        let jobs: Vec<u32> = (0..10).collect();
        let in_flight = AtomicUsize::new(0);
        let peak = AtomicUsize::new(0);

        run_parallel_chunks(&jobs, 2, "test", test_logger(), |_| {
            let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(10));
            in_flight.fetch_sub(1, Ordering::SeqCst);
            Ok(())
        })
        .unwrap();

        assert!(
            peak.load(Ordering::SeqCst) <= 2,
            "peak in-flight workers exceeded parallelism bound"
        );
    }

    #[test]
    fn propagates_first_error() {
        // A single failing job should fail the batch. The job index returned
        // in the error payload asserts the failing worker is the one the
        // caller receives (not silently swallowed by a later success).
        let jobs: Vec<u32> = (0..4).collect();
        let result = run_parallel_chunks(&jobs, 2, "test", test_logger(), |job| {
            if *job == 2 {
                Err(anyhow!("job 2 failed"))
            } else {
                Ok(*job)
            }
        });
        let err = result.unwrap_err();
        assert!(
            err.to_string().contains("job 2 failed"),
            "unexpected error: {}",
            err
        );
    }

    /// A failed chunk must not silently swallow its completed siblings'
    /// work or the batch's additional failures: every job in the chunk
    /// still runs, the extra failure is logged, and a warn summarizes the
    /// partial progress (succeeded-in-batch / total-completed / never-started).
    #[test]
    fn failed_chunk_reports_partial_progress_and_sibling_failures() {
        let jobs: Vec<u32> = (0..8).collect();
        let executed = AtomicUsize::new(0);
        let (log, cap) = StageLogger::with_capture("test", crate::log::Verbosity::Quiet);

        // parallelism=4 → chunk [0,1,2,3]: jobs 1 and 3 fail, 0 and 2 succeed;
        // chunks [4..] must never start.
        let result = run_parallel_chunks(&jobs, 4, "partial-stage", &log, |job| {
            executed.fetch_add(1, Ordering::SeqCst);
            if *job == 1 || *job == 3 {
                Err(anyhow!("job {} failed", job)
                    .context("POST https://uploads.example/secret-token failed"))
            } else {
                Ok(*job)
            }
        });

        let err = result.unwrap_err();
        assert!(
            format!("{err:#}").contains("job 1 failed"),
            "the FIRST error (submission order) must propagate: {err:#}"
        );
        assert_eq!(
            executed.load(Ordering::SeqCst),
            4,
            "the whole failed chunk runs; later chunks never start"
        );
        let warns = cap.warn_messages();
        assert!(
            warns
                .iter()
                .any(|m| m.contains("job 3 failed") && m.contains("only the first is propagated")),
            "the sibling failure must be logged, not dropped: {warns:?}"
        );
        assert!(
            !warns.iter().any(|m| m.contains("uploads.example")),
            "the sibling warn must carry the root cause only, never the \
             unredacted context chain: {warns:?}"
        );
        let details: Vec<String> = cap
            .all_messages()
            .into_iter()
            .filter(|(lvl, _)| *lvl == crate::log::LogLevel::Verbose)
            .map(|(_, m)| m)
            .collect();
        assert!(
            details
                .iter()
                .any(|m| m.contains("uploads.example") && m.contains("job 3 failed")),
            "the full chain must still be available at verbose: {details:?}"
        );
        assert!(
            warns.iter().any(|m| m.contains(
                "partial-stage: 2 of 4 item(s) in this batch succeeded before the failure"
            ) && m.contains("2 of 8 total completed")
                && m.contains("4 never started")),
            "partial-progress summary must be warned: {warns:?}"
        );
    }

    /// A clean run must emit NO partial-progress warns — the summary is a
    /// failure-path diagnostic, not routine chatter.
    #[test]
    fn successful_run_emits_no_warns() {
        let jobs: Vec<u32> = (0..6).collect();
        let (log, cap) = StageLogger::with_capture("test", crate::log::Verbosity::Quiet);
        let out = run_parallel_chunks(&jobs, 3, "test", &log, |job| Ok(*job)).unwrap();
        assert_eq!(out.len(), 6);
        assert_eq!(cap.warn_count(), 0, "no warns on a clean run");
    }

    #[test]
    fn fan_out_workers_attribute_backoff_to_the_callers_retry_scope() {
        // A worker thread enters the pool holding no scope of its own, so
        // without inheritance a stage's parallel backoff is filed as
        // unattributed and the run summary can no longer name the slow remote.
        let scope = "parallel-fan-out-attribution-b7c2";
        let jobs: Vec<u64> = vec![3_000_000_000, 3_000_000_001];
        {
            let _guard = crate::retry::RetryScope::enter(scope);
            run_parallel_chunks(&jobs, 2, "test", test_logger(), |job| {
                crate::retry::record_retry_backoff(std::time::Duration::from_millis(*job));
                Ok(*job)
            })
            .unwrap();
        }
        let recorded = crate::retry::retry_scope_breakdown()
            .into_iter()
            .find(|(k, _, _)| k == scope);
        assert_eq!(
            recorded.map(|(_, retries, backoff)| (retries, backoff)),
            Some((2, std::time::Duration::from_millis(6_000_000_001))),
            "both workers' backoff must land under the caller's scope"
        );
    }

    #[test]
    fn zero_parallelism_clamps_to_one() {
        // `ctx.options.parallelism` can legitimately be 0 (unset) —
        // callers must not need to pre-clamp. Verify the helper runs
        // sequentially in that case rather than spawning 0 threads.
        let jobs: Vec<u32> = (0..3).collect();
        let out = run_parallel_chunks(&jobs, 0, "test", test_logger(), |job| Ok(*job + 1)).unwrap();
        assert_eq!(out, vec![1, 2, 3]);
    }

    #[test]
    fn empty_jobs_returns_empty() {
        let out: Vec<u32> =
            run_parallel_chunks::<u32, u32, _>(&[], 4, "test", test_logger(), |_| Ok(0)).unwrap();
        assert!(out.is_empty());
    }

    #[test]
    fn panic_in_worker_becomes_anyhow_error() {
        // A panicking worker must not take down the whole thread::scope
        // silently — the error must be attributable, carrying the stage name.
        let jobs: Vec<u32> = vec![1, 2, 3];
        let result = run_parallel_chunks(
            &jobs,
            2,
            "explode-stage",
            test_logger(),
            |job| -> Result<u32> {
                if *job == 2 {
                    panic!("boom");
                }
                Ok(*job)
            },
        );
        let err = result.unwrap_err();
        assert!(
            err.to_string()
                .contains("explode-stage worker thread panicked"),
            "unexpected error: {}",
            err
        );
    }

    // ---------- lock_recover ----------

    #[test]
    fn lock_recover_returns_inner_when_unpoisoned() {
        // Happy path: an unpoisoned Mutex yields its guard, the helper
        // adds no observable behavior over a bare `.lock().unwrap()`.
        let log = test_logger();
        let m = Mutex::new(0u32);
        {
            let mut g = lock_recover(&m, log, "test");
            *g = 42;
        }
        assert_eq!(*m.lock().unwrap(), 42);
    }

    #[test]
    fn lock_recover_recovers_from_poison() {
        // A poisoned Mutex (sibling thread panicked while holding the
        // guard) must yield the inner state rather than panicking the
        // recovering thread too.
        let log = test_logger();
        let m = std::sync::Arc::new(Mutex::new(7u32));
        let m_for_thread = std::sync::Arc::clone(&m);
        let h = std::thread::spawn(move || {
            let _g = m_for_thread.lock().unwrap();
            panic!("poison the mutex");
        });
        let _ = h.join();
        assert!(m.is_poisoned(), "test setup: mutex should be poisoned");
        let g = lock_recover(&m, log, "test");
        assert_eq!(*g, 7);
    }

    // ---------- join_panic_to_err ----------

    #[test]
    fn join_panic_to_err_passes_through_success() {
        let h = std::thread::spawn(|| 42u32);
        let r = join_panic_to_err(h.join(), "worker").unwrap();
        assert_eq!(r, 42);
    }

    #[test]
    fn join_panic_to_err_translates_str_panic() {
        // The most common panic shape in this codebase is `panic!("msg")`
        // which produces a `&'static str` payload — verify the message
        // survives into the surfaced anyhow chain.
        let h = std::thread::spawn(|| -> u32 {
            panic!("kaboom");
        });
        let err = join_panic_to_err(h.join(), "worker").unwrap_err();
        let s = err.to_string();
        assert!(
            s.contains("worker worker thread panicked") && s.contains("kaboom"),
            "unexpected error: {}",
            s
        );
    }

    #[test]
    fn join_panic_to_err_translates_string_panic() {
        // The other common panic shape — `format!()`-derived `String`
        // payloads — must also be downcast rather than printing as `Any`.
        let h = std::thread::spawn(|| -> u32 {
            panic!("{}", String::from("string-panic"));
        });
        let err = join_panic_to_err(h.join(), "worker").unwrap_err();
        assert!(
            err.to_string().contains("string-panic"),
            "unexpected error: {}",
            err
        );
    }

    #[test]
    fn join_panic_to_err_works_on_scoped_handle() {
        // ScopedJoinHandle::join returns the same Result shape as
        // JoinHandle::join — verify a single helper covers both so
        // callers using `std::thread::scope` don't need a second variant.
        let out: Result<u32> = std::thread::scope(|s| {
            let h = s.spawn(|| 99u32);
            join_panic_to_err(h.join(), "scoped")
        });
        assert_eq!(out.unwrap(), 99);
    }
}