queuey_rabbitmq/
codec.rs

1//! Pure mapping between [`Envelope`] and AMQP [`BasicProperties`] / headers.
2//!
3//! Nothing here touches the broker, so every function can be unit-tested
4//! without a running RabbitMQ.
5
6use lapin::{
7    BasicProperties,
8    types::{AMQPValue, FieldTable, LongString, MAX_SHORT_STRING_LENGTH, ShortString},
9};
10use queuey_core::Envelope;
11
12use crate::topology::{
13    HEADER_ATTEMPT, HEADER_ATTEMPTS, HEADER_DEATH_REASON, HEADER_DEFERRALS, HEADER_ORIGINAL_QUEUE,
14};
15
16/// `content-type` set on every published message.
17pub const CONTENT_TYPE_JSON: &str = "application/json";
18
19/// `delivery-mode` for a persistent message.
20pub const DELIVERY_MODE_PERSISTENT: u8 = 2;
21
22/// Reason recorded on bodies that could not be decoded as an [`Envelope`].
23pub const REASON_MALFORMED: &str = "malformed envelope";
24
25/// Headers carried by every published envelope.
26///
27/// `x-attempt` and `x-deferrals` mirror [`Envelope::attempt`] and
28/// [`Envelope::deferrals`], so an operator can read both counters in the
29/// management UI without decoding the body. The body stays the source of truth:
30/// nothing in this crate reads these back.
31#[must_use]
32pub fn base_headers(envelope: &Envelope) -> FieldTable {
33    let mut headers = FieldTable::default();
34    headers.insert(HEADER_ATTEMPT.into(), AMQPValue::LongUInt(envelope.attempt));
35    headers.insert(
36        HEADER_DEFERRALS.into(),
37        AMQPValue::LongUInt(envelope.deferrals),
38    );
39    headers
40}
41
42/// AMQP properties for publishing `envelope`.
43///
44/// * `content-type` is `application/json`, matching [`Envelope::to_bytes`].
45/// * `delivery-mode` is `2` (persistent).
46/// * `message-id` is the job id, stable across retries.
47/// * `type` is the job type.
48/// * `priority` is [`Envelope::priority`], always set. Normal work carries `0`;
49///   a deferred envelope carries its queue's top level so it overtakes the
50///   backlog. A queue declared without `x-max-priority` ignores the property,
51///   and a priority above the queue's `x-max-priority` is treated by the broker
52///   as that maximum, so this is safe to set unconditionally.
53/// * `expiration` is **never** set, not even for a message bound for a hold
54///   queue. The wait is the hold queue's queue-wide `x-message-ttl`; a
55///   per-message expiration on top of it would reintroduce exactly the
56///   mixed-TTL head-of-line blocking that hold queues exist to avoid, and a
57///   shorter one would release the job early.
58#[must_use]
59pub fn props_for(envelope: &Envelope) -> BasicProperties {
60    BasicProperties::default()
61        .with_content_type(CONTENT_TYPE_JSON.into())
62        .with_delivery_mode(DELIVERY_MODE_PERSISTENT)
63        .with_message_id(clamped(&envelope.job_id.to_string()))
64        .with_type(clamped(&envelope.job_type))
65        .with_priority(envelope.priority)
66        .with_headers(base_headers(envelope))
67}
68
69/// Headers recorded on a message routed to `q.dead`.
70///
71/// Extends [`base_headers`] with `x-death-reason`, `x-original-queue` and
72/// `x-attempts`.
73#[must_use]
74pub fn dead_letter_headers(envelope: &Envelope, reason: &str) -> FieldTable {
75    let mut headers = base_headers(envelope);
76    headers.insert(
77        HEADER_DEATH_REASON.into(),
78        AMQPValue::LongString(LongString::from(reason)),
79    );
80    headers.insert(
81        HEADER_ORIGINAL_QUEUE.into(),
82        AMQPValue::LongString(LongString::from(envelope.queue.as_str())),
83    );
84    headers.insert(
85        HEADER_ATTEMPTS.into(),
86        AMQPValue::LongUInt(envelope.attempt),
87    );
88    headers
89}
90
91/// AMQP properties for publishing `envelope` to its dead-letter queue.
92#[must_use]
93pub fn dead_letter_props(envelope: &Envelope, reason: &str) -> BasicProperties {
94    props_for(envelope).with_headers(dead_letter_headers(envelope, reason))
95}
96
97/// AMQP properties for a body that could not be decoded as an [`Envelope`].
98///
99/// The original bytes are forwarded verbatim, so there is no attempt counter and
100/// no job metadata to carry, only where it came from and why it was rejected.
101#[must_use]
102pub fn malformed_props(original_queue: &str, reason: &str) -> BasicProperties {
103    let mut headers = FieldTable::default();
104    headers.insert(
105        HEADER_DEATH_REASON.into(),
106        AMQPValue::LongString(LongString::from(reason)),
107    );
108    headers.insert(
109        HEADER_ORIGINAL_QUEUE.into(),
110        AMQPValue::LongString(LongString::from(original_queue)),
111    );
112    BasicProperties::default()
113        .with_delivery_mode(DELIVERY_MODE_PERSISTENT)
114        .with_headers(headers)
115}
116
117/// Convert to a [`ShortString`], truncating at a UTF-8 boundary if needed.
118///
119/// AMQP short strings are capped at 255 bytes and `ShortString::from` panics
120/// past that. Library code must never panic on user-supplied job types, so an
121/// over-long value is truncated rather than rejected.
122fn clamped(value: &str) -> ShortString {
123    ShortString::from(truncate_at_boundary(value, MAX_SHORT_STRING_LENGTH))
124}
125
126/// The longest prefix of `value` that is at most `max` bytes and still valid
127/// UTF-8 (i.e. it never splits a multi-byte character).
128pub(crate) fn truncate_at_boundary(value: &str, max: usize) -> &str {
129    if value.len() <= max {
130        return value;
131    }
132    let mut end = max;
133    while end > 0 && !value.is_char_boundary(end) {
134        end -= 1;
135    }
136    &value[..end]
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142
143    use queuey_core::Envelope;
144    use serde_json::json;
145
146    fn envelope() -> Envelope {
147        Envelope {
148            job_id: "67e55044-10b1-426f-9247-bb680e5fe0c8".parse().unwrap(),
149            job_type: "myapp::jobs::SendEmail".to_owned(),
150            queue: "myapp.emails".to_owned(),
151            attempt: 3,
152            enqueued_at_ms: 1_700_000_000_000,
153            deferrals: 0,
154            priority: 0,
155            payload: json!({ "to": "a@b.c" }),
156        }
157    }
158
159    /// The same envelope after two deferrals onto a 10-level priority queue.
160    fn deferred_envelope() -> Envelope {
161        envelope().deferred(10).deferred(10)
162    }
163
164    #[test]
165    fn props_carry_content_type_and_persistence() {
166        let props = props_for(&envelope());
167        assert_eq!(
168            props.content_type().as_ref().map(ShortString::to_string),
169            Some("application/json".to_owned())
170        );
171        assert_eq!(*props.delivery_mode(), Some(2));
172    }
173
174    #[test]
175    fn props_carry_message_id_and_type() {
176        let props = props_for(&envelope());
177        assert_eq!(
178            props.message_id().as_ref().map(ShortString::to_string),
179            Some("67e55044-10b1-426f-9247-bb680e5fe0c8".to_owned())
180        );
181        assert_eq!(
182            props.kind().as_ref().map(ShortString::to_string),
183            Some("myapp::jobs::SendEmail".to_owned())
184        );
185    }
186
187    #[test]
188    fn props_carry_the_attempt_and_deferrals_headers() {
189        let props = props_for(&envelope());
190        let headers = props.headers().as_ref().expect("headers");
191        assert_eq!(
192            headers.inner().get(HEADER_ATTEMPT),
193            Some(&AMQPValue::LongUInt(3))
194        );
195        assert_eq!(
196            headers.inner().get(HEADER_DEFERRALS),
197            Some(&AMQPValue::LongUInt(0))
198        );
199        assert_eq!(headers.inner().len(), 2);
200    }
201
202    #[test]
203    fn the_deferrals_header_tracks_the_envelope() {
204        let props = props_for(&deferred_envelope());
205        let headers = props.headers().as_ref().expect("headers");
206        assert_eq!(
207            headers.inner().get(HEADER_DEFERRALS),
208            Some(&AMQPValue::LongUInt(2))
209        );
210        // A deferral is not an attempt.
211        assert_eq!(
212            headers.inner().get(HEADER_ATTEMPT),
213            Some(&AMQPValue::LongUInt(3))
214        );
215    }
216
217    #[test]
218    fn props_carry_the_envelope_priority() {
219        // Normal work is priority 0, and the property is always set so a queue
220        // with `x-max-priority` orders every message the same way.
221        assert_eq!(*props_for(&envelope()).priority(), Some(0));
222        assert_eq!(*props_for(&deferred_envelope()).priority(), Some(10));
223        assert_eq!(
224            *dead_letter_props(&deferred_envelope(), "boom").priority(),
225            Some(10)
226        );
227    }
228
229    #[test]
230    fn props_never_carry_an_expiration_because_the_hold_queue_times_the_wait() {
231        // A per-message expiration would fight the hold queue's x-message-ttl
232        // and bring back head-of-line blocking between different delays.
233        assert!(props_for(&envelope()).expiration().is_none());
234        assert!(props_for(&deferred_envelope()).expiration().is_none());
235    }
236
237    #[test]
238    fn dead_letter_headers_record_reason_queue_and_attempts() {
239        let headers = dead_letter_headers(&envelope(), "handler returned Fatal");
240        assert_eq!(
241            headers.inner().get(HEADER_DEATH_REASON),
242            Some(&AMQPValue::LongString(LongString::from(
243                "handler returned Fatal"
244            )))
245        );
246        assert_eq!(
247            headers.inner().get(HEADER_ORIGINAL_QUEUE),
248            Some(&AMQPValue::LongString(LongString::from("myapp.emails")))
249        );
250        assert_eq!(
251            headers.inner().get(HEADER_ATTEMPTS),
252            Some(&AMQPValue::LongUInt(3))
253        );
254        assert_eq!(
255            headers.inner().get(HEADER_ATTEMPT),
256            Some(&AMQPValue::LongUInt(3))
257        );
258    }
259
260    #[test]
261    fn dead_letter_props_keep_identity_and_drop_expiration() {
262        let props = dead_letter_props(&envelope(), "boom");
263        assert_eq!(
264            props.message_id().as_ref().map(ShortString::to_string),
265            Some("67e55044-10b1-426f-9247-bb680e5fe0c8".to_owned())
266        );
267        assert!(props.expiration().is_none());
268        let headers = props.headers().as_ref().expect("headers");
269        assert!(headers.contains_key(HEADER_DEATH_REASON));
270    }
271
272    #[test]
273    fn malformed_props_record_origin_and_reason_only() {
274        let props = malformed_props("myapp.emails", REASON_MALFORMED);
275        assert_eq!(*props.delivery_mode(), Some(2));
276        assert!(props.message_id().is_none());
277        let headers = props.headers().as_ref().expect("headers");
278        assert_eq!(
279            headers.inner().get(HEADER_DEATH_REASON),
280            Some(&AMQPValue::LongString(LongString::from(
281                "malformed envelope"
282            )))
283        );
284        assert_eq!(
285            headers.inner().get(HEADER_ORIGINAL_QUEUE),
286            Some(&AMQPValue::LongString(LongString::from("myapp.emails")))
287        );
288        assert!(!headers.contains_key(HEADER_ATTEMPT));
289    }
290
291    #[test]
292    fn over_long_job_type_is_truncated_not_panicked() {
293        let mut env = envelope();
294        env.job_type = "é".repeat(400);
295        let props = props_for(&env);
296        let kind = props.kind().as_ref().expect("type").to_string();
297        assert!(
298            kind.len() <= MAX_SHORT_STRING_LENGTH,
299            "len was {}",
300            kind.len()
301        );
302        // 'é' is two bytes, so truncation must land on an even byte offset.
303        assert_eq!(kind.len(), 254);
304        assert!(kind.chars().all(|c| c == 'é'));
305    }
306
307    #[test]
308    fn truncation_never_splits_a_character() {
309        assert_eq!(truncate_at_boundary("héllo", 2), "h");
310        assert_eq!(truncate_at_boundary("héllo", 3), "hé");
311        assert_eq!(truncate_at_boundary("héllo", 99), "héllo");
312        assert_eq!(truncate_at_boundary("é", 1), "");
313    }
314
315    #[test]
316    fn exactly_max_length_is_kept_verbatim() {
317        let mut env = envelope();
318        env.job_type = "a".repeat(MAX_SHORT_STRING_LENGTH);
319        let props = props_for(&env);
320        assert_eq!(
321            props.kind().as_ref().map(ShortString::to_string),
322            Some(env.job_type)
323        );
324    }
325}
queuey_rabbitmq/codec.rs

queuey_rabbitmq/
codec.rs
