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
140
141
142
//! Where queued jobs wait — [`QueueDriver`].
use rainier_support::setting_enum;
setting_enum! {
/// Which [`Queue`](crate::Queue) to build.
///
/// ```
/// use rainier_queue::QueueDriver;
/// use rainier_support::Setting;
///
/// assert!(!QueueDriver::Sync.is_deferred(), "sync runs the job inline");
/// assert!(QueueDriver::parse("database").unwrap().survives_a_restart());
/// ```
pub enum QueueDriver: "queue driver" {
/// Run the job inline, on the thread that dispatched it.
///
/// The default because it needs no worker: a fresh clone dispatches a
/// job and the job happens. Also the reason a slow job makes a slow
/// request — which is the thing a queue exists to avoid, so this is a
/// development setting.
#[default]
Sync = "sync",
/// A queue in this process's memory.
///
/// For tests that want a job to be *queued* rather than *run*, so they
/// can assert on it. Lost on restart, invisible to other processes.
Memory = "memory",
/// Two tables in the application's own database.
///
/// Durable, shared, and needs no new infrastructure — the reason it is
/// the usual production answer. Needs the migrations
/// (`DatabaseQueue::migrations()`) and a `queue:work` process.
Database = "database",
/// Redis streams.
///
/// Durable-ish, and the qualifier is the point: Redis acknowledges a
/// write before it is on disk and before a replica has it, so a job
/// can be lost to a crash or a failover. Right for work you can afford
/// to lose, wrong for work you cannot — see the
/// [driver's docs](crate::redis). Needs the `redis` feature.
Redis = "redis",
/// Amazon SQS.
///
/// Durable and managed, with the visibility timeout doing the
/// reservation. Needs the `sqs` feature.
Sqs = "sqs",
}
}
impl QueueDriver {
/// Whether dispatching returns before the job has run.
///
/// `false` for [`Sync`](Self::Sync), which is the whole difference between
/// it and the others — and the reason a test that asserts on a side effect
/// passes under `sync` and hangs under everything else.
pub fn is_deferred(&self) -> bool {
!matches!(self, Self::Sync)
}
/// Whether a job dispatched now survives **this** process exiting.
///
/// True for anything that stores the job somewhere else. It is not the
/// same question as whether the job is safe — see
/// [`may_lose_an_accepted_job`](Self::may_lose_an_accepted_job), which is
/// the one that catches Redis.
pub fn survives_a_restart(&self) -> bool {
matches!(self, Self::Database | Self::Redis | Self::Sqs)
}
/// Whether a dispatch this driver **accepted** can still be lost.
///
/// True only for [`Redis`](Self::Redis), and it is the distinction the
/// obvious question misses: a Redis job survives your process restarting,
/// because it lives in another one — and can still vanish, because Redis
/// acknowledges a write before it is on disk (`appendfsync everysec`
/// leaves a window of about a second) and before any replica has it (
/// replication is asynchronous, so a failover can drop confirmed writes).
///
/// A database queue's insert is committed before it returns; SQS is
/// replicated across availability zones before it acknowledges. Neither
/// can tell you a job is queued and then lose it.
///
/// So this is the predicate a deploy check should look at when the work is
/// a payment rather than a cache warm.
pub fn may_lose_an_accepted_job(&self) -> bool {
matches!(self, Self::Redis)
}
/// Whether this driver needs a `queue:work` process to make progress.
pub fn needs_a_worker(&self) -> bool {
self.is_deferred()
}
}
#[cfg(test)]
mod tests {
use super::*;
use rainier_support::Setting;
#[test]
fn sync_is_the_only_driver_that_runs_inline() {
assert!(!QueueDriver::Sync.is_deferred());
assert!(!QueueDriver::Sync.needs_a_worker());
for driver in QueueDriver::ALL.iter().filter(|d| **d != QueueDriver::Sync) {
assert!(driver.is_deferred(), "{driver} should defer");
assert!(driver.needs_a_worker(), "{driver} should need a worker");
}
}
#[test]
fn only_the_backed_drivers_survive_a_restart() {
assert!(QueueDriver::Database.survives_a_restart());
assert!(QueueDriver::Sqs.survives_a_restart());
assert!(QueueDriver::Redis.survives_a_restart(), "it lives in another process");
assert!(!QueueDriver::Memory.survives_a_restart());
}
#[test]
fn only_redis_can_lose_a_job_it_already_accepted() {
// The distinction `survives_a_restart` cannot make: the job outlives
// your process *and* can still be gone, because Redis acknowledges
// before the write is durable or replicated.
assert!(QueueDriver::Redis.may_lose_an_accepted_job());
assert!(!QueueDriver::Database.may_lose_an_accepted_job());
assert!(!QueueDriver::Sqs.may_lose_an_accepted_job());
// Memory loses everything, but `survives_a_restart` already says so —
// this predicate is for the ones that look safe.
assert!(!QueueDriver::Memory.may_lose_an_accepted_job());
// Sync has already run the job by the time the process exits, which is
// not the same as surviving — there was nothing left to survive.
assert!(!QueueDriver::Sync.survives_a_restart());
}
}