haz-exec 0.1.0

//! Per-task lookup-then-spawn pipeline stages.
//!
//! The scheduler runs each admitted task through two awaitable
//! stages whose futures share one [`FuturesUnordered`] queue:
//!
//! - [`run_lookup_step`] performs the cache lookup
//!   ([`crate::run_task::cache_lookup_phase`]) and, on a hit,
//!   drives [`crate::run_task::restore_from_hit`] inline. The
//!   future never touches the mutex hold set per `MUTEX-007`. On
//!   a miss it returns the cache key, the task's declared mutex,
//!   and its bearing project name so the scheduler can evaluate
//!   `EXEC-006` condition 3 before deciding whether to acquire
//!   the mutex and dispatch a spawn-step future.
//! - [`run_spawn_step`] runs [`crate::run_task::run_fresh`] with
//!   the mutex acquired per `MUTEX-006`.
//!
//! Both futures report their completion as an
//! [`InFlightCompletion`] variant; the scheduler's main `match`
//! routes lookup completions and spawn completions through
//! distinct paths because spawn completions MUST trigger a mutex
//! release before downstream bookkeeping.
//!
//! [`FuturesUnordered`]: futures::stream::FuturesUnordered

use std::collections::BTreeMap;
use std::future::Future;
use std::pin::Pin;

use haz_cache::CacheKey;
use haz_domain::mutex::Mutex;
use haz_domain::name::ProjectName;
use haz_domain::task_id::TaskId;
use haz_vfs::WritableFilesystem;

use crate::cache_key::PredecessorStreamHashes;
use crate::process::ProcessSpawner;
use crate::run_task::{
    CompletedRecord, RunContext, RunObserver, RunTaskError, cache_lookup_phase, restore_from_hit,
    run_fresh,
};

/// Outcome of the lookup step in the scheduler's
/// lookup-then-spawn pipeline.
///
/// On a cache hit the lookup-step future runs
/// [`restore_from_hit`] inline and the variant carries the
/// finished [`CompletedRecord`]. On a cache miss the future
/// stops short of [`run_fresh`] and yields the data the
/// scheduler needs to evaluate `EXEC-006` condition 3 against
/// its live [`crate::hold_set::HoldSet`] before deciding whether
/// to acquire the mutex and drive the spawn step.
#[derive(Debug)]
pub(super) enum LookupStepOutcome {
    /// Cache hit; restoration was driven inline and the
    /// resulting [`CompletedRecord`] is final.
    Hit(CompletedRecord),
    /// Cache miss; the scheduler MUST check `HoldSet`
    /// compatibility before invoking [`run_fresh`].
    Miss {
        /// Computed cache key, threaded into [`run_fresh`] so
        /// the key is not re-derived in the spawn step.
        key: CacheKey,
        /// Task's declared mutex, cloned out of the workspace
        /// borrow so the lookup-step future does not need to
        /// carry workspace lifetimes across the await boundary.
        mutex: Option<Mutex>,
        /// Bearing project name, paired with `mutex` when the
        /// scheduler later releases the hold.
        project_name: ProjectName,
    },
}

/// Per-future report sent back to the scheduler's main loop.
///
/// The two variants distinguish a freshly-completed lookup step
/// from a freshly-completed spawn step. The main-loop `match`
/// routes each variant through the appropriate handler:
/// `Lookup` results may either be terminal (hit / lookup-error)
/// or trigger a push of a spawn-step future after the mutex
/// dance; `Spawn` results MUST be paired with a `HoldSet`
/// release before processing.
#[derive(Debug)]
pub(super) enum InFlightCompletion {
    /// A lookup-step future has just completed.
    Lookup {
        task: TaskId,
        result: Result<LookupStepOutcome, RunTaskError>,
    },
    /// A spawn-step future has just completed. The scheduler
    /// MUST release the corresponding mutex hold before
    /// processing the result.
    Spawn {
        task: TaskId,
        result: Result<CompletedRecord, RunTaskError>,
    },
}

/// Boxed-dyn alias for the heterogeneously-typed lookup-step
/// and spawn-step futures the scheduler stores in its
/// [`futures::stream::FuturesUnordered`]. Each push site
/// produces an async block with a distinct concrete future type;
/// boxing to `dyn Future<Output = InFlightCompletion> + 'a`
/// lets the queue hold both shapes uniformly.
pub(super) type InFlightFuture<'a> = Pin<Box<dyn Future<Output = InFlightCompletion> + 'a>>;

/// Drive the lookup step: cache lookup and, on hit, restoration.
/// On miss the future yields the cache key, declared mutex, and
/// bearing project name so the scheduler can decide whether to
/// acquire the mutex and drive [`run_fresh`].
///
/// The body is synchronous today (both [`cache_lookup_phase`]
/// and [`restore_from_hit`] are sync), but kept `async fn` so it
/// produces the same future shape as [`run_spawn_step`]; that
/// lets the scheduler box both into a single
/// [`futures::stream::FuturesUnordered`] queue and lets future
/// work (cancellation checks, async cache I/O) introduce real
/// awaits here without touching call sites.
#[allow(clippy::unused_async)]
pub(super) async fn run_lookup_step<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    task: TaskId,
    predecessor_streams: BTreeMap<TaskId, PredecessorStreamHashes>,
) -> InFlightCompletion
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let lookup = match cache_lookup_phase(ctx, &task, &predecessor_streams) {
        Ok(l) => l,
        Err(err) => {
            return InFlightCompletion::Lookup {
                task,
                result: Err(err),
            };
        }
    };
    if let Some(manifest) = lookup.manifest.as_ref() {
        let outcome = match restore_from_hit(ctx, &task, manifest) {
            Ok(o) => o,
            Err(err) => {
                return InFlightCompletion::Lookup {
                    task,
                    result: Err(err),
                };
            }
        };
        return InFlightCompletion::Lookup {
            task,
            result: Ok(LookupStepOutcome::Hit(outcome)),
        };
    }
    InFlightCompletion::Lookup {
        task,
        result: Ok(LookupStepOutcome::Miss {
            key: lookup.key,
            mutex: lookup.task_def.mutex.clone(),
            project_name: lookup.project.name.clone(),
        }),
    }
}

/// Drive the spawn step: spawn the task's command and capture
/// both streams via [`run_fresh`]. The scheduler MUST have
/// acquired the task's mutex (when one is declared) before
/// pushing this future.
pub(super) async fn run_spawn_step<F, S, O>(
    ctx: &RunContext<'_, F, S, O>,
    task: TaskId,
    key: CacheKey,
    created_at_unix: u64,
) -> InFlightCompletion
where
    F: WritableFilesystem,
    S: ProcessSpawner,
    O: RunObserver,
{
    let project = ctx
        .workspace
        .projects
        .get(&task.project)
        .expect("spawn-step task's project must exist (passed the lookup step)");
    let task_def = project
        .tasks
        .get(&task.task)
        .expect("spawn-step task name must exist in its project (passed the lookup step)");
    let result = run_fresh(ctx, &task, project, task_def, &key, created_at_unix).await;
    InFlightCompletion::Spawn { task, result }
}

#[cfg(test)]
mod tests {
    //! Section-specific mutex tests exercise the lookup-then-spawn
    //! pipeline (`EXEC-006` condition 3, `EXEC-007`, `MUTEX-001..007`):
    //! admission is cap-only, the mutex check sits between the lookup
    //! step and the spawn step, cache hits never touch the hold set
    //! (`MUTEX-007`), and a task that loses the mutex check returns to
    //! ready with a single `Started` event ever fired
    //! (`MUTEX-005`, `S3`).

    use std::collections::{BTreeMap, BTreeSet};
    use std::str::FromStr;

    use haz_domain::mutex::{Mutex as DomainMutex, MutexMode, MutexScope};
    use haz_domain::name::MutexName;

    use crate::mock_impl::MockProcessSpawner;
    use crate::run_graph::scheduler::run_graph;
    use crate::run_graph::test_fixtures::*;
    use crate::run_task::{RunSource, RunState, run_task};

    fn workspace_mutex(name: &str, mode: MutexMode) -> DomainMutex {
        DomainMutex {
            scope: MutexScope::Workspace,
            name: MutexName::from_str(name).unwrap(),
            mode,
        }
    }

    fn project_scoped_mutex(name: &str, mode: MutexMode) -> DomainMutex {
        DomainMutex {
            scope: MutexScope::Project,
            name: MutexName::from_str(name).unwrap(),
            mode,
        }
    }

    #[tokio::test]
    async fn mutex_007_cache_hit_does_not_block_on_held_exclusive() {
        // Two tasks declare the same workspace-exclusive mutex but
        // carry distinct commands (so their cache keys differ).
        // Pre-run task `b` alone to populate its cache entry; then
        // run both in one graph. Task `a` misses, acquires the
        // mutex, and proceeds to spawn; task `b` hits the cache
        // mid-flight. MUTEX-007 forbids `b` from waiting on the
        // hold set: its restoration completes before `a`'s spawn
        // finishes, so `Finished(b)` precedes `Finished(a)` in
        // the observer log.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));

        let graph_b_only = make_graph(vec![tid("p", "b")], vec![]);
        let mut fixture = Fixture::new(ws, graph_b_only);

        // Pre-warm: run `b` once via run_task to populate the cache.
        {
            let spawner = MockProcessSpawner::new();
            push_n_default_specs(&spawner, 1);
            let observer = Recorder::default();
            let ctx = make_ctx(&fixture, &spawner, &observer);
            let outcome = run_task(&ctx, &tid("p", "b"), &BTreeMap::new(), 1)
                .await
                .unwrap();
            assert_eq!(outcome.source, RunSource::FreshRun);
            assert_eq!(outcome.state, RunState::Succeeded);
        }

        // Second run: both tasks. `a` is the only fresh run; `b`
        // is a cache hit and so does not need the mutex.
        fixture.graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 1); // only `a` spawns
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 2).await.unwrap();

        let record_a = completed_for(&result.outcomes, &tid("p", "a"));
        let record_b = completed_for(&result.outcomes, &tid("p", "b"));
        assert_eq!(record_a.source, RunSource::FreshRun);
        assert_eq!(record_a.state, RunState::Succeeded);
        assert_eq!(record_b.source, RunSource::CacheHit);
        assert_eq!(record_b.state, RunState::Succeeded);
        assert_eq!(
            spawner.spawns().len(),
            1,
            "only `a` should spawn; `b` is a cache hit",
        );

        let events = observer.events();
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            finished_b < finished_a,
            "cache-hit `b` must finish before exclusive-holder `a`: \
             finished_b={finished_b}, finished_a={finished_a}, events={events:?}",
        );
    }

    #[tokio::test]
    async fn mutex_003_two_exclusive_serialise() {
        // Two tasks with the same workspace-exclusive mutex, both
        // missing the cache. Under MUTEX-003, the second task
        // cannot acquire while the first holds; the scheduler
        // yields it. The spawn count is 2, and the finish order
        // is canonical (`a` before `b`).
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        let result = run_graph(&ctx, 1).await.unwrap();

        assert_eq!(result.outcomes.len(), 2);
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "a")).state,
            RunState::Succeeded,
        );
        assert_eq!(
            completed_for(&result.outcomes, &tid("p", "b")).state,
            RunState::Succeeded,
        );
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            finished_a < finished_b,
            "exclusive holder `a` must finish before contended `b`: \
             finished_a={finished_a}, finished_b={finished_b}, events={events:?}",
        );
    }

    #[tokio::test]
    async fn mutex_004_two_shared_proceed_in_parallel() {
        // Two tasks declaring the same workspace-shared mutex.
        // MUTEX-004 allows them to acquire concurrently, so both
        // spawn before either finishes: every Started precedes
        // every Finished in the observer log.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Shared)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Shared)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let started_a = pos_started(&events, &tid("p", "a"));
        let started_b = pos_started(&events, &tid("p", "b"));
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            started_a < finished_a && started_b < finished_a,
            "both Started events must precede the first Finished under MUTEX-004 \
             shared/shared compatibility: events={events:?}",
        );
        assert!(
            started_a < finished_b && started_b < finished_b,
            "both Started events must precede the second Finished too: events={events:?}",
        );
    }

    #[tokio::test]
    async fn mutex_004_shared_blocks_incoming_exclusive() {
        // Task `a` declares shared `db`; task `b` declares
        // exclusive `db`. With `a` admitted and holding shared,
        // `b`'s incoming exclusive is incompatible per MUTEX-004
        // and yields. After `a` finishes and releases, `b`
        // proceeds.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Shared)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            finished_a < finished_b,
            "shared holder `a` must finish before incoming-exclusive `b`: \
             finished_a={finished_a}, finished_b={finished_b}, events={events:?}",
        );
        assert_eq!(count_started(&events, &tid("p", "b")), 1);
    }

    #[tokio::test]
    async fn mutex_004_exclusive_blocks_incoming_shared() {
        // Mirror of the previous: `a` exclusive, `b` shared.
        // Incoming shared is incompatible with an exclusive
        // holder per MUTEX-004; `b` must wait.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Shared)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            finished_a < finished_b,
            "exclusive holder `a` must finish before incoming-shared `b`: \
             finished_a={finished_a}, finished_b={finished_b}, events={events:?}",
        );
        assert_eq!(count_started(&events, &tid("p", "b")), 1);
    }

    #[tokio::test]
    async fn mutex_001_workspace_and_project_scopes_are_distinct() {
        // Task `a` declares `mutex: db` (workspace scope); task
        // `b` declares `mutex: ~:db` (project scope). MUTEX-001
        // says these are distinct keys even though the bare name
        // coincides; both should acquire and spawn in parallel.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(project_scoped_mutex("db", MutexMode::Exclusive)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let started_a = pos_started(&events, &tid("p", "a"));
        let started_b = pos_started(&events, &tid("p", "b"));
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        assert!(
            started_a < finished_a && started_b < finished_a,
            "workspace and project scopes are distinct keys; both should \
             spawn before either finishes: events={events:?}",
        );
        assert!(started_a < finished_b && started_b < finished_b);
    }

    #[tokio::test]
    async fn mutex_001_project_scoped_same_name_in_different_projects_does_not_serialise() {
        // `p1:t` declares `~:db`; `p2:t` declares `~:db`. Per
        // MUTEX-001, project-scoped names bind to the bearing
        // project, so the two acquisitions resolve to distinct
        // HoldKeys; both proceed in parallel.
        let task_p1 = make_task_with(
            "t",
            &["echo", "p1"],
            Some(project_scoped_mutex("db", MutexMode::Exclusive)),
        );
        let task_p2 = make_task_with(
            "t",
            &["echo", "p2"],
            Some(project_scoped_mutex("db", MutexMode::Exclusive)),
        );
        let p1 = make_project("p1", BTreeSet::new(), vec![task_p1]);
        let p2 = make_project("p2", BTreeSet::new(), vec![task_p2]);
        let ws = make_workspace(vec![p1, p2], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p1", "t"), tid("p2", "t")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        let finished_p1 = pos_finished(&events, &tid("p1", "t"));
        let finished_p2 = pos_finished(&events, &tid("p2", "t"));
        let started_p1 = pos_started(&events, &tid("p1", "t"));
        let started_p2 = pos_started(&events, &tid("p2", "t"));
        assert!(
            started_p1 < finished_p1
                && started_p2 < finished_p1
                && started_p1 < finished_p2
                && started_p2 < finished_p2,
            "project-scoped `~:db` in distinct projects must not serialise: \
             events={events:?}",
        );
    }

    #[tokio::test]
    async fn mutex_005_contended_task_yields_and_resumes_after_release() {
        // The yield-on-contention invariant of MUTEX-005 + S3:
        // when `b`'s lookup misses while `a` holds the mutex,
        // `b` returns to ready WITHOUT a duplicate Started
        // event and without consuming a slot during the yield.
        // After `a` finishes and releases, `b` is re-admitted
        // and runs. The original Started(b) from the first
        // admission still applies; no second Started fires.
        let task_a = make_task_with(
            "a",
            &["echo", "a"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let task_b = make_task_with(
            "b",
            &["echo", "b"],
            Some(workspace_mutex("db", MutexMode::Exclusive)),
        );
        let project = make_project("p", BTreeSet::new(), vec![task_a, task_b]);
        let ws = make_workspace(vec![project], workspace_settings_with(fixed_cap(2)));
        let graph = make_graph(vec![tid("p", "a"), tid("p", "b")], vec![]);
        let fixture = Fixture::new(ws, graph);

        let spawner = MockProcessSpawner::new();
        push_n_default_specs(&spawner, 2);
        let observer = Recorder::default();
        let ctx = make_ctx(&fixture, &spawner, &observer);

        run_graph(&ctx, 1).await.unwrap();
        assert_eq!(spawner.spawns().len(), 2);

        let events = observer.events();
        assert_eq!(
            count_started(&events, &tid("p", "a")),
            1,
            "Started(a) must fire exactly once: events={events:?}",
        );
        assert_eq!(
            count_started(&events, &tid("p", "b")),
            1,
            "Started(b) must fire exactly once even after the yield: \
             events={events:?}",
        );
        let started_a = pos_started(&events, &tid("p", "a"));
        let started_b = pos_started(&events, &tid("p", "b"));
        let finished_a = pos_finished(&events, &tid("p", "a"));
        let finished_b = pos_finished(&events, &tid("p", "b"));
        // Canonical-order admission: both Started events fire
        // before either Finished.
        assert!(started_a < finished_a);
        assert!(started_b < finished_a);
        // Mutex-serialised completion: `a` finishes before `b`.
        assert!(finished_a < finished_b, "events={events:?}");
    }
}