ruststream_fred/
list.rs

1//! Redis list transport: a competing-consumers work queue.
2//!
3//! A producer `LPUSH`es onto the list; consumers pop from the right (`BRPOP`), so delivery is FIFO
4//! and each entry goes to exactly one consumer (no fan-out, no replay, no groups). Two modes:
5//!
6//! * Simple (default) - `BRPOP`, at-most-once. `ack` / `nack` report [`AckError::Unsupported`]: once
7//!   popped, the entry is gone, so a crash mid-handler loses it.
8//! * Reliable ([`RedisList::reliable`]) - `LMOVE` the entry to a per-consumer processing list, then
9//!   `LREM` it on `ack` (at-least-once). `nack(requeue = true)` returns it to the main list;
10//!   `nack(requeue = false)` removes it.
11//!
12//! Reliable mode has no native idle/pending tracking, so a consumer that dies after `LMOVE` but
13//! before settling leaves its entry stranded on the processing list. Opting into a recovery ZSET
14//! with [`RedisList::recovery_zset`] (and [`RedisList::min_idle`]) starts a watchdog that returns
15//! such orphans to the main list; without it (the default) reliable lists have no orphan recovery,
16//! and Redis Streams ([`crate::RedisStream`]) remain the recommended durable path. See
17//! [`crate::recovery`].
18//!
19//! Headers travel in a frame around the payload (see [`crate::envelope`]): a lossless binary frame
20//! by default, or a readable codec-serialized envelope when a codec is set with
21//! [`RedisList::codec`] / [`RedisListPublish::codec`].
22
23use std::fmt::{Debug, Formatter};
24use std::sync::Arc;
25use std::time::Duration;
26
27use bytes::Bytes;
28use fred::clients::Pool;
29use fred::error::ErrorKind;
30use fred::interfaces::{KeysInterface, ListInterface};
31use fred::types::lists::LMoveDirection;
32use futures::Stream;
33use futures::stream::unfold;
34use ruststream::codec::Codec;
35use ruststream::runtime::RETRY_COUNT_HEADER;
36use ruststream::{
37    AckError, Headers, IncomingMessage, PairError, Partitioned, PublishPolicy, SubscriptionSource,
38};
39
40use crate::broker::{ConnectedRedisBroker, RedisCore};
41use crate::deadletter::{self, PoisonPolicy, REASON_DROPPED, REASON_MAX_DELIVERIES};
42use crate::envelope::{SharedEnvelope, frame, unframe};
43use crate::recovery::{self, RecoveryConfig};
44use crate::{error::RedisError, message::PARTITION_KEY_HEADER};
45
46const DEFAULT_BLOCK: Duration = Duration::from_secs(5);
47/// Suffix appended to the list key to form the default per-consumer processing list (reliable mode).
48const PROCESSING_SUFFIX: &str = ".processing";
49
50fn block_secs(block: Duration) -> f64 {
51    block.as_secs_f64()
52}
53
54/// Normalizes a blocking pop (`BRPOP` / `BLMOVE`) result: fred reports a timed-out pop with nothing
55/// available as a timeout error rather than an empty reply, so treat that as "no entry this round"
56/// and let the read loop retry. Any other error propagates.
57fn empty_on_timeout<T>(
58    result: Result<Option<T>, fred::error::Error>,
59) -> Result<Option<T>, RedisError> {
60    match result {
61        Ok(value) => Ok(value),
62        Err(err) if matches!(err.kind(), ErrorKind::Timeout) => Ok(None),
63        Err(err) => Err(RedisError::stream(err)),
64    }
65}
66
67/// Describes one list subscription against a [`ConnectedRedisBroker`].
68///
69/// # Examples
70///
71/// ```
72/// use std::time::Duration;
73/// use ruststream_fred::RedisList;
74///
75/// let simple = RedisList::new("jobs");
76/// let reliable = RedisList::new("jobs").reliable().block(Duration::from_secs(2));
77/// # let _ = (simple, reliable);
78/// ```
79#[derive(Clone)]
80#[must_use]
81pub struct RedisList {
82    key: String,
83    reliable: bool,
84    processing: Option<String>,
85    block: Option<Duration>,
86    codec: Option<SharedEnvelope>,
87    dead_letter: Option<String>,
88    max_deliveries: Option<u64>,
89    min_idle: Option<Duration>,
90    recovery_zset: Option<String>,
91    recovery_ttl: Option<Duration>,
92}
93
94impl Debug for RedisList {
95    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
96        f.debug_struct("RedisList")
97            .field("key", &self.key)
98            .field("reliable", &self.reliable)
99            .field("processing", &self.processing)
100            .field("codec", &self.codec.is_some())
101            .field("dead_letter", &self.dead_letter)
102            .field("max_deliveries", &self.max_deliveries)
103            .field("recovery_zset", &self.recovery_zset)
104            .field("recovery_ttl", &self.recovery_ttl)
105            .finish_non_exhaustive()
106    }
107}
108
109impl RedisList {
110    /// A simple (at-most-once) `BRPOP` work-queue consumer on `key`.
111    pub fn new(key: impl Into<String>) -> Self {
112        Self {
113            key: key.into(),
114            reliable: false,
115            processing: None,
116            block: None,
117            codec: None,
118            dead_letter: None,
119            max_deliveries: None,
120            min_idle: None,
121            recovery_zset: None,
122            recovery_ttl: None,
123        }
124    }
125
126    /// Switches to reliable (at-least-once) mode: entries move to a processing list and are removed
127    /// on `ack`.
128    pub const fn reliable(mut self) -> Self {
129        self.reliable = true;
130        self
131    }
132
133    /// Sets the processing-list key used in reliable mode. Defaults to `<key>.processing`.
134    pub fn processing(mut self, key: impl Into<String>) -> Self {
135        self.processing = Some(key.into());
136        self
137    }
138
139    /// How long one blocking pop waits before looping. Defaults to 5 seconds.
140    pub const fn block(mut self, block: Duration) -> Self {
141        self.block = Some(block);
142        self
143    }
144
145    /// Decodes the header/payload envelope with `codec` (must match the publisher). Without it the
146    /// default lossless binary framing is used.
147    pub fn codec(mut self, codec: impl Codec + 'static) -> Self {
148        self.codec = Some(Arc::new(codec));
149        self
150    }
151
152    /// In reliable mode, routes dropped and poison entries to the named dead-letter list (`LPUSH`)
153    /// instead of discarding them, tagged with
154    /// [`DEAD_LETTER_REASON_HEADER`](crate::DEAD_LETTER_REASON_HEADER). Off by default. Has no effect
155    /// on a simple list, which cannot ack. See [`crate::deadletter`].
156    pub fn dead_letter(mut self, key: impl Into<String>) -> Self {
157        self.dead_letter = Some(key.into());
158        self
159    }
160
161    /// In reliable mode, caps how many times an entry may be `nack(requeue = true)`-ed before it is
162    /// treated as poison (dead-lettered or, with no dead-letter list, discarded). Off by default.
163    ///
164    /// Lists have no native delivery counter, so this tracks the framework retry-count header carried
165    /// in the entry's envelope.
166    pub const fn max_deliveries(mut self, max: u64) -> Self {
167        self.max_deliveries = Some(max);
168        self
169    }
170
171    /// How long a claimed reliable-mode entry may sit idle on the processing list before the
172    /// recovery watchdog returns it to the main list. Required for (and only meaningful with)
173    /// [`recovery_zset`](Self::recovery_zset).
174    ///
175    /// It has no default and must exceed the longest legitimate handler runtime: set it too low and
176    /// a healthy consumer's in-flight entry gets recovered and processed twice.
177    pub const fn min_idle(mut self, min_idle: Duration) -> Self {
178        self.min_idle = Some(min_idle);
179        self
180    }
181
182    /// Opts reliable mode into orphan recovery, naming the ZSET key that tracks in-flight claims.
183    ///
184    /// Off by default (a dead consumer's entry stays stranded on the processing list). The key has
185    /// no sane default, so it is named explicitly here; pair it with [`min_idle`](Self::min_idle),
186    /// which is required when recovery is on. Reliable mode is implied. See [`crate::recovery`].
187    pub fn recovery_zset(mut self, key: impl Into<String>) -> Self {
188        self.recovery_zset = Some(key.into());
189        self.reliable = true;
190        self
191    }
192
193    /// An optional auto-cleanup TTL on the recovery ZSET key (refreshed on every claim).
194    ///
195    /// When set it must exceed [`min_idle`](Self::min_idle) (and the longest legitimate handler
196    /// runtime), or in-flight tracking is dropped before the watchdog can act.
197    pub const fn recovery_ttl(mut self, ttl: Duration) -> Self {
198        self.recovery_ttl = Some(ttl);
199        self
200    }
201
202    /// The list key this subscription consumes.
203    #[must_use]
204    pub fn key(&self) -> &str {
205        &self.key
206    }
207
208    pub(crate) const fn is_reliable(&self) -> bool {
209        self.reliable
210    }
211
212    pub(crate) fn processing_or_default(&self) -> String {
213        self.processing
214            .clone()
215            .unwrap_or_else(|| format!("{}{PROCESSING_SUFFIX}", self.key))
216    }
217
218    pub(crate) fn block_or_default(&self) -> Duration {
219        self.block.unwrap_or(DEFAULT_BLOCK)
220    }
221
222    pub(crate) fn codec_handle(&self) -> Option<SharedEnvelope> {
223        self.codec.clone()
224    }
225
226    pub(crate) fn poison_policy(&self) -> PoisonPolicy {
227        PoisonPolicy {
228            dead_letter: self.dead_letter.clone(),
229            max_deliveries: self.max_deliveries,
230        }
231    }
232
233    /// Resolves the recovery settings, or `None` when recovery was not opted into.
234    ///
235    /// # Errors
236    ///
237    /// Returns [`RedisError::InvalidOptions`] when a recovery ZSET is named without a
238    /// [`min_idle`](Self::min_idle), which has no sane default.
239    pub(crate) fn recovery_config(&self) -> Result<Option<RecoveryConfig>, RedisError> {
240        let Some(zset_key) = self.recovery_zset.clone() else {
241            return Ok(None);
242        };
243        let min_idle = self.min_idle.ok_or_else(|| {
244            RedisError::InvalidOptions(format!(
245                "reliable list recovery on `{}` needs a min_idle: call .min_idle(duration) \
246                 alongside .recovery_zset(key)",
247                self.key
248            ))
249        })?;
250        Ok(Some(RecoveryConfig {
251            zset_key,
252            min_idle,
253            ttl: self.recovery_ttl,
254        }))
255    }
256}
257
258impl SubscriptionSource<ConnectedRedisBroker> for RedisList {
259    type Subscriber = RedisListSubscriber;
260
261    fn name(&self) -> &str {
262        self.key()
263    }
264
265    async fn subscribe(
266        self,
267        connected: &ConnectedRedisBroker,
268    ) -> Result<Self::Subscriber, RedisError> {
269        connected.subscribe_list(self).await
270    }
271}
272
273#[cfg(feature = "testing")]
274impl SubscriptionSource<crate::testing::ConnectedRedisTestBroker> for RedisList {
275    type Subscriber = crate::testing::RedisTestSubscriber;
276
277    fn name(&self) -> &str {
278        self.key()
279    }
280
281    async fn subscribe(
282        self,
283        connected: &crate::testing::ConnectedRedisTestBroker,
284    ) -> Result<Self::Subscriber, RedisError> {
285        connected.subscribe(self.key()).await
286    }
287}
288
289/// A list-backed work-queue subscription.
290pub struct RedisListSubscriber {
291    pool: Pool,
292    key: String,
293    reliable: bool,
294    processing: String,
295    block: Duration,
296    codec: Option<SharedEnvelope>,
297    policy: PoisonPolicy,
298    recovery: Option<RecoveryConfig>,
299}
300
301impl Debug for RedisListSubscriber {
302    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
303        f.debug_struct("RedisListSubscriber")
304            .field("key", &self.key)
305            .field("reliable", &self.reliable)
306            .field("poison", &self.policy.is_active())
307            .field("recovery", &self.recovery.is_some())
308            .finish_non_exhaustive()
309    }
310}
311
312impl RedisListSubscriber {
313    #[allow(
314        clippy::too_many_arguments,
315        reason = "internal constructor mirroring the descriptor"
316    )]
317    pub(crate) fn new(
318        pool: Pool,
319        key: String,
320        reliable: bool,
321        processing: String,
322        block: Duration,
323        codec: Option<SharedEnvelope>,
324        policy: PoisonPolicy,
325        recovery: Option<RecoveryConfig>,
326    ) -> Self {
327        Self {
328            pool,
329            key,
330            reliable,
331            processing,
332            block,
333            codec,
334            policy,
335            recovery,
336        }
337    }
338
339    fn simple_message(&self, raw: &[u8]) -> RedisListMessage {
340        let (payload, headers) = unframe(self.codec.as_ref(), raw);
341        RedisListMessage {
342            payload,
343            headers,
344            ack: None,
345        }
346    }
347
348    fn reliable_message(&self, raw: Vec<u8>, recovery: Option<RecoveryHandle>) -> RedisListMessage {
349        let (payload, headers) = unframe(self.codec.as_ref(), &raw);
350        RedisListMessage {
351            payload,
352            headers,
353            ack: Some(ListAck {
354                pool: self.pool.clone(),
355                main_key: self.key.clone(),
356                processing_key: self.processing.clone(),
357                value: raw,
358                codec: self.codec.clone(),
359                policy: self.policy.clone(),
360                recovery,
361            }),
362        }
363    }
364
365    /// Blocks for the next entry, returning `None` when the pop times out (the caller loops). When
366    /// recovery is enabled, first returns any orphaned entries to the main list so this same pop can
367    /// pick them up.
368    async fn next_entry(&self) -> Result<Option<RedisListMessage>, RedisError> {
369        let secs = block_secs(self.block);
370        if self.reliable {
371            if let Some(cfg) = &self.recovery {
372                recovery::sweep_orphans(&self.pool, cfg, &self.key, &self.processing).await?;
373            }
374            let value: Option<Vec<u8>> = empty_on_timeout(
375                self.pool
376                    .blmove(
377                        self.key.as_str(),
378                        self.processing.as_str(),
379                        LMoveDirection::Right,
380                        LMoveDirection::Left,
381                        secs,
382                    )
383                    .await,
384            )?;
385            let Some(value) = value else {
386                return Ok(None);
387            };
388            let handle = match &self.recovery {
389                Some(cfg) => {
390                    let member = recovery::record_claim(&self.pool, cfg, &value).await?;
391                    Some(RecoveryHandle {
392                        zset_key: cfg.zset_key.clone(),
393                        member,
394                    })
395                }
396                None => None,
397            };
398            Ok(Some(self.reliable_message(value, handle)))
399        } else {
400            let popped: Option<(String, Vec<u8>)> =
401                empty_on_timeout(self.pool.brpop(self.key.as_str(), secs).await)?;
402            Ok(popped.map(|(_, v)| self.simple_message(&v)))
403        }
404    }
405}
406
407impl ruststream::Subscriber for RedisListSubscriber {
408    type Message = RedisListMessage;
409    type Error = RedisError;
410
411    /// Yields one message per popped entry.
412    ///
413    /// # Cancel safety
414    ///
415    /// Dropping the returned stream between items is safe. In reliable mode an entry already moved
416    /// to the processing list but not yet settled stays there until acked or recovered manually.
417    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
418        unfold(&*self, |s| async move {
419            loop {
420                match s.next_entry().await {
421                    Ok(Some(msg)) => return Some((Ok(msg), s)),
422                    Ok(None) => {}
423                    Err(err) => return Some((Err(err), s)),
424                }
425            }
426        })
427    }
428}
429
430/// Settlement handle for a reliable-mode list delivery.
431struct ListAck {
432    pool: Pool,
433    main_key: String,
434    processing_key: String,
435    /// The raw wire value (framed), needed verbatim to `LREM` it from the processing list.
436    value: Vec<u8>,
437    /// The framing codec, so a poison-policy requeue can re-frame with an updated retry count.
438    codec: Option<SharedEnvelope>,
439    policy: PoisonPolicy,
440    /// Set when orphan recovery is enabled: the ZSET key and the member tracking this claim, so
441    /// settling removes its recovery tracking.
442    recovery: Option<RecoveryHandle>,
443}
444
445/// The recovery-ZSET coordinates for one in-flight reliable-list claim.
446struct RecoveryHandle {
447    zset_key: String,
448    member: Vec<u8>,
449}
450
451/// A list-queue delivery. In simple mode `ack` / `nack` are unsupported; in reliable mode `ack`
452/// removes the entry from the processing list and `nack` either returns it or drops it.
453pub struct RedisListMessage {
454    payload: Bytes,
455    headers: Headers,
456    ack: Option<ListAck>,
457}
458
459impl Debug for RedisListMessage {
460    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
461        f.debug_struct("RedisListMessage")
462            .field("payload_len", &self.payload.len())
463            .field("reliable", &self.ack.is_some())
464            .finish_non_exhaustive()
465    }
466}
467
468impl IncomingMessage for RedisListMessage {
469    fn payload(&self) -> &[u8] {
470        &self.payload
471    }
472
473    fn headers(&self) -> &Headers {
474        &self.headers
475    }
476
477    async fn ack(self) -> Result<(), AckError> {
478        let Some(handle) = self.ack else {
479            return Err(AckError::Unsupported);
480        };
481        settle(&handle).await
482    }
483
484    async fn nack(self, requeue: bool) -> Result<(), AckError> {
485        let Some(handle) = self.ack else {
486            return Err(AckError::Unsupported);
487        };
488        if requeue {
489            if handle.policy.is_active() {
490                let next = next_retry_count(&self.headers);
491                if handle.policy.is_poison(next) {
492                    list_dead_letter(&handle, &self.payload, &self.headers, REASON_MAX_DELIVERIES)
493                        .await?;
494                } else {
495                    // Re-frame with the incremented retry count and return it to the main list,
496                    // before removing the original from processing (a crash leaves a duplicate).
497                    let mut headers = self.headers.clone();
498                    headers.insert(RETRY_COUNT_HEADER, next.to_string());
499                    let body = frame(handle.codec.as_ref(), &self.payload, &headers);
500                    lpush(&handle.pool, handle.main_key.as_str(), body).await?;
501                }
502            } else {
503                // No poison policy: return the original entry verbatim to the main list.
504                lpush(&handle.pool, handle.main_key.as_str(), handle.value.clone()).await?;
505            }
506        } else if handle.policy.is_active() {
507            list_dead_letter(&handle, &self.payload, &self.headers, REASON_DROPPED).await?;
508        }
509        settle(&handle).await
510    }
511}
512
513fn ack_broker(err: fred::error::Error) -> AckError {
514    AckError::Broker(Box::new(err))
515}
516
517/// The next framework retry-count value (the current envelope header plus one, or one when absent).
518fn next_retry_count(headers: &Headers) -> u64 {
519    headers
520        .get_str(RETRY_COUNT_HEADER)
521        .and_then(|v| v.parse::<u64>().ok())
522        .unwrap_or(0)
523        + 1
524}
525
526async fn lpush(pool: &Pool, key: &str, body: Vec<u8>) -> Result<(), AckError> {
527    let _: i64 = pool.lpush(key, body).await.map_err(ack_broker)?;
528    Ok(())
529}
530
531/// `LPUSH`es a tagged copy onto the configured dead-letter list, or does nothing when none is set
532/// (the caller's `LREM` then discards the entry). Runs before the `LREM`, so a crash leaves a
533/// duplicate rather than a loss.
534async fn list_dead_letter(
535    handle: &ListAck,
536    payload: &[u8],
537    headers: &Headers,
538    reason: &'static str,
539) -> Result<(), AckError> {
540    if let Some(dlq) = handle.policy.dead_letter_key() {
541        let body = frame(
542            handle.codec.as_ref(),
543            payload,
544            &deadletter::with_reason(headers, reason),
545        );
546        lpush(&handle.pool, dlq, body).await?;
547    }
548    Ok(())
549}
550
551/// Removes the entry from the processing list and, when recovery is enabled, drops its tracking from
552/// the recovery ZSET.
553async fn settle(handle: &ListAck) -> Result<(), AckError> {
554    let _: i64 = handle
555        .pool
556        .lrem(handle.processing_key.as_str(), 1, handle.value.clone())
557        .await
558        .map_err(ack_broker)?;
559    if let Some(rec) = &handle.recovery {
560        recovery::forget(&handle.pool, &rec.zset_key, &rec.member).await?;
561    }
562    Ok(())
563}
564
565impl Partitioned for RedisListMessage {
566    fn partition_key(&self) -> Option<&[u8]> {
567        self.headers().get(PARTITION_KEY_HEADER)
568    }
569}
570
571/// The declaration half of the list publisher: envelope codec and key TTL, no connection.
572///
573/// Constructible anywhere, it pairs into a [`RedisListPublisher`] against a
574/// [`ConnectedRedisBroker`].
575///
576/// # Examples
577///
578/// ```
579/// use std::time::Duration;
580/// use ruststream::codec::JsonCodec;
581/// use ruststream_fred::RedisListPublish;
582///
583/// let publish = RedisListPublish::new()
584///     .codec(JsonCodec)
585///     .ttl(Duration::from_secs(300));
586/// # let _ = publish;
587/// ```
588#[derive(Clone, Default)]
589#[must_use]
590pub struct RedisListPublish {
591    codec: Option<SharedEnvelope>,
592    ttl: Option<Duration>,
593}
594
595impl Debug for RedisListPublish {
596    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
597        f.debug_struct("RedisListPublish")
598            .field("codec", &self.codec.is_some())
599            .field("ttl", &self.ttl)
600            .finish()
601    }
602}
603
604impl RedisListPublish {
605    /// A policy with the default binary framing and no key TTL. Equivalent to [`Self::default`].
606    pub fn new() -> Self {
607        Self::default()
608    }
609
610    /// Serializes the header/payload envelope with `codec` (must match the subscriber). Without it
611    /// the default lossless binary framing is used.
612    pub fn codec(mut self, codec: impl Codec + 'static) -> Self {
613        self.codec = Some(Arc::new(codec));
614        self
615    }
616
617    /// Sets a time-to-live on the list key, refreshed (`PEXPIRE`) on every publish, so an idle
618    /// queue auto-expires. Off by default: without it the list lives until drained or deleted.
619    ///
620    /// This is a per-key TTL on the whole list, not per-entry: Redis lists have no per-element
621    /// expiry, only the key can expire. Each publish pushes the entry and re-arms the key's TTL in
622    /// one pipeline, so an actively used queue never expires and only an idle one does. A sub-
623    /// millisecond `ttl` is clamped up to 1ms, since `PEXPIRE 0` would delete the key outright.
624    pub const fn ttl(mut self, ttl: Duration) -> Self {
625        self.ttl = Some(ttl);
626        self
627    }
628}
629
630impl PublishPolicy<ConnectedRedisBroker> for RedisListPublish {
631    type Live = RedisListPublisher;
632
633    async fn pair(self, connected: &ConnectedRedisBroker) -> Result<Self::Live, PairError> {
634        Ok(connected.list_publisher(self))
635    }
636}
637
638/// Publishes onto a list with `LPUSH`, so right-popping consumers see FIFO order: a
639/// [`RedisListPublish`] policy paired with a connection.
640///
641/// Obtain it from
642/// [`ConnectedRedisBroker::list_publisher`](crate::ConnectedRedisBroker::list_publisher), or by
643/// pairing the policy. Publishing after the connection was shut down reports
644/// [`RedisError::ShutDown`].
645#[derive(Clone)]
646pub struct RedisListPublisher {
647    core: Arc<RedisCore>,
648    codec: Option<SharedEnvelope>,
649    ttl: Option<Duration>,
650}
651
652impl Debug for RedisListPublisher {
653    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
654        f.debug_struct("RedisListPublisher")
655            .field("codec", &self.codec.is_some())
656            .field("ttl", &self.ttl)
657            .finish_non_exhaustive()
658    }
659}
660
661impl RedisListPublisher {
662    pub(crate) fn new(core: Arc<RedisCore>, publish: RedisListPublish) -> Self {
663        Self {
664            core,
665            codec: publish.codec,
666            ttl: publish.ttl,
667        }
668    }
669}
670
671/// Converts a TTL to the positive millisecond count `PEXPIRE` expects, clamping a sub-millisecond
672/// value up to 1 (a `PEXPIRE 0` deletes the key instead of expiring it).
673fn ttl_millis(ttl: Duration) -> i64 {
674    i64::try_from(ttl.as_millis()).unwrap_or(i64::MAX).max(1)
675}
676
677impl ruststream::Publisher for RedisListPublisher {
678    type Error = RedisError;
679
680    async fn publish(&self, msg: ruststream::OutgoingMessage<'_>) -> Result<(), Self::Error> {
681        let pool = self.core.pool()?;
682        let body = frame(self.codec.as_ref(), msg.payload(), msg.headers());
683        let Some(ttl) = self.ttl else {
684            let _: i64 = pool
685                .lpush(msg.name(), body)
686                .await
687                .map_err(RedisError::publish)?;
688            return Ok(());
689        };
690        // Push the entry and re-arm the key TTL in one pipeline, so an actively used queue keeps
691        // resetting its expiry and only an idle one is allowed to lapse.
692        let pipeline = pool.next().pipeline();
693        let _: () = pipeline
694            .lpush(msg.name(), body)
695            .await
696            .map_err(RedisError::publish)?;
697        let _: () = pipeline
698            .pexpire(msg.name(), ttl_millis(ttl), None)
699            .await
700            .map_err(RedisError::publish)?;
701        let _: Vec<fred::types::Value> = pipeline.all().await.map_err(RedisError::publish)?;
702        Ok(())
703    }
704}
705
706#[cfg(test)]
707mod tests {
708    use super::*;
709
710    #[test]
711    fn ttl_millis_converts_and_clamps() {
712        assert_eq!(ttl_millis(Duration::from_secs(60)), 60_000);
713        assert_eq!(ttl_millis(Duration::from_millis(1)), 1);
714        // A sub-millisecond TTL must not become PEXPIRE 0 (which deletes the key).
715        assert_eq!(ttl_millis(Duration::from_nanos(1)), 1);
716        assert_eq!(ttl_millis(Duration::ZERO), 1);
717    }
718}
ruststream_fred/list.rs

ruststream_fred/
list.rs
