some_executor 0.7.2

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Exposes the task registry through exfiltrate's `snapshot` command.
//!
//! Two subsystems, both under the one command rather than commands of their
//! own — see `exfiltrate::provider` for why:
//!
//! * `tasks` — one row per retained task; `--id <task id>` narrows to one.
//! * `executor` — lifetime counters, and which executors are installed.
//!
//! Call [`install`] once, after `exfiltrate::begin()`.

use exfiltrate::provider::{Provider, ProviderResult, Row, SnapshotRequest};

use super::registry;

/// Registers both providers. Idempotent: re-registering replaces.
///
/// ```no_run
/// # // no_run because: `begin` opens exfiltrate's listening socket and installs
/// # // process-global state, which a doctest process must not do.
/// exfiltrate::begin();
/// some_executor::task::exfiltrate::install();
/// ```
pub fn install() {
    exfiltrate::provider::add_provider(Tasks);
    exfiltrate::provider::add_provider(Executor);
}

struct Tasks;

impl Provider for Tasks {
    fn subsystem(&self) -> &'static str {
        "tasks"
    }

    fn description(&self) -> &'static str {
        "spawned tasks: id, label, priority, hint, state, and poll activity"
    }

    fn snapshot(&self, request: &SnapshotRequest<'_>) -> ProviderResult {
        let Some(entries) = registry::entries() else {
            // The registry was locked. Saying so beats an empty list, which
            // would read as "no tasks".
            return ProviderResult::Busy;
        };
        let stats = registry::stats();

        let wanted: Option<u64> = match request.selector() {
            Some(selector) => match selector.parse::<u64>() {
                Ok(id) => Some(id),
                Err(_) => {
                    return ProviderResult::Unavailable(format!(
                        "--id must be a task id; got {selector:?}"
                    ));
                }
            },
            None => None,
        };

        let now = crate::sys::Instant::now();
        let mut rows = Vec::new();
        // Oldest first: the oldest un-completed task is usually the bug.
        for entry in entries {
            if request.should_stop() {
                return ProviderResult::Partial(rows, "deadline".to_string());
            }
            if let Some(id) = wanted
                && entry.task_id.to_u64() != id
            {
                continue;
            }
            let mut row = Row::new()
                .support("task_id", entry.task_id.to_u64())
                .support("state", entry.state.name())
                .support("priority", format!("{:?}", entry.priority))
                .support("hint", format!("{:?}", entry.hint))
                .support("polls", entry.polls)
                .support(
                    "age_ms",
                    now.duration_since(entry.spawned_at).as_millis() as u64,
                )
                // The label is chosen by whoever spawned the task and can name
                // anything in the application's own vocabulary, so it is
                // local-only rather than support-safe.
                .local("label", entry.label.clone());
            row = match entry.last_poll {
                Some(at) => row.support(
                    "since_last_poll_ms",
                    now.duration_since(at).as_millis() as u64,
                ),
                // Distinguishable from "polled a long time ago", which is a
                // different bug.
                None => row.support("never_polled", true),
            };
            rows.push(row);
        }

        // Bounded retention is reported, never implied. A caller must be able
        // to tell "nothing else is running" from "I lost it".
        if stats.dropped > 0 {
            return ProviderResult::Partial(
                rows,
                format!(
                    "the registry has dropped {} task(s) to stay within its capacity of {}; \
                     raise SOME_EXECUTOR_TASK_REGISTRY_CAPACITY to keep more",
                    stats.dropped, stats.capacity
                ),
            );
        }
        ProviderResult::Rows(rows)
    }
}

struct Executor;

impl Provider for Executor {
    fn subsystem(&self) -> &'static str {
        "executor"
    }

    fn description(&self) -> &'static str {
        "task counters, registry capacity, and which executors are installed"
    }

    fn snapshot(&self, _request: &SnapshotRequest<'_>) -> ProviderResult {
        let stats = registry::stats();
        let global = crate::global_executor::global_executor(|executor| executor.is_some());
        let thread = crate::thread_executor::thread_executor(|executor| executor.is_some());

        ProviderResult::Rows(vec![
            Row::new()
                .support("spawned", stats.spawned)
                .support("completed", stats.completed)
                .support("cancelled", stats.cancelled)
                .support("dropped", stats.dropped)
                .support("retained", stats.retained as u64)
                .support("capacity", stats.capacity as u64)
                // "Why did my task not run" is very often one of these being
                // absent, and it is invisible from outside otherwise.
                .support("global_executor_installed", global)
                .support("thread_executor_installed", thread),
        ])
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::task::TaskID;
    use exfiltrate_internal::command::{Command, Response};
    use exfiltrate_internal::snapshot::{SnapshotOutcome, SnapshotSet, SnapshotValue};

    fn ask(args: &[&str]) -> SnapshotSet {
        let response = exfiltrate::provider::Snapshot
            .execute(args.iter().map(|arg| arg.to_string()).collect())
            .expect("snapshot should not fail");
        match response {
            Response::Bytes(bytes) => rmp_serde::from_slice(&bytes).expect("decode"),
            other => panic!("expected a structured response, got {other:?}"),
        }
    }

    /// Shares the registry's own lock: a lock per module would let the two
    /// suites interleave over one global registry.
    fn session(capacity: usize) -> std::sync::MutexGuard<'static, ()> {
        let guard = registry::test_session(capacity);
        install();
        guard
    }

    fn spawn(id: u64, label: &str) {
        registry::record_spawn(
            TaskID::from_u64(id),
            label,
            priority::Priority::unit_test(),
            crate::hint::Hint::Unknown,
        );
    }

    /// The task label is the application's own vocabulary, so a remote view
    /// does not get it — but the field keeps its slot.
    #[test]
    fn the_task_label_is_local_only() {
        let _session = session(8);
        spawn(1, "a label someone chose");

        let set = ask(&["--subsystem", "tasks"]);
        let answer = &set.snapshots[0];
        let row = &answer.rows[0];
        assert_eq!(
            row.get("task_id").unwrap().value,
            Some(SnapshotValue::U64(1))
        );
        assert_eq!(
            row.get("label").unwrap().value,
            None,
            "a remote view must not see the label"
        );

        let set = ask(&["--subsystem", "tasks", "--view", "local"]);
        assert_eq!(
            set.snapshots[0].rows[0].get("label").unwrap().value,
            Some(SnapshotValue::String("a label someone chose".to_string()))
        );
    }

    /// A never-polled task is distinguishable from one polled long ago, because
    /// they are different bugs.
    #[test]
    fn a_never_polled_task_says_so() {
        let _session = session(8);
        spawn(7, "unpolled");
        let set = ask(&["--subsystem", "tasks", "--id", "7"]);
        let row = &set.snapshots[0].rows[0];
        assert_eq!(
            row.get("never_polled").unwrap().value,
            Some(SnapshotValue::Bool(true))
        );
        assert!(row.get("since_last_poll_ms").is_none());
    }

    #[test]
    fn the_selector_narrows_to_one_task() {
        let _session = session(8);
        spawn(1, "one");
        spawn(2, "two");
        let set = ask(&["--subsystem", "tasks", "--id", "2"]);
        let answer = &set.snapshots[0];
        assert_eq!(answer.returned, 1);
        assert_eq!(
            answer.rows[0].get("task_id").unwrap().value,
            Some(SnapshotValue::U64(2))
        );
    }

    /// The thing the maintainer asked for explicitly: when the registry has
    /// dropped tasks, the answer says so rather than looking complete.
    #[test]
    fn dropped_tasks_are_reported_explicitly() {
        let _session = session(2);
        for id in 1..=5 {
            spawn(id, "churn");
        }
        let set = ask(&["--subsystem", "tasks"]);
        let answer = &set.snapshots[0];
        assert_eq!(answer.outcome, SnapshotOutcome::Partial);
        let reason = answer.reason.as_deref().unwrap();
        assert!(reason.contains("dropped"), "{reason}");
        assert!(
            reason.contains("SOME_EXECUTOR_TASK_REGISTRY_CAPACITY"),
            "the reason should say how to keep more: {reason}"
        );
    }

    /// "Why did my task not run" is very often an executor that was never
    /// installed, which is invisible from outside otherwise.
    #[test]
    fn the_executor_subsystem_reports_counters_and_installation() {
        let _session = session(8);
        spawn(1, "one");
        registry::record_terminal(TaskID::from_u64(1), registry::TaskState::Completed);

        let set = ask(&["--subsystem", "executor"]);
        let row = &set.snapshots[0].rows[0];
        assert_eq!(
            row.get("spawned").unwrap().value,
            Some(SnapshotValue::U64(1))
        );
        assert_eq!(
            row.get("completed").unwrap().value,
            Some(SnapshotValue::U64(1))
        );
        assert!(matches!(
            row.get("global_executor_installed").unwrap().value,
            Some(SnapshotValue::Bool(_))
        ));
    }
}