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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
//! 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",
/// A Kafka topic.
///
/// **Read [the driver's docs](crate::kafka) before choosing it.** A log
/// is not a queue: concurrency is the partition count rather than the
/// worker count, a delayed job blocks its partition, and a retry goes
/// to the end of the topic instead of back where it was. Right when the
/// jobs are already events on a topic; wrong as a general job queue.
/// Needs the `kafka` feature.
Kafka = "kafka",
}
}
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 | Self::Kafka)
}
/// 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.
///
/// Kafka is produced to with `acks=all`, so a push returns once every
/// in-sync replica holds the record — which is the same guarantee, *and is
/// only as good as the replication*. A single-broker cluster has one
/// in-sync replica and Kafka does not fsync each write, so a development
/// cluster can lose an accepted job even though this reports it cannot.
/// `replication.factor` and `min.insync.replicas` of at least two are what
/// make the answer true.
///
/// 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::Kafka.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());
// `acks=all`, so the push returns once the replicas have it — which is
// a claim about a replicated cluster, as the doc comment says.
assert!(!QueueDriver::Kafka.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());
}
}