graphile_worker 0.13.7

High performance Rust/PostgreSQL job queue (also suitable for getting jobs generated by PostgreSQL triggers/functions out into a different work queue)
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

use graphile_worker::sql::{
    batch_get_jobs::batch_get_jobs, return_jobs::batch::return_jobs,
    task_identifiers::get_tasks_details,
};
use graphile_worker::JobSpec;
use helpers::with_test_db;
use serde_json::json;

mod helpers;

// Run explicitly; database latency makes this unsuitable as a performance assertion.
/// Measures mixed-queue fetch/return cycles while checking claims and lock release.
///
/// This manual database workload has no performance threshold because latency
/// noise would make a speed assertion unreliable.
#[tokio::test]
#[ignore = "manual database fetch benchmark"]
async fn benchmark_cached_fetch_with_mixed_queues() {
    with_test_db(|db| async move {
        db.worker_utils().migrate().await.unwrap();
        for i in 0..100 {
            db.add_job(
                "benchmark",
                json!({}),
                JobSpec {
                    queue_name: if i < 20 {
                        Some(format!("queue_{i}"))
                    } else {
                        None
                    },
                    run_at: Some(chrono::Utc::now() - chrono::Duration::minutes(1)),
                    ..Default::default()
                },
            )
            .await;
        }
        let tasks = get_tasks_details(&db.database, "graphile_worker", vec!["benchmark".into()])
            .await
            .unwrap();
        let start = std::time::Instant::now();
        for _ in 0..1000 {
            let jobs = batch_get_jobs(
                &db.database,
                &tasks,
                "graphile_worker",
                "benchmark",
                &[],
                100,
                None,
            )
            .await
            .unwrap();
            assert_eq!(jobs.len(), 100);
            assert!(jobs.iter().all(|job| *job.attempts() == 1));
            return_jobs(&db.database, &jobs, "graphile_worker", "benchmark")
                .await
                .unwrap();
        }
        eprintln!(
            "1000 fetch/return cycles, 100 jobs (20 named queues + 80 unnamed): {:?}",
            start.elapsed()
        );
        assert!(db
            .get_jobs()
            .await
            .iter()
            .all(|job| job.locked_by.is_none() && job.attempts == 0));
    })
    .await;
}