tinyflow-framework 0.1.1

Streaming runtime engine — checkpoint, chained task execution, and state store
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

//! Spawn helpers: forward task `Err`, join failures, and panics to the job `error_tx`.

use std::future::Future;

use tinyflow_api::error::{StreamError, StreamResult, task_failed};

/// Log and send a task failure to the job error channel (logs again if send fails).
pub async fn report_task_failure(
    error_tx: &tokio::sync::mpsc::Sender<StreamError>,
    task_id: u32,
    task_name: &str,
    err: StreamError,
) {
    log::error!("[{task_name}] task failed: {err}");
    if error_tx.send(err).await.is_err() {
        log::error!(
            "[{task_name}] could not report failure to job (task_id={task_id}, error_tx closed)"
        );
    }
}

/// Run `run` on an inner task; propagate `Err`, panic, or cancellation to `error_tx`.
pub fn spawn_monitored_task<F, Fut>(
    task_id: u32,
    task_name: impl Into<String>,
    error_tx: tokio::sync::mpsc::Sender<StreamError>,
    run: F,
) -> tokio::task::JoinHandle<()>
where
    F: FnOnce() -> Fut + Send + 'static,
    Fut: Future<Output = StreamResult<()>> + Send + 'static,
{
    let task_name = task_name.into();
    tokio::spawn(async move {
        let inner = tokio::spawn(run());
        match inner.await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => report_task_failure(&error_tx, task_id, &task_name, e).await,
            Err(join_err) => {
                report_task_failure(
                    &error_tx,
                    task_id,
                    &task_name,
                    task_failed(task_id, format!("task panicked or cancelled: {join_err}")),
                )
                .await;
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use tinyflow_api::error::StreamError;

    #[tokio::test]
    async fn report_task_failure_logs_when_channel_closed() {
        let (tx, rx) = tokio::sync::mpsc::channel(1);
        drop(rx);
        report_task_failure(
            &tx,
            1,
            "test-task",
            StreamError::TaskFailed {
                task_id: 1,
                reason: "boom".into(),
            },
        )
        .await;
    }
}