taskvisor 0.4.1

//! # Single task attempt runner.
//!
//! Runs one attempt of a [`Task`].
//!
//! `run_once` is called by [`TaskActor`](super::actor::TaskActor).
//! It does not decide whether the task should restart.
//! It only:
//! - creates a child cancellation token for this attempt,
//! - calls [`Task::spawn`](crate::Task::spawn),
//! - applies the optional per-attempt timeout,
//! - catches task panics and turns them into retryable failures,
//! - publishes attempt-level lifecycle events.
//!
//! ## Event Flow
//!
//! ```text
//! Ok(())
//!   -> TaskStopped
//!   -> return Ok(())
//!
//! Err(TaskError::Canceled)
//!   -> TaskCanceled
//!   -> return Err(Canceled)
//!
//! Err(TaskError::Fail | TaskError::Fatal)
//!   -> TaskFailed
//!   -> return the same error
//!
//! panic in spawn() or task future
//!   -> TaskFailed(reason="task panicked: ...")
//!   -> return Err(TaskError::Fail)
//!
//! timeout
//!   -> cancel child token
//!   -> TimeoutHit
//!   -> TaskFailed
//!   -> return Err(TaskError::Timeout)
//! ```
//!
//! ## Rules
//!
//! - Exactly one terminal attempt event is published per call: `TaskStopped`, `TaskCanceled`, or `TaskFailed`.
//! - `TimeoutHit` is extra context. It is published before `TaskFailed` and is not terminal by itself.
//! - `TaskError::Canceled` is treated as cooperative cancellation, not failure.
//! - A child token is created per attempt. Parent cancellation reaches the child, but cancelling the child does not cancel the parent.
//! - Panics inside the task body become retryable [`TaskError::Fail`] values. The actor's restart policy decides what happens next.

use std::future::Future;
use std::panic::AssertUnwindSafe;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use tokio::time;
use tokio_util::sync::CancellationToken;

use crate::{
    error::TaskError,
    events::{Bus, Event, EventKind},
    identity::TaskId,
    tasks::{BoxTaskFuture, Task, TaskContext},
};

/// Future adapter that catches panics while polling a task future.
///
/// A panic is converted to a retryable [`TaskError::Fail`].
/// This keeps task-body panics on the normal failure/retry path instead of letting them unwind through the actor task.
struct CatchPanic(BoxTaskFuture);

impl Future for CatchPanic {
    type Output = Result<(), TaskError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        std::panic::catch_unwind(AssertUnwindSafe(|| self.0.as_mut().poll(cx)))
            .unwrap_or_else(|payload| Poll::Ready(Err(panic_to_error(payload.as_ref()))))
    }
}

/// Converts a panic payload into a retryable [`TaskError::Fail`].
fn panic_to_error(payload: &(dyn std::any::Any + Send)) -> TaskError {
    let msg = payload
        .downcast_ref::<&'static str>()
        .copied()
        .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
        .unwrap_or("non-string panic payload");
    TaskError::fail(format!("task panicked: {msg}"))
}

/// Executes one attempt of `task`.
///
/// This function publishes attempt-level lifecycle events and returns the raw attempt result to the actor.
/// The actor applies restart policy and backoff.
///
/// ### Steps
///
/// 1. Create a child cancellation token.
/// 2. Call [`Task::spawn`](crate::Task::spawn).
/// 3. Run the returned future, optionally wrapped in `tokio::time::timeout`.
/// 4. Publish the attempt event.
/// 5. Return the attempt result.
///
/// ### Timeout
///
/// If `timeout` is `Some(dur)` and `dur > Duration::ZERO`, the attempt is bounded by `dur`.
///
/// On timeout, the child token is cancelled, `TimeoutHit` is published, and the final attempt event is `TaskFailed` with [`TaskError::Timeout`].
///
/// `None` and `Some(Duration::ZERO)` both mean no timeout.
///
/// ### Cancellation
///
/// Parent cancellation propagates to the child token.
/// A cooperative task should observe [`TaskContext::cancelled`](crate::TaskContext::cancelled) and return [`TaskError::Canceled`].
///
/// `TaskError::Canceled` publishes `TaskCanceled`, not `TaskFailed`.
///
/// ### Panic Handling
///
/// Panics from `spawn()` or from polling the task future are caught and returned as retryable [`TaskError::Fail`] values with reason `task panicked: ...`.
pub async fn run_once<T: Task + ?Sized>(
    task: &T,
    parent: &CancellationToken,
    timeout: Option<Duration>,
    attempt: u32,
    id: TaskId,
    bus: &Bus,
) -> Result<(), TaskError> {
    let started = Instant::now();
    let child = parent.child_token();
    let ctx = TaskContext::from_token(child.clone());

    let fut = match std::panic::catch_unwind(AssertUnwindSafe(move || task.spawn(ctx))) {
        Ok(fut) => CatchPanic(fut),
        Err(payload) => {
            let e = panic_to_error(payload.as_ref());
            publish_failed(bus, id, task.name(), attempt, &e, started.elapsed());
            return Err(e);
        }
    };

    let res = if let Some(dur) = timeout.filter(|d| *d > Duration::ZERO) {
        match time::timeout(dur, fut).await {
            Ok(r) => r,
            Err(_elapsed) => {
                child.cancel();
                publish_timeout(bus, id, task.name(), dur, attempt, started.elapsed());
                Err(TaskError::Timeout { timeout: dur })
            }
        }
    } else {
        fut.await
    };

    match res {
        Ok(()) => {
            publish_stopped(bus, id, task.name(), attempt, started.elapsed());
            Ok(())
        }
        Err(TaskError::Canceled) => {
            publish_canceled(bus, id, task.name(), attempt, started.elapsed());
            Err(TaskError::Canceled)
        }
        Err(e) => {
            publish_failed(bus, id, task.name(), attempt, &e, started.elapsed());
            Err(e)
        }
    }
}

/// Publishes `TaskStopped` for a successful attempt.
fn publish_stopped(bus: &Bus, id: TaskId, name: &str, attempt: u32, duration: Duration) {
    bus.publish(
        Event::new(EventKind::TaskStopped)
            .with_task(name)
            .with_id(id)
            .with_attempt(attempt)
            .with_duration(duration),
    );
}

/// Publishes `TaskCanceled` for a cooperative cancellation attempt.
fn publish_canceled(bus: &Bus, id: TaskId, name: &str, attempt: u32, duration: Duration) {
    bus.publish(
        Event::new(EventKind::TaskCanceled)
            .with_task(name)
            .with_id(id)
            .with_attempt(attempt)
            .with_duration(duration),
    );
}

/// Publishes `TaskFailed` with error details and attempt duration.
fn publish_failed(
    bus: &Bus,
    id: TaskId,
    name: &str,
    attempt: u32,
    err: &TaskError,
    duration: Duration,
) {
    let mut ev = Event::new(EventKind::TaskFailed)
        .with_task(name)
        .with_id(id)
        .with_attempt(attempt)
        .with_duration(duration)
        .with_reason(err.to_string());
    if let Some(code) = err.exit_code() {
        ev = ev.with_exit_code(code);
    }
    bus.publish(ev);
}

/// Publishes `TimeoutHit` before the final timeout `TaskFailed` event.
fn publish_timeout(
    bus: &Bus,
    id: TaskId,
    name: &str,
    dur: Duration,
    attempt: u32,
    duration: Duration,
) {
    bus.publish(
        Event::new(EventKind::TimeoutHit)
            .with_task(name)
            .with_id(id)
            .with_timeout(dur)
            .with_attempt(attempt)
            .with_duration(duration),
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::future::Future;
    use std::pin::Pin;

    type BoxFut = Pin<Box<dyn Future<Output = Result<(), TaskError>> + Send + 'static>>;

    struct SlowTask;

    impl Task for SlowTask {
        fn name(&self) -> &str {
            "slow-task"
        }

        fn spawn(&self, _ctx: TaskContext) -> BoxFut {
            Box::pin(async {
                tokio::time::sleep(Duration::from_secs(3600)).await;
                Ok(())
            })
        }
    }

    struct OkTask;

    impl Task for OkTask {
        fn name(&self) -> &str {
            "ok-task"
        }

        fn spawn(&self, _ctx: TaskContext) -> BoxFut {
            Box::pin(async { Ok(()) })
        }
    }

    struct FailTask;

    impl Task for FailTask {
        fn name(&self) -> &str {
            "fail-task"
        }

        fn spawn(&self, _ctx: TaskContext) -> BoxFut {
            Box::pin(async { Err(TaskError::fail("boom")) })
        }
    }

    #[tokio::test]
    async fn timeout_returns_timeout_variant_not_fail() {
        let bus = Bus::new(16);
        let parent = CancellationToken::new();
        let timeout = Some(Duration::from_millis(50));

        let result = run_once(&SlowTask, &parent, timeout, 1, TaskId::next(), &bus).await;

        match result {
            Err(TaskError::Timeout { timeout: dur }) => {
                assert_eq!(dur, Duration::from_millis(50));
            }
            Err(TaskError::Fail { reason, .. }) => {
                panic!("timeout should return TaskError::Timeout, not TaskError::Fail: {reason}");
            }
            other => {
                panic!("expected TaskError::Timeout, got: {other:?}");
            }
        }
    }

    #[tokio::test]
    async fn success_publishes_stopped_with_attempt() {
        let bus = Bus::new(16);
        let mut rx = bus.subscribe();
        let parent = CancellationToken::new();

        let _ = run_once(&OkTask, &parent, None, 3, TaskId::next(), &bus).await;

        let mut stopped_attempt = None;
        while let Ok(ev) = rx.try_recv() {
            if matches!(ev.kind, EventKind::TaskStopped) {
                stopped_attempt = ev.attempt;
            }
        }
        assert_eq!(
            stopped_attempt,
            Some(3),
            "TaskStopped must carry the attempt number"
        );
    }

    #[tokio::test]
    async fn stopped_event_carries_measured_attempt_duration() {
        struct SleepOk;
        impl Task for SleepOk {
            fn name(&self) -> &str {
                "sleep-ok"
            }
            fn spawn(&self, _ctx: TaskContext) -> BoxFut {
                Box::pin(async {
                    tokio::time::sleep(Duration::from_millis(30)).await;
                    Ok(())
                })
            }
        }

        let bus = Bus::new(16);
        let mut rx = bus.subscribe();
        let parent = CancellationToken::new();

        run_once(&SleepOk, &parent, None, 1, TaskId::next(), &bus)
            .await
            .expect("task succeeds");

        let mut duration_ms = None;
        while let Ok(ev) = rx.try_recv() {
            if ev.kind == EventKind::TaskStopped {
                duration_ms = ev.duration_ms;
            }
        }
        let measured = duration_ms.expect("TaskStopped must carry the attempt duration");
        assert!(
            measured >= 20,
            "attempt duration must reflect the ~30ms of work, got {measured}ms"
        );
    }

    #[tokio::test]
    async fn success_returns_ok() {
        let bus = Bus::new(16);
        let parent = CancellationToken::new();

        let result = run_once(&OkTask, &parent, None, 1, TaskId::next(), &bus).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn failure_returns_fail_variant() {
        let bus = Bus::new(16);
        let parent = CancellationToken::new();

        let result = run_once(&FailTask, &parent, None, 1, TaskId::next(), &bus).await;

        assert!(
            matches!(result, Err(TaskError::Fail { .. })),
            "expected TaskError::Fail, got: {result:?}"
        );
    }

    #[tokio::test]
    async fn timeout_publishes_timeout_hit_event() {
        let bus = Bus::new(16);
        let mut rx = bus.subscribe();
        let parent = CancellationToken::new();

        let _ = run_once(
            &SlowTask,
            &parent,
            Some(Duration::from_millis(50)),
            1,
            TaskId::next(),
            &bus,
        )
        .await;

        let mut saw_timeout_hit = false;
        while let Ok(ev) = rx.try_recv() {
            if matches!(ev.kind, EventKind::TimeoutHit) {
                saw_timeout_hit = true;
            }
        }
        assert!(saw_timeout_hit, "expected TimeoutHit event to be published");
    }
}