boson-runtime 0.1.1

Boson worker engine, builder, and enqueue orchestration
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! Shared run finish logic for worker backends.

use std::time::Duration;

use async_trait::async_trait;
use boson_core::{BosonError, Job, JobStatus, Result, RunStatus, TaskConfig, RetryPolicy};
use tokio::time::sleep;

use crate::telemetry;

/// Persistence and retry operations used by [`finish_job_execution`].
#[async_trait]
pub trait RunLifecycleHost: Send + Sync {
    async fn record_run_finish(
        &self,
        run_id: &str,
        status: RunStatus,
        duration_ms: Option<i64>,
        error_message: Option<String>,
    ) -> Result<()>;

    async fn put_job(&self, job: Job);

    async fn load_task_config(&self, task_name: &str) -> Result<TaskConfig>;

    async fn schedule_retry(&self, job: Job, delay_ms: u64);
}

pub async fn finish_job_execution<H: RunLifecycleHost>(
    host: &H,
    run_id: String,
    job: Job,
    result: std::result::Result<(), BosonError>,
    duration_ms: i64,
) {
    match result {
        Ok(()) => {
            let _ = host
                .record_run_finish(&run_id, RunStatus::Success, Some(duration_ms), None)
                .await;
            telemetry::record_task_completed(&job.task_name, &job.job_id, &run_id, duration_ms);
            let mut finished = job;
            finished.status = JobStatus::Success;
            host.put_job(finished).await;
        }
        Err(e) => {
            let err_msg = e.to_string();
            let config = host.load_task_config(&job.task_name).await.ok();
            let retry_policy = config.as_ref().map(|c| &c.retry_policy);
            let should_retry = retry_policy
                .is_some_and(|r| job.attempt.cast_unsigned() < r.max_attempts);
            if should_retry {
                let policy = retry_policy.unwrap();
                let delay_ms = compute_retry_delay_ms(policy, job.attempt);
                let _ = host
                    .record_run_finish(
                        &run_id,
                        RunStatus::Failed,
                        Some(duration_ms),
                        Some(err_msg.clone()),
                    )
                    .await;
                telemetry::record_task_failed(
                    &job.task_name,
                    &job.job_id,
                    &run_id,
                    &err_msg,
                    true,
                );
                host.schedule_retry(job, delay_ms).await;
            } else {
                let _ = host
                    .record_run_finish(
                        &run_id,
                        RunStatus::Failed,
                        Some(duration_ms),
                        Some(err_msg.clone()),
                    )
                    .await;
                telemetry::record_task_failed(
                    &job.task_name,
                    &job.job_id,
                    &run_id,
                    &err_msg,
                    false,
                );
                let mut failed = job;
                failed.status = JobStatus::Failed;
                host.put_job(failed).await;
            }
        }
    }
}

pub async fn sleep_retry_delay(delay_ms: u64) {
    sleep(Duration::from_millis(delay_ms)).await;
}

fn ms_u64_to_f64(ms: u64) -> f64 {
    f64::from(u32::try_from(ms.min(u64::from(u32::MAX))).unwrap_or(u32::MAX))
}

fn round_f64_to_u64_at_most(max: u64, value: f64) -> u64 {
    if value <= 0.0 || !value.is_finite() {
        return 0;
    }
    let max = max.min(u64::from(u32::MAX));
    if value >= f64::from(u32::try_from(max).unwrap_or(u32::MAX)) {
        return max;
    }
    let target = value.round();
    let mut lo = 0u64;
    let mut hi = max;
    while lo < hi {
        let mid = lo + (hi - lo).div_ceil(2);
        if f64::from(u32::try_from(mid).unwrap_or(u32::MAX)) <= target {
            lo = mid;
        } else {
            hi = mid - 1;
        }
    }
    lo
}

fn compute_retry_delay_ms(policy: &RetryPolicy, attempt: i32) -> u64 {
    let exponent = i32::max(attempt - 1, 0);
    let scaled =
        ms_u64_to_f64(policy.base_delay_ms) * policy.backoff_multiplier.powi(exponent);
    round_f64_to_u64_at_most(policy.max_delay_ms, scaled)
}