moirai-executor 0.9.1

//! Borrowing scopes: nesting, recursion, panics, and quiescence on exit.

use super::super::scope_state::{SchedulerScopeState, ScopedTaskCompletion};
use super::*;

#[test]
fn independent_scopes_share_wake_mechanisms_without_sharing_completion() {
    const CALLERS: usize = 8;
    const ITEMS: usize = 4096;
    let scheduler = ThreadScheduler::new(4, "independent-scopes").expect("scheduler");
    let visits: [[AtomicUsize; ITEMS]; CALLERS] =
        std::array::from_fn(|_| std::array::from_fn(|_| AtomicUsize::new(0)));
    let barrier = Barrier::new(CALLERS);
    std::thread::scope(|threads| {
        for row in &visits {
            let scheduler = &scheduler;
            let barrier = &barrier;
            threads.spawn(move || {
                barrier.wait();
                scheduler
                    .for_each_indexed::<SyncTask, _>(Priority::Normal, None, ITEMS, |index| {
                        row[index].fetch_add(1, Ordering::Relaxed);
                    })
                    .expect("independent scope completes");
                assert_eq!(
                    row.iter()
                        .map(|count| count.load(Ordering::Acquire))
                        .sum::<usize>(),
                    ITEMS
                );
            });
        }
    });
    assert!(
        visits
            .iter()
            .flatten()
            .all(|count| count.load(Ordering::Acquire) == 1)
    );
}

#[test]
fn scheduler_scope_runs_borrowing_jobs_before_return() {
    let scheduler = ThreadScheduler::new(2, "test-scope").unwrap();
    let sum = AtomicUsize::new(0);

    scheduler
        .scope::<SyncTask, _>(Priority::Normal, None, |scope| {
            for value in 1..=16 {
                let sum = &sum;
                scope.spawn(move |_| {
                    sum.fetch_add(value, Ordering::Relaxed);
                })?;
            }
            Ok(())
        })
        .unwrap();

    scheduler.shutdown();
    assert_eq!(sum.load(Ordering::Relaxed), 136);
}

#[test]
fn scheduler_scope_nested_saturation_completes() {
    // Regression guard for ISSUE-208: a scoped job that itself opens a nested
    // scope must complete. Before help-while-waiting, `scope().wait()` parked the
    // caller without running scheduler work, so nested fork-join deadlocked
    // (provably with one worker) and corrupted the heap under concurrent nesting
    // (STATUS_HEAP_CORRUPTION). Value-semantic: every inner increment must land.
    for &workers in &[1usize, 2, 4] {
        let scheduler = ThreadScheduler::new(workers, "test-nested-scope").unwrap();
        let outer = 32usize;
        let inner = 16usize;
        let counter = AtomicUsize::new(0);

        scheduler
            .scope::<SyncTask, _>(Priority::Normal, None, |outer_scope| {
                for _ in 0..outer {
                    let scheduler = &scheduler;
                    let counter = &counter;
                    outer_scope.spawn(move |_| {
                        scheduler
                            .scope::<SyncTask, _>(Priority::Normal, None, |inner_scope| {
                                for _ in 0..inner {
                                    let counter = &counter;
                                    inner_scope.spawn(move |_| {
                                        counter.fetch_add(1, Ordering::Relaxed);
                                    })?;
                                }
                                Ok(())
                            })
                            .expect("nested scope must complete");
                    })?;
                }
                Ok(())
            })
            .unwrap();

        assert_eq!(
            counter.load(Ordering::Relaxed),
            outer * inner,
            "nested saturation lost increments at {workers} worker(s)"
        );
        scheduler.shutdown();
    }
}

fn recursive_scope_sum(scheduler: &ThreadScheduler, lo: u64, hi: u64) -> u64 {
    if hi.saturating_sub(lo) <= 1024 {
        return (lo..hi).sum();
    }
    let mid = lo + (hi - lo) / 2;
    let mut left = 0u64;
    let mut right = 0u64;
    {
        let left = &mut left;
        let right = &mut right;
        scheduler
            .scope::<SyncTask, _>(Priority::Normal, None, |scope| {
                scope.spawn(|_| {
                    *left = recursive_scope_sum(scheduler, lo, mid);
                })?;
                scope.spawn(|_| {
                    *right = recursive_scope_sum(scheduler, mid, hi);
                })?;
                Ok(())
            })
            .expect("recursive scope must complete");
    }
    left + right
}

#[test]
fn scheduler_scope_recursive_fork_join_is_sound() {
    // ISSUE-208 corruption guard: the recursive two-branch fork-join is the exact
    // shape of `moirai_iter` `drive` (log2-depth nested scopes, each branch stolen
    // by a peer worker that dereferences the parent scope's stack-owned state).
    // Before help-while-waiting this deadlocked (one worker) and corrupted the
    // heap (STATUS_HEAP_CORRUPTION) under concurrent nesting. Analytical oracle:
    // the arithmetic series sum, asserted value-semantically.
    const N: u64 = 200_000;
    let expected = N * (N - 1) / 2;
    for &workers in &[1usize, 2, 4] {
        let scheduler = ThreadScheduler::new(workers, "test-recursive-scope").unwrap();
        assert_eq!(
            recursive_scope_sum(&scheduler, 0, N),
            expected,
            "recursive fork-join sum diverged at {workers} worker(s)"
        );
        scheduler.shutdown();
    }
}

#[test]
fn scheduler_scope_nested_leaves_scheduler_quiescent() {
    // Accounting guard for help-while-waiting (ADR-019): a worker waiter runs
    // jobs via a re-entrant `execute_job`, which mutates the global
    // `pending_tasks`/`active_workers` counters. A leaked increment would leave
    // `join()` unable to observe quiescence (hang → nextest terminates). Assert
    // the scheduler returns to a consistent quiescent state with no spurious
    // failures after a nested workload.
    for &workers in &[1usize, 2, 4] {
        let scheduler = ThreadScheduler::new(workers, "test-nested-quiescent").unwrap();
        let outer = 16usize;
        let inner = 8usize;
        let counter = AtomicUsize::new(0);

        scheduler
            .scope::<SyncTask, _>(Priority::Normal, None, |outer_scope| {
                for _ in 0..outer {
                    let scheduler = &scheduler;
                    let counter = &counter;
                    outer_scope.spawn(move |_| {
                        scheduler
                            .scope::<SyncTask, _>(Priority::Normal, None, |inner_scope| {
                                for _ in 0..inner {
                                    let counter = &counter;
                                    inner_scope.spawn(move |_| {
                                        counter.fetch_add(1, Ordering::Relaxed);
                                    })?;
                                }
                                Ok(())
                            })
                            .expect("nested scope must complete");
                    })?;
                }
                Ok(())
            })
            .unwrap();

        // Terminates only if the help path leaked no pending/active count.
        scheduler.join().expect("scheduler must reach quiescence");
        let metrics = scheduler.metrics();

        assert_eq!(counter.load(Ordering::Relaxed), outer * inner);
        assert_eq!(
            metrics.pending_tasks, 0,
            "leaked pending count at {workers} worker(s)"
        );
        assert_eq!(
            metrics.active_workers, 0,
            "leaked active-worker count at {workers} worker(s)"
        );
        assert_eq!(
            metrics.failed_tasks, 0,
            "spurious job failure at {workers} worker(s)"
        );
        scheduler.shutdown();
    }
}

#[test]
fn scheduler_scope_nested_panic_propagates_and_pool_survives() {
    // Adversarial guard for the help-while-waiting scope (ADR-019): when a nested
    // scoped job panics, the nested scope must report SpawnFailed(Panicked), its
    // sibling job must still run, and the outer scope must complete without
    // deadlock or corruption — i.e. a panic on a help-stealing worker unwinds
    // only its own job, never the waiter's help loop.
    for &workers in &[1usize, 2, 4] {
        let scheduler = ThreadScheduler::new(workers, "test-nested-panic").unwrap();
        let outer = 8usize;
        let sibling_ran = AtomicUsize::new(0);
        let nested_panics_reported = AtomicUsize::new(0);

        scheduler
            .scope::<SyncTask, _>(Priority::Normal, None, |outer_scope| {
                for _ in 0..outer {
                    let scheduler = &scheduler;
                    let sibling_ran = &sibling_ran;
                    let nested_panics_reported = &nested_panics_reported;
                    outer_scope.spawn(move |_| {
                        let result =
                            scheduler.scope::<SyncTask, _>(Priority::Normal, None, |inner| {
                                inner.spawn(|_| panic!("nested scoped job panic"))?;
                                inner.spawn(move |_| {
                                    sibling_ran.fetch_add(1, Ordering::Relaxed);
                                })?;
                                Ok(())
                            });
                        if matches!(result, Err(ExecutorError::SpawnFailed(TaskError::Panicked))) {
                            nested_panics_reported.fetch_add(1, Ordering::Relaxed);
                        }
                    })?;
                }
                Ok(())
            })
            .unwrap();

        assert_eq!(
            nested_panics_reported.load(Ordering::Relaxed),
            outer,
            "each nested scope must report its panic at {workers} worker(s)"
        );
        assert_eq!(
            sibling_ran.load(Ordering::Relaxed),
            outer,
            "sibling of a panicking nested job must still run at {workers} worker(s)"
        );
        scheduler.shutdown();
    }
}

#[test]
fn scheduler_scope_reports_panicked_job() {
    let scheduler = ThreadScheduler::new(1, "test-scope-panic").unwrap();
    let completed = AtomicUsize::new(0);

    let result = scheduler.scope::<SyncTask, _>(Priority::Normal, None, |scope| {
        scope.spawn(|_| panic!("scoped job panic"))?;
        let completed = &completed;
        scope.spawn(move |_| {
            completed.fetch_add(1, Ordering::Relaxed);
        })?;
        Ok(())
    });

    scheduler.shutdown();
    assert_eq!(result, Err(ExecutorError::SpawnFailed(TaskError::Panicked)));
    assert_eq!(completed.load(Ordering::Relaxed), 1);
}

#[test]
fn scheduler_scope_completes_registered_jobs_before_body_error_returns() {
    let scheduler = ThreadScheduler::new(2, "test-scope-body-error").unwrap();
    let completed = AtomicUsize::new(0);

    let result = scheduler.scope::<SyncTask, _>(Priority::Normal, None, |scope| {
        for _ in 0..8 {
            let completed = &completed;
            scope.spawn(move |_| {
                completed.fetch_add(1, Ordering::Relaxed);
            })?;
        }

        Err(ExecutorError::SpawnFailed(TaskError::InvalidOperation))
    });

    scheduler.shutdown();
    assert_eq!(
        result,
        Err(ExecutorError::SpawnFailed(TaskError::InvalidOperation))
    );
    assert_eq!(completed.load(Ordering::Relaxed), 8);
}

#[test]
fn scheduler_scope_completes_registered_jobs_before_resuming_body_panic() {
    let scheduler = ThreadScheduler::new(2, "test-scope-body-panic").unwrap();
    let completed = AtomicUsize::new(0);

    let result = catch_unwind(AssertUnwindSafe(|| {
        scheduler
            .scope::<SyncTask, _>(Priority::Normal, None, |scope| {
                for _ in 0..8 {
                    let completed = &completed;
                    scope.spawn(move |_| {
                        completed.fetch_add(1, Ordering::Relaxed);
                    })?;
                }

                panic!("scope body panic");
            })
            .unwrap();
    }));

    scheduler.shutdown();
    let Err(payload) = result else {
        panic!("invariant: the scope body panic must resume once the registered jobs complete");
    };
    assert_eq!(
        payload.downcast_ref::<&str>().copied(),
        Some("scope body panic"),
        "the body's own panic, not a job's, resumes after the scope drains"
    );
    assert_eq!(completed.load(Ordering::Relaxed), 8);
}

#[test]
fn scope_opened_from_another_schedulers_worker_completes() {
    // The worker-id cache is process-wide. A worker of the larger scheduler
    // carries an id past the smaller scheduler's worker table, so a drain keyed
    // on that id indexed out of bounds and unwound `scope` while its jobs still
    // borrowed the opener's frame.
    let large = ThreadScheduler::new(4, "test-scope-large").unwrap();
    let small = ThreadScheduler::new(1, "test-scope-small").unwrap();
    let (sender, receiver) = std::sync::mpsc::sync_channel(1);
    let target = small.clone();

    large
        .schedule::<SyncTask, _>(Priority::Normal, Some(3), move |_| {
            let sum = AtomicUsize::new(0);
            let outcome = target.scope::<SyncTask, _>(Priority::Normal, None, |scope| {
                for value in 1..=8 {
                    let sum = &sum;
                    scope.spawn(move |_| {
                        sum.fetch_add(value, Ordering::Relaxed);
                    })?;
                }
                Ok(())
            });
            sender
                .send((outcome.is_ok(), sum.load(Ordering::Relaxed)))
                .unwrap();
        })
        .unwrap();

    let (completed, sum) = receiver
        .recv()
        .expect("the opening job must survive the scope call");
    assert!(completed);
    assert_eq!(sum, 36);
    large.shutdown();
    small.shutdown();
}

/// A scoped job registered on `state`, holding its completion token.
fn registered_job<'scope>(
    state: &'scope SchedulerScopeState<'scope>,
    ran: &'scope AtomicUsize,
) -> crate::schedule::job::ScheduledJob {
    state.register_task();
    let completion = ScopedTaskCompletion::new(state);
    // SAFETY: every caller runs or drops the job before `state` and `ran` end.
    unsafe {
        crate::schedule::job::ScheduledJob::new_scoped_with_completion(
            move |_| {
                ran.fetch_add(1, Ordering::Relaxed);
            },
            move |succeeded| completion.finish(succeeded),
        )
    }
}

#[test]
fn scoped_job_dropped_before_running_fails_its_scope() {
    let wait_lock = std::sync::Mutex::new(());
    let wait_signal = std::sync::Condvar::new();
    let state = SchedulerScopeState::new(&wait_lock, &wait_signal);
    let ran = AtomicUsize::new(0);

    drop(registered_job(&state, &ran));
    state.wait();

    assert_eq!(ran.load(Ordering::Relaxed), 0);
    assert!(!state.has_panicked());
    assert_eq!(
        state.unrun_job_result(),
        Err(ExecutorError::SpawnFailed(TaskError::Cancelled))
    );
}

/// The waiter returns and destroys the stack-owned state as soon as the last
/// release publishes zero, while the releasing thread is still inside its own
/// call. Run under Miri with many seeds: a reference to the state held across
/// the release is reported as a deallocation while protected.
#[test]
fn the_waiter_may_destroy_the_state_while_the_last_release_returns() {
    static WAIT_LOCK: Mutex<()> = Mutex::new(());
    static WAIT_SIGNAL: Condvar = Condvar::new();
    let state = Box::new(SchedulerScopeState::new(&WAIT_LOCK, &WAIT_SIGNAL));
    state.register_task();
    // SAFETY: the lifetime is extended only so the token can move to another
    // thread. The token reads the state only until its release publishes zero,
    // and `wait` returns after that, which is the point the state is destroyed.
    let extended: &'static SchedulerScopeState<'static> = unsafe { &*std::ptr::from_ref(&*state) };
    let completion = ScopedTaskCompletion::new(extended);
    let releaser = std::thread::spawn(move || completion.finish(true));

    state.wait();
    assert!(!state.has_panicked());
    drop(state);

    releaser
        .join()
        .expect("the releasing thread must not panic");
}

#[test]
fn scoped_job_that_ran_leaves_its_scope_whole() {
    let wait_lock = std::sync::Mutex::new(());
    let wait_signal = std::sync::Condvar::new();
    let state = SchedulerScopeState::new(&wait_lock, &wait_signal);
    let ran = AtomicUsize::new(0);

    let _ = registered_job(&state, &ran).execute(0);
    state.wait();

    assert_eq!(ran.load(Ordering::Relaxed), 1);
    assert!(!state.has_panicked());
    assert_eq!(state.unrun_job_result(), Ok(()));
}

#[test]
fn refused_scoped_job_is_forgiven_and_an_admitted_drop_is_not() {
    let wait_lock = std::sync::Mutex::new(());
    let wait_signal = std::sync::Condvar::new();
    let state = SchedulerScopeState::new(&wait_lock, &wait_signal);
    let ran = AtomicUsize::new(0);

    // Admission drops a refused job and its submitter answers the refusal.
    drop(registered_job(&state, &ran));
    state.forgive_refused_job();
    // A second job is dropped after admission, and nothing answers for it.
    drop(registered_job(&state, &ran));
    state.wait();

    assert_eq!(
        state.unrun_job_result(),
        Err(ExecutorError::SpawnFailed(TaskError::Cancelled))
    );
}