Skip to main content

yo_kv/stream/
groups.rs

1//! Consumer groups and the pending entries list (`08` section 7).
2//!
3//! A consumer group is a bookmark plus a ledger. The bookmark is one ID saying
4//! how far the group has read, and the ledger is every entry the group handed
5//! out and has not been told is finished with. Redis calls the ledger the PEL,
6//! the pending entries list, and an entry in it a NACK.
7//!
8//! ```text
9//! group "workers"  last 990-0  read 41823
10//! +----------------------------------------------------------+
11//! | PEL, in ID order                                          |
12//! | 971-0 -> owner c2, handed out 3 times, last at 09:14:02   |
13//! | 984-0 -> owner c1, handed out 1 time,  last at 09:14:07   |
14//! | 990-0 -> owner c1, handed out 1 time,  last at 09:14:07   |
15//! +----------------------------------------------------------+
16//!     consumer c1 holds 984-0, 990-0
17//!     consumer c2 holds 971-0
18//! ```
19//!
20//! The point of the ledger is that a consumer can die holding work. `XPENDING`
21//! finds entries nobody has touched in a while and `XCLAIM` moves them to a
22//! consumer that is still alive, which is the whole reason to use a group
23//! rather than plain `XREAD`.
24//!
25//! # A NACK is in two indexes and owned by neither
26//!
27//! Every pending entry has to be reachable two ways. `XPENDING` and `XAUTOCLAIM`
28//! walk the group's entries in ID order, and `XINFO CONSUMERS` and consumer
29//! deletion need everything one consumer holds. Redis keeps a rax per group and
30//! a rax per consumer holding pointers to the same NACK, which means a claim
31//! updates a pointer in two trees and the NACK belongs to whichever one frees it
32//! last.
33//!
34//! Here the NACK lives in the group's map and the consumer holds only IDs. A
35//! claim moves an ID between two [`BTreeSet`]s and rewrites one field, nothing
36//! is shared and nothing has to be freed carefully. It costs one extra lookup
37//! when going from a consumer's ID to its NACK, which happens on consumer
38//! deletion and nowhere on a hot path.
39//!
40//! # Why a B-tree and not a sorted deque
41//!
42//! The log next door is a sorted deque because entries are appended in order and
43//! trimmed from the front, and never touched in the middle. A PEL is the same
44//! shape most of the time: `XREADGROUP >` appends increasing IDs and `XACK`
45//! usually takes the oldest. But `XCLAIM` and a slow consumer both put holes in
46//! the middle, and an ack of an arbitrary ID is a normal thing to do rather than
47//! a pathology, so the middle is not the rare case here that it is in the log.
48//!
49//! A [`BTreeMap`] keyed by [`Id`] holds the key inline in sixteen bytes with no
50//! allocation per entry and about eleven entries a node, and it gives the
51//! ordered walk `XPENDING` and `XAUTOCLAIM` need. That is already well ahead of
52//! a rax over sixteen byte string keys. Whether the sorted deque with tombstones
53//! would beat it is a real question and the benchmark is there to answer it, but
54//! it is not worth guessing at before the feature works.
55//!
56//! # Consumers are a vector
57//!
58//! A group has a handful of consumers, usually as many as there are processes,
59//! and a name is looked up once per command rather than once per entry. A linear
60//! scan over a vector beats a hash map at that size and brings no dependency and
61//! no hashing with it. A slot is never reused while the group lives, so the
62//! index a NACK holds stays valid.
63
64use std::collections::{BTreeMap, BTreeSet, HashSet};
65
66use super::Id;
67use crate::frozen::{self, Broken};
68
69/// Which pending entries a caller wants, which is every filter `XPENDING` takes.
70///
71/// A struct rather than five more arguments because the command parses them as
72/// a group and they travel together from the wire to here. The default is the
73/// whole list, so a caller that only wants a window sets `start` and `end` and
74/// leaves the rest alone.
75#[derive(Debug, Clone, Copy)]
76pub struct Filter {
77    /// The low end of the ID window, included.
78    pub start: Id,
79    /// The high end, included.
80    pub end: Id,
81    /// At most this many, or every one in the window.
82    pub count: Option<usize>,
83    /// Only what this consumer is holding.
84    pub owner: Option<u32>,
85    /// Only what has been sitting at least this many milliseconds.
86    pub min_idle: u64,
87}
88
89impl Default for Filter {
90    fn default() -> Filter {
91        Filter {
92            start: Id::MIN,
93            end: Id::MAX,
94            count: None,
95            owner: None,
96            min_idle: 0,
97        }
98    }
99}
100
101/// One entry handed out and not yet acknowledged.
102///
103/// Redis calls this a NACK, for not acknowledged.
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct Nack {
106    /// When it was last handed out, in milliseconds.
107    ///
108    /// Set on delivery and reset on a claim, because the point of it is how
109    /// long the entry has been sitting with somebody who is not finishing it.
110    time: u64,
111    /// How many times it has been handed out.
112    ///
113    /// `XCLAIM RETRYCOUNT` sets it and `XPENDING` reports it, so a consumer can
114    /// give up on a message that has killed several workers already.
115    count: u64,
116    /// Which consumer slot holds it, or [`Nack::NOBODY`].
117    owner: u32,
118}
119
120impl Nack {
121    /// The slot of an entry that is pending and that nobody holds.
122    ///
123    /// `XNACK` hands work back to the group without giving it to anybody, so the
124    /// pending list has to be able to hold an entry with no consumer against it.
125    /// Redis reports one as an empty consumer name and an idle time of minus
126    /// one, and treats it as idle for longer than any `min-idle-time` a claim can
127    /// name, which is what makes it the next thing `XAUTOCLAIM` picks up.
128    const NOBODY: u32 = u32::MAX;
129
130    /// When it was last handed out.
131    #[must_use]
132    #[inline]
133    pub fn time(&self) -> u64 {
134        self.time
135    }
136
137    /// How many times it has been handed out.
138    #[must_use]
139    #[inline]
140    pub fn count(&self) -> u64 {
141        self.count
142    }
143
144    /// Which consumer holds it, or `None` for one that has been released.
145    #[must_use]
146    #[inline]
147    pub fn owner(&self) -> Option<u32> {
148        (self.owner != Nack::NOBODY).then_some(self.owner)
149    }
150
151    /// How long it has been sitting, which is what `min-idle-time` is compared to.
152    ///
153    /// Saturating, because a NACK whose time was set forward by `XCLAIM TIME` is
154    /// something a caller is allowed to ask for and is not idle at all. An entry
155    /// nobody holds has been idle for as long as there is, so that every claim
156    /// and every `XPENDING IDLE` filter picks it up whatever they asked for.
157    #[must_use]
158    #[inline]
159    pub fn idle(&self, now: u64) -> u64 {
160        if self.owner == Nack::NOBODY {
161            return u64::MAX;
162        }
163        now.saturating_sub(self.time)
164    }
165}
166
167/// What releasing an entry does to its delivery count.
168///
169/// `XNACK` takes one of three words for this and they only differ here. A worker
170/// that could not do the job because the machine it was on went away wants the
171/// attempt not to count, one that failed the way work sometimes fails wants the
172/// count left alone, and one that has decided the message itself is the problem
173/// wants nobody to try it again.
174#[derive(Debug, Clone, Copy, PartialEq, Eq)]
175pub enum Retry {
176    /// `SILENT`: take one off the count, as if the delivery had not happened.
177    Down,
178    /// `FAIL`: leave the count where it is, and start a new entry at zero.
179    Keep,
180    /// `FATAL`: put the count as high as it goes.
181    Max,
182    /// `RETRYCOUNT n`: put the count at exactly this, whatever the word said.
183    At(u64),
184}
185
186impl Retry {
187    /// The count an entry that was on `had` ends up with.
188    ///
189    /// [`Retry::Down`] takes one off rather than putting the count back to zero,
190    /// which is worth saying because the two look the same on an entry that has
191    /// only been handed out once and that is the entry most people try it on. A
192    /// message that has killed four workers and is then released by a fifth for
193    /// a reason that was nothing to do with the message reads as three, not as
194    /// new. It saturates, so a released entry that is released again stays at
195    /// zero rather than wrapping.
196    ///
197    /// [`Retry::Max`] is [`i64::MAX`] and not [`u64::MAX`] because that is the
198    /// number Redis reports, and a client that reads the count into a signed
199    /// integer, which is what the protocol hands it, has to be able to hold it.
200    #[must_use]
201    pub fn applied(self, had: u64) -> u64 {
202        match self {
203            Retry::Down => had.saturating_sub(1),
204            Retry::Keep => had,
205            Retry::Max => i64::MAX as u64,
206            Retry::At(n) => n,
207        }
208    }
209}
210
211/// One consumer inside a group.
212#[derive(Debug, Clone, PartialEq, Eq)]
213pub struct Consumer {
214    name: Vec<u8>,
215    /// When this consumer was last heard from at all.
216    seen: u64,
217    /// When it last read something, as opposed to asking and getting nothing.
218    ///
219    /// Redis separates the two because a consumer polling an empty stream is
220    /// alive but idle, and telling those apart is the difference between a
221    /// worker that is stuck and one that has nothing to do.
222    ///
223    /// `None` for a consumer that has never had anything, which is what
224    /// `XGROUP CREATECONSUMER` makes and what an `XREADGROUP` that found nothing
225    /// leaves behind. Redis reports that as an active time of minus one rather
226    /// than as the moment the consumer turned up, and `XINFO CONSUMERS` passes
227    /// it through to `inactive`, so a fresh consumer reads as never active and
228    /// not as active a moment ago.
229    active: Option<u64>,
230    /// What it holds, in ID order.
231    pending: BTreeSet<Id>,
232}
233
234impl Consumer {
235    /// Its name.
236    #[must_use]
237    #[inline]
238    pub fn name(&self) -> &[u8] {
239        &self.name
240    }
241
242    /// When it was last heard from.
243    #[must_use]
244    #[inline]
245    pub fn seen(&self) -> u64 {
246        self.seen
247    }
248
249    /// When it last actually read something, or `None` if it never has.
250    #[must_use]
251    #[inline]
252    pub fn active(&self) -> Option<u64> {
253        self.active
254    }
255
256    /// How many entries it is holding.
257    #[must_use]
258    #[inline]
259    pub fn len(&self) -> usize {
260        self.pending.len()
261    }
262
263    /// Whether it is holding nothing.
264    #[must_use]
265    #[inline]
266    pub fn is_empty(&self) -> bool {
267        self.pending.is_empty()
268    }
269
270    /// What it is holding, oldest first.
271    pub fn pending(&self) -> impl Iterator<Item = Id> + '_ {
272        self.pending.iter().copied()
273    }
274}
275
276/// A consumer group over one stream.
277#[derive(Debug, Clone, Default, PartialEq, Eq)]
278pub struct Group {
279    /// The last ID handed out, which `XREADGROUP >` reads after.
280    last: Id,
281    /// How many entries the group has read, for the lag.
282    ///
283    /// An `Option` because it is not always knowable. `XSETID` without
284    /// `ENTRIESREAD` and a `SETID` to a point nobody can count from both leave
285    /// it unknown, and Redis reports a null lag rather than a made up one.
286    read: Option<u64>,
287    /// Everything handed out and not acknowledged, in ID order.
288    pending: BTreeMap<Id, Nack>,
289    /// The consumers. A slot is emptied on deletion and never reused.
290    consumers: Vec<Option<Consumer>>,
291    /// How many of the pending entries nobody holds.
292    ///
293    /// Kept rather than counted because `XINFO STREAM FULL` reports it and that
294    /// command takes a `COUNT` precisely so that it never walks a long pending
295    /// list. Every line that moves a NACK on or off [`Nack::NOBODY`] is in this
296    /// file and adjusts this, and a test at the bottom checks the number against
297    /// a full scan after a run of mixed operations.
298    nacked: usize,
299}
300
301impl Group {
302    /// A group reading after `last`, having read `read` entries.
303    #[must_use]
304    pub fn new(last: Id, read: Option<u64>) -> Group {
305        Group {
306            last,
307            read,
308            pending: BTreeMap::new(),
309            consumers: Vec::new(),
310            nacked: 0,
311        }
312    }
313
314    /// The last ID handed out.
315    #[must_use]
316    #[inline]
317    pub fn last_id(&self) -> Id {
318        self.last
319    }
320
321    /// How many entries the group has read, when that is known.
322    #[must_use]
323    #[inline]
324    pub fn entries_read(&self) -> Option<u64> {
325        self.read
326    }
327
328    /// Move the bookmark, which is `XGROUP SETID`.
329    ///
330    /// The PEL is left alone, because the entries in it were handed to somebody
331    /// who has not finished and moving the bookmark says nothing about them.
332    pub fn set_id(&mut self, last: Id, read: Option<u64>) {
333        self.last = last;
334        self.read = read;
335    }
336
337    /// How many entries are pending across the whole group.
338    #[must_use]
339    #[inline]
340    pub fn pending_len(&self) -> usize {
341        self.pending.len()
342    }
343
344    /// How many of those nobody is holding, which `XNACK` is what makes nonzero.
345    #[must_use]
346    #[inline]
347    pub fn nacked_len(&self) -> usize {
348        self.nacked
349    }
350
351    /// The lowest and highest pending IDs, which is the `XPENDING` summary.
352    #[must_use]
353    pub fn pending_bounds(&self) -> Option<(Id, Id)> {
354        let low = *self.pending.keys().next()?;
355        let high = *self.pending.keys().next_back()?;
356        Some((low, high))
357    }
358
359    /// One pending entry.
360    #[must_use]
361    #[inline]
362    pub fn nack(&self, id: Id) -> Option<&Nack> {
363        self.pending.get(&id)
364    }
365
366    /// The consumer slot for `name`, if there is one.
367    #[must_use]
368    pub fn slot(&self, name: &[u8]) -> Option<u32> {
369        self.consumers
370            .iter()
371            .position(|c| c.as_ref().is_some_and(|c| c.name == name))
372            .map(|at| at as u32)
373    }
374
375    /// A consumer by slot.
376    #[must_use]
377    #[inline]
378    pub fn consumer(&self, slot: u32) -> Option<&Consumer> {
379        self.consumers.get(slot as usize)?.as_ref()
380    }
381
382    /// A consumer by name.
383    #[must_use]
384    pub fn consumer_named(&self, name: &[u8]) -> Option<&Consumer> {
385        self.consumers
386            .iter()
387            .flatten()
388            .find(|c| c.name.as_slice() == name)
389    }
390
391    /// Every consumer, in the order they were created.
392    pub fn consumers(&self) -> impl Iterator<Item = &Consumer> + '_ {
393        self.consumers.iter().flatten()
394    }
395
396    /// The slot for `name`, making the consumer if it is not there yet.
397    ///
398    /// This is what `XREADGROUP` does, since a consumer exists because it turned
399    /// up rather than because anybody declared it.
400    pub fn consumer_or_create(&mut self, name: &[u8], now: u64) -> u32 {
401        if let Some(at) = self.slot(name) {
402            let c = self.consumers[at as usize]
403                .as_mut()
404                .expect("the slot the search just found");
405            c.seen = now;
406            return at;
407        }
408        self.consumers.push(Some(Consumer {
409            name: name.to_vec(),
410            seen: now,
411            active: None,
412            pending: BTreeSet::new(),
413        }));
414        (self.consumers.len() - 1) as u32
415    }
416
417    /// Make a consumer and say whether it was not already there.
418    ///
419    /// `XGROUP CREATECONSUMER`, which answers 1 when it made one.
420    pub fn create_consumer(&mut self, name: &[u8], now: u64) -> bool {
421        if self.slot(name).is_some() {
422            return false;
423        }
424        self.consumer_or_create(name, now);
425        true
426    }
427
428    /// Take a consumer out and say how many entries it was holding.
429    ///
430    /// Those entries stop being pending at all, which is Redis's behaviour and
431    /// is the point of the command: deleting a consumer is how you give up on
432    /// the work it was holding when you would rather lose it than claim it.
433    pub fn delete_consumer(&mut self, name: &[u8]) -> u64 {
434        let Some(at) = self.slot(name) else {
435            return 0;
436        };
437        let gone = self.consumers[at as usize]
438            .take()
439            .expect("the slot the search just found");
440        for id in &gone.pending {
441            self.pending.remove(id);
442        }
443        gone.pending.len() as u64
444    }
445
446    /// Mark that a consumer was heard from.
447    ///
448    /// `read` says whether it got anything, which is what separates seen from
449    /// active.
450    pub fn touch(&mut self, slot: u32, now: u64, read: bool) {
451        if let Some(Some(c)) = self.consumers.get_mut(slot as usize) {
452            c.seen = now;
453            if read {
454                c.active = Some(now);
455            }
456        }
457    }
458
459    /// Hand an entry to a consumer for the first time.
460    ///
461    /// The bookmark moves, since this is the `>` path and the entry is new to
462    /// the group. Answers false if the slot is empty, which a caller that got
463    /// its slot from [`Group::consumer_or_create`] cannot hit.
464    pub fn deliver(&mut self, slot: u32, id: Id, now: u64) -> bool {
465        let Some(Some(c)) = self.consumers.get_mut(slot as usize) else {
466            return false;
467        };
468        c.pending.insert(id);
469        self.pending.insert(
470            id,
471            Nack {
472                time: now,
473                count: 1,
474                owner: slot,
475            },
476        );
477        if id > self.last {
478            self.last = id;
479        }
480        true
481    }
482
483    /// Move the bookmark past an entry without writing it into the ledger.
484    ///
485    /// This is `XREADGROUP ... NOACK`, which is a consumer saying it does not
486    /// want the work tracked. The group still counts the entry as read, because
487    /// the lag is about how far behind the group is and not about how much of
488    /// it is outstanding, so a NOACK reader that has caught up reports a lag of
489    /// zero the same as any other.
490    pub fn skip(&mut self, id: Id) {
491        if id > self.last {
492            self.last = id;
493        }
494    }
495
496    /// Hand an entry to whoever already holds it, which is a history read.
497    ///
498    /// `XREADGROUP` with an ID rather than `>` is a consumer asking for what it
499    /// was already given, and Redis treats that as a real delivery: the time is
500    /// reset and the count goes up, exactly as if the entry had been handed out
501    /// again. Checked against Redis 8.10.1, where a history read of an entry
502    /// idle for 2006 milliseconds left it idle for 2 with its count up by one.
503    ///
504    /// It reads as surprising until you think about what the count is for. It
505    /// counts how many times a consumer has been told to do this work, and a
506    /// consumer re-reading its backlog after a restart has been told again.
507    pub fn redeliver(&mut self, id: Id, now: u64) -> bool {
508        let Some(nack) = self.pending.get_mut(&id) else {
509            return false;
510        };
511        nack.time = now;
512        nack.count += 1;
513        true
514    }
515
516    /// Put the read counter where the stream has worked out it belongs.
517    ///
518    /// The counter is a fact about the stream and not about the group, since
519    /// what a delivery does to it depends on whether anything has been deleted
520    /// ahead of the group. [`crate::stream::Stream::read_group`] is the one
521    /// caller, and it is the one that can see both.
522    pub fn set_read(&mut self, read: Option<u64>) {
523        self.read = read;
524    }
525
526    /// Finish with an entry, which is `XACK`.
527    ///
528    /// Answers whether it was pending. Acknowledging something twice is not an
529    /// error, it just does nothing the second time, because a consumer that
530    /// crashed between doing the work and sending the ack will send it again.
531    pub fn ack(&mut self, id: Id) -> bool {
532        let Some(nack) = self.pending.remove(&id) else {
533            return false;
534        };
535        match self.consumers.get_mut(nack.owner as usize) {
536            Some(Some(c)) => {
537                c.pending.remove(&id);
538            }
539            // Either the slot was emptied under it, which cannot happen because
540            // deleting a consumer takes its entries with it, or nobody held it.
541            _ => self.nacked -= usize::from(nack.owner == Nack::NOBODY),
542        }
543        true
544    }
545
546    /// Hand an entry back to the group without acknowledging it, which is `XNACK`.
547    ///
548    /// The entry stays pending and stops belonging to anybody, so it reads as
549    /// idle for longer than any claim can ask for and the next `XAUTOCLAIM`
550    /// takes it. `retry` is what the delivery count becomes, which is the whole
551    /// difference between the three words `XNACK` takes.
552    ///
553    /// Answers whether it was pending. The bookmark does not move, so a `>` read
554    /// will not hand it out again: releasing an entry offers it to a claim and
555    /// not to the group's next reader, which is Redis's behaviour and the only
556    /// one that keeps a released entry from being delivered twice over.
557    pub fn release(&mut self, id: Id, retry: Retry) -> bool {
558        let Some(nack) = self.pending.get_mut(&id) else {
559            return false;
560        };
561        let was = std::mem::replace(&mut nack.owner, Nack::NOBODY);
562        nack.count = retry.applied(nack.count);
563        // Zero rather than now, because the delivery time of an entry nobody
564        // holds is never read as a time: `XPENDING` reports minus one for it and
565        // `XINFO STREAM FULL` reports the zero.
566        nack.time = 0;
567        if was == Nack::NOBODY {
568            return true;
569        }
570        self.nacked += 1;
571        if let Some(Some(c)) = self.consumers.get_mut(was as usize) {
572            c.pending.remove(&id);
573        }
574        true
575    }
576
577    /// Make a released entry out of one that was not pending, which is
578    /// `XNACK ... FORCE`.
579    ///
580    /// The caller has to have checked that the entry is really in the stream,
581    /// for the same reason [`Group::force`] does. A count of zero is where a
582    /// released entry that has never been delivered starts, whatever word was
583    /// used, because there is no earlier count for `FAIL` to keep.
584    pub fn force_release(&mut self, id: Id, retry: Retry) {
585        if self.release(id, retry) {
586            return;
587        }
588        self.pending.insert(
589            id,
590            Nack {
591                time: 0,
592                count: retry.applied(0),
593                owner: Nack::NOBODY,
594            },
595        );
596        self.nacked += 1;
597    }
598
599    /// Drop a pending entry without it having been acknowledged.
600    ///
601    /// What happens to a NACK whose entry is no longer in the stream. `XCLAIM`
602    /// and `XAUTOCLAIM` both clear those out as they find them, because a
603    /// pending entry nobody can ever read is work no consumer can ever finish.
604    pub fn forget(&mut self, id: Id) -> bool {
605        self.ack(id)
606    }
607
608    /// Move an entry to another consumer, which is the middle of `XCLAIM`.
609    ///
610    /// `time` is when it should count as having been handed out, which is now
611    /// for a plain claim and something a caller chose for `IDLE` or `TIME`.
612    /// `count` replaces the delivery count when it is given, which is
613    /// `RETRYCOUNT`, and otherwise the count goes up by one unless `bump` says
614    /// not to, which is `JUSTID`.
615    ///
616    /// Answers false when the entry was not pending or the slot is empty.
617    pub fn claim(&mut self, id: Id, slot: u32, time: u64, count: Option<u64>, bump: bool) -> bool {
618        if !matches!(self.consumers.get(slot as usize), Some(Some(_))) {
619            return false;
620        }
621        let Some(nack) = self.pending.get_mut(&id) else {
622            return false;
623        };
624        let was = nack.owner;
625        nack.owner = slot;
626        nack.time = time;
627        if let Some(n) = count {
628            nack.count = n;
629        } else if bump {
630            nack.count += 1;
631        }
632        if was != slot {
633            if was == Nack::NOBODY {
634                // A claim is how a released entry gets an owner again, and it is
635                // the only way, since a `>` read never looks below the bookmark.
636                self.nacked -= 1;
637            } else if let Some(Some(c)) = self.consumers.get_mut(was as usize) {
638                c.pending.remove(&id);
639            }
640            if let Some(Some(c)) = self.consumers.get_mut(slot as usize) {
641                c.pending.insert(id);
642            }
643        }
644        true
645    }
646
647    /// Make a pending entry that was not pending, which is `XCLAIM FORCE`.
648    ///
649    /// The caller has to have checked that the entry is really in the stream,
650    /// because this cannot see the stream and creating a NACK for an entry that
651    /// is not there is exactly the state [`Group::forget`] exists to clean up.
652    pub fn force(&mut self, id: Id, slot: u32, time: u64, count: u64) -> bool {
653        let Some(Some(c)) = self.consumers.get_mut(slot as usize) else {
654            return false;
655        };
656        c.pending.insert(id);
657        self.pending.insert(
658            id,
659            Nack {
660                time,
661                count,
662                owner: slot,
663            },
664        );
665        true
666    }
667
668    /// Pending entries in `want`, oldest first.
669    ///
670    /// The callback answers whether to carry on, and gets `None` for the owner
671    /// of an entry that has been released, which `XPENDING` writes as an empty
672    /// name. A consumer filter never matches one of those, since asking what a
673    /// named consumer is holding is asking about entries that have an owner.
674    pub fn pending_range<F>(&self, want: Filter, now: u64, mut f: F) -> usize
675    where
676        F: FnMut(Id, &Nack, Option<&Consumer>) -> bool,
677    {
678        let mut seen = 0;
679        for (&id, nack) in self.pending.range(want.start..=want.end) {
680            if want.count.is_some_and(|n| seen >= n) {
681                break;
682            }
683            if want.owner.is_some_and(|c| Some(c) != nack.owner()) {
684                continue;
685            }
686            if nack.idle(now) < want.min_idle {
687                continue;
688            }
689            let who = match nack.owner() {
690                Some(slot) => match self.consumers.get(slot as usize) {
691                    Some(Some(c)) => Some(c),
692                    // A slot that has been emptied under a NACK, which deleting
693                    // a consumer cannot leave behind and nothing else can make.
694                    _ => continue,
695                },
696                None => None,
697            };
698            seen += 1;
699            if !f(id, nack, who) {
700                break;
701            }
702        }
703        seen
704    }
705
706    /// How many entries each consumer is holding, for the `XPENDING` summary.
707    pub fn pending_counts(&self) -> impl Iterator<Item = (&[u8], usize)> + '_ {
708        self.consumers
709            .iter()
710            .flatten()
711            .filter(|c| !c.pending.is_empty())
712            .map(|c| (c.name.as_slice(), c.pending.len()))
713    }
714
715    /// The IDs an `XAUTOCLAIM` would take, from `start` and idle at least
716    /// `min_idle`, and where a following call should carry on from.
717    ///
718    /// Only the scan, because deciding what to do with each one needs the
719    /// stream and this does not have it. The cursor is `None` when the scan
720    /// reached the end, which is the `0-0` Redis answers with.
721    #[must_use]
722    pub fn claimable(
723        &self,
724        start: Id,
725        min_idle: u64,
726        now: u64,
727        limit: usize,
728        out: &mut Vec<Id>,
729    ) -> Option<Id> {
730        // Redis charges attempts rather than hits, so a scan over a PEL full of
731        // entries that are not idle enough still ends and hands back a cursor
732        // instead of walking a million NACKs inside one command.
733        for (tried, (&id, nack)) in self.pending.range(start..).enumerate() {
734            if out.len() >= limit || tried >= limit * 10 {
735                return Some(id);
736            }
737            if nack.idle(now) >= min_idle {
738                out.push(id);
739            }
740        }
741        None
742    }
743
744    /// Write the group out as bytes, for [`super::Stream::freeze`].
745    ///
746    /// The consumer slots go out including the empty ones, because a slot is
747    /// emptied on deletion and never reused and a NACK names its owner by slot
748    /// number. Renumbering them on the way through would hand every pending
749    /// entry to the wrong consumer.
750    ///
751    /// What each consumer is holding is not written. It is a partition of the
752    /// pending list by owner, so it is rebuilt from the pending list on the way
753    /// back in and the two cannot come back disagreeing.
754    pub(super) fn freeze(&self, out: &mut Vec<u8>) {
755        frozen::put_uint(out, self.last.ms);
756        frozen::put_uint(out, self.last.seq);
757        put_opt(out, self.read);
758
759        frozen::put_uint(out, self.consumers.len() as u64);
760        for slot in &self.consumers {
761            match slot {
762                None => out.push(0),
763                Some(c) => {
764                    out.push(1);
765                    frozen::put_bytes(out, &c.name);
766                    frozen::put_uint(out, c.seen);
767                    put_opt(out, c.active);
768                }
769            }
770        }
771
772        frozen::put_uint(out, self.pending.len() as u64);
773        for (id, nack) in &self.pending {
774            frozen::put_uint(out, id.ms);
775            frozen::put_uint(out, id.seq);
776            frozen::put_uint(out, nack.time);
777            frozen::put_uint(out, nack.count);
778            // The owner goes out as itself rather than as an index with a spare
779            // value for nobody, because [`Nack::NOBODY`] already is one.
780            frozen::put_uint(out, u64::from(nack.owner));
781        }
782    }
783
784    /// Read back a group [`Group::freeze`] wrote.
785    pub(super) fn thaw(cut: &mut frozen::Cut<'_>) -> Result<Group, Broken> {
786        let last = Id::new(cut.uint()?, cut.uint()?);
787        let read = take_opt(cut)?;
788
789        let n = usize::try_from(cut.uint()?).map_err(|_| Broken::Short)?;
790        // A slot is a byte at the very least, so a count past what is left is a
791        // short body and not a reason to reserve that many.
792        if n > cut.rest().len() {
793            return Err(Broken::Short);
794        }
795        let mut consumers: Vec<Option<Consumer>> = Vec::with_capacity(n);
796        let mut names = HashSet::with_capacity(n);
797        for _ in 0..n {
798            match cut.byte()? {
799                0 => consumers.push(None),
800                1 => {
801                    let name = cut.bytes()?;
802                    // Two consumers under one name would make every lookup find
803                    // the first and leave the second unreachable, holding
804                    // entries nothing can claim back.
805                    if !names.insert(name) {
806                        return Err(Broken::Body);
807                    }
808                    consumers.push(Some(Consumer {
809                        name: name.to_vec(),
810                        seen: cut.uint()?,
811                        active: take_opt(cut)?,
812                        pending: BTreeSet::new(),
813                    }));
814                }
815                _ => return Err(Broken::Body),
816            }
817        }
818
819        let n = usize::try_from(cut.uint()?).map_err(|_| Broken::Short)?;
820        // Five numbers each, so one byte apiece is already generous.
821        if n > cut.rest().len() {
822            return Err(Broken::Short);
823        }
824        let mut group = Group {
825            last,
826            read,
827            pending: BTreeMap::new(),
828            consumers,
829            nacked: 0,
830        };
831        let mut prev = None;
832        for _ in 0..n {
833            let id = Id::new(cut.uint()?, cut.uint()?);
834            // Written in ID order out of a map, so anything else is bytes that
835            // did not come from `freeze`, and a repeat would silently drop an
836            // entry somebody is holding.
837            if prev.is_some_and(|p| p >= id) {
838                return Err(Broken::Body);
839            }
840            prev = Some(id);
841            let nack = Nack {
842                time: cut.uint()?,
843                count: cut.uint()?,
844                owner: u32::try_from(cut.uint()?).map_err(|_| Broken::Body)?,
845            };
846            if nack.owner == Nack::NOBODY {
847                group.nacked += 1;
848            } else {
849                match group.consumers.get_mut(nack.owner as usize) {
850                    Some(Some(c)) => {
851                        c.pending.insert(id);
852                    }
853                    // An owner that is off the end or an emptied slot would be
854                    // an entry held by a consumer that cannot be named, so
855                    // neither `XPENDING` nor a claim would ever reach it.
856                    _ => return Err(Broken::Body),
857                }
858            }
859            group.pending.insert(id, nack);
860        }
861        Ok(group)
862    }
863
864    /// How many bytes this group takes, not counting the struct itself.
865    ///
866    /// The pending map is counted per entry at the size of a key and a value
867    /// plus a share of the node around them, rather than exactly, because a
868    /// [`BTreeMap`] does not say how many nodes it has and the answer is only
869    /// ever read by `MEMORY USAGE` and the eviction total. A B-tree node here
870    /// holds eleven entries and some overhead, and a sixteenth of an entry is
871    /// close enough for both.
872    #[must_use]
873    pub fn memory_bytes(&self) -> usize {
874        let each = std::mem::size_of::<(Id, Nack)>();
875        let pending = self.pending.len() * (each + each / 16);
876        let consumers: usize = self
877            .consumers
878            .iter()
879            .map(|slot| {
880                std::mem::size_of::<Option<Consumer>>()
881                    + slot.as_ref().map_or(0, |c| {
882                        let each = std::mem::size_of::<Id>();
883                        c.name.capacity() + c.pending.len() * (each + each / 16)
884                    })
885            })
886            .sum();
887        pending + consumers
888    }
889}
890
891/// Append a count that may not be known, as a flag byte and then the number.
892///
893/// A byte rather than the usual trick of writing one more than the number and
894/// keeping zero for nothing, because both of the counts this is used for are a
895/// `u64` and adding one to the top of the range wraps. The flag costs a byte
896/// and is right everywhere.
897fn put_opt(out: &mut Vec<u8>, v: Option<u64>) {
898    match v {
899        None => out.push(0),
900        Some(n) => {
901            out.push(1);
902            frozen::put_uint(out, n);
903        }
904    }
905}
906
907/// Read back what [`put_opt`] wrote.
908fn take_opt(cut: &mut frozen::Cut<'_>) -> Result<Option<u64>, Broken> {
909    match cut.byte()? {
910        0 => Ok(None),
911        1 => Ok(Some(cut.uint()?)),
912        _ => Err(Broken::Body),
913    }
914}
915
916#[cfg(test)]
917mod tests {
918    use super::*;
919
920    fn group() -> Group {
921        Group::new(Id::MIN, Some(0))
922    }
923
924    #[test]
925    fn a_consumer_appears_by_turning_up() {
926        let mut g = group();
927        assert_eq!(g.slot(b"alice"), None);
928        let at = g.consumer_or_create(b"alice", 100);
929        assert_eq!(g.slot(b"alice"), Some(at));
930        assert_eq!(g.consumer_or_create(b"alice", 200), at);
931        assert_eq!(g.consumers().count(), 1);
932        assert_eq!(g.consumer(at).expect("alice").seen(), 200);
933    }
934
935    #[test]
936    fn creating_a_consumer_twice_says_so() {
937        let mut g = group();
938        assert!(g.create_consumer(b"alice", 1));
939        assert!(!g.create_consumer(b"alice", 2));
940    }
941
942    #[test]
943    fn delivering_moves_the_bookmark_and_fills_both_indexes() {
944        let mut g = group();
945        let a = g.consumer_or_create(b"alice", 10);
946        assert!(g.deliver(a, Id::new(5, 0), 10));
947        assert!(g.deliver(a, Id::new(7, 0), 12));
948
949        assert_eq!(g.last_id(), Id::new(7, 0));
950        assert_eq!(g.pending_len(), 2);
951        assert_eq!(
952            g.consumer(a).expect("alice").pending().collect::<Vec<_>>(),
953            vec![Id::new(5, 0), Id::new(7, 0)]
954        );
955        let nack = g.nack(Id::new(5, 0)).expect("a nack");
956        assert_eq!((nack.count(), nack.time()), (1, 10));
957    }
958
959    #[test]
960    fn acking_takes_it_out_of_both_indexes() {
961        let mut g = group();
962        let a = g.consumer_or_create(b"alice", 1);
963        g.deliver(a, Id::new(5, 0), 1);
964        assert!(g.ack(Id::new(5, 0)));
965        assert_eq!(g.pending_len(), 0);
966        assert!(g.consumer(a).expect("alice").is_empty());
967        // Twice is not an error, because a consumer that crashed after doing the
968        // work and before sending the ack will send it again.
969        assert!(!g.ack(Id::new(5, 0)));
970    }
971
972    #[test]
973    fn acking_does_not_move_the_bookmark() {
974        let mut g = group();
975        let a = g.consumer_or_create(b"alice", 1);
976        g.deliver(a, Id::new(5, 0), 1);
977        g.ack(Id::new(5, 0));
978        assert_eq!(g.last_id(), Id::new(5, 0));
979    }
980
981    #[test]
982    fn a_claim_moves_it_between_consumers() {
983        let mut g = group();
984        let a = g.consumer_or_create(b"alice", 1);
985        let b = g.consumer_or_create(b"bob", 1);
986        g.deliver(a, Id::new(5, 0), 100);
987
988        assert!(g.claim(Id::new(5, 0), b, 500, None, true));
989        assert!(g.consumer(a).expect("alice").is_empty());
990        assert_eq!(
991            g.consumer(b).expect("bob").pending().collect::<Vec<_>>(),
992            vec![Id::new(5, 0)]
993        );
994        let nack = g.nack(Id::new(5, 0)).expect("a nack");
995        assert_eq!((nack.count(), nack.time()), (2, 500));
996    }
997
998    #[test]
999    fn a_claim_that_does_not_bump_is_justid() {
1000        let mut g = group();
1001        let a = g.consumer_or_create(b"alice", 1);
1002        let b = g.consumer_or_create(b"bob", 1);
1003        g.deliver(a, Id::new(5, 0), 100);
1004        g.claim(Id::new(5, 0), b, 500, None, false);
1005        assert_eq!(g.nack(Id::new(5, 0)).expect("a nack").count(), 1);
1006    }
1007
1008    #[test]
1009    fn a_retry_count_replaces_rather_than_adds() {
1010        let mut g = group();
1011        let a = g.consumer_or_create(b"alice", 1);
1012        g.deliver(a, Id::new(5, 0), 100);
1013        g.claim(Id::new(5, 0), a, 500, Some(9), true);
1014        assert_eq!(g.nack(Id::new(5, 0)).expect("a nack").count(), 9);
1015    }
1016
1017    #[test]
1018    fn claiming_back_to_the_same_consumer_keeps_it_there() {
1019        let mut g = group();
1020        let a = g.consumer_or_create(b"alice", 1);
1021        g.deliver(a, Id::new(5, 0), 100);
1022        assert!(g.claim(Id::new(5, 0), a, 500, None, true));
1023        assert_eq!(
1024            g.consumer(a).expect("alice").pending().collect::<Vec<_>>(),
1025            vec![Id::new(5, 0)]
1026        );
1027    }
1028
1029    #[test]
1030    fn nothing_pending_cannot_be_claimed_without_force() {
1031        let mut g = group();
1032        let a = g.consumer_or_create(b"alice", 1);
1033        assert!(!g.claim(Id::new(5, 0), a, 500, None, true));
1034        assert!(g.force(Id::new(5, 0), a, 500, 1));
1035        assert_eq!(g.pending_len(), 1);
1036    }
1037
1038    #[test]
1039    fn deleting_a_consumer_gives_up_its_work() {
1040        let mut g = group();
1041        let a = g.consumer_or_create(b"alice", 1);
1042        let b = g.consumer_or_create(b"bob", 1);
1043        g.deliver(a, Id::new(5, 0), 1);
1044        g.deliver(a, Id::new(6, 0), 1);
1045        g.deliver(b, Id::new(7, 0), 1);
1046
1047        assert_eq!(g.delete_consumer(b"alice"), 2);
1048        assert_eq!(g.pending_len(), 1);
1049        assert!(g.nack(Id::new(5, 0)).is_none());
1050        assert!(g.nack(Id::new(7, 0)).is_some());
1051        assert_eq!(g.delete_consumer(b"alice"), 0);
1052        // The bookmark is untouched, so the entries are not handed out again.
1053        assert_eq!(g.last_id(), Id::new(7, 0));
1054    }
1055
1056    #[test]
1057    fn a_deleted_slot_is_not_reused() {
1058        let mut g = group();
1059        let a = g.consumer_or_create(b"alice", 1);
1060        g.delete_consumer(b"alice");
1061        let b = g.consumer_or_create(b"bob", 1);
1062        assert_ne!(a, b);
1063        assert_eq!(g.consumers().count(), 1);
1064    }
1065
1066    #[test]
1067    fn idle_is_measured_from_the_last_hand_out() {
1068        let mut g = group();
1069        let a = g.consumer_or_create(b"alice", 1);
1070        g.deliver(a, Id::new(5, 0), 1_000);
1071        assert_eq!(g.nack(Id::new(5, 0)).expect("a nack").idle(4_000), 3_000);
1072        // A time set into the future is something XCLAIM TIME allows, and it is
1073        // not idle rather than idle by a negative amount.
1074        assert_eq!(g.nack(Id::new(5, 0)).expect("a nack").idle(500), 0);
1075    }
1076
1077    #[test]
1078    fn the_summary_is_the_two_ends_and_the_counts() {
1079        let mut g = group();
1080        let a = g.consumer_or_create(b"alice", 1);
1081        let b = g.consumer_or_create(b"bob", 1);
1082        for ms in [3u64, 5, 9] {
1083            g.deliver(a, Id::new(ms, 0), 1);
1084        }
1085        g.deliver(b, Id::new(11, 0), 1);
1086
1087        assert_eq!(g.pending_bounds(), Some((Id::new(3, 0), Id::new(11, 0))));
1088        let counts: Vec<_> = g
1089            .pending_counts()
1090            .map(|(n, c)| (String::from_utf8_lossy(n).into_owned(), c))
1091            .collect();
1092        assert_eq!(counts, vec![("alice".into(), 3), ("bob".into(), 1)]);
1093    }
1094
1095    #[test]
1096    fn a_consumer_with_nothing_is_left_out_of_the_summary() {
1097        let mut g = group();
1098        g.consumer_or_create(b"alice", 1);
1099        assert_eq!(g.pending_counts().count(), 0);
1100        assert_eq!(g.pending_bounds(), None);
1101    }
1102
1103    #[test]
1104    fn the_pending_range_takes_both_ends_and_the_filters() {
1105        let mut g = group();
1106        let a = g.consumer_or_create(b"alice", 1);
1107        let b = g.consumer_or_create(b"bob", 1);
1108        g.deliver(a, Id::new(3, 0), 100);
1109        g.deliver(b, Id::new(5, 0), 100);
1110        g.deliver(a, Id::new(9, 0), 900);
1111
1112        let seen = |g: &Group, start, end, owner, idle| {
1113            let mut out = Vec::new();
1114            let want = Filter {
1115                start,
1116                end,
1117                owner,
1118                min_idle: idle,
1119                ..Filter::default()
1120            };
1121            g.pending_range(want, 1_000, |id, _, c| {
1122                out.push((
1123                    id,
1124                    String::from_utf8_lossy(c.expect("an owner").name()).into_owned(),
1125                ));
1126                true
1127            });
1128            out
1129        };
1130
1131        assert_eq!(seen(&g, Id::MIN, Id::MAX, None, 0).len(), 3);
1132        assert_eq!(seen(&g, Id::new(4, 0), Id::new(9, 0), None, 0).len(), 2);
1133        assert_eq!(
1134            seen(&g, Id::MIN, Id::MAX, Some(a), 0),
1135            vec![
1136                (Id::new(3, 0), "alice".into()),
1137                (Id::new(9, 0), "alice".into())
1138            ]
1139        );
1140        // Only the two handed out at 100 have been sitting 500 milliseconds.
1141        assert_eq!(seen(&g, Id::MIN, Id::MAX, None, 500).len(), 2);
1142    }
1143
1144    #[test]
1145    fn a_count_stops_the_pending_range() {
1146        let mut g = group();
1147        let a = g.consumer_or_create(b"alice", 1);
1148        for ms in 1..=10u64 {
1149            g.deliver(a, Id::new(ms, 0), 1);
1150        }
1151        let mut out = Vec::new();
1152        let want = Filter {
1153            count: Some(4),
1154            ..Filter::default()
1155        };
1156        let seen = g.pending_range(want, 1, |id, _, _| {
1157            out.push(id);
1158            true
1159        });
1160        assert_eq!((seen, out.len()), (4, 4));
1161    }
1162
1163    #[test]
1164    fn the_callback_can_stop_the_pending_range() {
1165        let mut g = group();
1166        let a = g.consumer_or_create(b"alice", 1);
1167        for ms in 1..=10u64 {
1168            g.deliver(a, Id::new(ms, 0), 1);
1169        }
1170        let mut out = Vec::new();
1171        g.pending_range(Filter::default(), 1, |id, _, _| {
1172            out.push(id);
1173            out.len() < 3
1174        });
1175        assert_eq!(out.len(), 3);
1176    }
1177
1178    #[test]
1179    fn claimable_takes_the_idle_ones_and_says_where_to_carry_on() {
1180        let mut g = group();
1181        let a = g.consumer_or_create(b"alice", 1);
1182        for ms in 1..=10u64 {
1183            g.deliver(a, Id::new(ms, 0), if ms <= 5 { 100 } else { 900 });
1184        }
1185        let mut out = Vec::new();
1186        let cursor = g.claimable(Id::MIN, 500, 1_000, 100, &mut out);
1187        assert_eq!(cursor, None, "the scan reached the end");
1188        assert_eq!(out, (1..=5).map(|ms| Id::new(ms, 0)).collect::<Vec<_>>());
1189
1190        // A limit hands back where the next call starts.
1191        out.clear();
1192        let cursor = g.claimable(Id::MIN, 500, 1_000, 3, &mut out);
1193        assert_eq!(out.len(), 3);
1194        assert_eq!(cursor, Some(Id::new(4, 0)));
1195    }
1196
1197    /// The read counter is set from outside and a delivery does not touch it.
1198    ///
1199    /// It looks like something the group should keep for itself, and it is not:
1200    /// what a delivery does to it depends on whether anything has been deleted
1201    /// ahead of the entry being handed over, which is a fact about the stream.
1202    /// The rule lives in [`crate::stream::Stream::read_group`] and this only
1203    /// holds the number.
1204    #[test]
1205    fn a_delivery_leaves_the_read_counter_to_the_stream() {
1206        let mut g = group();
1207        let a = g.consumer_or_create(b"alice", 1);
1208        g.deliver(a, Id::new(1, 0), 1);
1209        assert_eq!(g.entries_read(), Some(0), "the group did not count it");
1210
1211        g.set_read(Some(1));
1212        assert_eq!(g.entries_read(), Some(1));
1213        g.set_read(None);
1214        assert_eq!(g.entries_read(), None, "and it can be given up on");
1215    }
1216
1217    #[test]
1218    fn setting_the_id_leaves_the_pending_list_alone() {
1219        let mut g = group();
1220        let a = g.consumer_or_create(b"alice", 1);
1221        g.deliver(a, Id::new(5, 0), 1);
1222        g.set_id(Id::MIN, Some(0));
1223        assert_eq!(g.last_id(), Id::MIN);
1224        assert_eq!(g.pending_len(), 1, "somebody is still holding it");
1225    }
1226
1227    /// A released entry is pending, owned by nobody, and idle for ever.
1228    #[test]
1229    fn releasing_takes_the_entry_out_of_the_consumers_hands() {
1230        let mut g = group();
1231        let a = g.consumer_or_create(b"alice", 1);
1232        g.deliver(a, Id::new(5, 0), 100);
1233
1234        assert!(g.release(Id::new(5, 0), Retry::Keep));
1235        assert_eq!(g.pending_len(), 1, "it is still the group's problem");
1236        assert_eq!(
1237            g.consumer(a).expect("alice").pending().count(),
1238            0,
1239            "and no longer alice's"
1240        );
1241        let nack = g.nack(Id::new(5, 0)).expect("a nack");
1242        assert_eq!(nack.owner(), None);
1243        assert_eq!(nack.count(), 1, "Keep left the count where it was");
1244        // Idle for longer than any min-idle-time a claim can name, which is what
1245        // puts it at the front of the next sweep.
1246        assert_eq!(nack.idle(100), u64::MAX);
1247        let mut out = Vec::new();
1248        assert_eq!(g.claimable(Id::MIN, u64::MAX, 100, 10, &mut out), None);
1249        assert_eq!(out, vec![Id::new(5, 0)]);
1250
1251        // Releasing it again is still true and does not count it twice.
1252        assert!(g.release(Id::new(5, 0), Retry::Keep));
1253        assert_eq!(g.nacked_len(), 1);
1254        // And nothing pending is false, however it is asked.
1255        assert!(!g.release(Id::new(9, 0), Retry::Keep));
1256    }
1257
1258    #[test]
1259    fn the_three_words_differ_only_in_the_delivery_count() {
1260        let count = |retry| {
1261            let mut g = group();
1262            let a = g.consumer_or_create(b"alice", 1);
1263            g.deliver(a, Id::new(5, 0), 1);
1264            g.claim(Id::new(5, 0), a, 2, None, true);
1265            g.release(Id::new(5, 0), retry);
1266            g.nack(Id::new(5, 0)).expect("a nack").count()
1267        };
1268        assert_eq!(count(Retry::Down), 1, "one off, not back to nothing");
1269        assert_eq!(count(Retry::Keep), 2);
1270        assert_eq!(count(Retry::Max), i64::MAX as u64);
1271        assert_eq!(count(Retry::At(7)), 7);
1272    }
1273
1274    /// Forcing makes the pending entry when there is not one, and does not make
1275    /// a second one when there is.
1276    #[test]
1277    fn forcing_a_release_is_the_same_call_twice() {
1278        let mut g = group();
1279        g.force_release(Id::new(5, 0), Retry::Keep);
1280        assert_eq!(g.pending_len(), 1);
1281        assert_eq!(
1282            g.nack(Id::new(5, 0)).expect("a nack").count(),
1283            0,
1284            "there was no earlier count to keep"
1285        );
1286
1287        g.force_release(Id::new(5, 0), Retry::At(4));
1288        assert_eq!(g.pending_len(), 1);
1289        assert_eq!(g.nacked_len(), 1);
1290        assert_eq!(g.nack(Id::new(5, 0)).expect("a nack").count(), 4);
1291    }
1292
1293    /// The counter behind `XINFO STREAM FULL`'s `nacked-count`, which is a field
1294    /// and not a walk, so every line that moves an entry on or off `NOBODY` has
1295    /// to keep it right. This is the walk, run against the field.
1296    #[test]
1297    fn the_nacked_count_matches_a_full_scan() {
1298        let mut g = group();
1299        let a = g.consumer_or_create(b"alice", 1);
1300        let b = g.consumer_or_create(b"bob", 1);
1301        for ms in 1..=6u64 {
1302            g.deliver(a, Id::new(ms, 0), 100);
1303        }
1304
1305        let scan = |g: &Group| {
1306            (1..=9u64)
1307                .filter(|&ms| g.nack(Id::new(ms, 0)).is_some_and(|n| n.owner().is_none()))
1308                .count()
1309        };
1310        let agrees = |g: &Group| assert_eq!(g.nacked_len(), scan(g), "the field drifted");
1311
1312        agrees(&g);
1313        g.release(Id::new(1, 0), Retry::Keep);
1314        g.release(Id::new(2, 0), Retry::Keep);
1315        agrees(&g);
1316
1317        // A claim takes one back into somebody's hands.
1318        g.claim(Id::new(1, 0), b, 200, None, true);
1319        agrees(&g);
1320        // An ack takes one out of the list altogether, released or not.
1321        assert!(g.ack(Id::new(2, 0)));
1322        assert!(g.ack(Id::new(3, 0)));
1323        agrees(&g);
1324        // And a forced release on an entry nobody was ever handed.
1325        g.force_release(Id::new(9, 0), Retry::Down);
1326        agrees(&g);
1327        assert_eq!(g.nacked_len(), 1);
1328    }
1329
1330    /// A consumer filter skips released entries, because a released entry has no
1331    /// consumer to match and `XPENDING key group - + n consumer` is a question
1332    /// about one consumer's work.
1333    #[test]
1334    fn a_released_entry_is_not_anybodys_pending_work() {
1335        let mut g = group();
1336        let a = g.consumer_or_create(b"alice", 1);
1337        g.deliver(a, Id::new(3, 0), 100);
1338        g.deliver(a, Id::new(5, 0), 100);
1339        g.release(Id::new(3, 0), Retry::Keep);
1340
1341        let seen = |g: &Group, owner| {
1342            let mut out = Vec::new();
1343            let want = Filter {
1344                owner,
1345                ..Filter::default()
1346            };
1347            g.pending_range(want, 1_000, |id, _, c| {
1348                out.push((id, c.map(|c| c.name().to_vec())));
1349                true
1350            });
1351            out
1352        };
1353
1354        assert_eq!(
1355            seen(&g, None),
1356            vec![
1357                (Id::new(3, 0), None),
1358                (Id::new(5, 0), Some(b"alice".to_vec()))
1359            ]
1360        );
1361        assert_eq!(
1362            seen(&g, Some(a)),
1363            vec![(Id::new(5, 0), Some(b"alice".to_vec()))]
1364        );
1365        // The summary counts it against nobody, so alice is down to one.
1366        assert_eq!(
1367            g.pending_counts()
1368                .map(|(name, n)| (name.to_vec(), n))
1369                .collect::<Vec<_>>(),
1370            vec![(b"alice".to_vec(), 1)]
1371        );
1372    }
1373
1374    #[test]
1375    fn a_frozen_group_with_a_pending_entry_nobody_could_hold_is_refused() {
1376        let mut g = group();
1377        let slot = g.consumer_or_create(b"alice", 1_000);
1378        g.deliver(slot, Id::new(1, 0), 1_000);
1379        let mut bytes = Vec::new();
1380        g.freeze(&mut bytes);
1381        assert_eq!(Group::thaw(&mut frozen::Cut::new(&bytes)), Ok(g));
1382
1383        // The owner is the last number in the body, and a slot past the end
1384        // would be an entry no consumer could ever be told to finish.
1385        let mut bad = bytes.clone();
1386        *bad.last_mut().expect("a body") = 7;
1387        assert_eq!(Group::thaw(&mut frozen::Cut::new(&bad)), Err(Broken::Body));
1388
1389        for cut in 0..bytes.len() {
1390            assert!(
1391                Group::thaw(&mut frozen::Cut::new(&bytes[..cut])).is_err(),
1392                "cut at {cut}"
1393            );
1394        }
1395    }
1396
1397    #[test]
1398    fn a_frozen_group_that_names_one_consumer_twice_is_refused() {
1399        let mut g = group();
1400        g.consumer_or_create(b"alice", 1_000);
1401        g.consumer_or_create(b"carol", 1_000);
1402        let mut bytes = Vec::new();
1403        g.freeze(&mut bytes);
1404        // Both names are five letters, so one name becomes the other without
1405        // the length in front of it moving.
1406        let at = bytes
1407            .windows(5)
1408            .position(|w| w == b"carol")
1409            .expect("the second name");
1410        bytes[at..at + 5].copy_from_slice(b"alice");
1411        assert_eq!(
1412            Group::thaw(&mut frozen::Cut::new(&bytes)),
1413            Err(Broken::Body)
1414        );
1415    }
1416}