ruststream_fred/
message.rs

1//! Delivered-message wrapper that implements [`IncomingMessage`].
2
3use std::fmt::{Debug, Formatter};
4use std::time::Duration;
5
6use bytes::Bytes;
7use fred::clients::Pool;
8use fred::interfaces::StreamsInterface;
9use ruststream::runtime::RETRY_COUNT_HEADER;
10use ruststream::{AckError, Headers, IncomingMessage, Partitioned, Positioned};
11
12use crate::convert::fields_for_publish;
13use crate::deadletter::{self, PoisonPolicy, REASON_DROPPED, REASON_MAX_DELIVERIES};
14use crate::delay::{self, DelayConfig};
15use crate::seek::{EntryId, RedisGroupPosition};
16
17/// The well-known header key for per-message routing / partitioning.
18///
19/// Set this header on outgoing messages to control key-based fan-out when the runtime is
20/// configured with `workers(N, by_key)`. The value is opaque bytes; the runtime hashes it to
21/// assign a dispatch lane. Redis has no native partition concept on a single stream, so the key
22/// travels as this header value and the sender is responsible for setting it.
23pub const PARTITION_KEY_HEADER: &str = "redis-partition-key";
24
25/// Everything a [`RedisMessage`] needs to settle itself against the stream it came from.
26struct AckHandle {
27    pool: Pool,
28    key: String,
29    group: String,
30    id: String,
31}
32
33/// A Redis Streams delivery, read from a consumer group via `XREADGROUP` or `XAUTOCLAIM`.
34///
35/// Settlement follows the republish-retry model: `ack` is `XACK`; `nack(requeue = true)`
36/// re-appends a copy of the entry to the same stream and then acks the original (at-least-once,
37/// so a duplicate is possible if the process crashes between the two); `nack(requeue = false)`
38/// acks the original to drop it.
39pub struct RedisMessage {
40    payload: Bytes,
41    headers: Headers,
42    ack: Option<AckHandle>,
43    /// The parsed form of the entry id, kept beside the wire form so
44    /// [`Positioned::position`] cannot fail. Both are derived from the same server-issued id.
45    entry: EntryId,
46    policy: PoisonPolicy,
47    /// Set when the subscription opted into a durable ZSET delay queue; makes `nack_after` native.
48    delay: Option<DelayConfig>,
49}
50
51impl Debug for RedisMessage {
52    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
53        let mut s = f.debug_struct("RedisMessage");
54        s.field("payload_len", &self.payload.len());
55        if let Some(ack) = &self.ack {
56            s.field("key", &ack.key).field("id", &ack.id);
57        }
58        s.finish_non_exhaustive()
59    }
60}
61
62impl RedisMessage {
63    #[allow(
64        clippy::too_many_arguments,
65        reason = "internal constructor mirroring the descriptor"
66    )]
67    pub(crate) fn new(
68        pool: Pool,
69        key: String,
70        group: String,
71        id: String,
72        entry: EntryId,
73        payload: Bytes,
74        headers: Headers,
75        policy: PoisonPolicy,
76        delay: Option<DelayConfig>,
77    ) -> Self {
78        Self {
79            payload,
80            headers,
81            ack: Some(AckHandle {
82                pool,
83                key,
84                group,
85                id,
86            }),
87            entry,
88            policy,
89            delay,
90        }
91    }
92
93    /// The stream entry ID (for example `1700000000000-0`) this message was read at.
94    #[must_use]
95    pub fn id(&self) -> Option<&str> {
96        self.ack.as_ref().map(|a| a.id.as_str())
97    }
98
99    /// The parsed entry id this message was read at.
100    #[must_use]
101    pub const fn entry_id(&self) -> EntryId {
102        self.entry
103    }
104
105    /// The consumer group this delivery was read through, or `None` once the message has settled.
106    #[must_use]
107    pub fn group(&self) -> Option<&str> {
108        self.ack.as_ref().map(|a| a.group.as_str())
109    }
110}
111
112impl Partitioned for RedisMessage {
113    fn partition_key(&self) -> Option<&[u8]> {
114        self.headers().get(PARTITION_KEY_HEADER)
115    }
116}
117
118/// The position of a delivery is the group cursor that redelivers it.
119///
120/// The cursor is exclusive (a group resumes *after* the id it holds), so the pinned position is
121/// the id immediately below this entry's - seeking to it delivers this message again, followed by
122/// the entries after it. Repositioning is group-wide; see
123/// [`RedisGroupSeeker`](crate::RedisGroupSeeker).
124impl Positioned for RedisMessage {
125    type Position = RedisGroupPosition;
126
127    fn position(&self) -> RedisGroupPosition {
128        RedisGroupPosition::after(self.entry.previous())
129    }
130}
131
132impl IncomingMessage for RedisMessage {
133    fn payload(&self) -> &[u8] {
134        &self.payload
135    }
136
137    fn headers(&self) -> &Headers {
138        &self.headers
139    }
140
141    async fn ack(mut self) -> Result<(), AckError> {
142        let handle = self.ack.take().expect("RedisMessage settled twice");
143        xack(&handle).await
144    }
145
146    async fn nack(mut self, requeue: bool) -> Result<(), AckError> {
147        let handle = self.ack.take().expect("RedisMessage settled twice");
148        if requeue {
149            if self.policy.is_active() {
150                let next = next_retry_count(&self.headers);
151                if self.policy.is_poison(next) {
152                    // The framework retry-count reached the cap: dead-letter (or discard) instead
153                    // of redelivering, then ack the original.
154                    deadletter::settle_poison_stream(
155                        &handle.pool,
156                        &self.policy,
157                        &self.payload,
158                        &self.headers,
159                        REASON_MAX_DELIVERIES,
160                    )
161                    .await
162                    .map_err(broker_err)?;
163                } else {
164                    let mut headers = self.headers.clone();
165                    headers.insert(RETRY_COUNT_HEADER, next.to_string());
166                    republish(&handle, &self.payload, &headers).await?;
167                }
168            } else {
169                // No poison policy: republish verbatim, the plain at-least-once retry.
170                republish(&handle, &self.payload, &self.headers).await?;
171            }
172        } else if self.policy.is_active() {
173            // Drop: dead-letter it (or discard when no dead-letter stream is set) before acking.
174            deadletter::settle_poison_stream(
175                &handle.pool,
176                &self.policy,
177                &self.payload,
178                &self.headers,
179                REASON_DROPPED,
180            )
181            .await
182            .map_err(broker_err)?;
183        }
184        xack(&handle).await
185    }
186
187    /// Native delayed redelivery is available only when the subscription opted into a durable ZSET
188    /// delay queue with [`RedisStream::delayed_retry`](crate::RedisStream::delayed_retry); otherwise
189    /// the runtime applies its broker-agnostic deferred-republish fallback.
190    fn supports_nack_after(&self) -> bool {
191        self.delay.is_some()
192    }
193
194    /// Schedules the message for redelivery no sooner than `delay` from now via the configured ZSET
195    /// delay queue (`ZADD` the delayed copy, then `XACK` the original), with the retry-count header
196    /// incremented. The subscriber's sweeper re-`XADD`s it to the source stream once due.
197    ///
198    /// # Errors
199    ///
200    /// Returns [`AckError::Unsupported`] when the subscription did not opt into a delay queue, or
201    /// [`AckError::Broker`] when the `ZADD` or `XACK` fails.
202    async fn nack_after(mut self, delay: Duration) -> Result<(), AckError> {
203        let handle = self.ack.take().expect("RedisMessage settled twice");
204        let Some(cfg) = self.delay.as_ref() else {
205            return Err(AckError::Unsupported);
206        };
207        // ZADD the delayed copy before XACK-ing the original, so a crash in between leaves a
208        // duplicate (the scheduled copy plus the still-pending original) rather than a loss.
209        delay::schedule(
210            &handle.pool,
211            cfg,
212            &handle.id,
213            &self.payload,
214            &self.headers,
215            delay,
216        )
217        .await?;
218        xack(&handle).await
219    }
220}
221
222/// The next framework retry-count value (the current header plus one, or one when absent).
223fn next_retry_count(headers: &Headers) -> u64 {
224    headers
225        .get_str(RETRY_COUNT_HEADER)
226        .and_then(|v| v.parse::<u64>().ok())
227        .unwrap_or(0)
228        + 1
229}
230
231fn broker_err(err: fred::error::Error) -> AckError {
232    AckError::Broker(Box::new(err))
233}
234
235/// Re-appends a copy of the message to the tail of its stream (the at-least-once retry). Runs before
236/// the caller's `XACK` so a crash leaves a duplicate rather than a loss.
237async fn republish(handle: &AckHandle, payload: &[u8], headers: &Headers) -> Result<(), AckError> {
238    let fields = fields_for_publish(payload, headers);
239    let _: String = handle
240        .pool
241        .xadd(handle.key.as_str(), false, None::<()>, "*", fields)
242        .await
243        .map_err(broker_err)?;
244    Ok(())
245}
246
247async fn xack(handle: &AckHandle) -> Result<(), AckError> {
248    let _: i64 = handle
249        .pool
250        .xack(
251            handle.key.as_str(),
252            handle.group.as_str(),
253            handle.id.as_str(),
254        )
255        .await
256        .map_err(broker_err)?;
257    Ok(())
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263    use crate::context::StreamContext;
264    use fred::clients::Pool;
265    use fred::types::config::Config;
266    use ruststream::BuildContext;
267
268    /// An unconnected pool (just client structs); `Pool::new` opens no sockets.
269    fn offline_pool() -> Pool {
270        Pool::new(Config::default(), None, None, None, 1).expect("offline pool")
271    }
272
273    fn delivery(id: &str) -> RedisMessage {
274        RedisMessage::new(
275            offline_pool(),
276            "orders".to_owned(),
277            "workers".to_owned(),
278            id.to_owned(),
279            id.parse().expect("valid entry id"),
280            Bytes::from_static(b"{}"),
281            Headers::new(),
282            PoisonPolicy::default(),
283            None,
284        )
285    }
286
287    #[test]
288    fn build_context_reads_entry_id_and_group() {
289        let cx = StreamContext::build(&delivery("1700000000000-0"));
290        assert_eq!(cx.entry_id(), Some("1700000000000-0"));
291        assert_eq!(cx.consumer_group(), Some("workers"));
292    }
293
294    // The group cursor is exclusive, so the position that redelivers an entry sits one id below
295    // it: pinning `<ms>-0` has to borrow from the millisecond half.
296    #[test]
297    fn position_pins_the_delivery_for_redelivery() {
298        assert_eq!(
299            delivery("1700000000000-4").position(),
300            RedisGroupPosition::after(EntryId::new(1_700_000_000_000, 3))
301        );
302        assert_eq!(
303            delivery("1700000000000-0").position(),
304            RedisGroupPosition::after(EntryId::new(1_699_999_999_999, u64::MAX))
305        );
306    }
307}
ruststream_fred/message.rs

ruststream_fred/
message.rs
