queuey_rabbitmq/options.rs
1//! Tunables for [`RabbitMqBackend`](crate::RabbitMqBackend).
2
3use std::time::Duration;
4
5use lapin::ConnectionProperties;
6
7use crate::topology::{DEFAULT_DEAD_SUFFIX, DEFAULT_DEFERRED_SUFFIX};
8
9/// Default for [`RabbitMqOptions::retry_granularity`] and
10/// [`RabbitMqOptions::deferred_granularity`].
11const DEFAULT_GRANULARITY: Duration = Duration::from_secs(1);
12
13/// Configuration for [`RabbitMqBackend::with_options`](crate::RabbitMqBackend::with_options).
14///
15/// ```
16/// use queuey_rabbitmq::RabbitMqOptions;
17///
18/// let options = RabbitMqOptions::default()
19/// .dead_suffix("-dlq")
20/// .declare_dead_letter_queues(false);
21/// assert_eq!(options.dead_suffix, "-dlq");
22/// ```
23#[derive(Clone, Debug)]
24pub struct RabbitMqOptions {
25 /// Handshake properties passed to `lapin::Connection::connect`.
26 pub connection_properties: ConnectionProperties,
27
28 /// Suffix appended to a queue name to name its dead-letter queue.
29 ///
30 /// Defaults to [`DEFAULT_DEAD_SUFFIX`] (`".dead"`).
31 pub dead_suffix: String,
32
33 /// Whether [`declare`](queuey_core::Backend::declare) also declares
34 /// the `q.dead` queues.
35 ///
36 /// Defaults to `true`. Set to `false` when dead-letter queues are managed
37 /// out of band (policies, an operator-owned topology, a different broker
38 /// vhost).
39 ///
40 /// This flag also selects *how* a message is dead-lettered, because this
41 /// backend never publishes to a queue it does not own:
42 ///
43 /// * `true`: [`Delivery::dead_letter`](queuey_core::Delivery::dead_letter)
44 /// publishes the envelope to `q.dead` with the `x-death-*` headers and
45 /// then acks the original.
46 /// * `false`: nothing is published. The original is rejected with
47 /// `requeue = false`, so the broker applies whatever
48 /// `x-dead-letter-exchange` policy the operator put on `q`, and drops the
49 /// message if there is none. The reason is logged at `WARN`, since it is
50 /// not recorded anywhere else.
51 ///
52 /// The same choice governs a message whose body is not a valid envelope.
53 pub declare_dead_letter_queues: bool,
54
55 /// Infix between a queue name and a hold queue's TTL.
56 ///
57 /// Defaults to [`DEFAULT_DEFERRED_SUFFIX`] (`".deferred"`), so a 30-second
58 /// wait on `myapp.emails` happens in `myapp.emails.deferred.30000`. Retries,
59 /// delayed enqueues and deferrals all wait in these hold queues; see
60 /// [`crate::topology`] for why the TTL is part of the name.
61 pub deferred_suffix: String,
62
63 /// Step that retry backoffs and
64 /// [`Producer::enqueue_after`](queuey_core::Producer::enqueue_after) delays
65 /// are rounded **up** to.
66 ///
67 /// Defaults to one second. Every distinct rounded delay gets its own hold
68 /// queue, so this is the knob that trades backoff precision for the number
69 /// of queues on the broker. It matters most for exponential backoff with
70 /// jitter, which produces a different delay for every retry: with the
71 /// default, a policy capped at five minutes can create at most 300 hold
72 /// queues per work queue, and a granularity of ten seconds brings that down
73 /// to 30. Idle hold queues delete themselves, so this bounds the number that
74 /// exist at once, not a total.
75 ///
76 /// Separate from [`deferred_granularity`](Self::deferred_granularity) on
77 /// purpose: a backoff is a heuristic that tolerates coarse rounding, a
78 /// `Retry-After` is a contract that may not.
79 ///
80 /// A retry is never released *early*: rounding is always up, a delay
81 /// shorter than the granularity still waits one full step, and a delay that
82 /// rounds up past
83 /// [`MAX_DEFERRAL_MS`](crate::topology::MAX_DEFERRAL_MS) (~24.8 days) is
84 /// refused instead of being shortened. A zero (or sub-millisecond) value is
85 /// clamped to one millisecond rather than rejected, exactly as for
86 /// [`deferred_granularity`](Self::deferred_granularity).
87 pub retry_granularity: Duration,
88
89 /// Step that deferral delays are rounded **up** to.
90 ///
91 /// Defaults to one second. Every distinct rounded delay gets its own hold
92 /// queue, so this is the knob that trades precision for the number of queues
93 /// on the broker: with the default, `Retry-After: 30` and a computed `29.2s`
94 /// delay share `q.deferred.30000`, and no deferral can create more than
95 /// `MAX_TTL_MS / 1000` queues per work queue.
96 ///
97 /// A deferral is never released *early*: rounding is always up, a delay
98 /// shorter than the granularity still waits one full step, and a delay that
99 /// rounds up past
100 /// [`MAX_DEFERRAL_MS`](crate::topology::MAX_DEFERRAL_MS) (~24.8 days) is
101 /// refused instead of being shortened.
102 ///
103 /// A zero (or sub-millisecond) value is clamped to one millisecond by
104 /// [`deferred_ttl_ms`](crate::topology::deferred_ttl_ms) rather than
105 /// rejected, because a backend constructor must not panic on a config value, but
106 /// one millisecond of granularity means up to one hold queue per distinct
107 /// millisecond, which is almost never what you want.
108 ///
109 /// Note what is *not* here: nothing tunes a hold queue's `x-expires`. Its
110 /// arguments are a pure function of its name (`x-expires = 2 * ttl`), so two
111 /// processes configured differently still agree on `q.deferred.30000`
112 /// instead of locking each other out with `PRECONDITION_FAILED`. Both
113 /// granularities are safe to tune because they only change *which* hold
114 /// queue a delay lands in, never that queue's arguments.
115 pub deferred_granularity: Duration,
116}
117
118impl Default for RabbitMqOptions {
119 fn default() -> Self {
120 Self {
121 connection_properties: ConnectionProperties::default(),
122 dead_suffix: DEFAULT_DEAD_SUFFIX.to_owned(),
123 declare_dead_letter_queues: true,
124 deferred_suffix: DEFAULT_DEFERRED_SUFFIX.to_owned(),
125 retry_granularity: DEFAULT_GRANULARITY,
126 deferred_granularity: DEFAULT_GRANULARITY,
127 }
128 }
129}
130
131impl RabbitMqOptions {
132 /// Replace the connection handshake properties.
133 #[must_use]
134 pub fn connection_properties(mut self, properties: ConnectionProperties) -> Self {
135 self.connection_properties = properties;
136 self
137 }
138
139 /// Replace the dead-letter queue suffix.
140 #[must_use]
141 pub fn dead_suffix(mut self, suffix: impl Into<String>) -> Self {
142 self.dead_suffix = suffix.into();
143 self
144 }
145
146 /// Enable or disable declaring `q.dead` queues.
147 #[must_use]
148 pub fn declare_dead_letter_queues(mut self, declare: bool) -> Self {
149 self.declare_dead_letter_queues = declare;
150 self
151 }
152
153 /// Replace the hold queue infix.
154 #[must_use]
155 pub fn deferred_suffix(mut self, suffix: impl Into<String>) -> Self {
156 self.deferred_suffix = suffix.into();
157 self
158 }
159
160 /// Replace the step retry backoffs and delayed enqueues are rounded up to.
161 ///
162 /// A zero or sub-millisecond value is *clamped* to one millisecond when the
163 /// TTL is computed, not rejected here: this is a builder, and library code
164 /// does not panic on configuration.
165 #[must_use]
166 pub fn retry_granularity(mut self, granularity: Duration) -> Self {
167 self.retry_granularity = granularity;
168 self
169 }
170
171 /// Replace the step deferral delays are rounded up to.
172 ///
173 /// A zero or sub-millisecond value is *clamped* to one millisecond when the
174 /// TTL is computed, not rejected here: this is a builder, and library code
175 /// does not panic on configuration.
176 #[must_use]
177 pub fn deferred_granularity(mut self, granularity: Duration) -> Self {
178 self.deferred_granularity = granularity;
179 self
180 }
181}
182
183#[cfg(test)]
184mod tests {
185 use super::*;
186
187 #[test]
188 fn defaults_match_the_documented_topology() {
189 let options = RabbitMqOptions::default();
190 assert_eq!(options.dead_suffix, ".dead");
191 assert!(options.declare_dead_letter_queues);
192 assert_eq!(options.deferred_suffix, ".deferred");
193 assert_eq!(options.retry_granularity, Duration::from_secs(1));
194 assert_eq!(options.deferred_granularity, Duration::from_secs(1));
195 }
196
197 #[test]
198 fn deferral_tunables_can_be_overridden() {
199 let options = RabbitMqOptions::default()
200 .deferred_suffix("-hold")
201 .deferred_granularity(Duration::from_millis(250));
202 assert_eq!(options.deferred_suffix, "-hold");
203 assert_eq!(options.deferred_granularity, Duration::from_millis(250));
204 // And they are independent of the retry / dead-letter tunables.
205 assert_eq!(options.retry_granularity, Duration::from_secs(1));
206 assert_eq!(options.dead_suffix, ".dead");
207 }
208
209 #[test]
210 fn retry_granularity_is_independent_of_the_deferral_granularity() {
211 // Coarsening backoff rounding must not touch `Retry-After` precision.
212 let options = RabbitMqOptions::default().retry_granularity(Duration::from_secs(10));
213 assert_eq!(options.retry_granularity, Duration::from_secs(10));
214 assert_eq!(options.deferred_granularity, Duration::from_secs(1));
215 }
216
217 #[test]
218 fn a_zero_granularity_is_accepted_and_clamped_later_not_panicked_on() {
219 let options = RabbitMqOptions::default().deferred_granularity(Duration::ZERO);
220 assert_eq!(options.deferred_granularity, Duration::ZERO);
221 // The clamp lives in `deferred_ttl_ms`, so nothing here can panic.
222 assert_eq!(
223 crate::topology::deferred_ttl_ms(
224 Duration::from_millis(7),
225 options.deferred_granularity
226 ),
227 Some(7)
228 );
229 }
230
231 #[test]
232 fn suffixes_can_be_overridden() {
233 let options = RabbitMqOptions::default()
234 .dead_suffix("-dlq")
235 .deferred_suffix("-hold");
236 assert_eq!(options.dead_suffix, "-dlq");
237 assert_eq!(options.deferred_suffix, "-hold");
238 assert!(options.declare_dead_letter_queues);
239 }
240
241 #[test]
242 fn dead_letter_declaration_can_be_disabled() {
243 let options = RabbitMqOptions::default().declare_dead_letter_queues(false);
244 assert!(!options.declare_dead_letter_queues);
245 // Turning declaration off must not change the names.
246 assert_eq!(options.dead_suffix, ".dead");
247 }
248
249 #[test]
250 fn connection_properties_can_be_replaced() {
251 let options = RabbitMqOptions::default()
252 .connection_properties(ConnectionProperties::default().with_locale("nl_NL".into()));
253 assert!(format!("{:?}", options.connection_properties).contains("nl_NL"));
254 }
255}