azums 0.2.0

High-performance job queue & streaming engine for Rust — from embedded to cloud
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

use serde_json::json;

mod common;
use common::setup_db;

use azums::jobs::{
    timeline::{self, TimelineEvent},
    AttemptsRepo, JobsRepo, PoliciesRepo, PolicyDecisionsRepo,
};

#[tokio::test]
async fn timeline_includes_policy_decision_event() {
    let Some(pool) = setup_db().await else {
        return;
    };

    let jobs = JobsRepo::new(pool.clone());
    let attempts = AttemptsRepo::new(pool.clone());
    let policies = PoliciesRepo::new(pool.clone());
    let policy_decisions = PolicyDecisionsRepo::new(pool.clone());

    // 1) set strict policy so we will throttle
    // max_in_flight = 0 => always throttles when trying to lease
    policies
        .upsert_policy("default", 10_000, 0, 250)
        .await
        .unwrap();

    // 2) enqueue a runnable job
    let job_id = jobs
        .enqueue_now("default", "ok_job", json!({"x": 1}))
        .await
        .unwrap();

    // 3) attempt leasing -> should be throttled and return None
    let leased = jobs.lease_one_job("default", "worker-1", 10).await.unwrap();
    assert!(leased.is_none(), "expected throttle to return None");

    // 4) timeline should include a PolicyDecision event in story
    let tl = timeline::build_timeline(&jobs, &attempts, &policy_decisions, job_id)
        .await
        .unwrap()
        .expect("job should exist");

    let has_policy = tl.story.iter().any(|e| {
        matches!(
            e,
            TimelineEvent::PolicyDecision {
                decision,
                reason_code,
                ..
            } if decision == "THROTTLED" && reason_code == "IN_FLIGHT_EXCEEDED"
        )
    });

    assert!(
        has_policy,
        "expected story to include THROTTLED/IN_FLIGHT_EXCEEDED"
    );

    // Optional: prove it was written to DB too (not just timeline merge)
    let rows = policy_decisions.list_for_job(job_id).await.unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].decision, "THROTTLED");
    assert_eq!(rows[0].reason_code, "IN_FLIGHT_EXCEEDED");
}