azums 0.1.0

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

// crates/postgresflow/tests/timeline.rs
mod common;

use common::setup_db;

use azums::jobs::timeline::build_timeline;
use azums::jobs::{AttemptsRepo, JobsRepo, PolicyDecisionsRepo};

use uuid::Uuid;

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

    let jobs = JobsRepo::new(pool.clone());
    let attempts = AttemptsRepo::new(pool.clone());
    let policy = PolicyDecisionsRepo::new(pool.clone()); // ✅ new

    // Insert a job directly (or use jobs.enqueue_now if you prefer)
    let job_id: Uuid = sqlx::query_scalar(
        r#"
        INSERT INTO jobs (queue, job_type, payload_json, run_at, status, priority, max_attempts)
        VALUES ('default', 'email_send', '{}'::jsonb, now(), 'queued', 0, 5)
        RETURNING id
        "#,
    )
    .fetch_one(&pool)
    .await
    .unwrap();

    // Attempt 1 fails
    let leased = jobs
        .lease_one_job("default", "worker-a", 30)
        .await
        .unwrap()
        .unwrap();

    let a1 = attempts.start_attempt(leased.id, "worker-a").await.unwrap();
    attempts
        .finish_failed(a1.id, 10, "timeout", "request timed out")
        .await
        .unwrap();

    // mimic retry scheduling
    jobs.reschedule_for_retry(
        job_id,
        chrono::Utc::now(),
        Some("timeout"),
        Some("request timed out"),
    )
    .await
    .unwrap();

    // Attempt 2 succeeds
    let leased2 = jobs
        .lease_one_job("default", "worker-b", 30)
        .await
        .unwrap()
        .unwrap();

    let a2 = attempts
        .start_attempt(leased2.id, "worker-b")
        .await
        .unwrap();
    attempts.finish_succeeded(a2.id, 5).await.unwrap();
    jobs.mark_succeeded(job_id, "worker-b").await.unwrap();

    let tl = build_timeline(&jobs, &attempts, &policy, job_id) // ✅ new arg
        .await
        .unwrap()
        .unwrap();

    assert_eq!(tl.job_id, job_id);
    assert_eq!(tl.status, "succeeded");
    assert_eq!(tl.last_worker_id.as_deref(), Some("worker-b"));
    assert_eq!(tl.attempts.len(), 2);

    assert_eq!(tl.attempts[0].attempt_no, 1);
    assert_eq!(tl.attempts[0].status, "failed");
    assert_eq!(tl.attempts[0].error_code.as_deref(), Some("timeout"));

    assert_eq!(tl.attempts[1].attempt_no, 2);
    assert_eq!(tl.attempts[1].status, "succeeded");
}