rustqueue 0.2.0

Background jobs without infrastructure — embeddable job queue with zero external dependencies
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
131
132
133
134
135
136
137
138
139

//! Smoke tests for the full RustQueue server startup.

use std::sync::Arc;

use reqwest::Client;
use serde_json::Value;

use rustqueue::api::{self, AppState};
use rustqueue::engine::queue::QueueManager;
use rustqueue::storage::{MemoryStorage, RedbStorage};

/// Start a full test server (HTTP + TCP) on random ports.
/// Returns the HTTP base URL and TCP address.
async fn start_full_server() -> (String, std::net::SocketAddr) {
    let dir = tempfile::tempdir().unwrap();
    let db_path = dir.path().join("smoke.redb");
    // Leak the tempdir so it outlives the spawned server tasks.
    let _keep = Box::leak(Box::new(dir));

    let (event_tx, _) = tokio::sync::broadcast::channel(1024);
    let storage = Arc::new(RedbStorage::new(&db_path).unwrap());
    let queue_manager = Arc::new(QueueManager::new(storage));

    // Build HTTP app
    let state = Arc::new(AppState {
        queue_manager: Arc::clone(&queue_manager),
        start_time: std::time::Instant::now(),
        metrics_handle: None,
        event_tx,
        auth_config: rustqueue::config::AuthConfig::default(),
        auth_rate_limiter: rustqueue::api::auth::AuthRateLimiter::new(),
        webhook_manager: None,
    });
    let app = api::router(state);

    // Bind HTTP listener on random port
    let http_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let http_addr = http_listener.local_addr().unwrap();

    // Bind TCP listener on random port
    let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let tcp_addr = tcp_listener.local_addr().unwrap();

    // Spawn HTTP server
    tokio::spawn(async move {
        axum::serve(http_listener, app).await.unwrap();
    });

    // Spawn TCP server (auth disabled for smoke tests)
    let auth_config = rustqueue::config::AuthConfig::default();
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
    // Leak the sender so it outlives the spawned server task (never sends shutdown).
    let _keep_tx = Box::leak(Box::new(shutdown_tx));
    tokio::spawn(async move {
        rustqueue::protocol::start_tcp_server(
            tcp_listener,
            queue_manager,
            auth_config,
            shutdown_rx,
        )
        .await;
    });

    (format!("http://{http_addr}"), tcp_addr)
}

#[tokio::test]
async fn test_server_starts_and_responds_to_health() {
    let (base_url, _tcp_addr) = start_full_server().await;
    let client = Client::new();

    let resp = client
        .get(format!("{base_url}/api/v1/health"))
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["ok"], true);
    assert_eq!(body["status"], "healthy");
    assert_eq!(body["version"], "0.1.0");
    assert!(body["uptime_seconds"].is_number());
}

#[tokio::test]
async fn test_server_cli_help() {
    #[allow(deprecated)]
    let mut cmd = assert_cmd::Command::cargo_bin("rustqueue").unwrap();
    cmd.arg("--help");
    cmd.assert()
        .success()
        .stdout(predicates::str::contains("Background jobs without infrastructure"));
}

#[tokio::test]
async fn test_cli_status_help() {
    #[allow(deprecated)]
    let mut cmd = assert_cmd::Command::cargo_bin("rustqueue").unwrap();
    cmd.arg("status").arg("--help");
    cmd.assert()
        .success()
        .stdout(predicates::str::contains("queue status"));
}

#[tokio::test]
async fn test_server_with_memory_backend() {
    let (event_tx, _) = tokio::sync::broadcast::channel(1024);
    let storage = Arc::new(MemoryStorage::new());
    let queue_manager = Arc::new(QueueManager::new(storage));

    let state = Arc::new(AppState {
        queue_manager: Arc::clone(&queue_manager),
        start_time: std::time::Instant::now(),
        metrics_handle: None,
        event_tx,
        auth_config: rustqueue::config::AuthConfig::default(),
        auth_rate_limiter: rustqueue::api::auth::AuthRateLimiter::new(),
        webhook_manager: None,
    });
    let app = api::router(state);

    let http_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let http_addr = http_listener.local_addr().unwrap();

    tokio::spawn(async move {
        axum::serve(http_listener, app).await.unwrap();
    });

    let client = Client::new();
    let resp = client
        .get(format!("http://{http_addr}/api/v1/health"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["ok"], true);
}