graphile_worker 0.14.0

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
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

use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use graphile_worker_database::{Database, Schema};
pub use graphile_worker_lifecycle_hooks::LocalQueueMode;
use graphile_worker_lifecycle_hooks::{HookRegistry, LocalQueueInitContext};
use graphile_worker_runtime as runtime;
use graphile_worker_shutdown_signal::ShutdownSignal;
use thiserror::Error;
use tracing::warn;

use graphile_worker_queries::flag_filter::JobFlagFilter;
use graphile_worker_queries::task_identifiers::SharedTaskDetails;

mod cache;
mod claims;
mod config;
mod fetch;
mod jobs;
mod refetch_delay;
mod release;
mod state;

pub use config::{
    LocalQueueConfig, LocalQueueConfigBuilder, LocalQueueConfigError, RefetchDelayConfig,
    RefetchDelayConfigBuilder,
};
use state::LocalQueueState;

/// Sender for notifying the runner that LocalQueue has cached work available.
pub type LocalQueueSignalSender = runtime::Sender<()>;

/// Receiver for LocalQueue cache notifications.
pub(crate) type LocalQueueSignalReceiver = runtime::Receiver<()>;

#[derive(Debug, Error)]
pub enum LocalQueueError {
    #[error("Failed to return jobs to database: {0}")]
    ReturnJobsError(String),
    #[error("Database error: {0}")]
    DatabaseError(#[from] crate::errors::GraphileWorkerError),
}

#[derive(Clone)]
pub struct LocalQueue(Arc<LocalQueueState>);

impl From<LocalQueueState> for LocalQueue {
    fn from(state: LocalQueueState) -> Self {
        Self(Arc::new(state))
    }
}

impl From<LocalQueueParams> for LocalQueue {
    fn from(params: LocalQueueParams) -> Self {
        LocalQueueState::new(params).into()
    }
}

pub struct LocalQueueParams {
    pub config: LocalQueueConfig,
    pub database: Database,
    pub schema: Schema,
    pub worker_id: String,
    pub task_details: SharedTaskDetails,
    pub poll_interval: Duration,
    pub continuous: bool,
    pub shutdown_signal: Option<ShutdownSignal>,
    pub hooks: Arc<HookRegistry>,
    pub job_signal_sender: LocalQueueSignalSender,
    pub use_local_time: bool,
}

impl LocalQueue {
    /// Starts a local fetch loop and, when supplied, its shutdown listener.
    ///
    /// The configuration must be valid for the polling interval. Call `release`
    /// to await cleanup and return cached claims before discarding the queue.
    pub fn new(params: LocalQueueParams) -> Self {
        Self::new_with_filter(params, JobFlagFilter::default())
    }

    /// Starts worker-owned prefetch with an immutable claim-time filter.
    pub(crate) fn new_with_filter(
        params: LocalQueueParams,
        filter: JobFlagFilter<'static>,
    ) -> Self {
        params
            .config
            .validate(params.poll_interval)
            .expect("invalid local queue config");

        let shutdown_signal = params.shutdown_signal.clone();
        let queue: LocalQueue = LocalQueueState::new_with_filter(params, filter).into();
        queue.0.run_complete.store(false, Ordering::Release);

        let queue_clone = queue.clone();
        let run_task = runtime::spawn(async move {
            queue_clone.run().await;
        });
        queue.0.run_task.replace_abort(run_task);

        if let Some(signal) = shutdown_signal {
            let queue_for_shutdown = queue.clone();
            let shutdown_task = runtime::spawn(async move {
                signal.await;
                if let Err(e) = queue_for_shutdown.release().await {
                    warn!(error = %e, "Error releasing LocalQueue on shutdown");
                }
            });
            queue.0.shutdown_task.replace_abort(shutdown_task);
        }

        queue
    }

    /// Runs fetching and records completion for every waiting release caller.
    async fn run(&self) {
        self.0
            .hooks
            .emit(LocalQueueInitContext {
                worker_id: self.0.worker_id.clone(),
            })
            .await;

        self.set_mode(LocalQueueMode::Polling).await;
        self.schedule_fetch().await;

        self.0.run_complete.store(true, Ordering::Release);
        self.0.run_complete_notify.notify_waiters();
    }
}

#[cfg(test)]
mod tests;

#[cfg(all(test, feature = "runtime-tokio"))]
mod return_tests;