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}