rust-job-queue-api-worker-system 0.1.0

A production-shaped Rust job queue: Axum API + async workers + Postgres SKIP LOCKED dequeue, retries with decorrelated jitter, idempotency, cooperative cancellation, OpenAPI, Prometheus metrics.
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
131
132
133
134
135

//! Integration test: API create / get / list. Hits the real Axum router via
//! reqwest against an ephemeral listener.

#![cfg(feature = "api")]

mod common;

use serde_json::{json, Value};

#[tokio::test]
async fn create_then_get_then_list() {
    let pool = common::fresh_pool().await;
    let server = common::spawn_api(pool).await;
    let client = reqwest::Client::new();

    let res = client
        .post(format!("{}/jobs", server.base_url))
        .json(&json!({
            "kind": "send_email",
            "payload": {"to": "a@b.c", "subject": "hi", "body": "hello"},
            "max_attempts": 3,
        }))
        .send()
        .await
        .expect("post");
    assert_eq!(res.status(), 201, "POST should return 201 Created");
    let body: Value = res.json().await.expect("json");
    let id = body["id"].as_str().expect("id is str").to_string();
    assert_eq!(body["status"], "queued");
    assert_eq!(body["kind"], "send_email");
    assert_eq!(body["attempts"], 0);
    assert_eq!(body["max_attempts"], 3);

    let res = client
        .get(format!("{}/jobs/{id}", server.base_url))
        .send()
        .await
        .expect("get by id");
    assert_eq!(res.status(), 200);
    let got: Value = res.json().await.expect("json");
    assert_eq!(got["id"].as_str(), Some(id.as_str()));

    let res = client
        .get(format!("{}/jobs", server.base_url))
        .send()
        .await
        .expect("list");
    assert_eq!(res.status(), 200);
    let list: Vec<Value> = res.json().await.expect("json");
    assert_eq!(list.len(), 1);
    assert_eq!(list[0]["id"].as_str(), Some(id.as_str()));

    server.shutdown().await;
}

#[tokio::test]
async fn get_unknown_id_returns_404() {
    let pool = common::fresh_pool().await;
    let server = common::spawn_api(pool).await;
    let client = reqwest::Client::new();

    let missing = uuid::Uuid::now_v7();
    let res = client
        .get(format!("{}/jobs/{missing}", server.base_url))
        .send()
        .await
        .expect("get");
    assert_eq!(res.status(), 404);
    let body: Value = res.json().await.expect("json");
    assert_eq!(body["error"], "not_found");

    server.shutdown().await;
}

#[tokio::test]
async fn invalid_payload_returns_422() {
    let pool = common::fresh_pool().await;
    let server = common::spawn_api(pool).await;
    let client = reqwest::Client::new();

    let res = client
        .post(format!("{}/jobs", server.base_url))
        .json(&json!({
            "kind": "send_email",
            "payload": {"to": "a@b.c"},   // missing subject + body
        }))
        .send()
        .await
        .expect("post");
    assert_eq!(res.status(), 422);
    let body: Value = res.json().await.expect("json");
    assert_eq!(body["error"], "validation");

    server.shutdown().await;
}

#[tokio::test]
async fn health_endpoint_returns_ok() {
    let pool = common::fresh_pool().await;
    let server = common::spawn_api(pool).await;
    let client = reqwest::Client::new();

    let res = client
        .get(format!("{}/health", server.base_url))
        .send()
        .await
        .expect("health");
    assert_eq!(res.status(), 200);
    let body: Value = res.json().await.expect("json");
    assert_eq!(body["status"], "ok");

    server.shutdown().await;
}

#[tokio::test]
async fn openapi_spec_renders() {
    let pool = common::fresh_pool().await;
    let server = common::spawn_api(pool).await;
    let client = reqwest::Client::new();

    let res = client
        .get(format!("{}/api-docs/openapi.json", server.base_url))
        .send()
        .await
        .expect("openapi");
    assert_eq!(res.status(), 200);
    let doc: Value = res.json().await.expect("json");
    // The spec should declare the four jobs endpoints.
    assert!(doc["paths"]["/jobs"].is_object());
    assert!(doc["paths"]["/jobs/{id}"].is_object());
    assert!(doc["paths"]["/jobs/{id}/cancel"].is_object());
    assert!(doc["paths"]["/health"].is_object());

    server.shutdown().await;
}