docbox_core/tasks/
background_task.rs

1use chrono::{DateTime, Utc};
2use docbox_database::{
3    DbPool, DbResult,
4    models::{
5        document_box::DocumentBoxScopeRaw,
6        tasks::{Task, TaskId, TaskStatus},
7    },
8};
9use std::{future::Future, time::Duration};
10use tokio::time::sleep;
11use tracing::Instrument;
12
13pub async fn background_task<Fut>(
14    db: DbPool,
15    scope: DocumentBoxScopeRaw,
16    future: Fut,
17) -> DbResult<(TaskId, DateTime<Utc>)>
18where
19    Fut: Future<Output = (TaskStatus, serde_json::Value)> + Send + 'static,
20{
21    // Create task for progression
22    let mut task = Task::create(&db, scope).await?;
23
24    let task_id = task.id;
25    let created_at = task.created_at;
26
27    let span = tracing::Span::current();
28
29    // Swap background task
30    tokio::spawn(
31        async move {
32            let (status, output) = future.await;
33
34            // Multiple retry attempts:
35            // We retry multiple times because things like database connection exhaustion could
36            // prevent a connection from being acquired to commit the state. But we need to make
37            // sure that this state is committed
38            for i in 1..5 {
39                // Update task completion
40                match task.complete_task(&db, status, Some(output.clone())).await {
41                    Ok(_) => break,
42                    Err(error) => {
43                        tracing::error!(?error, "failed to mark task as complete");
44                        sleep(Duration::from_secs(60 * (i * i))).await;
45                    }
46                }
47            }
48        }
49        .instrument(span),
50    );
51
52    Ok((task_id, created_at))
53}
docbox_core/tasks/background_task.rs

docbox_core/tasks/
background_task.rs
