graphile_worker 0.13.3

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

use std::{fmt, future::Future, time::Duration};

use futures::FutureExt;
use graphile_worker_shutdown_signal::ShutdownSignal;

/// Worker shutdown behavior.
///
/// These options apply to ordinary graceful shutdown, even when worker recovery
/// heartbeat/sweeping is disabled.
#[derive(Clone)]
pub struct WorkerShutdownConfig {
    /// Whether to install OS-level shutdown signal handlers.
    pub listen_os_shutdown_signals: bool,
    /// Time to let in-flight tasks finish after a shutdown signal.
    pub grace_period: Duration,
    /// Delay before a shutdown-aborted job is eligible to run again.
    pub interrupted_job_retry_delay: Duration,
    /// Custom application shutdown signal.
    pub shutdown_signal: Option<ShutdownSignal>,
}

impl Default for WorkerShutdownConfig {
    fn default() -> Self {
        Self {
            listen_os_shutdown_signals: true,
            grace_period: Duration::from_secs(5),
            interrupted_job_retry_delay: Duration::from_secs(30),
            shutdown_signal: None,
        }
    }
}

impl fmt::Debug for WorkerShutdownConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WorkerShutdownConfig")
            .field(
                "listen_os_shutdown_signals",
                &self.listen_os_shutdown_signals,
            )
            .field("grace_period", &self.grace_period)
            .field(
                "interrupted_job_retry_delay",
                &self.interrupted_job_retry_delay,
            )
            .field("shutdown_signal", &self.shutdown_signal.is_some())
            .finish()
    }
}

impl WorkerShutdownConfig {
    pub fn listen_os_shutdown_signals(mut self, value: bool) -> Self {
        self.listen_os_shutdown_signals = value;
        self
    }

    pub fn grace_period(mut self, value: Duration) -> Self {
        self.grace_period = value;
        self
    }

    pub fn interrupted_job_retry_delay(mut self, value: Duration) -> Self {
        self.interrupted_job_retry_delay = value;
        self
    }

    /// Sets a custom application shutdown signal.
    ///
    /// The future should complete when the host application requests shutdown.
    /// The worker still owns graceful draining after the signal is received.
    pub fn shutdown_signal(mut self, signal: impl Future<Output = ()> + Send + 'static) -> Self {
        self.shutdown_signal = Some(future_to_shutdown_signal(signal));
        self
    }
}

pub(crate) fn future_to_shutdown_signal(
    signal: impl Future<Output = ()> + Send + 'static,
) -> ShutdownSignal {
    signal.boxed().shared()
}

#[cfg(test)]
mod tests {
    use super::WorkerShutdownConfig;

    #[test]
    fn debug_reports_shutdown_signal_presence() {
        let without_signal = format!("{:?}", WorkerShutdownConfig::default());
        assert!(without_signal.contains("shutdown_signal: false"));

        let with_signal = format!(
            "{:?}",
            WorkerShutdownConfig::default().shutdown_signal(std::future::pending::<()>())
        );
        assert!(with_signal.contains("shutdown_signal: true"));
    }
}