ruststream_fred/seek.rs
1//! Repositioning a consumer group over the stream: [`EntryId`], [`RedisGroupPosition`], and the
2//! [`RedisGroupSeeker`] handle behind the core `Seekable` capability.
3//!
4//! Redis Streams keep every entry until the stream is trimmed, so a consumer group can be moved
5//! back over history or forward past a region. The move is `XGROUP SETID`, which rewrites the
6//! group's cursor - and a group has one cursor, so the move applies to **every consumer of that
7//! group**, not just the subscription that asked. The type names say `Group` for exactly that
8//! reason; see [`RedisGroupSeeker`] for the full contract.
9
10use std::fmt::{Display, Formatter};
11use std::str::FromStr;
12use std::sync::Arc;
13use std::sync::atomic::{AtomicU64, Ordering};
14
15use fred::clients::Pool;
16use fred::interfaces::StreamsInterface;
17use ruststream::Seeker;
18
19use crate::error::RedisError;
20
21/// A Redis Streams entry id: the `<milliseconds>-<sequence>` pair the server assigns to every
22/// entry.
23///
24/// Parsed on construction, so a value of this type is always a well-formed id and comparisons
25/// order entries the way the stream does. The `<milliseconds>` half alone is accepted too
26/// (`"1700000000000"` means `1700000000000-0`), matching what Redis accepts on the wire.
27///
28/// # Examples
29///
30/// ```
31/// use ruststream_fred::EntryId;
32///
33/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
34/// let id: EntryId = "1700000000000-4".parse()?;
35/// assert_eq!(id.milliseconds(), 1_700_000_000_000);
36/// assert_eq!(id.sequence(), 4);
37/// assert_eq!(id.to_string(), "1700000000000-4");
38/// # Ok(())
39/// # }
40/// ```
41#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
42pub struct EntryId {
43 milliseconds: u64,
44 sequence: u64,
45}
46
47impl EntryId {
48 /// The lowest id the stream can express, below every real entry (Redis rejects `0-0` as an
49 /// entry id, so nothing can sit at it).
50 pub const ZERO: Self = Self::new(0, 0);
51
52 /// Builds an id from its two halves.
53 #[must_use]
54 pub const fn new(milliseconds: u64, sequence: u64) -> Self {
55 Self {
56 milliseconds,
57 sequence,
58 }
59 }
60
61 /// The `<milliseconds>` half: the entry's server-side timestamp.
62 #[must_use]
63 pub const fn milliseconds(&self) -> u64 {
64 self.milliseconds
65 }
66
67 /// The `<sequence>` half: the counter distinguishing entries added within one millisecond.
68 #[must_use]
69 pub const fn sequence(&self) -> u64 {
70 self.sequence
71 }
72
73 /// The id immediately below this one, saturating at [`ZERO`](Self::ZERO).
74 ///
75 /// Sequence numbers are dense within a millisecond, so the predecessor of `<ms>-0` is
76 /// `<ms - 1>-<u64::MAX>`. Used to turn "resume *at* this entry" into the exclusive cursor
77 /// `XGROUP SETID` takes.
78 #[must_use]
79 pub const fn previous(&self) -> Self {
80 match (self.milliseconds, self.sequence) {
81 (0, 0) => Self::ZERO,
82 (ms, 0) => Self::new(ms - 1, u64::MAX),
83 (ms, seq) => Self::new(ms, seq - 1),
84 }
85 }
86}
87
88impl Display for EntryId {
89 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
90 write!(f, "{}-{}", self.milliseconds, self.sequence)
91 }
92}
93
94impl FromStr for EntryId {
95 type Err = RedisError;
96
97 fn from_str(raw: &str) -> Result<Self, Self::Err> {
98 let invalid = || {
99 RedisError::InvalidOptions(format!(
100 "`{raw}` is not a redis stream entry id (expected `<milliseconds>-<sequence>`)"
101 ))
102 };
103 let (ms, seq) = match raw.split_once('-') {
104 Some((ms, seq)) => (ms, seq),
105 // Redis accepts a bare `<ms>`, which means `<ms>-0`.
106 None => (raw, "0"),
107 };
108 Ok(Self::new(
109 ms.parse().map_err(|_| invalid())?,
110 seq.parse().map_err(|_| invalid())?,
111 ))
112 }
113}
114
115/// Where a consumer group's cursor sits, for [`RedisGroupSeeker::seek`] and the
116/// `start_at(..)` clause of `#[subscriber]`.
117///
118/// The cursor is **group-wide**: it belongs to the consumer group, not to one subscription, so
119/// moving it moves it for every consumer reading that group.
120///
121/// # Examples
122///
123/// ```
124/// use ruststream_fred::{EntryId, RedisGroupPosition};
125///
126/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
127/// let replay_all = RedisGroupPosition::beginning();
128/// let skip_backlog = RedisGroupPosition::end();
129/// let resume = RedisGroupPosition::after("1700000000000-4".parse::<EntryId>()?);
130/// # let _ = (replay_all, skip_backlog, resume);
131/// # Ok(())
132/// # }
133/// ```
134#[derive(Debug, Clone, Copy, PartialEq, Eq)]
135#[non_exhaustive]
136pub enum RedisGroupPosition {
137 /// The oldest entry the stream still retains: the group replays everything it holds.
138 Beginning,
139 /// The tail: only entries added after the seek are delivered.
140 End,
141 /// The first entry *after* `id`. The cursor is exclusive, exactly like the id argument of
142 /// `XGROUP SETID`, so the entry at `id` itself is not delivered again.
143 After(EntryId),
144}
145
146impl RedisGroupPosition {
147 /// The oldest retained entry. Constructor form of [`Beginning`](Self::Beginning).
148 #[must_use]
149 pub const fn beginning() -> Self {
150 Self::Beginning
151 }
152
153 /// Only entries added from now on. Constructor form of [`End`](Self::End).
154 #[must_use]
155 pub const fn end() -> Self {
156 Self::End
157 }
158
159 /// Resume with the entry following `id`. Constructor form of [`After`](Self::After).
160 #[must_use]
161 pub const fn after(id: EntryId) -> Self {
162 Self::After(id)
163 }
164
165 /// The id `XGROUP SETID` takes for this position.
166 fn as_xid(self) -> String {
167 match self {
168 Self::Beginning => EntryId::ZERO.to_string(),
169 Self::End => "$".to_owned(),
170 Self::After(id) => id.to_string(),
171 }
172 }
173}
174
175/// Moves a consumer group's cursor.
176///
177/// The handle behind the core `Seekable` capability, minted with
178/// [`Seekable::seeker`](ruststream::Seekable::seeker) or injected into a handler as a
179/// `Seek(seeker): Seek<RedisGroupSeeker>` parameter.
180///
181/// # A seek is group-wide
182///
183/// Redis keeps one cursor per consumer group, so `XGROUP SETID` moves the read position for
184/// **every consumer of that group**, not only for the subscription this seeker came from. That is
185/// unlike a partitioned log, where a seek is scoped to one consumer instance. Treat a seek as an
186/// operation on the group: replaying a range replays it for the whole worker pool, and skipping
187/// forward skips for all of them.
188///
189/// # What a seek does not do
190///
191/// * It does not clear the pending entries list. Entries already delivered and not acknowledged
192/// stay pending and remain reachable through the reclaim path
193/// ([`RedisStream::reclaim`](crate::RedisStream::reclaim)), whichever way the cursor moved.
194/// * It does not cancel delayed retries. Copies already scheduled in a
195/// [`DelayedRetry::DurableZset`](crate::DelayedRetry::DurableZset) queue are keyed by their due
196/// time, not by the cursor, so they are appended to the stream when they fall due regardless of
197/// where the group is reading.
198/// * It does not reset delivery counts. A replayed entry is delivered again, so its **native**
199/// delivery count (the one the reclaim path reads and
200/// [`RedisStream::max_deliveries`](crate::RedisStream::max_deliveries) caps) grows with each
201/// replay; the framework retry-count header only moves on an actual `nack`.
202///
203/// # Timing
204///
205/// The cursor changes as soon as `seek` returns, but a subscription parked in a blocking
206/// `XREADGROUP` observes it only on its next read - within one
207/// [`RedisStream::block`](crate::RedisStream::block) interval. Entries selected under the old
208/// cursor are discarded rather than delivered, so a seek never yields a message from the position
209/// it moved away from.
210///
211/// # Examples
212///
213/// ```no_run
214/// use ruststream::{Broker, Seekable, Seeker};
215/// use ruststream_fred::{RedisBroker, RedisGroupPosition, RedisStream};
216///
217/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
218/// let connected = RedisBroker::standalone("redis://localhost:6379").connect().await?;
219/// let subscriber = connected
220/// .subscribe(RedisStream::new("orders").group("workers"))
221/// .await?;
222///
223/// // Minted before the stream opens; usable while it runs.
224/// let seeker = subscriber.seeker();
225/// seeker.seek(RedisGroupPosition::beginning()).await?;
226/// # Ok(())
227/// # }
228/// ```
229#[derive(Clone)]
230pub struct RedisGroupSeeker {
231 pool: Pool,
232 key: String,
233 group: String,
234 /// Shared with the subscription: bumped on every seek so entries selected under the old
235 /// cursor are recognised as stale and dropped instead of delivered.
236 generation: Arc<AtomicU64>,
237}
238
239impl std::fmt::Debug for RedisGroupSeeker {
240 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
241 f.debug_struct("RedisGroupSeeker")
242 .field("key", &self.key)
243 .field("group", &self.group)
244 .finish_non_exhaustive()
245 }
246}
247
248impl RedisGroupSeeker {
249 pub(crate) fn new(pool: Pool, key: String, group: String, generation: Arc<AtomicU64>) -> Self {
250 Self {
251 pool,
252 key,
253 group,
254 generation,
255 }
256 }
257}
258
259impl Seeker for RedisGroupSeeker {
260 type Position = RedisGroupPosition;
261 type Error = RedisError;
262
263 /// Moves the group cursor with `XGROUP SETID`.
264 ///
265 /// # Errors
266 ///
267 /// Returns [`RedisError::Stream`] when the group or the stream does not exist, or the
268 /// command fails.
269 async fn seek(&self, to: RedisGroupPosition) -> Result<(), RedisError> {
270 let _: String = self
271 .pool
272 .xgroup_setid(self.key.as_str(), self.group.as_str(), to.as_xid())
273 .await
274 .map_err(RedisError::stream)?;
275 // Bumped after the cursor moved, so a reader that observes the new generation is
276 // guaranteed to read under the new cursor.
277 self.generation.fetch_add(1, Ordering::Release);
278 Ok(())
279 }
280}
281
282#[cfg(test)]
283mod tests {
284 use super::*;
285
286 #[test]
287 fn entry_ids_parse_and_render() {
288 let id: EntryId = "1700000000000-4".parse().expect("valid id");
289 assert_eq!(id, EntryId::new(1_700_000_000_000, 4));
290 assert_eq!(id.to_string(), "1700000000000-4");
291 // A bare millisecond half means sequence zero, as on the wire.
292 assert_eq!(
293 "5".parse::<EntryId>().expect("valid id"),
294 EntryId::new(5, 0)
295 );
296 }
297
298 #[test]
299 fn malformed_entry_ids_are_rejected() {
300 for raw in ["", "-", "abc", "5-x", "5-6-7"] {
301 assert!(
302 raw.parse::<EntryId>().is_err(),
303 "`{raw}` must not parse as an entry id"
304 );
305 }
306 }
307
308 // The predecessor is what turns a captured delivery into the exclusive cursor that redelivers
309 // it, so the sequence and millisecond borrows must both be right.
310 #[test]
311 fn previous_walks_back_one_id() {
312 assert_eq!(EntryId::new(5, 3).previous(), EntryId::new(5, 2));
313 assert_eq!(EntryId::new(5, 0).previous(), EntryId::new(4, u64::MAX));
314 assert_eq!(EntryId::ZERO.previous(), EntryId::ZERO);
315 assert_eq!(EntryId::new(0, 1).previous(), EntryId::ZERO);
316 }
317
318 #[test]
319 fn positions_map_to_setid_arguments() {
320 assert_eq!(RedisGroupPosition::beginning().as_xid(), "0-0");
321 assert_eq!(RedisGroupPosition::end().as_xid(), "$");
322 assert_eq!(
323 RedisGroupPosition::after(EntryId::new(7, 2)).as_xid(),
324 "7-2"
325 );
326 }
327}