tinyflow_framework/runtime/
task_lifecycle.rs

1//! Spawn helpers: forward task `Err`, join failures, and panics to the job `error_tx`.
2
3use std::future::Future;
4
5use tinyflow_api::error::{StreamError, StreamResult, task_failed};
6
7/// Log and send a task failure to the job error channel (logs again if send fails).
8pub async fn report_task_failure(
9    error_tx: &tokio::sync::mpsc::Sender<StreamError>,
10    task_id: u32,
11    task_name: &str,
12    err: StreamError,
13) {
14    log::error!("[{task_name}] task failed: {err}");
15    if error_tx.send(err).await.is_err() {
16        log::error!(
17            "[{task_name}] could not report failure to job (task_id={task_id}, error_tx closed)"
18        );
19    }
20}
21
22/// Run `run` on an inner task; propagate `Err`, panic, or cancellation to `error_tx`.
23pub fn spawn_monitored_task<F, Fut>(
24    task_id: u32,
25    task_name: impl Into<String>,
26    error_tx: tokio::sync::mpsc::Sender<StreamError>,
27    run: F,
28) -> tokio::task::JoinHandle<()>
29where
30    F: FnOnce() -> Fut + Send + 'static,
31    Fut: Future<Output = StreamResult<()>> + Send + 'static,
32{
33    let task_name = task_name.into();
34    tokio::spawn(async move {
35        let inner = tokio::spawn(run());
36        match inner.await {
37            Ok(Ok(())) => {}
38            Ok(Err(e)) => report_task_failure(&error_tx, task_id, &task_name, e).await,
39            Err(join_err) => {
40                report_task_failure(
41                    &error_tx,
42                    task_id,
43                    &task_name,
44                    task_failed(task_id, format!("task panicked or cancelled: {join_err}")),
45                )
46                .await;
47            }
48        }
49    })
50}
51
52#[cfg(test)]
53mod tests {
54    use super::*;
55    use tinyflow_api::error::StreamError;
56
57    #[tokio::test]
58    async fn report_task_failure_logs_when_channel_closed() {
59        let (tx, rx) = tokio::sync::mpsc::channel(1);
60        drop(rx);
61        report_task_failure(
62            &tx,
63            1,
64            "test-task",
65            StreamError::TaskFailed {
66                task_id: 1,
67                reason: "boom".into(),
68            },
69        )
70        .await;
71    }
72}
tinyflow_framework/runtime/task_lifecycle.rs

tinyflow_framework/runtime/
task_lifecycle.rs
