rainier_queue/driver.rs
1//! Where queued jobs wait — [`QueueDriver`].
2
3use rainier_support::setting_enum;
4
5setting_enum! {
6 /// Which [`Queue`](crate::Queue) to build.
7 ///
8 /// ```
9 /// use rainier_queue::QueueDriver;
10 /// use rainier_support::Setting;
11 ///
12 /// assert!(!QueueDriver::Sync.is_deferred(), "sync runs the job inline");
13 /// assert!(QueueDriver::parse("database").unwrap().survives_a_restart());
14 /// ```
15 pub enum QueueDriver: "queue driver" {
16 /// Run the job inline, on the thread that dispatched it.
17 ///
18 /// The default because it needs no worker: a fresh clone dispatches a
19 /// job and the job happens. Also the reason a slow job makes a slow
20 /// request — which is the thing a queue exists to avoid, so this is a
21 /// development setting.
22 #[default]
23 Sync = "sync",
24
25 /// A queue in this process's memory.
26 ///
27 /// For tests that want a job to be *queued* rather than *run*, so they
28 /// can assert on it. Lost on restart, invisible to other processes.
29 Memory = "memory",
30
31 /// Two tables in the application's own database.
32 ///
33 /// Durable, shared, and needs no new infrastructure — the reason it is
34 /// the usual production answer. Needs the migrations
35 /// (`DatabaseQueue::migrations()`) and a `queue:work` process.
36 Database = "database",
37
38 /// Redis streams.
39 ///
40 /// Durable-ish, and the qualifier is the point: Redis acknowledges a
41 /// write before it is on disk and before a replica has it, so a job
42 /// can be lost to a crash or a failover. Right for work you can afford
43 /// to lose, wrong for work you cannot — see the
44 /// [driver's docs](crate::redis). Needs the `redis` feature.
45 Redis = "redis",
46
47 /// Amazon SQS.
48 ///
49 /// Durable and managed, with the visibility timeout doing the
50 /// reservation. Needs the `sqs` feature.
51 Sqs = "sqs",
52 }
53}
54
55impl QueueDriver {
56 /// Whether dispatching returns before the job has run.
57 ///
58 /// `false` for [`Sync`](Self::Sync), which is the whole difference between
59 /// it and the others — and the reason a test that asserts on a side effect
60 /// passes under `sync` and hangs under everything else.
61 pub fn is_deferred(&self) -> bool {
62 !matches!(self, Self::Sync)
63 }
64
65 /// Whether a job dispatched now survives **this** process exiting.
66 ///
67 /// True for anything that stores the job somewhere else. It is not the
68 /// same question as whether the job is safe — see
69 /// [`may_lose_an_accepted_job`](Self::may_lose_an_accepted_job), which is
70 /// the one that catches Redis.
71 pub fn survives_a_restart(&self) -> bool {
72 matches!(self, Self::Database | Self::Redis | Self::Sqs)
73 }
74
75 /// Whether a dispatch this driver **accepted** can still be lost.
76 ///
77 /// True only for [`Redis`](Self::Redis), and it is the distinction the
78 /// obvious question misses: a Redis job survives your process restarting,
79 /// because it lives in another one — and can still vanish, because Redis
80 /// acknowledges a write before it is on disk (`appendfsync everysec`
81 /// leaves a window of about a second) and before any replica has it (
82 /// replication is asynchronous, so a failover can drop confirmed writes).
83 ///
84 /// A database queue's insert is committed before it returns; SQS is
85 /// replicated across availability zones before it acknowledges. Neither
86 /// can tell you a job is queued and then lose it.
87 ///
88 /// So this is the predicate a deploy check should look at when the work is
89 /// a payment rather than a cache warm.
90 pub fn may_lose_an_accepted_job(&self) -> bool {
91 matches!(self, Self::Redis)
92 }
93
94 /// Whether this driver needs a `queue:work` process to make progress.
95 pub fn needs_a_worker(&self) -> bool {
96 self.is_deferred()
97 }
98}
99
100#[cfg(test)]
101mod tests {
102 use super::*;
103 use rainier_support::Setting;
104
105 #[test]
106 fn sync_is_the_only_driver_that_runs_inline() {
107 assert!(!QueueDriver::Sync.is_deferred());
108 assert!(!QueueDriver::Sync.needs_a_worker());
109
110 for driver in QueueDriver::ALL.iter().filter(|d| **d != QueueDriver::Sync) {
111 assert!(driver.is_deferred(), "{driver} should defer");
112 assert!(driver.needs_a_worker(), "{driver} should need a worker");
113 }
114 }
115
116 #[test]
117 fn only_the_backed_drivers_survive_a_restart() {
118 assert!(QueueDriver::Database.survives_a_restart());
119 assert!(QueueDriver::Sqs.survives_a_restart());
120 assert!(QueueDriver::Redis.survives_a_restart(), "it lives in another process");
121 assert!(!QueueDriver::Memory.survives_a_restart());
122 }
123
124 #[test]
125 fn only_redis_can_lose_a_job_it_already_accepted() {
126 // The distinction `survives_a_restart` cannot make: the job outlives
127 // your process *and* can still be gone, because Redis acknowledges
128 // before the write is durable or replicated.
129 assert!(QueueDriver::Redis.may_lose_an_accepted_job());
130
131 assert!(!QueueDriver::Database.may_lose_an_accepted_job());
132 assert!(!QueueDriver::Sqs.may_lose_an_accepted_job());
133
134 // Memory loses everything, but `survives_a_restart` already says so —
135 // this predicate is for the ones that look safe.
136 assert!(!QueueDriver::Memory.may_lose_an_accepted_job());
137
138 // Sync has already run the job by the time the process exits, which is
139 // not the same as surviving — there was nothing left to survive.
140 assert!(!QueueDriver::Sync.survives_a_restart());
141 }
142}