1use crate::{Error, Result, Timescale, Timestamp, coding};
17use crate::{broadcast, cache, group, stats};
18
19use super::{Datagram, Requests};
20
21use super::Cap;
22use super::resume::Successor;
23pub use super::subscription::{Position, Subscription};
24
25use std::{
26 collections::{BTreeMap, HashSet, VecDeque},
27 ops::{Bound, RangeBounds},
28 sync::Arc,
29 sync::OnceLock,
30 sync::atomic::{AtomicBool, Ordering},
31 task::{Poll, ready},
32 time::Duration,
33};
34
35pub const DEFAULT_MAX_AGE: Duration = Duration::from_secs(5);
37
38const DEFAULT_PRIORITY: u8 = 127;
41
42const MAX_DATAGRAMS: usize = 64;
48
49const EVICT_SLACK: usize = 64;
52
53const EVICT_SCAN: usize = 4;
57
58#[derive(Clone, Copy)]
60pub(super) struct ExpiryScan {
61 start: usize,
62 width: usize,
65 now: u64,
66 max_ticks: u64,
67 gc: bool,
69}
70
71#[derive(Clone, Debug)]
82#[non_exhaustive]
83pub struct Info {
84 pub timescale: Timescale,
91 pub max_age: Duration,
111 pub priority: u8,
115}
116
117impl Default for Info {
118 fn default() -> Self {
119 Self {
120 timescale: Timescale::default(),
121 max_age: DEFAULT_MAX_AGE,
122 priority: DEFAULT_PRIORITY,
123 }
124 }
125}
126
127impl Info {
128 pub fn with_timescale(mut self, timescale: Timescale) -> Self {
133 self.timescale = timescale;
134 self
135 }
136
137 pub fn with_max_age(mut self, max_age: Duration) -> Self {
139 self.max_age = max_age;
140 self
141 }
142
143 pub fn with_priority(mut self, priority: u8) -> Self {
145 self.priority = priority;
146 self
147 }
148}
149
150#[derive(Default)]
151pub(crate) struct TrackState {
152 info: Option<Info>,
155 published: bool,
158 claimed: bool,
161
162 broadcast: Arc<broadcast::Info>,
165
166 cache: Arc<cache::Track>,
170
171 lookup: BTreeMap<u64, Slot>,
177
178 arrival: VecDeque<(u64, u32)>,
183
184 evict: VecDeque<(u64, u32)>,
192
193 debt: u64,
198
199 datagrams: VecDeque<Datagram>,
202
203 datagram_offset: usize,
206
207 offset: usize,
210
211 max_sequence: Option<u64>,
214
215 latest_group: Option<u64>,
221
222 next_stamp: u32,
224
225 final_sequence: Option<u64>,
227
228 sealed: bool,
231
232 closed: bool,
236
237 start_sequence: Option<u64>,
241
242 start_pending: bool,
247
248 resume: Option<Position>,
252
253 abort: Option<Error>,
255
256 settled: bool,
259
260 subscriptions: kio::Shared<Subscriptions>,
264
265 fetch: kio::Shared<FetchState>,
268}
269
270struct Slot {
276 group: group::Producer,
277
278 stamp: u32,
283
284 visible: bool,
287
288 protection: Option<cache::Protection>,
291}
292
293pub(crate) const CACHE_OVERHEAD: u64 = 2 * (size_of::<u64>() + size_of::<Slot>() + 2 * size_of::<(u64, u32)>()) as u64;
302
303type Subscriptions = Vec<kio::Consumer<Subscription>>;
305
306pub(crate) type FetchState = Requests<u64, PendingFetch>;
311
312pub(crate) struct PendingFetch {
314 priority: u8,
316
317 frame_start: u64,
321
322 result: kio::Producer<FetchOutcome>,
327}
328
329#[derive(Default)]
332pub(crate) struct FetchOutcome {
333 pub(crate) rejected: Option<Error>,
334}
335
336impl TrackState {
337 fn normalize_info(broadcast: &broadcast::Info, mut info: Info) -> Info {
338 info.max_age = info.max_age.min(broadcast.cache_duration);
339 info
340 }
341
342 fn accept(&mut self, info: Info) {
343 self.published = true;
344 self.install(info);
345 }
346
347 fn poll_info(&self) -> Poll<Result<Info>> {
348 if let Some(info) = &self.info {
349 Poll::Ready(Ok(info.clone()))
350 } else if let Some(err) = &self.abort {
351 Poll::Ready(Err(err.clone()))
354 } else {
355 Poll::Pending
356 }
357 }
358
359 fn poll_recv_group(&self, index: usize, min_sequence: u64) -> Poll<Result<Option<(group::Producer, usize)>>> {
365 let start = index.saturating_sub(self.offset);
366 for (i, (sequence, stamp)) in self.arrival.iter().enumerate().skip(start) {
367 if *sequence >= min_sequence
368 && let Some(slot) = self.lookup.get(sequence)
369 && slot.stamp == *stamp
370 && !slot.group.is_aborted()
371 {
372 return Poll::Ready(Ok(Some((slot.group.clone(), self.offset + i))));
373 }
374 }
375
376 if self.is_complete() {
378 Poll::Ready(Ok(None))
379 } else if let Some(err) = &self.abort {
380 Poll::Ready(Err(err.clone()))
381 } else {
382 Poll::Pending
383 }
384 }
385
386 fn poll_recv_datagram(&self, index: usize) -> Poll<Result<Option<(Datagram, usize)>>> {
392 let start = index.saturating_sub(self.datagram_offset);
393 if let Some(datagram) = self.datagrams.get(start) {
394 return Poll::Ready(Ok(Some((datagram.clone(), self.datagram_offset + start))));
395 }
396
397 if self.is_complete() {
399 Poll::Ready(Ok(None))
400 } else if let Some(err) = &self.abort {
401 Poll::Ready(Err(err.clone()))
402 } else {
403 Poll::Pending
404 }
405 }
406
407 fn push_datagram(&mut self, datagram: Datagram) {
409 if self.datagrams.len() == MAX_DATAGRAMS {
410 self.datagrams.pop_front();
411 self.datagram_offset += 1;
412 }
413 self.datagrams.push_back(datagram);
414 }
415
416 fn poll_next_in_range(
426 &self,
427 next_sequence: u64,
428 end_sequence: Option<u64>,
429 ) -> Poll<Result<Option<group::Producer>>> {
430 let closed = self.sealed || self.abort.is_some();
434 let end = || match &self.abort {
437 Some(err) if !self.settled => Err(err.clone()),
438 _ => Ok(None),
439 };
440
441 if let Some(cap) = end_sequence
446 && cap <= next_sequence
447 {
448 if closed {
449 return Poll::Ready(end());
450 }
451 return Poll::Pending;
452 }
453
454 let best = self
455 .lookup
456 .range(next_sequence..)
457 .map(|(_, slot)| &slot.group)
458 .take_while(|group| super::subscription::before_end(group.sequence, end_sequence))
459 .find(|group| !group.is_aborted());
460
461 if let Some(group) = best {
462 return Poll::Ready(Ok(Some(group.clone())));
467 }
468
469 if closed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) {
477 return Poll::Ready(end());
478 }
479 Poll::Pending
480 }
481
482 fn covering_group(&self, sequence: u64, frame_start: u64) -> Option<&group::Producer> {
489 let slot = self.lookup.get(&sequence)?;
490 let first = slot.group.live_first_frame()?;
491 (first as u64 <= frame_start).then_some(&slot.group)
492 }
493
494 fn max_age_bound(&self) -> Option<Duration> {
497 self.info.as_ref().map(|info| info.max_age)
498 }
499
500 fn live_edge(&self, cap: Option<u64>) -> Option<PresentationEdge> {
515 self.lookup
516 .range(..)
517 .rev()
518 .filter(|(seq, _)| super::subscription::before_end(**seq, cap))
519 .find_map(|(_, slot)| {
520 if !slot.visible || slot.group.is_aborted() {
521 return None;
522 }
523 let timestamp = slot.group.timestamp()?;
526 Some(PresentationEdge {
527 sequence: slot.group.sequence,
528 stamp: slot.stamp,
529 timestamp: slot.group.latest().unwrap_or(timestamp),
530 })
531 })
532 }
533
534 fn drift_edge(&self, cap: Option<u64>, outer: Option<(u64, Timestamp)>, successor: Option<Timestamp>) -> Edge {
538 Edge {
539 presentation: self.live_edge(cap),
540 outer,
541 cap,
542 successor,
543 }
544 }
545
546 fn holds(&self, sequence: u64, stamp: u32) -> bool {
548 self.lookup
549 .get(&sequence)
550 .is_some_and(|slot| slot.stamp == stamp && !slot.group.is_aborted())
551 }
552
553 fn reach(&self, sequence: u64, cap: Option<u64>, beyond: Option<Timestamp>) -> Option<Timestamp> {
568 match self.first_servable(sequence.saturating_add(1), cap) {
569 Some(group) => group.timestamp(),
570 None => beyond,
571 }
572 }
573
574 fn first_servable(&self, from: u64, cap: Option<u64>) -> Option<&group::Producer> {
576 self.lookup
577 .range(from..)
578 .map(|(_, slot)| slot)
579 .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap))
580 .find(|slot| slot.visible && !slot.group.is_aborted())
581 .map(|slot| &slot.group)
582 }
583
584 fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool {
611 if !self.lookup.contains_key(&sequence) {
612 return false;
613 }
614
615 let local = edge
619 .presentation
620 .filter(|live| self.holds(live.sequence, live.stamp))
621 .map(|live| (live.sequence, live.timestamp));
622 let Some((_, timestamp)) = local
626 .into_iter()
627 .chain(edge.outer)
628 .filter(|(live, _)| *live > sequence)
629 .max_by_key(|(live, _)| *live)
630 else {
631 return false;
632 };
633 self.reach(sequence, edge.cap, edge.successor)
634 .is_some_and(|reach| matches!(timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget))
635 }
636
637 fn poll_fetch_cached(&self, sequence: u64, frame_start: u64) -> Poll<Result<group::Consumer>> {
642 if let Some(group) = self.covering_group(sequence, frame_start)
645 && let Some(consumer) = group.try_consume()
646 {
647 group.cache_refresh();
651 return Poll::Ready(Ok(consumer));
652 }
653
654 if let Some(err) = &self.abort {
655 return Poll::Ready(Err(err.clone()));
656 }
657
658 if self.final_sequence.is_some_and(|fin| sequence >= fin) {
660 return Poll::Ready(Err(Error::NotFound));
661 }
662
663 Poll::Pending
664 }
665
666 pub(super) fn evict_expired(&mut self) {
681 let scan = self.expiry_scan();
682 self.evict_expired_scan(scan);
683 }
684
685 pub(super) fn expiry_scan(&self) -> ExpiryScan {
687 ExpiryScan {
688 start: self.cache.next_expiry_scan(EVICT_SCAN),
689 width: EVICT_SCAN,
690 now: self.cache.pool().now(),
691 max_ticks: self.cache.pool().expiry_ticks(),
692 gc: false,
693 }
694 }
695
696 pub(super) fn expiry_scan_drain(&self) -> ExpiryScan {
699 ExpiryScan {
700 start: 0,
701 width: self.evict.len(),
702 now: self.cache.pool().now(),
703 max_ticks: self.cache.pool().expiry_ticks(),
704 gc: true,
705 }
706 }
707
708 #[cfg(test)]
709 pub(super) fn date_cache_accesses(&self, now: u64) {
710 for slot in self.lookup.values() {
711 slot.group.cache_accessed_tick(Some(now));
712 }
713 }
714
715 pub(super) fn expiry_mutation_due(&self, scan: ExpiryScan) -> bool {
720 let len = self.evict.len();
721 if len > 0 {
722 let start = scan.start % len;
723 let mut retained = 0;
724 for step in 0..len.min(scan.width) {
725 let (sequence, stamp) = self.evict[(start + step) % len];
726 let Some(slot) = self.lookup.get(&sequence) else {
727 continue;
728 };
729 if slot.stamp != stamp {
730 continue;
731 }
732 if slot.group.is_aborted()
733 || (!self.protects(sequence)
734 && slot
735 .group
736 .cache_accessed_tick(scan.gc.then_some(scan.now))
737 .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks))
738 {
739 return true;
740 }
741 retained += 1;
742 if !scan.gc && retained >= EVICT_SCAN {
743 break;
744 }
745 }
746 }
747
748 self.arrival
749 .front()
750 .is_some_and(|(sequence, stamp)| !self.is_current(*sequence, *stamp))
751 || self
752 .evict
753 .front()
754 .is_some_and(|(sequence, stamp)| !self.is_current(*sequence, *stamp))
755 || self.evict.len() > 2 * self.lookup.len() + EVICT_SLACK
756 }
757
758 pub(super) fn expire_closed(&self, scan: ExpiryScan) {
764 for (sequence, stamp) in &self.evict {
765 let Some(slot) = self.lookup.get(sequence) else {
766 continue;
767 };
768 if slot.stamp == *stamp
769 && !slot.group.is_aborted()
770 && !slot.group.cache_protected()
771 && slot
772 .group
773 .cache_accessed_tick(scan.gc.then_some(scan.now))
774 .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks)
775 {
776 let _ = slot.group.clone().abort(Error::Old);
777 }
778 }
779 }
780
781 pub(super) fn evict_expired_scan(&mut self, scan: ExpiryScan) {
784 let len = self.evict.len();
785 if len > 0 {
786 let start = scan.start % len;
787 let mut retained = 0;
788 for step in 0..len.min(scan.width) {
789 let (sequence, stamp) = self.evict[(start + step) % len];
790 let Some(slot) = self.lookup.get(&sequence) else {
791 continue;
792 };
793 if slot.stamp != stamp {
794 continue;
796 }
797 if slot.group.is_aborted() {
800 self.lookup.remove(&sequence);
801 continue;
802 }
803 if self.protects(sequence)
804 || slot
805 .group
806 .cache_accessed_tick(scan.gc.then_some(scan.now))
807 .is_none_or(|tick| scan.now.saturating_sub(tick) <= scan.max_ticks)
808 {
809 retained += 1;
812 if !scan.gc && retained >= EVICT_SCAN {
813 break;
814 }
815 continue;
816 }
817 let slot = self.lookup.remove(&sequence).unwrap();
821 let _ = slot.group.abort(Error::Old);
822 }
823 }
824
825 while let Some((sequence, stamp)) = self.arrival.front() {
828 if self.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp) {
829 break;
830 }
831 self.arrival.pop_front();
832 self.offset += 1;
833 }
834
835 while let Some((sequence, stamp)) = self.evict.front() {
837 if self.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp) {
838 break;
839 }
840 self.evict.pop_front();
841 }
842
843 if self.evict.len() > 2 * self.lookup.len() + EVICT_SLACK {
846 let lookup = &self.lookup;
847 self.evict
848 .retain(|(sequence, stamp)| lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp));
849 }
850 }
851
852 fn is_current(&self, sequence: u64, stamp: u32) -> bool {
854 self.lookup.get(&sequence).is_some_and(|slot| slot.stamp == stamp)
855 }
856
857 fn protects(&self, sequence: u64) -> bool {
860 !self.closed && Some(sequence) == self.latest_group
861 }
862
863 fn close_cache(&mut self) {
867 if std::mem::replace(&mut self.closed, true) {
868 return;
869 }
870 if let Some(latest) = self.latest_group
871 && let Some(slot) = self.lookup.get_mut(&latest)
872 {
873 slot.group.cache_demote();
874 slot.protection = None;
875 self.evict.push_back((latest, slot.stamp));
876 }
877 }
878
879 fn drop_open_groups(&mut self) {
883 self.lookup.retain(|_, slot| slot.group.is_finished());
884 }
885
886 fn install(&mut self, info: Info) {
892 let info = Self::normalize_info(&self.broadcast, info);
893 self.info = Some(info);
894 }
895
896 fn spawn(broadcast: Arc<broadcast::Info>) -> kio::Producer<Self> {
903 let state = kio::Producer::new(Self {
904 broadcast: broadcast.clone(),
905 ..Default::default()
906 });
907 let cache = cache::Track::new(broadcast.pool.clone(), state.downgrade());
908 state.write().ok().expect("a new track is open").cache = cache;
909 state
910 }
911
912 fn claim_sequence(&mut self, sequence: u64, frame_start: u64) -> Result<()> {
921 if let Some(slot) = self.lookup.get(&sequence) {
922 if slot
926 .group
927 .live_first_frame()
928 .is_some_and(|first| first as u64 <= frame_start)
929 {
930 return Err(Error::Duplicate);
931 }
932 self.lookup.remove(&sequence);
933 }
934 Ok(())
935 }
936
937 fn insert_group(&mut self, group: &group::Producer, visible: bool) {
944 let sequence = group.sequence;
945 self.next_stamp = self.next_stamp.wrapping_add(1);
946 let stamp = self.next_stamp;
947
948 if self.latest_group.is_none_or(|latest| sequence >= latest) {
952 if let Some(latest) = self.latest_group
955 && sequence > latest
956 && let Some(prev) = self.lookup.get_mut(&latest)
957 {
958 prev.group.cache_demote();
959 prev.protection = None;
960 self.evict.push_back((latest, prev.stamp));
961 }
962 self.latest_group = Some(sequence);
963 }
964 if !self.protects(sequence) {
965 group.cache_demote();
966 self.evict.push_back((sequence, stamp));
967 }
968
969 self.max_sequence = Some(self.max_sequence.map_or(sequence, |max| max.max(sequence)));
970 self.lookup.insert(
971 sequence,
972 Slot {
973 group: group.clone(),
974 stamp,
975 visible,
976 protection: self.protects(sequence).then(|| group.cache_protect()),
977 },
978 );
979 if visible {
980 self.arrival.push_back((sequence, stamp));
981 }
982 }
983
984 fn commit_group(&mut self, group: &group::Producer, visible: bool) {
988 self.charge_debt();
989 self.insert_group(group, visible);
990 self.evict_expired();
991 }
992
993 pub(super) fn charge_debt(&mut self) {
1006 let written = self.cache.take_written();
1007 let pool = self.cache.pool().clone();
1008 match pool.accrue(written) {
1009 Some(mut accrued) => {
1010 if self.oldest_is_stale(&pool) {
1011 accrued = accrued.saturating_mul(2);
1012 }
1013 self.debt = self.debt.saturating_add(accrued).min(pool.used());
1016 self.pay_debt(&pool, written.saturating_mul(2));
1019 }
1020 None => self.debt = 0,
1023 }
1024 }
1025
1026 fn oldest_is_stale(&self, pool: &cache::Pool) -> bool {
1030 let Some(average) = pool.average() else {
1031 return false;
1032 };
1033 let Some((sequence, stamp)) = self.evict.front() else {
1034 return false;
1035 };
1036 let Some(slot) = self.lookup.get(sequence) else {
1037 return false;
1038 };
1039 slot.stamp == *stamp && !slot.group.is_aborted() && slot.group.cache_accessed() <= average
1040 }
1041
1042 fn pay_debt(&mut self, pool: &cache::Pool, cap: u64) {
1055 let average = pool.average().unwrap_or(0);
1056 let mut paid = 0u64;
1057 let mut scanned = 0usize;
1058 for _ in 0..self.evict.len() {
1059 if self.debt == 0 || paid >= cap || scanned >= EVICT_SCAN {
1060 return;
1061 }
1062 let Some((sequence, stamp)) = self.evict.pop_front() else {
1063 return;
1064 };
1065 let Some(slot) = self.lookup.get(&sequence) else {
1066 continue;
1068 };
1069 if slot.stamp != stamp {
1070 continue;
1072 }
1073 if slot.group.is_aborted() {
1074 self.lookup.remove(&sequence);
1076 continue;
1077 }
1078 if self.protects(sequence) {
1079 self.evict.push_back((sequence, stamp));
1081 continue;
1082 }
1083
1084 scanned += 1;
1085 if slot.group.cache_accessed() > average {
1089 self.evict.push_back((sequence, stamp));
1090 continue;
1091 }
1092 let size = slot.group.cache_size();
1095 if size > self.debt {
1096 self.evict.push_front((sequence, stamp));
1097 return;
1098 }
1099
1100 self.debt -= size;
1101 paid = paid.saturating_add(size);
1102 let slot = self.lookup.remove(&sequence).unwrap();
1103 let _ = slot.group.abort(Error::Evicted);
1104 }
1105 }
1106
1107 fn set_start(&mut self, start_sequence: Option<u64>, pending: bool) {
1112 self.start_sequence = start_sequence;
1113 self.start_pending = pending;
1114 }
1115
1116 fn set_final(&mut self, final_sequence: u64) -> Result<()> {
1119 if self.final_sequence.is_some() {
1120 return Err(Error::Closed);
1121 }
1122 if let Some(max) = self.max_sequence
1123 && final_sequence <= max
1124 {
1125 return Err(Error::ProtocolViolation);
1126 }
1127 self.final_sequence = Some(final_sequence);
1128 Ok(())
1129 }
1130
1131 fn is_complete(&self) -> bool {
1141 let reached = self.sealed
1144 || self
1145 .final_sequence
1146 .is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin);
1147 reached && (self.abort.is_none() || self.settled)
1148 }
1149
1150 fn is_settled(&self) -> bool {
1153 let Some(fin) = self.final_sequence else {
1154 return false;
1155 };
1156 self.is_complete() && self.lookup.range(..fin).all(|(_, slot)| slot.group.is_finished())
1157 }
1158
1159 fn resume_position(&self) -> Option<Position> {
1165 if self.resume.is_some() {
1168 return self.resume;
1169 }
1170
1171 let max = self.latest_group?;
1172 match self.lookup.get(&max).and_then(|slot| slot.group.resume_frame()) {
1173 Some(frame) => Some(Position {
1176 group: max,
1177 frame: frame as u64,
1178 }),
1179 None => Some(Position::group(max.saturating_add(1))),
1184 }
1185 }
1186
1187 fn poll_finished(&self) -> Poll<Result<u64>> {
1188 if let Some(fin) = self.final_sequence {
1189 Poll::Ready(Ok(fin))
1190 } else if let Some(err) = &self.abort {
1191 Poll::Ready(Err(err.clone()))
1192 } else if self.sealed {
1193 Poll::Ready(Err(Error::Closed))
1194 } else {
1195 Poll::Pending
1196 }
1197 }
1198
1199 fn poll_end(&self) -> Poll<Result<Option<u64>>> {
1201 if self.sealed && self.final_sequence.is_none() && self.abort.is_none() {
1202 Poll::Ready(Ok(None))
1203 } else {
1204 self.poll_finished().map(|res| res.map(Some))
1205 }
1206 }
1207
1208 fn modify(producer: &kio::Producer<Self>) -> Result<kio::Mut<'_, Self>> {
1209 producer.write().map_err(|r| r.abort.clone().unwrap_or(Error::Dropped))
1210 }
1211
1212 pub(crate) fn insert_group_request(
1218 &mut self,
1219 sequence: u64,
1220 frame_start: u64,
1221 info: Option<Info>,
1222 ) -> Result<group::Producer> {
1223 if let Some(err) = &self.abort {
1224 return Err(err.clone());
1225 }
1226 if let Some(fin) = self.final_sequence
1227 && sequence >= fin
1228 {
1229 return Err(Error::Closed);
1230 }
1231
1232 if self.info.is_none() {
1236 self.install(info.unwrap_or_default());
1237 }
1238 let info = self.info.clone().unwrap();
1239
1240 self.claim_sequence(sequence, frame_start)?;
1242
1243 let mut group = group::Producer::new(group::Info { sequence }, info, self.cache.clone());
1244 group.start_at(frame_start)?;
1248 group.cache_refresh();
1253 self.commit_group(&group, false);
1254 Ok(group)
1255 }
1256}
1257
1258fn commit_abort(mut state: kio::Mut<'_, TrackState>, err: Error) {
1261 state.resume = state.resume_position();
1264 state.settled = state.is_settled();
1266 state.abort = Some(err);
1267 state.drop_open_groups();
1271 state.close_cache();
1272 state.close();
1273}
1274
1275#[derive(Clone)]
1277pub struct Producer {
1278 name: Arc<str>,
1279 info: Info,
1280 broadcast: Arc<broadcast::Info>,
1283 state: kio::Producer<TrackState>,
1284 prev_subscription: Option<Subscription>,
1285 alive: Arc<Alive>,
1287 stats: stats::Scope,
1291}
1292
1293impl Producer {
1294 pub(crate) fn publisher_priority(&self) -> u8 {
1296 self.info.priority
1297 }
1298
1299 pub(crate) fn new(
1307 broadcast: Arc<broadcast::Info>,
1308 name: impl Into<Arc<str>>,
1309 info: impl Into<Option<Info>>,
1310 ) -> Self {
1311 let name = name.into();
1312 let info = TrackState::normalize_info(&broadcast, info.into().unwrap_or_default());
1313 let state = TrackState::spawn(broadcast.clone());
1314 state.write().ok().expect("a new track is open").accept(info.clone());
1315 let alive = Alive::new(name.clone(), state.clone());
1316 alive.publish(None);
1317 Self {
1318 name,
1319 info,
1320 state,
1321 broadcast,
1322 prev_subscription: None,
1323 alive,
1324 stats: stats::Scope::default(),
1325 }
1326 }
1327
1328 pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
1332 self.alive.publish(Some(&scope));
1333 self.stats = scope;
1334 self
1335 }
1336
1337 pub fn name(&self) -> &str {
1339 &self.name
1340 }
1341
1342 pub fn broadcast(&self) -> &broadcast::Info {
1344 &self.broadcast
1345 }
1346
1347 pub(crate) fn adopt_group(&mut self, group: group::Producer, visible: bool) -> Result<()> {
1352 let mut state = self.modify()?;
1353 if let Some(fin) = state.final_sequence
1354 && group.sequence >= fin
1355 {
1356 return Err(Error::Closed);
1357 }
1358 if state.lookup.contains_key(&group.sequence) {
1359 return Err(Error::Duplicate);
1360 }
1361 state.insert_group(&group, visible);
1362 Ok(())
1363 }
1364
1365 pub fn create_group(&self, group: group::Info) -> Result<group::Producer> {
1367 let mut state = self.modify()?;
1368 if let Some(fin) = state.final_sequence
1369 && group.sequence >= fin
1370 {
1371 return Err(Error::Closed);
1372 }
1373 let track = state.info.clone().unwrap();
1374
1375 state.claim_sequence(group.sequence, 0)?;
1377
1378 let group = group::Producer::new(group, track, state.cache.clone()).with_meter(self.stats.meter());
1379 state.commit_group(&group, true);
1380
1381 Ok(group)
1382 }
1383
1384 pub fn append_group(&self) -> Result<group::Producer> {
1386 let mut state = self.modify()?;
1387 let sequence = match state.max_sequence {
1388 Some(s) => s.checked_add(1).ok_or(coding::BoundsExceeded)?,
1389 None => 0,
1390 };
1391 if let Some(fin) = state.final_sequence
1392 && sequence >= fin
1393 {
1394 return Err(Error::Closed);
1395 }
1396
1397 let track = state.info.clone().unwrap();
1398
1399 let group =
1400 group::Producer::new(group::Info { sequence }, track, state.cache.clone()).with_meter(self.stats.meter());
1401 state.commit_group(&group, true);
1402
1403 Ok(group)
1404 }
1405
1406 pub fn append_datagram<B: crate::IntoBytes>(&mut self, timestamp: Timestamp, payload: B) -> Result<u64> {
1417 let payload = payload.into_bytes();
1418 if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD {
1419 return Err(Error::FrameTooLarge);
1420 }
1421 let meter = self.stats.meter();
1423 let mut state = self.modify()?;
1424 let timescale = state.info.as_ref().unwrap().timescale;
1426 let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?;
1427 let sequence = match state.max_sequence {
1428 Some(s) => s.checked_add(1).ok_or(coding::BoundsExceeded)?,
1429 None => 0,
1430 };
1431 if let Some(fin) = state.final_sequence
1432 && sequence >= fin
1433 {
1434 return Err(Error::Closed);
1435 }
1436 state.max_sequence = Some(sequence);
1437 meter.datagram(payload.len() as u64);
1438 state.push_datagram(Datagram {
1439 sequence,
1440 timestamp,
1441 payload,
1442 });
1443 let cache = state.cache.clone();
1444 drop(state);
1445 cache.settle(None);
1446 Ok(sequence)
1447 }
1448
1449 pub fn insert_datagram<B: crate::IntoBytes>(
1455 &mut self,
1456 sequence: u64,
1457 timestamp: Timestamp,
1458 payload: B,
1459 ) -> Result<()> {
1460 let payload = payload.into_bytes();
1461 if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD {
1462 return Err(Error::FrameTooLarge);
1463 }
1464 let meter = self.stats.meter();
1466 let mut state = self.modify()?;
1467 let timescale = state.info.as_ref().unwrap().timescale;
1469 let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?;
1470 if let Some(fin) = state.final_sequence
1471 && sequence >= fin
1472 {
1473 return Err(Error::Closed);
1474 }
1475 state.max_sequence = Some(state.max_sequence.unwrap_or(0).max(sequence));
1476 meter.datagram(payload.len() as u64);
1477 state.push_datagram(Datagram {
1478 sequence,
1479 timestamp,
1480 payload,
1481 });
1482 let cache = state.cache.clone();
1483 drop(state);
1484 cache.settle(None);
1485 Ok(())
1486 }
1487
1488 pub fn write_frame<B: crate::IntoBytes>(&mut self, timestamp: Timestamp, frame: B) -> Result<()> {
1493 let frame = crate::IntoBytes::into_bytes(frame);
1494 if frame.len() as u64 > group::MAX_CACHE_BYTES {
1495 return Err(Error::FrameTooLarge);
1496 }
1497 let mut group = self.append_group()?;
1498 group.write_frame(timestamp, frame)?;
1499 group.finish()?;
1500 Ok(())
1501 }
1502
1503 pub(crate) fn close(self) -> Result<()> {
1505 let mut state = self.modify()?;
1506 state.sealed = true;
1507 state.resume = state.resume_position();
1509 state.drop_open_groups();
1510 state.close_cache();
1511 state.close();
1512 Ok(())
1513 }
1514
1515 pub(crate) fn abort_session(self, err: Error) -> Result<()> {
1517 let state = self.modify()?;
1518 let open: Vec<_> = state
1519 .lookup
1520 .values()
1521 .filter(|slot| !slot.group.is_finished())
1522 .map(|slot| slot.group.clone())
1523 .collect();
1524 commit_abort(state, err.clone());
1526 for group in open {
1527 let _ = group.abort(err.clone());
1528 }
1529 Ok(())
1530 }
1531
1532 pub fn finish(&self) -> Result<()> {
1538 let mut state = self.modify()?;
1539 let final_sequence = match state.max_sequence {
1540 Some(max) => max.checked_add(1).ok_or(coding::BoundsExceeded)?,
1541 None => 0,
1542 };
1543 state.set_final(final_sequence)
1544 }
1545
1546 pub fn finish_at(&mut self, final_sequence: u64) -> Result<()> {
1559 self.modify()?.set_final(final_sequence)
1560 }
1561
1562 pub fn start_at(&mut self, sequence: impl Into<Option<u64>>) -> Result<()> {
1573 self.modify()?.set_start(sequence.into(), false);
1574 Ok(())
1575 }
1576
1577 pub(crate) fn request_start(&mut self, sequence: Option<u64>) -> Result<()> {
1582 self.modify()?.set_start(sequence, true);
1583 Ok(())
1584 }
1585
1586 #[cfg(test)]
1589 pub(crate) fn start_sequence(&self) -> Option<u64> {
1590 self.state.read().start_sequence
1591 }
1592
1593 pub fn final_sequence(&self) -> Option<u64> {
1598 self.state.read().final_sequence
1599 }
1600
1601 pub fn abort(self, err: Error) -> Result<()> {
1616 commit_abort(self.modify()?, err);
1617 Ok(())
1618 }
1619
1620 #[expect(
1628 clippy::result_large_err,
1629 reason = "return the owned producer without allocating on an idle check"
1630 )]
1631 pub fn abort_unused(self, err: Error) -> std::result::Result<(), Self> {
1632 match self.state.write_unused() {
1633 kio::Unused::Idle(guard) => {
1634 commit_abort(guard, err);
1635 return Ok(());
1636 }
1637 kio::Unused::Closed => return Ok(()),
1638 kio::Unused::Used => {}
1639 }
1640 Err(self)
1641 }
1642
1643 pub fn is_used(&self) -> bool {
1650 !self.is_closed() && self.state.is_used()
1651 }
1652
1653 pub async fn unused(&self) -> Result<()> {
1655 self.state.unused().await.map_err(|_| self.abort_reason())
1656 }
1657
1658 pub async fn used(&self) -> Result<()> {
1660 self.state.used().await.map_err(|_| self.abort_reason())
1661 }
1662
1663 pub async fn closed(&self) -> Error {
1665 kio::wait(|waiter| self.poll_closed(waiter)).await
1666 }
1667
1668 pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll<Error> {
1670 self.state.poll_closed(waiter).map(|()| self.abort_reason())
1671 }
1672
1673 fn abort_reason(&self) -> Error {
1675 self.state.read().abort.clone().unwrap_or(Error::Dropped)
1676 }
1677
1678 pub fn is_closed(&self) -> bool {
1680 self.state.read().is_closed()
1681 }
1682
1683 pub fn latest(&self) -> Option<u64> {
1685 self.state.read().max_sequence
1686 }
1687
1688 pub fn is_clone(&self, other: &Self) -> bool {
1690 self.state.same_channel(&other.state)
1691 }
1692
1693 pub(crate) fn weak(&self) -> TrackWeak {
1695 TrackWeak {
1696 name: self.name.clone(),
1697 state: self.state.weak(),
1698 }
1699 }
1700
1701 pub fn demand(&self) -> Demand {
1709 Demand {
1710 name: self.name.clone(),
1711 state: self.state.weak(),
1712 }
1713 }
1714
1715 pub fn consume(&self) -> Consumer {
1720 Consumer::plain(self.name.clone(), self.state.consume())
1721 }
1722
1723 pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> Subscriber {
1733 let preferences = subscription.into().unwrap_or_default();
1734
1735 let info = self.info.clone();
1740 let min_sequence = floor_of(&preferences);
1741 let subscription = kio::Producer::new(preferences);
1742 register_subscription(self.state.read(), &subscription);
1743 let drift_anchor = kio::Producer::new(Anchor::default());
1744
1745 let broadcast = self.state.read().broadcast.clone();
1748 Subscriber {
1749 name: self.name.clone(),
1750 broadcast,
1751 info,
1752 inner: SubscriberKind::Plain(PlainSubscriber {
1753 state: self.state.consume(),
1754 subscription,
1755 min_sequence,
1756 index: 0,
1757 datagram_index: 0,
1758 next_sequence: 0,
1759 end_sequence: None,
1760 parked: BTreeMap::new(),
1761 stale_cap: None,
1762 drift_anchor,
1763 stale: stats::Content::default(),
1764 seek_pending: BTreeMap::new(),
1765 }),
1766 stats: stats::Scope::default(),
1768 _stats_sub: stats::Subscription::default(),
1769 }
1770 }
1771
1772 pub async fn subscription_changed(&mut self) -> Result<Option<Subscription>> {
1778 kio::wait(|waiter| self.poll_subscription_changed(waiter)).await
1779 }
1780
1781 pub fn subscription(&self) -> Option<Subscription> {
1789 let state = self.state.read();
1790 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
1791 drop(state);
1792 snapshot_subscription(&subs, bound)
1793 }
1794
1795 pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Subscription>>> {
1799 if self.state.poll_closed(waiter).is_ready() {
1802 let abort = self.state.read().abort.clone();
1803 return Poll::Ready(Err(abort.unwrap_or(Error::Dropped)));
1804 }
1805
1806 let state = self.state.read();
1808 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
1809 drop(state);
1810
1811 poll_combined_changed(&subs, bound, &mut self.prev_subscription, waiter).map(Ok)
1812 }
1813
1814 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
1816 self.state.poll_unused(waiter).map(|used| match used {
1817 Some(()) => Ok(()),
1818 None => Err(self.abort_reason()),
1819 })
1820 }
1821
1822 pub fn dynamic(&self) -> Dynamic {
1826 Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
1827 }
1828
1829 fn modify(&self) -> Result<kio::Mut<'_, TrackState>> {
1830 TrackState::modify(&self.state)
1831 }
1832}
1833
1834fn poll_requested_group(
1838 state: &kio::Producer<TrackState>,
1839 fetch: &kio::Shared<FetchState>,
1840 waiter: &kio::Waiter,
1841) -> Poll<Result<group::Request>> {
1842 if let Poll::Ready(mut guard) = fetch.poll(waiter, |fetch| {
1844 if fetch.has_queued() {
1845 Poll::Ready(())
1846 } else {
1847 Poll::Pending
1848 }
1849 }) {
1850 let sequence = guard.pop().expect("predicate guaranteed a request");
1851 let pending = guard.get(&sequence).expect("popped key must be pending");
1855 let priority = pending.priority;
1856 let frame_start = pending.frame_start;
1857 let result = pending.result.clone();
1858 drop(guard);
1859 return Poll::Ready(Ok(group::Request {
1860 state: state.clone(),
1861 fetch: fetch.clone(),
1862 sequence,
1863 priority,
1864 frame_start,
1865 result,
1866 done: false,
1867 }));
1868 }
1869
1870 match state.poll_ref(waiter, |state| match &state.abort {
1872 Some(err) => Poll::Ready(err.clone()),
1873 None => Poll::Pending,
1874 }) {
1875 Poll::Ready(Ok(err)) => Poll::Ready(Err(err)),
1876 Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
1877 Poll::Pending => Poll::Pending,
1878 }
1879}
1880
1881pub struct Dynamic {
1891 name: Arc<str>,
1892 state: kio::Producer<TrackState>,
1894 fetch: kio::Shared<FetchState>,
1896 alive: Arc<Alive>,
1899}
1900
1901impl Dynamic {
1902 fn new(name: Arc<str>, state: kio::Producer<TrackState>, alive: Arc<Alive>) -> Self {
1903 let fetch = state.read().fetch.clone();
1904 fetch.lock().add_handler();
1905 Self {
1906 name,
1907 state,
1908 fetch,
1909 alive,
1910 }
1911 }
1912
1913 pub fn name(&self) -> &str {
1915 &self.name
1916 }
1917
1918 pub async fn requested_group(&self) -> Result<group::Request> {
1924 kio::wait(|waiter| self.poll_requested_group(waiter)).await
1925 }
1926
1927 pub fn poll_requested_group(&self, waiter: &kio::Waiter) -> Poll<Result<group::Request>> {
1929 poll_requested_group(&self.state, &self.fetch, waiter)
1930 }
1931
1932 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
1934 self.state.poll_unused(waiter).map(|_| ())
1935 }
1936}
1937
1938impl Clone for Dynamic {
1939 fn clone(&self) -> Self {
1940 self.fetch.lock().add_handler();
1942 Self {
1943 name: self.name.clone(),
1944 state: self.state.clone(),
1945 fetch: self.fetch.clone(),
1946 alive: self.alive.clone(),
1947 }
1948 }
1949}
1950
1951impl Drop for Dynamic {
1952 fn drop(&mut self) {
1953 let mut fetch = self.fetch.lock();
1959 if fetch.remove_handler() {
1960 fetch.drain_queued();
1961 }
1962 }
1963}
1964
1965struct Alive {
1974 name: Arc<str>,
1975 state: kio::Producer<TrackState>,
1976
1977 published: AtomicBool,
1980
1981 stats: OnceLock<stats::Subscription>,
1984}
1985
1986impl Alive {
1987 fn new(name: Arc<str>, state: kio::Producer<TrackState>) -> Arc<Self> {
1988 Arc::new(Self {
1989 name,
1990 state,
1991 published: Default::default(),
1992 stats: Default::default(),
1993 })
1994 }
1995
1996 fn publish(&self, stats: Option<&stats::Scope>) {
2000 self.published.store(true, Ordering::Relaxed);
2001 if let Some(scope) = stats {
2002 let _ = self.stats.set(scope.subscribe());
2005 }
2006 }
2007}
2008
2009impl Drop for Alive {
2010 fn drop(&mut self) {
2011 if !self.published.load(Ordering::Relaxed) {
2013 return;
2014 }
2015 match self.state.write() {
2022 Ok(mut state) => {
2023 if state.final_sequence.is_some() {
2024 state.sealed = true;
2027 state.close_cache();
2028 return;
2029 }
2030 if state.abort.is_some() {
2031 return;
2032 }
2033 tracing::warn!(
2034 track = %self.name,
2035 "track::Producer dropped without finish() or abort()"
2036 );
2037 state.resume = state.resume_position();
2039 state.drop_open_groups();
2040 state.close_cache();
2041 }
2042 Err(state) => {
2043 if state.sealed || state.final_sequence.is_some() || state.abort.is_some() {
2044 return;
2045 }
2046 tracing::warn!(
2047 track = %self.name,
2048 "track::Producer dropped without finish() or abort()"
2049 );
2050 }
2051 }
2052 }
2053}
2054
2055fn poll_combined_changed(
2061 subs: &kio::Shared<Subscriptions>,
2062 bound: Option<Duration>,
2063 prev: &mut Option<Subscription>,
2064 waiter: &kio::Waiter,
2065) -> Poll<Option<Subscription>> {
2066 loop {
2067 let mut next = None;
2068 let mut departed = false;
2069 let mut guard = ready!(subs.poll(waiter, |subs| {
2070 (next, departed) = combined_subscription(subs, bound, waiter);
2071 if departed || next != *prev {
2072 Poll::Ready(())
2073 } else {
2074 Poll::Pending
2075 }
2076 }));
2077 if departed {
2078 guard.retain(|sub| !sub.is_closed());
2079 }
2080 drop(guard);
2081 if next != *prev {
2082 *prev = next.clone();
2083 return Poll::Ready(next);
2084 }
2085 }
2087}
2088
2089fn combined_subscription(
2096 subs: &Subscriptions,
2097 bound: Option<Duration>,
2098 waiter: &kio::Waiter,
2099) -> (Option<Subscription>, bool) {
2100 let mut combined = None;
2101 let mut departed = false;
2102 for sub in subs.iter() {
2103 if sub.is_closed() {
2108 departed = true;
2109 continue;
2110 }
2111 let _ = sub.poll_closed(waiter);
2118 let _ = sub.poll(waiter, |_| Poll::<()>::Pending);
2119 if let Poll::Ready(merged) = sub.read().poll_combined(&combined) {
2120 combined = Some(merged);
2121 }
2122 }
2123 (clamp_combined(combined, bound), departed)
2124}
2125
2126fn snapshot_subscription(subs: &kio::Shared<Subscriptions>, bound: Option<Duration>) -> Option<Subscription> {
2128 let mut combined: Option<Subscription> = None;
2129 for sub in subs.read().iter() {
2130 if sub.is_closed() {
2132 continue;
2133 }
2134 if let Poll::Ready(merged) = sub.read().poll_combined(&combined) {
2135 combined = Some(merged);
2136 }
2137 }
2138 clamp_combined(combined, bound)
2139}
2140
2141fn servable_cap(cursor: Option<u64>, outer: Option<u64>) -> Option<u64> {
2156 super::subscription::min_some(cursor, outer)
2157}
2158
2159fn floor_of(subscription: &Subscription) -> u64 {
2168 subscription.start.map(|start| start.group).unwrap_or(0)
2169}
2170
2171fn clamp_max_age(mut max_age: Duration, bound: Option<Duration>) -> Duration {
2181 if let Some(bound) = bound {
2182 max_age = max_age.min(bound);
2183 }
2184 max_age
2185}
2186
2187fn clamp_combined(combined: Option<Subscription>, bound: Option<Duration>) -> Option<Subscription> {
2189 let mut combined = combined?;
2190 combined.max_age = clamp_max_age(combined.max_age, bound);
2191 Some(combined)
2192}
2193
2194fn register_subscription(state: kio::Ref<'_, TrackState>, subscription: &kio::Producer<Subscription>) {
2202 if state.is_closed() {
2203 return;
2204 }
2205 let subs = state.subscriptions.clone();
2206 drop(state);
2207 let mut subs = subs.lock();
2208 if subs.len() == subs.capacity() {
2209 subs.retain(|sub| !sub.is_closed());
2210 let live = subs.len();
2212 subs.reserve(live);
2213 }
2214 subs.push(subscription.consume());
2215}
2216
2217#[derive(Clone)]
2219pub(crate) struct TrackWeak {
2220 name: Arc<str>,
2221 state: kio::ProducerWeak<TrackState>,
2222}
2223
2224impl TrackWeak {
2225 pub fn try_consume(&self) -> Option<Consumer> {
2232 Some(Consumer::plain(self.name.clone(), self.state.try_consume()?))
2233 }
2234
2235 pub(crate) fn name(&self) -> &Arc<str> {
2238 &self.name
2239 }
2240
2241 pub(crate) fn reject(&self, err: Error) -> bool {
2250 let Some(producer) = self.state.produce() else {
2251 return false;
2252 };
2253 let Ok(mut state) = producer.write() else {
2254 return false;
2255 };
2256 if state.published || state.claimed || state.abort.is_some() {
2257 return false;
2258 }
2259 state.abort = Some(err);
2260 state.close();
2261 true
2262 }
2263
2264 pub(crate) fn is_used(&self) -> bool {
2267 !self.state.is_closed() && self.state.is_used()
2268 }
2269
2270 pub(crate) fn poll_used(&self, waiter: &kio::Waiter) {
2273 let _ = self.state.poll_used(waiter);
2274 }
2275
2276 pub(crate) fn poll_unused(&self, waiter: &kio::Waiter) {
2279 let _ = self.state.poll_unused(waiter);
2280 }
2281}
2282
2283impl super::WeakEntry for TrackWeak {
2284 fn is_closed(&self) -> bool {
2285 self.state.is_closed()
2286 }
2287
2288 fn same_channel(&self, other: &Self) -> bool {
2289 self.state.same_channel(&other.state)
2290 }
2291}
2292
2293#[derive(Clone)]
2302pub struct Demand {
2303 name: Arc<str>,
2304 state: kio::ProducerWeak<TrackState>,
2305}
2306
2307impl Demand {
2308 pub fn name(&self) -> &str {
2310 &self.name
2311 }
2312
2313 pub async fn used(&self) -> Result<()> {
2315 self.state.used().await.map_err(|_| self.abort_reason())
2316 }
2317
2318 pub async fn unused(&self) -> Result<()> {
2320 self.state.unused().await.map_err(|_| self.abort_reason())
2321 }
2322
2323 pub async fn closed(&self) -> Error {
2325 self.state.closed().await;
2326 self.abort_reason()
2327 }
2328
2329 pub(crate) fn priority(&self) -> u8 {
2331 self.state
2333 .read()
2334 .info
2335 .as_ref()
2336 .map_or(DEFAULT_PRIORITY, |info| info.priority)
2337 }
2338
2339 pub fn is_used(&self) -> bool {
2341 self.state.is_used()
2342 }
2343
2344 pub fn poll_used(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
2346 self.state.poll_used(waiter).map(|used| match used {
2347 Some(()) => Ok(()),
2348 None => Err(self.abort_reason()),
2349 })
2350 }
2351
2352 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
2354 self.state.poll_unused(waiter).map(|used| match used {
2355 Some(()) => Ok(()),
2356 None => Err(self.abort_reason()),
2357 })
2358 }
2359
2360 pub(crate) fn is_closed(&self) -> bool {
2362 self.state.is_closed()
2363 }
2364
2365 pub(crate) fn poll_state(&self, waiter: &kio::Waiter) -> DemandState {
2372 loop {
2373 match self.state.poll_used(waiter) {
2374 Poll::Ready(None) => return DemandState::Closed,
2375 Poll::Pending => return DemandState::Idle,
2377 Poll::Ready(Some(())) => match self.state.poll_unused(waiter) {
2379 Poll::Ready(None) => return DemandState::Closed,
2380 Poll::Pending => return DemandState::Active,
2381 Poll::Ready(Some(())) => continue,
2384 },
2385 }
2386 }
2387 }
2388
2389 fn abort_reason(&self) -> Error {
2391 self.state.read().abort.clone().unwrap_or(Error::Dropped)
2392 }
2393}
2394
2395#[derive(Copy, Clone, Debug, Eq, PartialEq)]
2397pub(crate) enum DemandState {
2398 Active,
2400 Idle,
2402 Closed,
2404}
2405
2406#[derive(Clone)]
2417pub struct Consumer {
2418 name: Arc<str>,
2419 broadcast: Arc<broadcast::Info>,
2423 inner: ConsumerKind,
2424 stats: stats::Scope,
2427}
2428
2429#[derive(Clone)]
2430enum ConsumerKind {
2431 Plain(kio::Consumer<TrackState>),
2432 Spliced(super::resume::Consumer),
2433}
2434
2435impl Consumer {
2436 fn plain(name: Arc<str>, state: kio::Consumer<TrackState>) -> Self {
2437 let broadcast = state.read().broadcast.clone();
2438 Self {
2439 name,
2440 broadcast,
2441 inner: ConsumerKind::Plain(state),
2442 stats: stats::Scope::default(),
2443 }
2444 }
2445
2446 pub(crate) fn spliced(name: Arc<str>, broadcast: Arc<broadcast::Info>, resume: super::resume::Consumer) -> Self {
2448 Self {
2449 name,
2450 broadcast,
2451 inner: ConsumerKind::Spliced(resume),
2452 stats: stats::Scope::default(),
2453 }
2454 }
2455
2456 pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
2459 self.stats = scope;
2460 self
2461 }
2462
2463 pub(crate) fn with_broadcast(mut self, broadcast: Arc<broadcast::Info>) -> Self {
2467 self.broadcast = broadcast;
2468 self
2469 }
2470
2471 pub(crate) fn cached_groups(&self) -> Vec<(group::Producer, bool)> {
2477 match &self.inner {
2478 ConsumerKind::Plain(state) => {
2479 let state = state.read();
2480 let mut out = Vec::with_capacity(state.lookup.len());
2481 for (sequence, stamp) in state.arrival.iter() {
2482 if let Some(slot) = state.lookup.get(sequence)
2483 && slot.stamp == *stamp
2484 && !slot.group.is_aborted()
2485 {
2486 out.push((slot.group.clone(), slot.visible));
2487 }
2488 }
2489 let mut copied: HashSet<u64> = out.iter().map(|(group, _)| group.sequence).collect();
2491 for (sequence, slot) in state.lookup.iter() {
2492 if !slot.group.is_aborted() && copied.insert(*sequence) {
2493 out.push((slot.group.clone(), slot.visible));
2494 }
2495 }
2496 out
2497 }
2498 ConsumerKind::Spliced(resume) => resume.cached_groups(),
2499 }
2500 }
2501
2502 pub(crate) fn cached_group(&self, sequence: u64, frame_start: u64, finished: bool) -> Option<group::Consumer> {
2509 match &self.inner {
2510 ConsumerKind::Plain(state) => {
2511 let state = state.read();
2512 let group = state.covering_group(sequence, frame_start)?;
2513 if finished && !group.is_finished() {
2514 return None;
2515 }
2516 group.cache_refresh();
2517 Some(group.consume())
2518 }
2519 ConsumerKind::Spliced(resume) => resume.cached_group(sequence, frame_start),
2520 }
2521 }
2522
2523 pub(crate) fn cached_info(&self) -> Option<Info> {
2525 match &self.inner {
2526 ConsumerKind::Plain(state) => state.read().info.clone(),
2527 ConsumerKind::Spliced(resume) => resume.cached_info(),
2528 }
2529 }
2530
2531 pub fn name(&self) -> &str {
2533 &self.name
2534 }
2535
2536 pub fn broadcast(&self) -> &broadcast::Info {
2540 &self.broadcast
2541 }
2542
2543 pub fn subscribe(&self, subscription: impl Into<Option<Subscription>>) -> kio::Pending<Subscribing> {
2554 let subscription = kio::Producer::new(subscription.into().unwrap_or_default());
2555
2556 let inner = match &self.inner {
2557 ConsumerKind::Plain(state) => {
2558 register_subscription(state.read(), &subscription);
2561 SubscribingKind::Plain(state.clone())
2562 }
2563 ConsumerKind::Spliced(resume) => SubscribingKind::Spliced(resume.clone()),
2565 };
2566
2567 kio::Pending::new(Subscribing {
2568 name: self.name.clone(),
2569 broadcast: self.broadcast.clone(),
2570 inner,
2571 subscription,
2572 stats: self.stats.clone(),
2573 })
2574 }
2575
2576 pub(crate) fn poll_start(&self, waiter: &kio::Waiter) -> Poll<Option<u64>> {
2582 match &self.inner {
2583 ConsumerKind::Plain(state) => Self::poll_state_start(state, waiter),
2584 ConsumerKind::Spliced(_) => Poll::Ready(None),
2585 }
2586 }
2587
2588 fn poll_state_start(state: &kio::Consumer<TrackState>, waiter: &kio::Waiter) -> Poll<Option<u64>> {
2590 let res = state.poll(waiter, |state| match state.start_pending && state.abort.is_none() {
2591 true => Poll::Pending,
2592 false => Poll::Ready(state.start_sequence),
2593 });
2594 match ready!(res) {
2595 Ok(start) => Poll::Ready(start),
2596 Err(state) => Poll::Ready(state.start_sequence),
2597 }
2598 }
2599
2600 pub(crate) fn peek_latest(&self) -> Option<group::Consumer> {
2604 match &self.inner {
2605 ConsumerKind::Plain(state) => {
2606 let sequence = state.read().max_sequence?;
2607 self.peek_group(sequence)
2608 }
2609 ConsumerKind::Spliced(resume) => resume.peek_latest(),
2610 }
2611 }
2612
2613 pub(crate) fn live_edge(&self, cap: Option<u64>) -> Option<LiveEdge> {
2616 match &self.inner {
2617 ConsumerKind::Plain(state) => {
2618 let edge = state.read().live_edge(cap)?;
2619 Some(LiveEdge {
2620 sequence: edge.sequence,
2621 timestamp: edge.timestamp,
2622 stamp: edge.stamp,
2623 track: state.weak(),
2624 })
2625 }
2626 ConsumerKind::Spliced(resume) => resume.live_edge(cap),
2627 }
2628 }
2629
2630 pub(crate) fn poll_first_start(
2635 &self,
2636 waiter: &kio::Waiter,
2637 from: u64,
2638 cap: Option<u64>,
2639 ) -> Option<Option<Timestamp>> {
2640 match &self.inner {
2641 ConsumerKind::Plain(state) => {
2642 let mut start = None;
2643 let _ = state.poll(waiter, |state| {
2644 while let Some(group) = state.first_servable(from, cap) {
2645 let _ = group.poll_closed(waiter);
2648 let _ = group.poll_timestamp(waiter);
2649 if !group.is_aborted() {
2650 start = Some(group.timestamp());
2651 break;
2652 }
2653 }
2654 Poll::<()>::Pending
2655 });
2656 start
2657 }
2658 ConsumerKind::Spliced(resume) => resume.poll_first_start(waiter, from, cap),
2659 }
2660 }
2661
2662 pub(crate) fn peek_before(&self, sequence: u64) -> Option<group::Consumer> {
2666 match &self.inner {
2667 ConsumerKind::Plain(state) => {
2668 let state = state.read();
2669 state
2670 .lookup
2671 .range(..sequence)
2672 .rev()
2673 .map(|(_, slot)| &slot.group)
2674 .find(|group| !group.is_aborted())
2675 .map(|group| group.consume())
2676 }
2677 ConsumerKind::Spliced(resume) => resume.peek_before(sequence),
2678 }
2679 }
2680
2681 pub(crate) fn poll_landed(&self, sequence: u64, waiter: &kio::Waiter) -> Poll<Option<group::Consumer>> {
2686 let ConsumerKind::Plain(state) = &self.inner else {
2687 return Poll::Ready(None);
2688 };
2689 let res = state.poll(waiter, |state| {
2690 if let Some(slot) = state.lookup.get(&sequence)
2691 && !slot.group.is_aborted()
2692 {
2693 return Poll::Ready(Some(slot.group.consume()));
2694 }
2695 match state.start_sequence.is_some_and(|start| start > sequence)
2696 || state.final_sequence.is_some_and(|fin| fin <= sequence)
2697 {
2698 true => Poll::Ready(None),
2699 false => Poll::Pending,
2700 }
2701 });
2702 match res {
2703 Poll::Ready(Ok(group)) => Poll::Ready(group),
2704 Poll::Ready(Err(_)) => Poll::Ready(None),
2705 Poll::Pending => Poll::Pending,
2706 }
2707 }
2708
2709 pub(crate) fn peek_group(&self, sequence: u64) -> Option<group::Consumer> {
2714 match &self.inner {
2715 ConsumerKind::Plain(state) => {
2716 let state = state.read();
2717 let slot = state.lookup.get(&sequence)?;
2718 if slot.group.is_aborted() {
2719 return None;
2720 }
2721 Some(slot.group.consume())
2722 }
2723 ConsumerKind::Spliced(resume) => resume.peek_group(sequence),
2724 }
2725 }
2726
2727 pub(crate) fn guard_group(
2729 &self,
2730 group: group::Consumer,
2731 subscription: kio::Consumer<Subscription>,
2732 anchor: kio::Consumer<Anchor>,
2733 bound: Option<u64>,
2734 ) -> group::Consumer {
2735 let ConsumerKind::Plain(state) = &self.inner else {
2736 return group;
2737 };
2738 let sequence = group.sequence;
2739 group.with_expiry(Arc::new(GroupExpiry {
2740 state: state.weak(),
2741 subscription,
2742 anchor,
2743 bound,
2744 sequence,
2745 }))
2746 }
2747
2748 pub(crate) fn poll_latest_group(&self, after: Option<u64>, waiter: &kio::Waiter) -> Poll<group::Producer> {
2752 let ConsumerKind::Plain(state) = &self.inner else {
2753 return Poll::Pending;
2754 };
2755 let res = state.poll(waiter, |state| {
2756 match state
2757 .latest_group
2758 .filter(|latest| after.is_none_or(|after| *latest > after))
2759 {
2760 Some(latest) => match state.lookup.get(&latest) {
2761 Some(slot) => Poll::Ready(slot.group.clone()),
2762 None => Poll::Pending,
2763 },
2764 None => Poll::Pending,
2765 }
2766 });
2767 match res {
2768 Poll::Ready(Ok(group)) => Poll::Ready(group),
2769 _ => Poll::Pending,
2770 }
2771 }
2772
2773 pub(crate) fn poll_peek_group(&self, sequence: u64, waiter: &kio::Waiter) -> Poll<Option<group::Consumer>> {
2781 let ConsumerKind::Plain(state) = &self.inner else {
2782 return Poll::Pending;
2784 };
2785
2786 let res = state.poll(waiter, |state| {
2787 match state.lookup.get(&sequence) {
2788 Some(slot) if !slot.group.is_aborted() => Poll::Ready(Some(slot.group.consume())),
2789 Some(_) => Poll::Ready(None),
2791 None if state.final_sequence.is_some_and(|fin| sequence >= fin) => Poll::Ready(None),
2793 None if state.start_sequence.is_some_and(|start| sequence < start) => Poll::Ready(None),
2795 None => Poll::Pending,
2796 }
2797 });
2798
2799 match res {
2800 Poll::Ready(Ok(res)) => Poll::Ready(res),
2801 Poll::Ready(Err(_)) => Poll::Ready(None),
2803 Poll::Pending => Poll::Pending,
2804 }
2805 }
2806
2807 pub(crate) fn poll_serving_group(&self, sequence: u64, index: u64, waiter: &kio::Waiter) -> Poll<()> {
2822 let ConsumerKind::Plain(state) = &self.inner else {
2823 return Poll::Pending;
2825 };
2826 let res = state.poll(waiter, |state| match state.lookup.get(&sequence) {
2827 Some(slot) if !slot.group.is_aborted() => {
2828 let mut group = slot.group.consume();
2831 group.start_at(index);
2832 match group.index() == index {
2833 true => Poll::Ready(()),
2834 false => Poll::Pending,
2835 }
2836 }
2837 _ => Poll::Pending,
2838 });
2839 match res {
2840 Poll::Ready(Ok(())) => Poll::Ready(()),
2841 Poll::Ready(Err(_)) | Poll::Pending => Poll::Pending,
2844 }
2845 }
2846
2847 pub(crate) fn is_live(&self) -> bool {
2850 match &self.inner {
2851 ConsumerKind::Plain(state) => {
2852 let state = state.read();
2853 state.published && !state.closed
2854 }
2855 ConsumerKind::Spliced(resume) => resume.is_live(),
2856 }
2857 }
2858
2859 pub fn fetch_group(&self, sequence: u64, options: impl Into<Option<group::Fetch>>) -> kio::Pending<Fetching> {
2872 let options = options.into().unwrap_or_default();
2873
2874 self.stats.fetch();
2878
2879 let state = match &self.inner {
2880 ConsumerKind::Plain(state) => state,
2881 ConsumerKind::Spliced(resume) => {
2884 return kio::Pending::new(Fetching {
2885 inner: FetchingKind::Spliced(resume.fetch_group(sequence, options)),
2886 stats: self.stats.clone(),
2887 });
2888 }
2889 };
2890
2891 let mut result = None;
2892
2893 let (fetch, cached) = {
2897 let state = state.read();
2898 (
2899 state.fetch.clone(),
2900 state.poll_fetch_cached(sequence, options.frame_start),
2901 )
2902 };
2903 let unresolved = cached.is_pending();
2904 let hit = match cached {
2907 Poll::Ready(Ok(group)) => Some(Box::new(group)),
2908 _ => None,
2909 };
2910
2911 if unresolved {
2912 let mut fetch = fetch.lock();
2913 if let Some(pending) = fetch.join(&sequence) {
2914 pending.priority = pending.priority.max(options.priority);
2924 pending.frame_start = pending.frame_start.min(options.frame_start);
2925 result = Some(pending.result.consume());
2926 } else {
2927 let producer = kio::Producer::<FetchOutcome>::default();
2931 let consumer = producer.consume();
2932 let attempt = PendingFetch {
2933 priority: options.priority,
2934 frame_start: options.frame_start,
2935 result: producer,
2936 };
2937 if fetch.insert(sequence, attempt).is_ok() {
2938 result = Some(consumer);
2939 }
2940 }
2941 }
2942
2943 kio::Pending::new(Fetching {
2944 inner: FetchingKind::Plain {
2945 state: state.clone(),
2946 fetch,
2947 sequence,
2948 frame_start: options.frame_start,
2949 result,
2950 hit,
2951 },
2952 stats: self.stats.clone(),
2953 })
2954 }
2955
2956 pub fn query(&self) -> kio::Pending<Querying> {
2963 kio::Pending::new(Querying {
2964 inner: match &self.inner {
2965 ConsumerKind::Plain(state) => QueryingKind::Plain(state.clone()),
2966 ConsumerKind::Spliced(resume) => QueryingKind::Spliced(resume.clone()),
2967 },
2968 })
2969 }
2970
2971 pub fn latest(&self) -> Option<u64> {
2973 match &self.inner {
2974 ConsumerKind::Plain(state) => state.read().max_sequence,
2975 ConsumerKind::Spliced(resume) => resume.latest(),
2976 }
2977 }
2978
2979 pub(crate) fn final_sequence(&self) -> Option<u64> {
2983 match &self.inner {
2984 ConsumerKind::Plain(state) => state.read().final_sequence,
2985 ConsumerKind::Spliced(resume) => resume.final_sequence(),
2986 }
2987 }
2988
2989 pub(crate) fn resume_position(&self) -> Option<Position> {
2994 match &self.inner {
2995 ConsumerKind::Plain(state) => state.read().resume_position(),
2996 ConsumerKind::Spliced(resume) => resume.resume_position(),
2997 }
2998 }
2999
3000 pub(crate) fn poll_complete(&self, waiter: &kio::Waiter) -> Poll<Result<()>> {
3005 let ConsumerKind::Plain(state) = &self.inner else {
3006 return Poll::Pending;
3008 };
3009 match ready!(state.poll(waiter, |state| {
3010 if state.final_sequence.is_some() && state.is_complete() {
3012 Poll::Ready(())
3013 } else {
3014 Poll::Pending
3015 }
3016 })) {
3017 Ok(_) => Poll::Ready(Ok(())),
3018 Err(closed) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
3021 }
3022 }
3023}
3024
3025pub struct Subscribing {
3028 name: Arc<str>,
3029 broadcast: Arc<broadcast::Info>,
3030 inner: SubscribingKind,
3031 subscription: kio::Producer<Subscription>,
3032 stats: stats::Scope,
3033}
3034
3035enum SubscribingKind {
3036 Plain(kio::Consumer<TrackState>),
3037 Spliced(super::resume::Consumer),
3038}
3039
3040impl Subscribing {
3041 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<Subscriber>> {
3044 match &self.inner {
3045 SubscribingKind::Plain(state) => {
3046 let info = ready!(state.poll(waiter, |state| state.poll_info()))
3048 .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??;
3049
3050 let drift_anchor = kio::Producer::new(Anchor::default());
3051 let min_sequence = floor_of(&self.subscription.read());
3052 Poll::Ready(Ok(Subscriber {
3053 name: self.name.clone(),
3054 broadcast: self.broadcast.clone(),
3055 info,
3056 inner: SubscriberKind::Plain(PlainSubscriber {
3057 state: state.clone(),
3058 subscription: self.subscription.clone(),
3059 min_sequence,
3060 index: 0,
3061 datagram_index: 0,
3062 next_sequence: 0,
3063 end_sequence: None,
3064 parked: BTreeMap::new(),
3065 stale_cap: None,
3066 drift_anchor,
3067 stale: stats::Content::default(),
3068 seek_pending: BTreeMap::new(),
3069 }),
3070 stats: self.stats.clone(),
3071 _stats_sub: self.stats.subscribe(),
3072 }))
3073 }
3074 SubscribingKind::Spliced(resume) => {
3075 let info = ready!(resume.poll_info(waiter))?;
3078
3079 Poll::Ready(Ok(Subscriber {
3080 name: self.name.clone(),
3081 broadcast: self.broadcast.clone(),
3082 info,
3083 inner: SubscriberKind::Spliced(Box::new(resume.subscribe_shared(self.subscription.clone()))),
3084 stats: self.stats.clone(),
3085 _stats_sub: self.stats.subscribe(),
3086 }))
3087 }
3088 }
3089 }
3090
3091 pub fn update(&mut self, subscription: Subscription) -> Result<()> {
3096 let mut state = self.subscription.write().map_err(|_| Error::Closed)?;
3097 *state = subscription;
3098 Ok(())
3099 }
3100}
3101
3102impl kio::Pollable for Subscribing {
3103 type Output = Result<Subscriber>;
3104
3105 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3106 self.poll_ok(waiter)
3107 }
3108}
3109
3110pub struct Querying {
3113 inner: QueryingKind,
3114}
3115
3116enum QueryingKind {
3117 Plain(kio::Consumer<TrackState>),
3118 Spliced(super::resume::Consumer),
3119}
3120
3121impl Querying {
3122 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<Info>> {
3124 match &self.inner {
3125 QueryingKind::Plain(state) => {
3126 let info = ready!(state.poll(waiter, |state| state.poll_info()))
3128 .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??;
3129 Poll::Ready(Ok(info))
3130 }
3131 QueryingKind::Spliced(resume) => resume.poll_info(waiter),
3132 }
3133 }
3134}
3135
3136impl kio::Pollable for Querying {
3137 type Output = Result<Info>;
3138
3139 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3140 self.poll_ok(waiter)
3141 }
3142}
3143
3144impl group::Request {
3145 pub fn sequence(&self) -> u64 {
3147 self.sequence
3148 }
3149
3150 pub fn priority(&self) -> u8 {
3152 self.priority
3153 }
3154
3155 pub fn frame_start(&self) -> u64 {
3165 self.frame_start
3166 }
3167
3168 pub fn accept(mut self, info: impl Into<Option<Info>>) -> Result<group::Producer> {
3176 self.done = true;
3177 let res = TrackState::modify(&self.state)
3181 .and_then(|mut state| state.insert_group_request(self.sequence, self.frame_start, info.into()));
3182 self.remove();
3183 if let Ok(mut outcome) = self.result.write() {
3187 outcome.rejected = Some(Error::NotFound);
3188 }
3189 res
3190 }
3191
3192 pub(crate) fn finish_track_at(&self, final_sequence: u64) {
3195 if let Ok(mut state) = TrackState::modify(&self.state)
3196 && state.final_sequence.is_none()
3197 {
3198 let _ = state.set_final(final_sequence);
3199 }
3200 }
3201
3202 pub fn reject(mut self, err: Error) {
3204 self.done = true;
3205 self.remove();
3208 if let Ok(mut outcome) = self.result.write() {
3209 outcome.rejected = Some(err);
3210 }
3211 }
3212
3213 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
3219 loop {
3220 ready!(self.result.poll_unused(waiter));
3221 let mut fetch = self.fetch.lock();
3224 if self.result.is_used() {
3225 continue;
3226 }
3227 fetch.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
3228 return Poll::Ready(());
3229 }
3230 }
3231
3232 fn remove(&self) {
3235 self.fetch
3236 .lock()
3237 .remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
3238 }
3239}
3240
3241impl Drop for group::Request {
3242 fn drop(&mut self) {
3243 if self.done {
3244 return;
3245 }
3246 self.remove();
3247 if let Ok(mut outcome) = self.result.write() {
3248 outcome.rejected = Some(Error::Dropped);
3249 }
3250 }
3251}
3252
3253pub struct Fetching {
3259 inner: FetchingKind,
3260 stats: stats::Scope,
3263}
3264
3265enum FetchingKind {
3266 Plain {
3267 state: kio::Consumer<TrackState>,
3268 fetch: kio::Shared<FetchState>,
3269 sequence: u64,
3270 frame_start: u64,
3273 result: Option<kio::Consumer<FetchOutcome>>,
3276 hit: Option<Box<group::Consumer>>,
3279 },
3280 Spliced(kio::Pending<super::resume::Fetching>),
3282}
3283
3284impl kio::Pollable for Fetching {
3285 type Output = Result<group::Consumer>;
3286
3287 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3288 let (state, fetch, sequence, frame_start, result, hit) = match &self.inner {
3289 FetchingKind::Plain {
3290 state,
3291 fetch,
3292 sequence,
3293 frame_start,
3294 result,
3295 hit,
3296 } => (state, fetch, *sequence, *frame_start, result.as_ref(), hit),
3297 FetchingKind::Spliced(spliced) => {
3298 return kio::Pollable::poll(&**spliced, waiter)
3301 .map(|res| res.map(|group| group.with_meter(self.stats.meter())));
3302 }
3303 };
3304
3305 let resolve = |mut group: group::Consumer| {
3308 group.start_at(frame_start);
3309 group.with_meter(self.stats.meter())
3310 };
3311 if let Some(group) = hit {
3312 return Poll::Ready(Ok(resolve((**group).clone())));
3313 }
3314
3315 let cached =
3318 |waiter: &kio::Waiter| match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) {
3319 Poll::Ready(Ok(res)) => Poll::Ready(res.map(resolve)),
3320 Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
3321 Poll::Pending => Poll::Pending,
3322 };
3323 if let Poll::Ready(res) = cached(waiter) {
3324 return Poll::Ready(res);
3325 }
3326
3327 let Some(result) = result else {
3329 return match fetch.poll(waiter, |fetch| match fetch.has_handlers() {
3332 false => Poll::Ready(()),
3333 true => Poll::Pending,
3334 }) {
3335 Poll::Ready(_guard) => Poll::Ready(Err(Error::NotFound)),
3336 Poll::Pending => Poll::Pending,
3337 };
3338 };
3339
3340 let err = match result.poll(waiter, |outcome| match &outcome.rejected {
3343 Some(err) => Poll::Ready(err.clone()),
3344 None => Poll::Pending,
3345 }) {
3346 Poll::Ready(Ok(err)) => err,
3347 Poll::Ready(Err(_closed)) => Error::NotFound,
3348 Poll::Pending => return Poll::Pending,
3349 };
3350 match cached(waiter) {
3353 Poll::Ready(res) => Poll::Ready(res),
3354 Poll::Pending => Poll::Ready(Err(err)),
3355 }
3356 }
3357}
3358
3359pub struct Subscriber {
3387 name: Arc<str>,
3388 broadcast: Arc<broadcast::Info>,
3390 info: Info,
3391 inner: SubscriberKind,
3392 stats: stats::Scope,
3395 _stats_sub: stats::Subscription,
3398}
3399
3400enum SubscriberKind {
3401 Plain(PlainSubscriber),
3402 Spliced(Box<super::resume::Subscriber>),
3404}
3405
3406#[derive(Clone)]
3411struct Drift {
3412 budget: Duration,
3413 edge: Edge,
3414 outer: Option<LiveEdge>,
3415 successor: Option<Successor>,
3416}
3417
3418struct GroupExpiry {
3420 state: kio::ConsumerWeak<TrackState>,
3426 subscription: kio::Consumer<Subscription>,
3427 anchor: kio::Consumer<Anchor>,
3428 bound: Option<u64>,
3429 sequence: u64,
3430}
3431
3432impl group::Expiry for GroupExpiry {
3433 fn is_expired(&self, waiter: &kio::Waiter) -> bool {
3434 let mut max_age = Duration::default();
3435 let _ = self.subscription.poll(waiter, |subscription| {
3436 max_age = subscription.max_age;
3437 Poll::<()>::Pending
3438 });
3439
3440 let mut anchor = Anchor::default();
3441 let _ = self.anchor.poll(waiter, |current| {
3442 anchor = (**current).clone();
3443 Poll::<()>::Pending
3444 });
3445 let anchor = anchor.capped(self.bound);
3446 let cap = anchor.cap;
3447 let outer = anchor
3449 .edge
3450 .filter(LiveEdge::is_live)
3451 .map(|live| (live.sequence, live.timestamp));
3452 let successor = anchor
3453 .successor
3454 .as_ref()
3455 .and_then(|successor| successor.poll_start(waiter));
3456
3457 let mut expired = false;
3458 let _ = self.state.poll(waiter, |state| {
3459 let budget = clamp_max_age(max_age, state.max_age_bound());
3460 loop {
3461 let next = loop {
3466 let next = state.first_servable(self.sequence.saturating_add(1), cap);
3467 match next {
3468 Some(group) if group.poll_closed(waiter).is_ready() && group.is_aborted() => continue,
3469 next => break next,
3470 }
3471 };
3472 let edge = state.drift_edge(cap, outer, successor);
3473 if let Some(live) = edge.presentation
3476 && let Some(slot) = state.lookup.get(&live.sequence)
3477 && slot.group.poll_closed(waiter).is_ready()
3478 && slot.group.is_aborted()
3479 {
3480 continue;
3481 }
3482 expired = state.is_stale(self.sequence, &edge, budget);
3483 if expired {
3484 break;
3485 }
3486
3487 let mut timestamp_raced = false;
3497 if let Some(slot) = state.lookup.get(&self.sequence) {
3498 let group = &slot.group;
3499 if group.timestamp().is_none()
3500 && group.poll_timestamp(waiter).is_ready()
3501 && group.timestamp().is_some()
3502 {
3503 timestamp_raced = true;
3504 }
3505 }
3506 let past = edge
3507 .presentation
3508 .map_or(self.sequence, |live| live.sequence.max(self.sequence));
3509 let beyond = state
3510 .lookup
3511 .range((std::ops::Bound::Excluded(past), std::ops::Bound::Unbounded))
3512 .map(|(_, slot)| slot)
3513 .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap))
3514 .filter(|slot| slot.visible && !slot.group.is_aborted())
3515 .map(|slot| &slot.group);
3516 for group in next.into_iter().chain(beyond) {
3517 if group.timestamp().is_none()
3518 && group.poll_timestamp(waiter).is_ready()
3519 && group.timestamp().is_some()
3520 {
3521 timestamp_raced = true;
3522 break;
3523 }
3524 }
3525 if !timestamp_raced {
3526 break;
3527 }
3528 }
3529
3530 Poll::<()>::Pending
3533 });
3534
3535 expired
3536 }
3537}
3538
3539#[derive(Clone, Copy)]
3542struct Edge {
3543 presentation: Option<PresentationEdge>,
3545 outer: Option<(u64, Timestamp)>,
3548 cap: Option<u64>,
3551 successor: Option<Timestamp>,
3553}
3554
3555#[derive(Clone, Default, PartialEq)]
3558pub(crate) struct Anchor {
3559 pub cap: Option<u64>,
3561 pub edge: Option<LiveEdge>,
3566 pub successor: Option<Successor>,
3572}
3573
3574impl Anchor {
3575 pub fn capped(mut self, cap: Option<u64>) -> Self {
3578 let capped = servable_cap(cap, self.cap);
3579 if capped != self.cap {
3580 self.cap = capped;
3581 self.successor = None;
3582 }
3583 self
3584 }
3585}
3586
3587#[derive(Clone)]
3590pub(crate) struct LiveEdge {
3591 pub sequence: u64,
3592 pub timestamp: Timestamp,
3593 stamp: u32,
3594 track: kio::ConsumerWeak<TrackState>,
3595}
3596
3597impl LiveEdge {
3598 fn is_live(&self) -> bool {
3603 self.track.read().holds(self.sequence, self.stamp)
3604 }
3605}
3606
3607impl PartialEq for LiveEdge {
3608 fn eq(&self, other: &Self) -> bool {
3609 self.sequence == other.sequence
3610 && self.timestamp == other.timestamp
3611 && self.stamp == other.stamp
3612 && self.track.same_channel(&other.track)
3613 }
3614}
3615
3616#[derive(Clone, Copy)]
3618struct PresentationEdge {
3619 sequence: u64,
3620 stamp: u32,
3623 timestamp: Timestamp,
3628}
3629
3630struct PlainSubscriber {
3632 state: kio::Consumer<TrackState>,
3633
3634 subscription: kio::Producer<Subscription>,
3635 index: usize,
3637 datagram_index: usize,
3639 min_sequence: u64,
3641 next_sequence: u64,
3644 end_sequence: Option<u64>,
3650 parked: BTreeMap<u64, group::Consumer>,
3655 stale_cap: Option<u64>,
3659 drift_anchor: kio::Producer<Anchor>,
3662 stale: stats::Content,
3668 seek_pending: BTreeMap<u64, stats::Content>,
3671}
3672
3673impl PlainSubscriber {
3674 fn anchor(&self, end: Option<u64>) -> Anchor {
3677 self.drift_anchor.read().clone().capped(end)
3678 }
3679
3680 fn update_drift_anchor(&mut self, outer: Anchor) {
3682 let anchor = outer.capped(self.end_sequence);
3683 if *self.drift_anchor.read() != anchor
3685 && let Ok(mut current) = self.drift_anchor.write()
3686 {
3687 *current = anchor;
3688 }
3689 }
3690
3691 fn poll<F, R>(&self, waiter: &kio::Waiter, f: F) -> Poll<Result<R>>
3693 where
3694 F: Fn(&kio::Ref<'_, TrackState>) -> Poll<Result<R>>,
3695 {
3696 Poll::Ready(match ready!(self.state.poll(waiter, f)) {
3697 Ok(res) => res,
3698 Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)),
3700 })
3701 }
3702
3703 fn take_stale(&mut self) -> stats::Content {
3705 std::mem::take(&mut self.stale)
3706 }
3707
3708 fn note_stale(&mut self, group: &group::Consumer) {
3711 self.stale.add(group.content());
3712 }
3713
3714 fn poll_drift(&self, anchor: Anchor, waiter: &kio::Waiter) -> Poll<Result<Drift>> {
3725 let mut max_age = Duration::default();
3726 let _ = self.subscription.poll(waiter, |subscription| {
3727 max_age = subscription.max_age;
3728 Poll::<()>::Pending
3729 });
3730 let cap = anchor.cap;
3731 let outer = anchor.edge;
3732 let successor = anchor.successor;
3733 self.poll(waiter, |state| {
3734 Poll::Ready(Ok(Drift {
3737 budget: clamp_max_age(max_age, state.max_age_bound()),
3738 edge: state.drift_edge(cap, None, None),
3739 outer: outer.clone(),
3740 successor: successor.clone(),
3741 }))
3742 })
3743 }
3744
3745 fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll<Result<bool>> {
3748 let outer = drift
3751 .outer
3752 .as_ref()
3753 .filter(|live| live.is_live())
3754 .map(|live| (live.sequence, live.timestamp));
3755 let successor = drift
3756 .successor
3757 .as_ref()
3758 .and_then(|successor| successor.poll_start(waiter));
3759 let presentation = drift.edge.presentation;
3760 let cap = drift.edge.cap;
3761 let budget = drift.budget;
3762 self.poll(waiter, move |state| {
3763 let edge = Edge {
3764 presentation,
3765 outer,
3766 cap,
3767 successor,
3768 };
3769 Poll::Ready(Ok(state.is_stale(group.sequence, &edge, budget)))
3770 })
3771 }
3772
3773 fn with_expiry(&self, group: group::Consumer) -> group::Consumer {
3774 let sequence = group.sequence;
3775 group.with_expiry(Arc::new(GroupExpiry {
3776 state: self.state.weak(),
3777 subscription: self.subscription.consume(),
3778 anchor: self.drift_anchor.consume(),
3779 bound: None,
3780 sequence,
3781 }))
3782 }
3783
3784 fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
3785 let watch = |group: &group::Consumer| match group.poll_closed(waiter) {
3793 Poll::Pending => true,
3794 Poll::Ready(()) => !group.is_aborted(),
3795 };
3796
3797 let min_sequence = self.min_sequence;
3802 self.parked
3803 .retain(|sequence, group| *sequence >= min_sequence && watch(group));
3804
3805 let drift = ready!(self.poll_drift(self.anchor(self.end_sequence), waiter))?;
3807
3808 loop {
3809 let consumer = match self.parked.keys().next().copied() {
3812 Some(sequence) if super::subscription::before_end(sequence, self.end_sequence) => {
3813 let group = self.parked.remove(&sequence).expect("just looked it up");
3814 group.cache_refresh();
3816 group
3817 }
3818 _ => {
3819 let Some((producer, found_index)) =
3820 ready!(self.poll(waiter, |state| state.poll_recv_group(self.index, self.min_sequence))?)
3821 else {
3822 if self.parked.is_empty() {
3825 return Poll::Ready(Ok(None));
3826 }
3827 return Poll::Pending;
3828 };
3829 let consumer = producer.consume();
3830 consumer.cache_refresh();
3833 self.index = found_index + 1;
3834
3835 if !super::subscription::before_end(consumer.sequence, self.end_sequence) {
3838 if watch(&consumer) {
3842 self.parked.insert(consumer.sequence, consumer);
3843 }
3844 continue;
3845 }
3846 consumer
3847 }
3848 };
3849
3850 if ready!(self.poll_stale(&consumer, &drift, waiter))? {
3853 self.stale.add(consumer.content());
3854 continue;
3855 }
3856 return Poll::Ready(Ok(Some(self.with_expiry(consumer))));
3857 }
3858 }
3859
3860 fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
3861 let Some((datagram, found_index)) =
3862 ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?)
3863 else {
3864 return Poll::Ready(Ok(None));
3865 };
3866
3867 self.datagram_index = found_index + 1;
3868 Poll::Ready(Ok(Some(datagram)))
3869 }
3870
3871 fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
3872 let floor = self.next_sequence.max(self.min_sequence);
3873 let Some(group) = ready!(self.poll_seek_group(floor, self.end_sequence, waiter))? else {
3874 return Poll::Ready(Ok(None));
3875 };
3876 self.next_sequence = group.sequence.saturating_add(1);
3877 self.commit_seek_stale(self.next_sequence);
3879 group.cache_refresh();
3881 Poll::Ready(Ok(Some(group)))
3882 }
3883
3884 fn poll_seek_group(
3897 &mut self,
3898 floor: u64,
3899 end: Option<u64>,
3900 waiter: &kio::Waiter,
3901 ) -> Poll<Result<Option<group::Consumer>>> {
3902 let mut floor = floor.max(self.min_sequence);
3903 let end = super::subscription::min_some(end, self.end_sequence);
3904 let drift = ready!(self.poll_drift(self.anchor(end), waiter))?;
3906
3907 loop {
3908 let Some(producer) = ready!(self.poll(waiter, |state| state.poll_next_in_range(floor, end))?) else {
3909 return Poll::Ready(Ok(None));
3916 };
3917 let group = producer.consume();
3918
3919 if ready!(self.poll_stale(&group, &drift, waiter))? {
3922 self.seek_pending.insert(group.sequence, group.content());
3927 floor = group.sequence.saturating_add(1);
3928 continue;
3929 }
3930
3931 self.seek_pending.remove(&group.sequence);
3934 return Poll::Ready(Ok(Some(self.with_expiry(group))));
3935 }
3936 }
3937
3938 fn commit_seek_stale(&mut self, committed: u64) {
3949 let keep = self.seek_pending.split_off(&committed);
3950 for (_, content) in std::mem::replace(&mut self.seek_pending, keep) {
3951 self.stale.add(content);
3952 }
3953 }
3954
3955 fn discard_seek_conviction(&mut self, sequence: u64) {
3959 self.seek_pending.remove(&sequence);
3960 }
3961}
3962
3963#[derive(Clone)]
3969pub struct Control {
3970 subscription: kio::Producer<Subscription>,
3971}
3972
3973impl Control {
3974 pub fn subscription(&self) -> Subscription {
3976 self.subscription.read().clone()
3977 }
3978
3979 pub fn update(&self, subscription: Subscription) -> Result<()> {
3984 let mut state = self.subscription.write().map_err(|_| Error::Closed)?;
3985 *state = subscription;
3986 Ok(())
3987 }
3988}
3989
3990impl Subscriber {
3991 pub fn info(&self) -> &Info {
3996 &self.info
3997 }
3998
3999 pub fn name(&self) -> &str {
4001 &self.name
4002 }
4003
4004 pub fn broadcast(&self) -> &broadcast::Info {
4008 &self.broadcast
4009 }
4010
4011 fn count_stale(&mut self, meter: &stats::Meter) {
4017 if meter.is_tracked() {
4022 meter.stale(self.take_stale());
4023 }
4024 }
4025
4026 pub(crate) fn take_stale(&mut self) -> stats::Content {
4029 match &mut self.inner {
4030 SubscriberKind::Plain(plain) => plain.take_stale(),
4031 SubscriberKind::Spliced(spliced) => spliced.take_stale(),
4032 }
4033 }
4034
4035 pub(crate) fn set_anchor(&mut self, anchor: Anchor) {
4046 match &mut self.inner {
4047 SubscriberKind::Plain(plain) => {
4048 plain.stale_cap = anchor.cap;
4049 plain.update_drift_anchor(anchor);
4050 }
4051 SubscriberKind::Spliced(spliced) => spliced.set_anchor(anchor),
4052 }
4053 }
4054
4055 pub(crate) fn commit_seek_stale(&mut self, committed: u64) {
4062 match &mut self.inner {
4063 SubscriberKind::Plain(plain) => plain.commit_seek_stale(committed),
4064 SubscriberKind::Spliced(spliced) => spliced.commit_seek_stale(committed),
4065 }
4066 }
4067
4068 pub(crate) fn discard_seek_conviction(&mut self, sequence: u64) {
4073 match &mut self.inner {
4074 SubscriberKind::Plain(plain) => plain.discard_seek_conviction(sequence),
4075 SubscriberKind::Spliced(spliced) => spliced.discard_seek_conviction(sequence),
4076 }
4077 }
4078
4079 pub(crate) fn poll_stale(&mut self, group: &group::Consumer, waiter: &kio::Waiter) -> Poll<Result<bool>> {
4085 let plain = match &mut self.inner {
4086 SubscriberKind::Plain(plain) => plain,
4087 SubscriberKind::Spliced(spliced) => return spliced.poll_stale(group, waiter),
4088 };
4089 let drift = ready!(plain.poll_drift(plain.anchor(plain.end_sequence), waiter))?;
4090 let stale = ready!(plain.poll_stale(group, &drift, waiter))?;
4091 if stale {
4092 plain.note_stale(group);
4093 }
4094 Poll::Ready(Ok(stale))
4095 }
4096
4097 pub fn control(&self) -> Control {
4099 Control {
4100 subscription: match &self.inner {
4101 SubscriberKind::Plain(plain) => plain.subscription.clone(),
4102 SubscriberKind::Spliced(spliced) => spliced.prefs(),
4103 },
4104 }
4105 }
4106
4107 pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4136 let meter = self.stats.meter();
4137 let res = match &mut self.inner {
4138 SubscriberKind::Plain(plain) => plain.poll_recv_group(waiter),
4139 SubscriberKind::Spliced(spliced) => spliced.poll_recv_group(waiter),
4140 };
4141 self.count_stale(&meter);
4142 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4143 }
4144
4145 pub async fn recv_group(&mut self) -> Result<Option<group::Consumer>> {
4152 kio::wait(|waiter| self.poll_recv_group(waiter)).await
4153 }
4154
4155 pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
4166 let meter = self.stats.meter();
4167 let res = match &mut self.inner {
4168 SubscriberKind::Plain(plain) => plain.poll_recv_datagram(waiter),
4169 SubscriberKind::Spliced(spliced) => spliced.poll_recv_datagram(waiter),
4170 };
4171 if let Poll::Ready(Ok(Some(datagram))) = &res {
4174 meter.datagram(datagram.payload.len() as u64);
4175 }
4176 res
4177 }
4178
4179 pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
4186 kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
4187 }
4188
4189 fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4191 let meter = self.stats.meter();
4192 let res = match &mut self.inner {
4193 SubscriberKind::Plain(plain) => plain.poll_next_group(waiter),
4194 SubscriberKind::Spliced(spliced) => spliced.poll_next_group(waiter),
4195 };
4196 self.count_stale(&meter);
4197 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4198 }
4199
4200 pub(crate) fn poll_seek_group(
4205 &mut self,
4206 floor: u64,
4207 end: Option<u64>,
4208 waiter: &kio::Waiter,
4209 ) -> Poll<Result<Option<group::Consumer>>> {
4210 let meter = self.stats.meter();
4211 let res = match &mut self.inner {
4212 SubscriberKind::Plain(plain) => plain.poll_seek_group(floor, end, waiter),
4213 SubscriberKind::Spliced(spliced) => spliced.poll_seek_group(floor, end, waiter),
4214 };
4215 self.count_stale(&meter);
4216 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4217 }
4218
4219 pub fn ordered(self) -> Ordered {
4229 Ordered { inner: self }
4230 }
4231
4232 pub fn is_clone(&self, other: &Self) -> bool {
4234 match (&self.inner, &other.inner) {
4235 (SubscriberKind::Plain(a), SubscriberKind::Plain(b)) => a.state.same_channel(&b.state),
4236 (SubscriberKind::Spliced(a), SubscriberKind::Spliced(b)) => a.is_clone(b),
4237 _ => false,
4238 }
4239 }
4240
4241 pub(crate) fn poll_start(&mut self, waiter: &kio::Waiter) -> Poll<Option<u64>> {
4245 match &mut self.inner {
4246 SubscriberKind::Plain(plain) => {
4247 let start = ready!(Consumer::poll_state_start(&plain.state, waiter));
4248 Poll::Ready(start.map(|start| start.max(plain.min_sequence)))
4249 }
4250 SubscriberKind::Spliced(spliced) => spliced.poll_start(waiter),
4251 }
4252 }
4253
4254 pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
4256 match &mut self.inner {
4257 SubscriberKind::Plain(plain) => plain.poll(waiter, |state| state.poll_finished()),
4258 SubscriberKind::Spliced(spliced) => spliced.poll_finished(waiter),
4259 }
4260 }
4261
4262 pub(crate) fn poll_end(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<u64>>> {
4264 match &mut self.inner {
4265 SubscriberKind::Plain(plain) => plain.poll(waiter, |state| state.poll_end()),
4266 SubscriberKind::Spliced(spliced) => spliced.poll_end(waiter),
4267 }
4268 }
4269
4270 pub async fn finished(&mut self) -> Result<u64> {
4279 kio::wait(|waiter| self.poll_finished(waiter)).await
4280 }
4281
4282 pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
4290 let (start, end) = super::subscription::sequence_bounds(groups);
4291 self.raise_start_to(start);
4292 self.end_at(end.map_or(Bound::Unbounded, Bound::Excluded));
4293 }
4294
4295 pub(crate) fn start_at(&mut self, sequence: u64) {
4302 match &mut self.inner {
4303 SubscriberKind::Plain(plain) => plain.min_sequence = sequence,
4304 SubscriberKind::Spliced(spliced) => spliced.start_at(sequence),
4305 }
4306 }
4307
4308 pub(crate) fn raise_start_to(&mut self, sequence: u64) {
4314 match &mut self.inner {
4315 SubscriberKind::Plain(plain) => plain.min_sequence = plain.min_sequence.max(sequence),
4316 SubscriberKind::Spliced(spliced) => spliced.raise_start_to(sequence),
4317 }
4318 }
4319
4320 pub(crate) fn end_at(&mut self, end: impl Into<Cap>) {
4334 let end = end.into();
4335 match &mut self.inner {
4336 SubscriberKind::Plain(plain) => {
4337 plain.end_sequence = end.exclusive();
4338 let outer = Anchor {
4341 cap: plain.stale_cap,
4342 ..plain.drift_anchor.read().clone()
4343 };
4344 plain.update_drift_anchor(outer);
4345 }
4346 SubscriberKind::Spliced(spliced) => spliced.end_at(end),
4347 }
4348 }
4349
4350 pub fn subscription(&self) -> Subscription {
4352 self.control().subscription()
4353 }
4354
4355 pub fn update(&mut self, subscription: Subscription) -> Result<()> {
4361 match &mut self.inner {
4362 SubscriberKind::Plain(plain) => {
4363 let mut state = plain.subscription.write().map_err(|_| Error::Closed)?;
4364 *state = subscription;
4365 }
4366 SubscriberKind::Spliced(spliced) => spliced.update(subscription),
4367 }
4368 Ok(())
4369 }
4370
4371 pub fn latest(&self) -> Option<u64> {
4373 match &self.inner {
4374 SubscriberKind::Plain(plain) => plain.state.read().max_sequence,
4375 SubscriberKind::Spliced(spliced) => spliced.latest(),
4376 }
4377 }
4378}
4379
4380pub struct Ordered {
4407 inner: Subscriber,
4408}
4409
4410impl Ordered {
4411 pub fn info(&self) -> &Info {
4413 self.inner.info()
4414 }
4415
4416 pub fn name(&self) -> &str {
4418 self.inner.name()
4419 }
4420
4421 pub fn broadcast(&self) -> &broadcast::Info {
4423 self.inner.broadcast()
4424 }
4425
4426 pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4439 self.inner.poll_next_group(waiter)
4440 }
4441
4442 pub async fn next_group(&mut self) -> Result<Option<group::Consumer>> {
4444 kio::wait(|waiter| self.poll_next_group(waiter)).await
4445 }
4446
4447 pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
4455 self.inner.poll_recv_datagram(waiter)
4456 }
4457
4458 pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
4465 kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
4466 }
4467
4468 pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
4472 self.inner.set_groups(groups);
4473 }
4474
4475 pub fn control(&self) -> Control {
4478 self.inner.control()
4479 }
4480
4481 pub fn subscription(&self) -> Subscription {
4483 self.inner.subscription()
4484 }
4485
4486 pub fn update(&mut self, subscription: Subscription) -> Result<()> {
4490 self.inner.update(subscription)
4491 }
4492
4493 pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
4495 self.inner.poll_finished(waiter)
4496 }
4497
4498 pub async fn finished(&mut self) -> Result<u64> {
4503 kio::wait(|waiter| self.poll_finished(waiter)).await
4504 }
4505
4506 pub fn latest(&self) -> Option<u64> {
4508 self.inner.latest()
4509 }
4510
4511 pub fn is_clone(&self, other: &Self) -> bool {
4513 self.inner.is_clone(&other.inner)
4514 }
4515}
4516
4517pub struct Request {
4529 name: Arc<str>,
4530 broadcast: Arc<broadcast::Info>,
4532 state: kio::Producer<TrackState>,
4533
4534 prev_subscription: Option<Subscription>,
4536
4537 alive: Arc<Alive>,
4540
4541 _dynamic: Dynamic,
4546
4547 stats: stats::Scope,
4550
4551 resolving_start: bool,
4554}
4555
4556impl Request {
4557 pub(crate) fn new(broadcast: Arc<broadcast::Info>, name: impl Into<Arc<str>>) -> Self {
4558 let name = name.into();
4559 let state = TrackState::spawn(broadcast.clone());
4560 let alive = Alive::new(name.clone(), state.clone());
4561 let dynamic = Dynamic::new(name.clone(), state.clone(), alive.clone());
4562 Self {
4563 name,
4564 broadcast,
4565 state,
4566 prev_subscription: None,
4567 alive,
4568 _dynamic: dynamic,
4569 stats: stats::Scope::default(),
4570 resolving_start: false,
4571 }
4572 }
4573
4574 pub(crate) fn resolving_start(mut self) -> Self {
4579 self.resolving_start = true;
4580 self
4581 }
4582
4583 pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
4586 self.stats = scope;
4587 self
4588 }
4589
4590 pub fn name(&self) -> &str {
4592 &self.name
4593 }
4594
4595 pub fn consume(&self) -> Consumer {
4597 Consumer::plain(self.name.clone(), self.state.consume())
4598 }
4599
4600 pub fn dynamic(&self) -> Dynamic {
4604 Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
4605 }
4606
4607 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
4610 self.state.poll_unused(waiter).map(|_| ())
4611 }
4612
4613 pub(crate) fn claim(self) -> Self {
4615 if let Ok(mut state) = self.state.write() {
4616 state.claimed = true;
4617 }
4618 self
4619 }
4620
4621 pub(crate) fn reject_unused(&self, err: Error) -> bool {
4625 match self.state.write_unused() {
4626 kio::Unused::Idle(guard) => {
4627 commit_abort(guard, err);
4628 true
4629 }
4630 kio::Unused::Closed => true,
4631 kio::Unused::Used => false,
4632 }
4633 }
4634
4635 pub fn accept(self, info: impl Into<Option<Info>>) -> Producer {
4642 let info = TrackState::normalize_info(&self.broadcast, info.into().unwrap_or_default());
4643 if let Ok(mut state) = self.state.write() {
4646 state.accept(info.clone());
4647 state.start_pending = self.resolving_start;
4648 }
4649 self.alive.publish(Some(&self.stats));
4652 Producer {
4653 name: self.name,
4654 info,
4655 broadcast: self.broadcast,
4656 state: self.state,
4657 prev_subscription: None,
4658 alive: self.alive,
4659 stats: self.stats,
4660 }
4661 }
4662
4663 pub fn reject(self, err: Error) {
4665 if let Ok(mut state) = self.state.write() {
4666 state.abort = Some(err);
4667 }
4668 }
4669
4670 pub fn subscription(&self) -> Option<Subscription> {
4673 let state = self.state.read();
4674 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
4675 drop(state);
4676 snapshot_subscription(&subs, bound)
4677 }
4678
4679 pub async fn subscription_changed(&mut self) -> Option<Subscription> {
4682 kio::wait(|waiter| self.poll_subscription_changed(waiter)).await
4683 }
4684
4685 pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll<Option<Subscription>> {
4687 let state = self.state.read();
4688 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
4689 drop(state);
4690
4691 poll_combined_changed(&subs, bound, &mut self.prev_subscription, waiter)
4692 }
4693
4694 pub(super) fn weak(&self) -> TrackWeak {
4695 TrackWeak {
4696 name: self.name.clone(),
4697 state: self.state.weak(),
4698 }
4699 }
4700}
4701
4702#[cfg(test)]
4703use futures::FutureExt;
4704
4705#[cfg(test)]
4706#[allow(missing_docs)] impl Subscriber {
4708 pub fn assert_group(&mut self) -> group::Consumer {
4709 self.recv_group()
4710 .now_or_never()
4711 .expect("group would have blocked")
4712 .expect("would have errored")
4713 .expect("track was closed")
4714 }
4715
4716 pub fn assert_no_group(&mut self) {
4717 assert!(
4718 self.recv_group().now_or_never().is_none(),
4719 "recv_group would not have blocked"
4720 );
4721 }
4722
4723 pub fn assert_not_closed(&mut self) {
4724 assert!(self.finished().now_or_never().is_none(), "should not be closed");
4725 }
4726
4727 pub fn assert_closed(&mut self) {
4728 assert!(self.finished().now_or_never().is_some(), "should be closed");
4729 }
4730
4731 pub fn assert_error(&mut self) {
4733 assert!(
4734 self.finished().now_or_never().expect("should not block").is_err(),
4735 "should be error"
4736 );
4737 }
4738
4739 pub fn assert_is_clone(&self, other: &Self) {
4740 assert!(self.is_clone(other), "should be clone");
4741 }
4742
4743 pub fn assert_not_clone(&self, other: &Self) {
4744 assert!(!self.is_clone(other), "should not be clone");
4745 }
4746}
4747
4748#[cfg(test)]
4749mod test {
4750 use super::*;
4751 use crate::frame;
4752 use crate::model::test_tracing::count_drop_warnings;
4753 use std::time::Duration;
4754
4755 fn track_producer(name: impl Into<Arc<str>>, info: impl Into<Option<Info>>) -> Producer {
4758 Producer::new(Arc::new(broadcast::Info::default()), name, info)
4759 }
4760
4761 fn replay() -> Subscription {
4763 Subscription::default().with_max_age(Duration::from_secs(30))
4764 }
4765
4766 fn live_groups(state: &TrackState) -> usize {
4768 state.lookup.len()
4769 }
4770
4771 fn first_live_sequence(state: &TrackState) -> u64 {
4773 state
4774 .arrival
4775 .iter()
4776 .find(|(sequence, stamp)| state.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp))
4777 .map(|(sequence, _)| *sequence)
4778 .unwrap()
4779 }
4780
4781 fn recv_datagram(dg: &mut Subscriber) -> Datagram {
4783 dg.recv_datagram()
4784 .now_or_never()
4785 .expect("datagram would have blocked")
4786 .expect("would have errored")
4787 .expect("track was closed")
4788 }
4789
4790 #[tokio::test]
4794 async fn peek_resolves_below_the_declared_start() {
4795 let mut producer = track_producer("test", None);
4796 let consumer = producer.consume();
4797
4798 let mut cached = producer.create_group(group::Info { sequence: 1 }).unwrap();
4799 cached.write_frame(Timestamp::ZERO, b"backfill".to_vec()).unwrap();
4800 cached.finish().unwrap();
4801
4802 let waiter = kio::Waiter::noop();
4804 assert!(consumer.poll_peek_group(0, &waiter).is_pending());
4805
4806 producer.start_at(3).unwrap();
4809 assert!(matches!(consumer.poll_peek_group(0, &waiter), Poll::Ready(None)));
4810 assert!(matches!(consumer.poll_peek_group(1, &waiter), Poll::Ready(Some(_))));
4811 assert!(consumer.poll_peek_group(3, &waiter).is_pending());
4812
4813 producer.start_at(4).unwrap();
4816 assert!(matches!(consumer.poll_peek_group(3, &waiter), Poll::Ready(None)));
4817 producer.start_at(0).unwrap();
4818 assert!(consumer.poll_peek_group(0, &waiter).is_pending());
4819 }
4820
4821 #[tokio::test]
4822 async fn append_datagram_shares_group_sequence() {
4823 let mut producer = track_producer("test", None);
4824 let ts = Timestamp::from_millis(10).unwrap();
4825
4826 assert_eq!(producer.append_group().unwrap().sequence, 0);
4828 assert_eq!(producer.append_datagram(ts, &b"a"[..]).unwrap(), 1);
4829 assert_eq!(producer.append_group().unwrap().sequence, 2);
4830 assert_eq!(producer.append_datagram(ts, &b"b"[..]).unwrap(), 3);
4831 assert_eq!(producer.latest(), Some(3));
4832 }
4833
4834 #[tokio::test]
4835 async fn append_datagram_roundtrip() {
4836 let mut producer = track_producer("test", None);
4837 let mut dg = producer.subscribe(None);
4838
4839 let ts = Timestamp::from_millis(42).unwrap();
4840 let seq = producer.append_datagram(ts, &b"hello"[..]).unwrap();
4841
4842 let got = recv_datagram(&mut dg);
4843 assert_eq!(got.sequence, seq);
4844 assert_eq!(got.timestamp, ts);
4845 assert_eq!(&got.payload[..], b"hello");
4846 }
4847
4848 #[tokio::test]
4849 async fn insert_datagram_preserves_sequence() {
4850 let mut producer = track_producer("test", None);
4851 let mut dg = producer.subscribe(None);
4852
4853 let ts = Timestamp::from_millis(5).unwrap();
4854 producer
4856 .insert_datagram(100, ts, bytes::Bytes::from_static(b"x"))
4857 .unwrap();
4858
4859 assert_eq!(recv_datagram(&mut dg).sequence, 100);
4860 assert_eq!(producer.append_group().unwrap().sequence, 101);
4862 }
4863
4864 #[tokio::test]
4865 async fn insert_datagram_leaves_a_gap() {
4866 let mut producer = track_producer("test", None);
4867 let mut dg = producer.subscribe(None);
4868 let ts = Timestamp::from_millis(0).unwrap();
4869
4870 producer
4871 .insert_datagram(10, ts, bytes::Bytes::from_static(b"gap"))
4872 .unwrap();
4873 assert_eq!(recv_datagram(&mut dg).sequence, 10);
4874 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
4875 assert_eq!(producer.append_group().unwrap().sequence, 12);
4876 }
4877
4878 #[tokio::test]
4879 async fn insert_datagram_out_of_order_does_not_rewind() {
4880 let mut producer = track_producer("test", None);
4881 let mut dg = producer.subscribe(None);
4882 let ts = Timestamp::from_millis(0).unwrap();
4883
4884 producer
4885 .insert_datagram(10, ts, bytes::Bytes::from_static(b"high"))
4886 .unwrap();
4887 producer
4888 .insert_datagram(5, ts, bytes::Bytes::from_static(b"low"))
4889 .unwrap();
4890
4891 assert_eq!(recv_datagram(&mut dg).sequence, 10);
4892 assert_eq!(recv_datagram(&mut dg).sequence, 5);
4893 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
4894 }
4895
4896 #[tokio::test]
4897 async fn insert_datagram_duplicate_is_best_effort() {
4898 let mut producer = track_producer("test", None);
4899 let mut dg = producer.subscribe(None);
4900 let ts = Timestamp::from_millis(0).unwrap();
4901
4902 producer
4903 .insert_datagram(3, ts, bytes::Bytes::from_static(b"first"))
4904 .unwrap();
4905 producer
4906 .insert_datagram(3, ts, bytes::Bytes::from_static(b"again"))
4907 .unwrap();
4908
4909 assert_eq!(&recv_datagram(&mut dg).payload[..], b"first");
4910 assert_eq!(&recv_datagram(&mut dg).payload[..], b"again");
4911 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 4);
4912 }
4913
4914 #[tokio::test]
4915 async fn insert_datagram_stale_does_not_rewind_after_append() {
4916 let mut producer = track_producer("test", None);
4917 let mut dg = producer.subscribe(None);
4918 let ts = Timestamp::from_millis(0).unwrap();
4919
4920 assert_eq!(producer.append_datagram(ts, &b"0"[..]).unwrap(), 0);
4921 assert_eq!(producer.append_datagram(ts, &b"1"[..]).unwrap(), 1);
4922 producer
4923 .insert_datagram(0, ts, bytes::Bytes::from_static(b"stale"))
4924 .unwrap();
4925
4926 assert_eq!(recv_datagram(&mut dg).sequence, 0);
4927 assert_eq!(recv_datagram(&mut dg).sequence, 1);
4928 assert_eq!(recv_datagram(&mut dg).sequence, 0);
4929 assert_eq!(producer.append_datagram(ts, &b"2"[..]).unwrap(), 2);
4930 assert_eq!(producer.append_group().unwrap().sequence, 3);
4931 }
4932
4933 #[tokio::test]
4934 async fn insert_datagram_cloned_producers_share_counter() {
4935 let mut producer = track_producer("test", None);
4936 let mut other = producer.clone();
4937 let mut dg = producer.subscribe(None);
4938 let ts = Timestamp::from_millis(0).unwrap();
4939
4940 producer
4941 .insert_datagram(4, ts, bytes::Bytes::from_static(b"a"))
4942 .unwrap();
4943 assert_eq!(other.append_datagram(ts, &b"b"[..]).unwrap(), 5);
4944 other.insert_datagram(8, ts, bytes::Bytes::from_static(b"c")).unwrap();
4945 assert_eq!(producer.append_group().unwrap().sequence, 9);
4946
4947 assert_eq!(recv_datagram(&mut dg).sequence, 4);
4948 assert_eq!(recv_datagram(&mut dg).sequence, 5);
4949 assert_eq!(recv_datagram(&mut dg).sequence, 8);
4950 }
4951
4952 #[test]
4953 fn insert_datagram_after_finish_is_closed() {
4954 let mut producer = track_producer("test", None);
4955 let ts = Timestamp::from_millis(0).unwrap();
4956 producer.finish().unwrap();
4957 assert!(matches!(
4958 producer.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")),
4959 Err(Error::Closed)
4960 ));
4961 assert!(matches!(producer.append_datagram(ts, &b"x"[..]), Err(Error::Closed)));
4962 }
4963
4964 #[test]
4965 fn insert_datagram_after_abort_fails() {
4966 let producer = track_producer("test", None);
4967 let mut other = producer.clone();
4968 let ts = Timestamp::from_millis(0).unwrap();
4969 producer.abort(Error::Cancel).unwrap();
4970 assert!(other.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")).is_err());
4971 }
4972
4973 #[test]
4974 fn insert_datagram_respects_finish_at() {
4975 let mut producer = track_producer("test", None);
4976 let ts = Timestamp::from_millis(0).unwrap();
4977 producer.finish_at(10).unwrap();
4978 producer
4979 .insert_datagram(5, ts, bytes::Bytes::from_static(b"ok"))
4980 .unwrap();
4981 assert!(matches!(
4982 producer.insert_datagram(10, ts, bytes::Bytes::from_static(b"late")),
4983 Err(Error::Closed)
4984 ));
4985 assert_eq!(producer.append_group().unwrap().sequence, 6);
4986 }
4987
4988 #[test]
4990 fn resume_position_uses_the_latest_group() {
4991 let mut datagram_only = track_producer("datagram-only", None);
4992 let datagram_only_consumer = datagram_only.consume();
4993 datagram_only
4994 .insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
4995 .unwrap();
4996 assert_eq!(
4997 datagram_only_consumer.resume_position(),
4998 None,
4999 "a datagram creates no group position to resume"
5000 );
5001
5002 let mut producer = track_producer("mixed", None);
5003 let consumer = producer.consume();
5004 let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
5005 group
5006 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
5007 .unwrap();
5008 producer
5009 .insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
5010 .unwrap();
5011
5012 assert_eq!(
5013 consumer.resume_position(),
5014 Some(Position { group: 3, frame: 1 }),
5015 "the replacement must continue the open group"
5016 );
5017 group.abort(Error::Cancel).unwrap();
5018 }
5019
5020 #[tokio::test]
5023 async fn recv_datagram_leaves_the_ordered_cursor_alone() {
5024 let mut producer = track_producer("test", None);
5025 let mut datagrams = producer.subscribe(None);
5026 let mut subscriber = producer.subscribe(None).ordered();
5027 let ts = Timestamp::from_millis(5).unwrap();
5028
5029 producer
5030 .insert_datagram(5, ts, bytes::Bytes::from_static(b"x"))
5031 .unwrap();
5032 assert_eq!(recv_datagram(&mut datagrams).sequence, 5);
5033
5034 producer.create_group(group::Info { sequence: 3 }).unwrap();
5035 producer.create_group(group::Info { sequence: 6 }).unwrap();
5036
5037 let mut next = || {
5038 subscriber
5039 .next_group()
5040 .now_or_never()
5041 .expect("group would have blocked")
5042 .expect("would have errored")
5043 .expect("track was closed")
5044 .sequence
5045 };
5046 assert_eq!(next(), 3, "the datagram at sequence 5 did not consume group 3");
5047 assert_eq!(next(), 6);
5048 }
5049
5050 #[tokio::test]
5051 async fn datagram_normalized_to_track_timescale() {
5052 let info = Info::default().with_timescale(Timescale::MICRO);
5053 let mut producer = track_producer("test", info);
5054 let mut dg = producer.subscribe(None);
5055
5056 producer
5058 .append_datagram(Timestamp::from_millis(2).unwrap(), &b"z"[..])
5059 .unwrap();
5060 let got = recv_datagram(&mut dg);
5061 assert_eq!(got.timestamp.scale(), Timescale::MICRO);
5062 assert_eq!(got.timestamp.value(), 2_000);
5063 }
5064
5065 #[tokio::test]
5066 async fn datagram_rejects_oversized() {
5067 let mut producer = track_producer("test", None);
5068 let big = bytes::Bytes::from(vec![0u8; crate::model::datagram::MAX_DATAGRAM_PAYLOAD + 1]);
5069 let ts = Timestamp::from_millis(0).unwrap();
5070 assert!(matches!(
5071 producer.append_datagram(ts, big.clone()),
5072 Err(Error::FrameTooLarge)
5073 ));
5074 assert!(matches!(
5075 producer.insert_datagram(0, ts, big),
5076 Err(Error::FrameTooLarge)
5077 ));
5078 }
5079
5080 #[tokio::test]
5081 async fn datagram_fanout_to_subscribers() {
5082 let mut producer = track_producer("test", None);
5083 let mut a = producer.subscribe(None);
5085 let mut b = producer.subscribe(None);
5086 let ts = Timestamp::from_millis(1).unwrap();
5087
5088 producer.append_datagram(ts, &b"first"[..]).unwrap();
5089 producer.append_datagram(ts, &b"second"[..]).unwrap();
5090
5091 assert_eq!(&recv_datagram(&mut a).payload[..], b"first");
5093 assert_eq!(&recv_datagram(&mut a).payload[..], b"second");
5094 assert_eq!(&recv_datagram(&mut b).payload[..], b"first");
5095 assert_eq!(&recv_datagram(&mut b).payload[..], b"second");
5096 }
5097
5098 #[test]
5099 fn datagram_buffer_drops_oldest_at_capacity() {
5100 let mut producer = track_producer("test", None);
5101 let mut slow = producer.subscribe(None);
5102 let mut fast = producer.subscribe(None);
5103 let count = MAX_DATAGRAMS * 3;
5104 for sequence in 0..count {
5105 producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();
5106 assert_eq!(recv_datagram(&mut fast).sequence, sequence as u64);
5107 }
5108 assert_eq!(producer.state.read().datagrams.len(), MAX_DATAGRAMS);
5109 for sequence in count - MAX_DATAGRAMS..count {
5110 assert_eq!(recv_datagram(&mut slow).sequence, sequence as u64);
5111 }
5112 assert!(slow.poll_recv_datagram(&kio::Waiter::noop()).is_pending());
5113 }
5114
5115 #[tokio::test]
5116 async fn datagram_recv_pends_until_written() {
5117 let mut producer = track_producer("test", None);
5118 let mut dg = producer.subscribe(None);
5119
5120 assert!(
5121 dg.recv_datagram().now_or_never().is_none(),
5122 "should block with no datagrams"
5123 );
5124
5125 producer
5126 .append_datagram(Timestamp::from_millis(0).unwrap(), &b"go"[..])
5127 .unwrap();
5128 assert_eq!(&recv_datagram(&mut dg).payload[..], b"go");
5129 }
5130
5131 #[tokio::test]
5135 async fn datagram_wire_roundtrip_between_tracks() {
5136 use crate::coding::{Decode, Encode};
5137 use crate::lite;
5138
5139 let version = lite::Version::Lite05;
5140
5141 let mut origin = track_producer("test", None);
5143 let mut origin_dg = origin.subscribe(None);
5144 let ts = Timestamp::from_millis(7).unwrap();
5145 let seq = origin.append_datagram(ts, &b"payload"[..]).unwrap();
5146
5147 let d = recv_datagram(&mut origin_dg);
5148 let body = lite::Datagram {
5149 subscribe: 5,
5150 sequence: d.sequence,
5151 timestamp: d.timestamp.value(),
5152 payload: d.payload.clone(),
5153 }
5154 .encode_bytes(version)
5155 .unwrap();
5156
5157 let mut slice = &body[..];
5159 let wire = lite::Datagram::decode(&mut slice, version).unwrap();
5160 let mut downstream = track_producer("test", None);
5161 let mut downstream_dg = downstream.subscribe(None);
5162 downstream
5163 .insert_datagram(
5164 wire.sequence,
5165 Timestamp::new(wire.timestamp, Timescale::MILLI).unwrap(),
5166 wire.payload,
5167 )
5168 .unwrap();
5169
5170 let got = recv_datagram(&mut downstream_dg);
5171 assert_eq!(got.sequence, seq);
5172 assert_eq!(got.timestamp, ts);
5173 assert_eq!(&got.payload[..], b"payload");
5174 }
5175
5176 #[tokio::test]
5177 async fn evict_expired_groups() {
5178 let producer = track_producer("test", None);
5179
5180 producer.append_group().unwrap(); producer.append_group().unwrap(); producer.append_group().unwrap(); {
5186 let state = producer.state.read();
5187 assert_eq!(live_groups(&state), 3);
5188 assert_eq!(state.offset, 0);
5189 }
5190
5191 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5193
5194 producer.append_group().unwrap(); {
5201 let state = producer.state.read();
5202 assert_eq!(live_groups(&state), 1);
5203 assert_eq!(first_live_sequence(&state), 3);
5204 assert_eq!(state.offset, 3);
5205 assert!(!state.lookup.contains_key(&0));
5206 assert!(!state.lookup.contains_key(&1));
5207 assert!(!state.lookup.contains_key(&2));
5208 assert!(state.lookup.contains_key(&3));
5209 }
5210 }
5211
5212 #[tokio::test]
5216 async fn aging_out_a_finished_group_keeps_the_clean_end() {
5217 let producer = track_producer("test", None);
5218 let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
5219 let mut consumer = group.consume();
5220
5221 group
5222 .write_frame(Timestamp::from_millis(0).unwrap(), b"hello".as_slice())
5223 .unwrap();
5224 assert_eq!(consumer.next_frame().await.unwrap().unwrap().size, 5);
5225
5226 crate::model::clock::advance(cache::DEFAULT_EXPIRY * 2);
5228 group.finish().unwrap();
5229 let _next = producer.create_group(group::Info { sequence: 1 }).unwrap();
5230
5231 assert!(consumer.next_frame().await.unwrap().is_none());
5232 }
5233
5234 #[tokio::test]
5238 async fn active_reader_survives_expiry() {
5239 let producer = track_producer("test", None);
5240 let mut subscriber = producer.subscribe(None);
5241
5242 let mut group = producer.create_group(0u64.into()).unwrap();
5244 for _ in 0..10 {
5245 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5246 }
5247 group.finish().unwrap();
5248 let mut reading = subscriber.assert_group();
5249
5250 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5252
5253 for seq in 2..12u64 {
5256 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5257 let frame = reading.next_frame().await;
5258 assert!(
5259 matches!(frame, Ok(Some(_))),
5260 "an actively-read group must not expire mid-read (step {seq})"
5261 );
5262 producer.create_group(seq.into()).unwrap().finish().unwrap();
5263 }
5264
5265 let state = producer.state.read();
5266 assert!(state.lookup.contains_key(&0), "the read group survived");
5267 assert!(!state.lookup.contains_key(&1), "the unread group still expired");
5268 }
5269
5270 #[tokio::test]
5275 async fn slow_prefetch_reader_survives_expiry() {
5276 let producer = track_producer("test", None);
5277 let mut subscriber = producer.subscribe(None);
5278
5279 let mut group = producer.create_group(0u64.into()).unwrap();
5280 for _ in 0..20 {
5281 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5282 }
5283 group.finish().unwrap();
5284 let mut reading = subscriber.assert_group();
5285
5286 for seq in 1..20u64 {
5289 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5290 let frame = reading.read_frame().await;
5291 assert!(
5292 matches!(frame, Ok(Some(_))),
5293 "a slow prefetch reader must not expire mid-read (step {seq})"
5294 );
5295 producer.create_group(seq.into()).unwrap().finish().unwrap();
5296 }
5297 }
5298
5299 #[tokio::test]
5303 async fn delivery_restarts_the_expiry_clock() {
5304 let producer = track_producer("test", None);
5305 let mut subscriber = producer.subscribe(replay());
5306
5307 let mut group = producer.create_group(0u64.into()).unwrap();
5308 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5309 group.finish().unwrap();
5310 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5312
5313 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5315 let mut reading = subscriber.assert_group();
5316
5317 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5320 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5321
5322 let frame = reading.read_frame().await.unwrap();
5323 assert!(frame.is_some(), "a just-delivered group must not expire unread");
5324 }
5325
5326 #[tokio::test]
5330 async fn streaming_frame_writes_keep_the_group_alive() {
5331 let producer = track_producer("test", None);
5332 let mut straggler = producer.create_group(0u64.into()).unwrap();
5333 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5335
5336 let mut frame = straggler
5337 .create_frame(frame::Info {
5338 size: 10,
5339 timestamp: Timestamp::ZERO,
5340 })
5341 .unwrap();
5342 for seq in 2..12u64 {
5345 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5346 frame.write(bytes::Bytes::from_static(b"x")).unwrap();
5347 producer.create_group(seq.into()).unwrap().finish().unwrap();
5348 }
5349 frame.finish().unwrap();
5350 straggler.finish().unwrap();
5351
5352 let state = producer.state.read();
5353 assert!(
5354 state.lookup.contains_key(&0),
5355 "a group streaming a frame survives expiry"
5356 );
5357 }
5358
5359 #[tokio::test]
5364 async fn coalesced_frame_completion_keeps_the_group_alive() {
5365 let producer = track_producer("test", None);
5366 let mut straggler = producer.create_group(0u64.into()).unwrap();
5367 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5369
5370 let mut frame = straggler
5371 .create_frame_owned(
5372 frame::Info {
5373 size: 3,
5374 timestamp: Timestamp::ZERO,
5375 },
5376 &Default::default(),
5377 )
5378 .unwrap();
5379
5380 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5384 frame.write(bytes::Bytes::from_static(b"abc")).unwrap();
5385 frame.finish().unwrap();
5386 straggler.finish().unwrap();
5387
5388 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5390
5391 let state = producer.state.read();
5392 assert!(
5393 state.lookup.contains_key(&0),
5394 "a group whose frame just completed must not expire"
5395 );
5396 }
5397
5398 #[tokio::test]
5401 async fn parked_reoffer_restarts_the_expiry_clock() {
5402 let producer = track_producer("test", None);
5403 let mut subscriber = producer.subscribe(None);
5404 subscriber.set_groups(..1);
5405
5406 for seq in 0..2u64 {
5407 let mut group = producer.create_group(seq.into()).unwrap();
5408 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5409 group.finish().unwrap();
5410 }
5411
5412 assert_eq!(subscriber.assert_group().sequence, 0);
5414 subscriber.assert_no_group();
5415
5416 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5418 subscriber.set_groups(..2);
5419 let mut reading = subscriber.assert_group();
5420 assert_eq!(reading.sequence, 1);
5421
5422 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5425 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5426
5427 let frame = reading.read_frame().await.unwrap();
5428 assert!(frame.is_some(), "a just-re-offered group must not expire unread");
5429 }
5430
5431 #[tokio::test]
5432 async fn evict_keeps_max_sequence() {
5433 let producer = track_producer("test", None);
5434 producer.append_group().unwrap(); crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5438
5439 producer.append_group().unwrap(); {
5443 let state = producer.state.read();
5444 assert_eq!(live_groups(&state), 1);
5445 assert_eq!(first_live_sequence(&state), 1);
5446 assert_eq!(state.offset, 1);
5447 }
5448 }
5449
5450 #[tokio::test]
5451 async fn no_eviction_when_fresh() {
5452 let producer = track_producer("test", None);
5453 producer.append_group().unwrap(); producer.append_group().unwrap(); producer.append_group().unwrap(); {
5458 let state = producer.state.read();
5459 assert_eq!(live_groups(&state), 3);
5460 assert_eq!(state.offset, 0);
5461 }
5462 }
5463
5464 #[tokio::test]
5465 async fn consumer_skips_evicted_groups() {
5466 let producer = track_producer("test", None);
5467 producer.append_group().unwrap(); let mut consumer = producer.subscribe(None);
5470
5471 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5472 producer.append_group().unwrap(); let group = consumer.assert_group();
5476 assert_eq!(group.sequence, 1);
5477 }
5478
5479 fn track_producer_expiring(name: impl Into<Arc<str>>, expiry: impl Into<Option<Duration>>) -> Producer {
5481 track_producer_pooled(name, cache::Pool::new(cache::Config::default().with_expiry(expiry)))
5482 }
5483
5484 fn track_producer_pooled(name: impl Into<Arc<str>>, pool: cache::Pool) -> Producer {
5486 Producer::new(
5487 Arc::new(broadcast::Info {
5488 pool,
5489 ..Default::default()
5490 }),
5491 name,
5492 None,
5493 )
5494 }
5495
5496 #[tokio::test]
5500 async fn pool_sweep_expires_without_a_write() {
5501 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5502 let producer = track_producer_pooled("test", pool.clone());
5503 let mut stalled = producer.append_group().unwrap(); stalled.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5505 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5508 pool.sweep();
5509
5510 assert!(
5511 !producer.state.read().lookup.contains_key(&0),
5512 "the sweep reclaimed an idle open group with no write behind it"
5513 );
5514 }
5515
5516 #[test]
5517 fn cache_gc_dates_activity_before_expiring_it() {
5518 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5519 let now = crate::model::clock::now();
5520 assert_eq!(pool.gc(now), Some(now + Duration::from_millis(500)));
5521 let producer = track_producer_pooled("test", pool.clone());
5522 let mut group = producer.append_group().unwrap();
5523 group.write_frame(Timestamp::ZERO, b"first".as_slice()).unwrap();
5524 producer.append_group().unwrap();
5525 let later = now + Duration::from_secs(60);
5527 pool.gc(later);
5528 assert!(producer.state.read().lookup.contains_key(&0));
5529 pool.gc(later + Duration::from_secs(2));
5530 assert!(!producer.state.read().lookup.contains_key(&0));
5531 }
5532
5533 #[test]
5534 fn cache_gc_reaches_old_entries_behind_a_fresh_front() {
5535 let expiry = Duration::from_secs(1);
5536 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
5537 let producer = track_producer_pooled("test", pool.clone());
5538 let now = crate::model::clock::now();
5539 let groups: Vec<_> = (0..EVICT_SCAN * 3).map(|_| producer.append_group().unwrap()).collect();
5540 producer.append_group().unwrap();
5541 pool.gc(now);
5542 for group in &groups[..EVICT_SCAN * 2] {
5544 group.cache_refresh();
5545 }
5546 pool.gc(now + expiry * 2);
5547 let state = producer.state.read();
5548 for sequence in 0..EVICT_SCAN * 2 {
5549 assert!(state.lookup.contains_key(&(sequence as u64)), "fresh front survives");
5550 }
5551 for sequence in EVICT_SCAN * 2..EVICT_SCAN * 3 {
5552 assert!(!state.lookup.contains_key(&(sequence as u64)), "old tail is reclaimed");
5553 }
5554 }
5555
5556 #[tokio::test]
5560 async fn pool_sweep_drains_a_deep_backlog() {
5561 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5562 let producer = track_producer_pooled("test", pool.clone());
5563
5564 let backlog = 4 * EVICT_SCAN;
5566 for _ in 0..backlog {
5567 let mut group = producer.append_group().unwrap();
5568 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5569 }
5570 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5573 pool.sweep();
5574
5575 let state = producer.state.read();
5576 let stale = (0..backlog as u64).filter(|seq| state.lookup.contains_key(seq)).count();
5577 assert_eq!(stale, 0, "one sweep reclaimed the whole idle backlog");
5578 }
5579
5580 #[tokio::test]
5581 async fn pool_expiry_controls_eviction() {
5582 let producer = track_producer_expiring("test", Duration::from_secs(1));
5584 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5588 producer.append_group().unwrap(); let state = producer.state.read();
5592 assert_eq!(live_groups(&state), 1);
5593 assert_eq!(first_live_sequence(&state), 1);
5594 }
5595
5596 #[tokio::test]
5597 async fn small_frame_write_expires_idle_siblings() {
5598 let producer = track_producer_expiring("test", Duration::from_secs(1));
5599 producer.append_group().unwrap().finish().unwrap(); let mut live = producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5603 live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5604
5605 let expired = !producer.state.read().lookup.contains_key(&0);
5606 assert!(expired, "a small frame write runs expiry");
5607 }
5608
5609 #[tokio::test]
5610 async fn fresh_expiry_scan_does_not_wake_track_consumers() {
5611 use std::sync::atomic::{AtomicBool, Ordering};
5612
5613 let producer = track_producer_expiring("test", cache::DEFAULT_EXPIRY);
5614 producer.append_group().unwrap().finish().unwrap();
5615 let mut live = producer.append_group().unwrap();
5616 let mut consumer = producer.subscribe(None);
5617 assert_eq!(consumer.assert_group().sequence, 0);
5618 assert_eq!(consumer.assert_group().sequence, 1);
5619
5620 let woken = Arc::new(AtomicBool::new(false));
5621 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
5622 assert!(consumer.poll_recv_group(&waiter).is_pending());
5623
5624 live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5625 assert!(
5626 !woken.load(Ordering::SeqCst),
5627 "a no-op expiry scan must not wake track consumers"
5628 );
5629 }
5630
5631 #[tokio::test]
5632 async fn streaming_frame_write_expires_idle_siblings() {
5633 let producer = track_producer_expiring("test", Duration::from_secs(1));
5634 producer.append_group().unwrap().finish().unwrap(); let mut live = producer.append_group().unwrap(); let mut frame = live
5637 .create_frame(frame::Info {
5638 size: 1,
5639 timestamp: Timestamp::ZERO,
5640 })
5641 .unwrap();
5642
5643 crate::model::clock::advance(Duration::from_secs(2));
5644 frame.write(b"x".as_slice()).unwrap();
5645
5646 let expired = !producer.state.read().lookup.contains_key(&0);
5647 assert!(expired, "a streamed chunk runs expiry");
5648 }
5649
5650 #[tokio::test]
5651 async fn appended_datagram_expires_idle_groups() {
5652 let mut producer = track_producer_expiring("test", Duration::from_secs(1));
5653 producer.append_group().unwrap().finish().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5657 producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();
5658
5659 let expired = !producer.state.read().lookup.contains_key(&0);
5660 assert!(expired, "an appended datagram runs expiry");
5661 }
5662
5663 #[tokio::test]
5664 async fn forwarded_datagram_expires_idle_groups() {
5665 let mut producer = track_producer_expiring("test", Duration::from_secs(1));
5666 producer.append_group().unwrap().finish().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5670 producer
5671 .insert_datagram(2, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
5672 .unwrap();
5673
5674 let expired = !producer.state.read().lookup.contains_key(&0);
5675 assert!(expired, "a forwarded datagram runs expiry");
5676 }
5677
5678 #[tokio::test]
5682 async fn max_age_does_not_drive_wall_eviction() {
5683 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(1)));
5684 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(10));
5688 producer.append_group().unwrap(); let state = producer.state.read();
5691 assert_eq!(live_groups(&state), 2, "max_age is media time, not a wall clock");
5692 }
5693
5694 #[tokio::test]
5696 async fn disabled_pool_expiry_never_reclaims() {
5697 let producer = track_producer_expiring("test", None);
5698 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(3600));
5701 producer.append_group().unwrap(); let state = producer.state.read();
5704 assert_eq!(live_groups(&state), 2);
5705 }
5706
5707 #[test]
5708 fn max_age_clamped_to_cache() {
5709 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5710
5711 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
5715 assert_eq!(subscriber.subscription().max_age, Duration::from_secs(10));
5716 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5717
5718 subscriber
5720 .update(Subscription::default().with_max_age(Duration::from_millis(500)))
5721 .unwrap();
5722 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_millis(500));
5723
5724 subscriber
5725 .update(Subscription::default().with_max_age(Duration::ZERO))
5726 .unwrap();
5727 assert_eq!(producer.subscription().unwrap().max_age, Duration::ZERO);
5728 }
5729
5730 fn track_producer_capped(name: impl Into<Arc<str>>, info: Info, cap: Duration) -> Producer {
5733 Producer::new(
5734 Arc::new(broadcast::Info {
5735 cache_duration: cap,
5736 ..Default::default()
5737 }),
5738 name,
5739 info,
5740 )
5741 }
5742
5743 #[test]
5744 fn origin_cache_duration_clamps_max_age() {
5745 let capped = track_producer_capped(
5748 "test",
5749 Info::default().with_max_age(Duration::from_secs(60)),
5750 Duration::from_secs(1),
5751 );
5752 assert_eq!(capped.state.read().max_age_bound(), Some(Duration::from_secs(1)));
5753 assert_eq!(capped.subscribe(None).info().max_age, Duration::from_secs(1));
5754
5755 let under = track_producer_capped(
5756 "test",
5757 Info::default().with_max_age(Duration::from_millis(500)),
5758 Duration::from_secs(1),
5759 );
5760 assert_eq!(under.state.read().max_age_bound(), Some(Duration::from_millis(500)));
5761 }
5762
5763 #[tokio::test]
5766 async fn origin_cache_duration_does_not_wall_evict() {
5767 let producer = track_producer_capped(
5768 "test",
5769 Info::default().with_max_age(Duration::from_secs(60)),
5770 Duration::from_secs(1),
5771 );
5772 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5776 producer.append_group().unwrap(); let state = producer.state.read();
5779 assert_eq!(live_groups(&state), 2);
5780 }
5781
5782 #[test]
5783 fn max_age_clamped_via_every_update_path() {
5784 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5785 let over = Subscription::default().with_max_age(Duration::from_secs(10));
5786
5787 let mut subscriber = producer.subscribe(over.clone());
5790 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5791
5792 subscriber.control().update(over.clone()).unwrap();
5793 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5794
5795 subscriber.update(over).unwrap();
5796 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5797 }
5798
5799 #[test]
5800 fn max_age_aggregate_clamps_across_subscribers() {
5801 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5802
5803 let _a = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
5806 let _b = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
5807
5808 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5809 }
5810
5811 #[test]
5812 fn churned_subscribers_do_not_accumulate() {
5813 let producer = track_producer("test", None);
5814 let consumer = producer.consume();
5815 let _steady = producer.subscribe(None);
5816
5817 for _ in 0..100 {
5819 drop(producer.subscribe(None));
5820 drop(consumer.subscribe(None));
5821 }
5822
5823 let subs = producer.state.read().subscriptions.clone();
5826 let len = subs.read().len();
5827 assert!(len <= 8, "departed subscribers accumulated: {len}");
5828 }
5829
5830 #[test]
5831 fn aggregate_poll_prunes_after_a_peak() {
5832 let mut producer = track_producer("test", None);
5833 let waiter = kio::Waiter::noop();
5834 let _steady = producer.subscribe(None);
5835 assert!(producer.poll_subscription_changed(&waiter).is_ready());
5836
5837 drop((0..1000).map(|_| producer.subscribe(None)).collect::<Vec<_>>());
5840 for _ in 0..100 {
5841 drop(producer.subscribe(None));
5842 assert!(producer.poll_subscription_changed(&waiter).is_pending());
5843
5844 let subs = producer.state.read().subscriptions.clone();
5845 assert_eq!(subs.read().len(), 1, "the poll walked past departed subscribers");
5846 }
5847 }
5848
5849 fn append_at(producer: &mut Producer, millis: u64) -> u64 {
5852 let mut group = producer.append_group().unwrap();
5853 group
5854 .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
5855 .unwrap();
5856 group.finish().unwrap();
5857 group.sequence
5858 }
5859
5860 fn drain(subscriber: &mut Subscriber) -> Vec<u64> {
5862 let mut sequences = Vec::new();
5863 while let Some(Ok(Some(group))) = subscriber.recv_group().now_or_never() {
5864 sequences.push(group.sequence);
5865 }
5866 sequences
5867 }
5868
5869 #[test]
5870 fn real_time_skips_a_backlog_to_the_live_edge() {
5871 let mut producer = track_producer("test", None);
5872 for second in 0..5 {
5873 append_at(&mut producer, second * 1000);
5874 }
5875
5876 let mut subscriber = producer.subscribe(None);
5880 assert_eq!(drain(&mut subscriber), vec![4]);
5881
5882 append_at(&mut producer, 5000);
5884 assert_eq!(drain(&mut subscriber), vec![5]);
5885 }
5886
5887 #[test]
5888 fn real_time_skips_a_backlog_after_catching_up() {
5889 let mut producer = track_producer("test", None);
5890 append_at(&mut producer, 0);
5891 let mut subscriber = producer.subscribe(None);
5892 assert_eq!(drain(&mut subscriber), vec![0]);
5893
5894 for second in 1..6 {
5897 append_at(&mut producer, second * 1000);
5898 }
5899 assert_eq!(drain(&mut subscriber), vec![5]);
5900 }
5901
5902 #[test]
5903 fn a_newer_edge_changes_an_active_catch_up() {
5904 let mut producer = track_producer("test", None);
5905 for second in 0..5 {
5906 append_at(&mut producer, second * 1000);
5907 }
5908 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
5909 assert_eq!(
5910 subscriber
5911 .recv_group()
5912 .now_or_never()
5913 .unwrap()
5914 .unwrap()
5915 .unwrap()
5916 .sequence,
5917 3
5918 );
5919
5920 append_at(&mut producer, 5000);
5923 append_at(&mut producer, 10000);
5924 assert_eq!(drain(&mut subscriber), vec![5, 6]);
5925 }
5926
5927 #[test]
5928 fn a_growing_edge_changes_an_active_catch_up() {
5929 let mut producer = track_producer("test", None);
5930 append_at(&mut producer, 0);
5931 append_at(&mut producer, 1000);
5932 let mut edge = producer.append_group().unwrap();
5933 edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"a"))
5934 .unwrap();
5935
5936 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
5937 assert_eq!(
5938 subscriber
5939 .recv_group()
5940 .now_or_never()
5941 .unwrap()
5942 .unwrap()
5943 .unwrap()
5944 .sequence,
5945 0
5946 );
5947
5948 edge.write_frame(Timestamp::from_millis(5000).unwrap(), bytes::Bytes::from_static(b"b"))
5951 .unwrap();
5952 assert_eq!(drain(&mut subscriber), vec![2]);
5953 }
5954
5955 #[test]
5956 fn a_budget_admits_groups_within_it() {
5957 let mut producer = track_producer("test", None);
5958 for second in 0..5 {
5959 append_at(&mut producer, second * 1000);
5960 }
5961
5962 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
5967 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
5968 }
5969
5970 #[test]
5971 fn a_budget_reaches_back_over_the_cache() {
5972 let mut producer = track_producer("test", None);
5973 for second in 0..5 {
5974 append_at(&mut producer, second * 1000);
5975 }
5976
5977 let mut live = producer.subscribe(None);
5980 assert_eq!(drain(&mut live), vec![4]);
5981
5982 let budget = Subscription::default().with_max_age(Duration::from_secs(2));
5987 let mut subscriber = producer.subscribe(budget);
5988 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
5989 }
5990
5991 #[test]
5992 fn a_named_start_is_a_floor_not_a_request() {
5993 let mut producer = track_producer("test", None);
5994 for second in 0..5 {
5995 append_at(&mut producer, second * 1000);
5996 }
5997
5998 let named = Subscription::default().with_start(Position::group(1));
6002 let mut subscriber = producer.subscribe(named);
6003 assert_eq!(drain(&mut subscriber), vec![4]);
6004
6005 let floored = Subscription::default()
6007 .with_start(Position::group(3))
6008 .with_max_age(Duration::from_secs(10));
6009 let mut subscriber = producer.subscribe(floored);
6010 assert_eq!(drain(&mut subscriber), vec![3, 4]);
6011
6012 let slack = Subscription::default()
6014 .with_start(Position::group(1))
6015 .with_max_age(Duration::from_secs(2));
6016 let mut subscriber = producer.subscribe(slack);
6017 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
6018 }
6019
6020 #[test]
6021 fn a_floor_above_the_live_edge_waits_there() {
6022 let mut producer = track_producer("test", None);
6023 for second in 0..3 {
6024 append_at(&mut producer, second * 1000);
6025 }
6026
6027 let resumed = Subscription::default()
6030 .with_start(Position::group(7))
6031 .with_max_age(Duration::from_secs(10));
6032 let mut subscriber = producer.subscribe(resumed);
6033 assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
6034 append_at(&mut producer, 3000); assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
6036 for second in 4..8 {
6037 append_at(&mut producer, second * 1000);
6038 }
6039 assert_eq!(drain(&mut subscriber), vec![7]);
6040 }
6041
6042 #[test]
6043 fn a_late_lower_group_within_the_budget_is_delivered() {
6044 let producer = track_producer("test", None);
6045 for (sequence, millis) in [(5, 0), (6, 1000), (7, 2000)] {
6046 let mut group = producer.create_group(group::Info { sequence }).unwrap();
6047 group
6048 .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
6049 .unwrap();
6050 group.finish().unwrap();
6051 }
6052
6053 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5)));
6054 assert_eq!(drain(&mut subscriber), vec![5, 6, 7]);
6055
6056 let mut late = producer.create_group(group::Info { sequence: 4 }).unwrap();
6060 late.write_frame(Timestamp::from_millis(500).unwrap(), bytes::Bytes::from_static(b"late"))
6061 .unwrap();
6062 late.finish().unwrap();
6063 assert_eq!(drain(&mut subscriber), vec![4]);
6064 }
6065
6066 #[test]
6067 fn drift_is_measured_in_presentation_time_not_arrival_time() {
6068 let mut producer = track_producer("test", None);
6069 for second in 0..4 {
6073 append_at(&mut producer, second * 1000);
6074 }
6075
6076 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)));
6077 assert_eq!(drain(&mut subscriber), vec![1, 2, 3]);
6078 }
6079
6080 #[tokio::test]
6081 async fn a_stamped_successor_expires_an_unstamped_group() {
6082 let mut producer = track_producer("test", None);
6083 let mut subscriber = producer.subscribe(None);
6084 producer.append_group().unwrap(); append_at(&mut producer, 1000); assert_eq!(drain(&mut subscriber), vec![1]);
6091 }
6092
6093 #[tokio::test]
6094 async fn a_handed_out_group_expires_while_its_first_frame_is_stalled() {
6095 let mut producer = track_producer("test", None);
6096 let mut subscriber = producer.subscribe(None);
6097 producer.append_group().unwrap();
6098
6099 let mut stalled = subscriber.recv_group().await.unwrap().expect("stalled group");
6100 let pending = tokio::spawn(async move { stalled.read_frame().await });
6101 tokio::task::yield_now().await;
6102 assert!(
6103 !pending.is_finished(),
6104 "the empty live edge still waits for its first frame"
6105 );
6106
6107 crate::model::clock::advance(Duration::from_secs(1));
6108 append_at(&mut producer, 1000);
6109
6110 let result = pending.await.unwrap();
6115 assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
6116 }
6117
6118 #[tokio::test]
6121 async fn aborted_stamped_successor_wakes_a_parked_read() {
6122 tokio::time::pause();
6123 let mut producer = track_producer("test", None);
6124 let mut head = producer.append_group().unwrap();
6125 head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap();
6126 let mut successor = producer.append_group().unwrap();
6127 successor
6128 .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice())
6129 .unwrap();
6130 append_at(&mut producer, 1000);
6131 append_at(&mut producer, 20_000);
6132 let mut sub = producer.subscribe(None);
6133 let mut reading = sub.recv_group().await.unwrap().unwrap();
6134 assert_eq!(reading.sequence, 0, "the rewound successor extends the reach");
6135 assert!(reading.read_frame().await.unwrap().is_some());
6136
6137 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6138 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6139 let mut cx = std::task::Context::from_waker(&waker);
6140 let mut next = std::pin::pin!(reading.read_frame());
6141 assert!(next.as_mut().poll(&mut cx).is_pending());
6142 successor.abort(Error::Cancel).unwrap();
6143 assert!(
6144 woken.load(Ordering::SeqCst),
6145 "the stamped successor's abort lost its wakeup"
6146 );
6147 let result = next.as_mut().poll(&mut cx);
6148 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6149 }
6150
6151 #[tokio::test]
6156 async fn a_handed_out_group_wakes_when_a_newer_group_gets_its_first_timestamp() {
6157 let mut producer = track_producer("test", None);
6158 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6159 let mut old = producer.append_group().unwrap();
6160 old.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"old"))
6161 .unwrap();
6162
6163 append_at(&mut producer, 1000);
6165
6166 let mut held = subscriber.recv_group().await.unwrap().expect("old group");
6167 assert!(held.read_frame().await.unwrap().is_some());
6168
6169 let mut live = producer.append_group().unwrap();
6170 let pending = tokio::spawn(async move { held.read_frame().await });
6171 tokio::task::yield_now().await;
6172 assert!(!pending.is_finished(), "the newer group has no timestamp yet");
6173
6174 live.write_frame(
6176 Timestamp::from_millis(2000).unwrap(),
6177 bytes::Bytes::from_static(b"live"),
6178 )
6179 .unwrap();
6180 tokio::task::yield_now().await;
6181
6182 assert!(pending.is_finished(), "the new presentation edge wakes the held reader");
6183 let result = pending.await.unwrap();
6185 assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
6186 }
6187
6188 #[test]
6193 fn a_parked_read_ignores_first_frames_between_its_successor_and_the_edge() {
6194 let mut producer = track_producer("test", None);
6195 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
6196 let mut head = producer.append_group().unwrap();
6197 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6198 .unwrap();
6199 append_at(&mut producer, 1000); let mut between = producer.append_group().unwrap();
6201 append_at(&mut producer, 3000); let mut held = subscriber
6204 .recv_group()
6205 .now_or_never()
6206 .unwrap()
6207 .unwrap()
6208 .expect("head group");
6209 assert_eq!(held.sequence, 0);
6210 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6211
6212 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6213 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6214 let mut cx = std::task::Context::from_waker(&waker);
6215 let mut next = std::pin::pin!(held.read_frame());
6216 assert!(next.as_mut().poll(&mut cx).is_pending());
6217
6218 between
6219 .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x"))
6220 .unwrap();
6221 assert!(
6222 !woken.load(Ordering::SeqCst),
6223 "a first frame below the edge moves neither the reach nor the edge"
6224 );
6225
6226 let mut beyond = producer.append_group().unwrap();
6228 assert!(next.as_mut().poll(&mut cx).is_pending());
6229 woken.store(false, Ordering::SeqCst);
6230 beyond
6231 .write_frame(Timestamp::from_millis(4000).unwrap(), bytes::Bytes::from_static(b"x"))
6232 .unwrap();
6233 assert!(woken.load(Ordering::SeqCst), "a new edge wakes the parked read");
6234 }
6235
6236 #[test]
6239 fn a_parked_read_watches_the_replacement_for_an_aborted_successor() {
6240 let mut producer = track_producer("test", None);
6241 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6242 let mut head = producer.append_group().unwrap();
6243 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6244 .unwrap();
6245 let successor = producer.append_group().unwrap();
6246 let mut replacement = producer.append_group().unwrap();
6247 append_at(&mut producer, 20_000); let mut held = subscriber
6250 .recv_group()
6251 .now_or_never()
6252 .unwrap()
6253 .unwrap()
6254 .expect("head group");
6255 assert_eq!(held.sequence, 0, "an unstamped successor leaves the reach unbounded");
6256 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6257
6258 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6259 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6260 let mut cx = std::task::Context::from_waker(&waker);
6261 let mut next = std::pin::pin!(held.read_frame());
6262 assert!(next.as_mut().poll(&mut cx).is_pending());
6263
6264 successor.abort(Error::Cancel).unwrap();
6265 assert!(
6266 woken.load(Ordering::SeqCst),
6267 "the successor's abort wakes the parked read"
6268 );
6269 woken.store(false, Ordering::SeqCst);
6270 assert!(
6271 next.as_mut().poll(&mut cx).is_pending(),
6272 "the unstamped replacement leaves the reach unbounded"
6273 );
6274
6275 replacement
6276 .write_frame(Timestamp::from_millis(1000).unwrap(), bytes::Bytes::from_static(b"x"))
6277 .unwrap();
6278 assert!(
6279 woken.load(Ordering::SeqCst),
6280 "the replacement's first frame bounds the reach"
6281 );
6282 let result = next.as_mut().poll(&mut cx);
6283 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6284 }
6285
6286 #[test]
6290 fn a_parked_read_watches_its_edge_abort() {
6291 let mut producer = track_producer("test", None);
6292 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5)));
6293 let mut head = producer.append_group().unwrap();
6294 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6295 .unwrap();
6296 append_at(&mut producer, 1000); let mut between = producer.append_group().unwrap();
6298 let mut edge = producer.append_group().unwrap();
6299 edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x"))
6300 .unwrap();
6301 edge.finish().unwrap();
6302
6303 let mut held = subscriber
6304 .recv_group()
6305 .now_or_never()
6306 .unwrap()
6307 .unwrap()
6308 .expect("head group");
6309 assert_eq!(held.sequence, 0);
6310 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6311
6312 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6313 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6314 let mut cx = std::task::Context::from_waker(&waker);
6315 let mut next = std::pin::pin!(held.read_frame());
6316 assert!(next.as_mut().poll(&mut cx).is_pending());
6317
6318 edge.abort(Error::Cancel).unwrap();
6319 between
6320 .write_frame(Timestamp::from_millis(10_000).unwrap(), bytes::Bytes::from_static(b"x"))
6321 .unwrap();
6322 assert!(
6323 woken.load(Ordering::SeqCst),
6324 "the edge's abort, then a new edge below it, lost the wakeup"
6325 );
6326 let result = next.as_mut().poll(&mut cx);
6327 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6328 }
6329
6330 #[tokio::test]
6341 async fn real_time_reads_a_live_stream_without_truncating_it() {
6342 let producer = track_producer("test", None);
6343 let mut subscriber = producer.subscribe(None);
6344
6345 let gop = |n: u64| {
6346 [
6347 Timestamp::from_millis(n * 2000).unwrap(),
6348 Timestamp::from_millis(n * 2000 + 1900).unwrap(),
6349 ]
6350 };
6351 let write = |group: &mut group::Producer, timestamp| {
6352 group.write_frame(timestamp, bytes::Bytes::from_static(b"x")).unwrap();
6353 };
6354
6355 let mut open = producer.append_group().unwrap();
6356 write(&mut open, gop(0)[0]);
6357 write(&mut open, gop(0)[1]);
6358 let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");
6359
6360 let mut read = Vec::new();
6361 for n in 1..5u64 {
6362 let sequence = reading.sequence;
6363 let mut frames = 0;
6364 while let Some(res) = reading.read_frame().now_or_never() {
6365 match res.expect("no truncation while draining") {
6366 Some(_) => frames += 1,
6367 None => panic!("group {sequence} ended early"),
6368 }
6369 }
6370 read.push((sequence, frames));
6371
6372 let next = {
6373 let mut end = std::pin::pin!(reading.read_frame());
6376 assert!(futures::poll!(end.as_mut()).is_pending(), "parked on the FIN");
6377
6378 let mut opened = producer.append_group().unwrap();
6381 write(&mut opened, gop(n)[0]);
6382 let verdict = futures::poll!(end.as_mut());
6383
6384 open.finish().unwrap();
6385 let res = match verdict {
6386 Poll::Ready(res) => res,
6387 Poll::Pending => end.await,
6388 };
6389 assert!(
6390 matches!(res, Ok(None)),
6391 "group {sequence} ends at the boundary rather than failing: {res:?}"
6392 );
6393 opened
6394 };
6395 let mut next = next;
6396 write(&mut next, gop(n)[1]);
6397
6398 reading = subscriber.recv_group().await.unwrap().expect("the next live group");
6399 open = next;
6400 }
6401
6402 assert_eq!(read, vec![(0, 2), (1, 2), (2, 2), (3, 2)], "every frame of every group");
6403 }
6404
6405 #[tokio::test]
6415 async fn a_budget_is_measured_from_the_readers_position() {
6416 let producer = track_producer("test", None);
6417 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
6418
6419 let mut open = producer.append_group().unwrap();
6420 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"key"))
6421 .unwrap();
6422 open.write_frame(
6423 Timestamp::from_millis(1900).unwrap(),
6424 bytes::Bytes::from_static(b"tail"),
6425 )
6426 .unwrap();
6427
6428 let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");
6429 assert!(reading.read_frame().await.unwrap().is_some());
6430 assert!(reading.read_frame().await.unwrap().is_some());
6431
6432 let mut end = std::pin::pin!(reading.read_frame());
6433 assert!(futures::poll!(end.as_mut()).is_pending(), "parked at 1900ms");
6434
6435 let mut next = producer.append_group().unwrap();
6437 next.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"key"))
6438 .unwrap();
6439 assert!(
6440 futures::poll!(end.as_mut()).is_pending(),
6441 "a reader inside its budget is not expired by the next group opening"
6442 );
6443
6444 open.write_frame(
6446 Timestamp::from_millis(1950).unwrap(),
6447 bytes::Bytes::from_static(b"late"),
6448 )
6449 .unwrap();
6450 let late = end.await.expect("the straggler is not truncated");
6451 assert_eq!(
6452 late.map(|frame| frame.timestamp),
6453 Some(Timestamp::from_millis(1950).unwrap())
6454 );
6455 }
6456
6457 #[tokio::test]
6461 async fn an_ended_group_stays_ended_when_probed_again() {
6462 let mut producer = track_producer("test", None);
6463 let mut subscriber = producer.subscribe(None);
6464 let mut open = producer.append_group().unwrap();
6465 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
6466 .unwrap();
6467
6468 let mut group = subscriber.recv_group().await.unwrap().expect("group");
6469 assert!(group.read_frame().await.unwrap().is_some());
6470
6471 let probes = tokio::spawn(async move {
6472 let first = group.read_frame().await;
6473 let second = group.read_frame().await;
6474 let finished = group.finished().await;
6475 (first, second, finished)
6476 });
6477 tokio::task::yield_now().await;
6478
6479 append_at(&mut producer, 1000);
6480
6481 let (first, second, finished) = probes.await.unwrap();
6482 assert!(matches!(first, Ok(None)), "the group ends: {first:?}");
6483 assert!(matches!(second, Ok(None)), "and stays ended: {second:?}");
6484 assert!(matches!(finished, Ok(1)), "reporting what it delivered: {finished:?}");
6485 }
6486
6487 #[tokio::test]
6488 async fn a_drained_group_finishes_cleanly_after_the_live_edge_advances() {
6489 let mut producer = track_producer("test", None);
6490 let mut subscriber = producer.subscribe(None);
6491 append_at(&mut producer, 0);
6492
6493 let mut group = subscriber.recv_group().await.unwrap().expect("first group");
6494 assert!(group.read_frame().await.unwrap().is_some());
6495
6496 append_at(&mut producer, 1000);
6497
6498 assert!(group.read_frame().await.unwrap().is_none());
6499 assert!(!group.latency_expired());
6500 }
6501
6502 #[tokio::test]
6503 async fn a_handed_out_partial_frame_expires_while_its_payload_is_stalled() {
6504 let mut producer = track_producer("test", None);
6505 let mut subscriber = producer.subscribe(None);
6506 let mut source = producer.append_group().unwrap();
6507 let mut writing = source
6508 .create_frame(frame::Info {
6509 size: 6,
6510 timestamp: Timestamp::ZERO,
6511 })
6512 .unwrap();
6513 writing.write(bytes::Bytes::from_static(b"old")).unwrap();
6514
6515 let mut group = subscriber.recv_group().await.unwrap().expect("partial group");
6516 let mut frame = group.next_frame().await.unwrap().expect("partial frame");
6517 assert_eq!(
6518 frame.read_chunk().await.unwrap(),
6519 Some(bytes::Bytes::from_static(b"old"))
6520 );
6521 let pending = tokio::spawn(async move { frame.read_chunk().await });
6522 tokio::task::yield_now().await;
6523 assert!(!pending.is_finished(), "the partial payload is still stalled");
6524
6525 crate::model::clock::advance(Duration::from_secs(1));
6526 append_at(&mut producer, 1000);
6527
6528 let result = pending.await.unwrap();
6529 assert!(
6530 matches!(result, Err(Error::Old)),
6531 "the in-flight frame expires: {result:?}"
6532 );
6533 writing.abort(Error::Cancel).unwrap();
6534 }
6535
6536 #[test]
6537 fn max_age_bounds_the_budget() {
6538 let mut producer = track_producer("test", Info::default().with_max_age(Duration::from_millis(500)));
6542 append_at(&mut producer, 0);
6543 append_at(&mut producer, 1000);
6544 append_at(&mut producer, 2000);
6545
6546 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
6549 assert_eq!(drain(&mut subscriber), vec![1, 2]);
6550 }
6551
6552 #[test]
6553 fn an_explicit_start_gets_no_exemption_from_the_budget() {
6554 let mut producer = track_producer("test", None);
6555 for second in 0..4 {
6556 append_at(&mut producer, second * 1000);
6557 }
6558
6559 let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(0)));
6562 subscriber.start_at(0);
6563 assert_eq!(drain(&mut subscriber), vec![3]);
6564
6565 let mut patient = producer.subscribe(
6566 Subscription::default()
6567 .with_start(Position::group(0))
6568 .with_max_age(Duration::from_secs(10)),
6569 );
6570 patient.start_at(0);
6571 assert_eq!(drain(&mut patient), vec![0, 1, 2, 3]);
6572 }
6573
6574 #[test]
6578 fn a_long_group_is_not_stale_while_its_tail_reaches_the_edge() {
6579 let mut producer = track_producer("test", None);
6580
6581 let mut long = producer.append_group().unwrap();
6583 for ms in [0u64, 500, 1000, 1500, 2000] {
6584 long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
6585 .unwrap();
6586 }
6587 long.finish().unwrap();
6588 append_at(&mut producer, 2000);
6589
6590 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6592 assert_eq!(drain(&mut sub), vec![0, 1]);
6593 }
6594
6595 #[test]
6601 fn a_long_group_is_stale_once_its_successor_falls_behind() {
6602 let mut producer = track_producer("test", None);
6603
6604 let mut long = producer.append_group().unwrap();
6605 for ms in [0u64, 500, 1000] {
6606 long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
6607 .unwrap();
6608 }
6609 long.finish().unwrap();
6610 append_at(&mut producer, 3000);
6613 append_at(&mut producer, 4000);
6614
6615 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6616 assert_eq!(drain(&mut sub), vec![1, 2]);
6617 }
6618
6619 #[tokio::test]
6623 async fn an_unstamped_immediate_successor_leaves_reach_unbounded() {
6624 let mut producer = track_producer("test", None);
6625 let mut subscriber = producer.subscribe(None);
6626
6627 append_at(&mut producer, 0); producer.append_group().unwrap(); append_at(&mut producer, 10_000); assert_eq!(drain(&mut subscriber), vec![0, 2]);
6635 }
6636
6637 #[test]
6643 fn reach_follows_the_immediate_successor_not_a_later_rewind() {
6644 let mut producer = track_producer("test", None);
6645
6646 append_at(&mut producer, 0);
6648 append_at(&mut producer, 10_000);
6649 append_at(&mut producer, 1_000);
6650 append_at(&mut producer, 2_000);
6651
6652 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6656 assert!(
6657 drain(&mut sub).contains(&0),
6658 "group 0 is bounded by its successor at 10s, not by a later rewind"
6659 );
6660 }
6661
6662 #[tokio::test]
6665 async fn ordered_carries_datagrams() {
6666 let mut producer = track_producer("test", None);
6667 let mut sub = producer.subscribe(None).ordered();
6668
6669 producer
6670 .insert_datagram(5, Timestamp::from_millis(5).unwrap(), bytes::Bytes::from_static(b"x"))
6671 .unwrap();
6672 producer.create_group(group::Info { sequence: 3 }).unwrap();
6673
6674 let datagram = sub
6675 .recv_datagram()
6676 .now_or_never()
6677 .expect("datagram would have blocked")
6678 .expect("would have errored")
6679 .expect("track was closed");
6680 assert_eq!(datagram.sequence, 5);
6681
6682 let group = sub
6684 .next_group()
6685 .now_or_never()
6686 .expect("group would have blocked")
6687 .expect("would have errored")
6688 .expect("track was closed");
6689 assert_eq!(group.sequence, 3);
6690 }
6691
6692 fn drain_ordered(subscriber: &mut Ordered) -> Vec<u64> {
6694 let mut sequences = Vec::new();
6695 while let Some(Ok(Some(group))) = subscriber.next_group().now_or_never() {
6696 sequences.push(group.sequence);
6697 }
6698 sequences
6699 }
6700
6701 #[test]
6705 fn next_group_sheds_a_stale_backlog() {
6706 let mut producer = track_producer("test", None);
6707 for second in 0..4 {
6708 append_at(&mut producer, second * 1000);
6709 }
6710
6711 let mut subscriber = producer.subscribe(None).ordered();
6712 assert_eq!(drain_ordered(&mut subscriber), vec![3]);
6713
6714 let mut arrival = producer.subscribe(None);
6716 assert_eq!(drain(&mut arrival), vec![3]);
6717 }
6718
6719 #[test]
6723 fn next_group_keeps_a_backlog_inside_the_budget() {
6724 let mut producer = track_producer("test", None);
6725 for second in 0..4 {
6726 append_at(&mut producer, second * 1000);
6727 }
6728
6729 let mut subscriber = producer
6732 .subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)))
6733 .ordered();
6734 assert_eq!(drain_ordered(&mut subscriber), vec![1, 2, 3]);
6735
6736 let mut replay = producer.subscribe(replay()).ordered();
6738 assert_eq!(drain_ordered(&mut replay), vec![0, 1, 2, 3]);
6739 }
6740
6741 #[test]
6744 fn next_group_keeps_a_group_with_no_proven_reach() {
6745 let mut producer = track_producer("test", None);
6746 append_at(&mut producer, 0); producer.append_group().unwrap(); append_at(&mut producer, 10_000); let mut subscriber = producer.subscribe(None).ordered();
6753 assert_eq!(drain_ordered(&mut subscriber), vec![0, 2]);
6754 }
6755
6756 #[tokio::test]
6757 async fn real_time_skips_older_sequences_with_equal_ages() {
6758 let mut producer = track_producer("test", None);
6759 append_at(&mut producer, 0);
6760 append_at(&mut producer, 0);
6761
6762 let mut subscriber = producer.subscribe(None);
6763 assert_eq!(drain(&mut subscriber), vec![1]);
6764 }
6765
6766 #[test]
6767 fn fetch_ignores_the_budget() {
6768 let mut producer = track_producer("test", None);
6769 for second in 0..4 {
6770 append_at(&mut producer, second * 1000);
6771 }
6772
6773 let consumer = producer.consume();
6776 let group = consumer.fetch_group(0, None).now_or_never().unwrap().unwrap();
6777 assert_eq!(group.sequence, 0);
6778 }
6779
6780 #[tokio::test]
6783 async fn fetched_group_is_not_a_live_drift_edge() {
6784 let mut producer = track_producer("test", None);
6785 let dynamic = producer.dynamic();
6786 let consumer = producer.consume();
6787 append_at(&mut producer, 0);
6788
6789 let pending = consumer.fetch_group(100, None);
6790 let req = dynamic
6791 .requested_group()
6792 .now_or_never()
6793 .expect("fetch request is ready")
6794 .unwrap();
6795 let mut fetched = req.accept(None).unwrap();
6796 fetched
6797 .write_frame(
6798 Timestamp::from_millis(100_000).unwrap(),
6799 bytes::Bytes::from_static(b"fetched"),
6800 )
6801 .unwrap();
6802 fetched.finish().unwrap();
6803 pending.await.unwrap();
6804
6805 let mut groups = producer.subscribe(None);
6806 assert_eq!(groups.assert_group().sequence, 0);
6807 groups.assert_no_group();
6808 }
6809
6810 #[test]
6815 fn an_evicted_live_edge_convicts_nothing() {
6816 let mut producer = track_producer("test", None);
6817 append_at(&mut producer, 0);
6818 let edge = append_at(&mut producer, 30_000);
6819
6820 let state = producer.state.read();
6821 let drift = Drift {
6822 budget: Duration::ZERO,
6823 edge: state.drift_edge(None, None, None),
6824 outer: None,
6825 successor: None,
6826 };
6827 assert!(
6828 state.is_stale(0, &drift.edge, drift.budget),
6829 "stale against a live edge"
6830 );
6831 drop(state);
6832
6833 let slot = producer.modify().unwrap().lookup.remove(&edge).unwrap();
6835 let _ = slot.group.abort(Error::Evicted);
6836
6837 let state = producer.state.read();
6838 assert!(
6839 !state.is_stale(0, &drift.edge, drift.budget),
6840 "a vanished edge is no reason to drop what is left"
6841 );
6842 }
6843
6844 #[tokio::test]
6848 async fn an_evicted_outer_edge_convicts_nothing() {
6849 let mut producer = track_producer("a", None);
6850 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
6851 let mut open = producer.append_group().unwrap();
6852 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
6853 .unwrap();
6854 let mut group = subscriber.recv_group().await.unwrap().expect("group");
6855 assert!(group.read_frame().await.unwrap().is_some());
6856 append_at(&mut producer, 10);
6858
6859 let mut next = track_producer("b", None);
6860 append_at(&mut next, 30_000);
6861 let edge = append_at(&mut next, 30_010);
6862 subscriber.set_anchor(Anchor {
6863 cap: None,
6864 edge: next.consume().live_edge(None),
6865 successor: None,
6866 });
6867
6868 let mut control = group.clone();
6869 assert!(
6870 matches!(control.read_frame().now_or_never(), Some(Ok(None))),
6871 "the open group ends against the outer edge"
6872 );
6873
6874 let slot = next.modify().unwrap().lookup.remove(&edge).unwrap();
6876 let _ = slot.group.abort(Error::Evicted);
6877
6878 assert!(
6879 group.read_frame().now_or_never().is_none(),
6880 "a vanished outer edge is no reason to drop what is left"
6881 );
6882 assert!(!group.latency_expired());
6883 open.finish().unwrap();
6884 }
6885
6886 #[tokio::test]
6887 async fn a_lower_sequence_is_never_the_live_edge() {
6888 let producer = track_producer("test", None);
6889 let mut straggler = producer.create_group(0u64.into()).unwrap();
6894 straggler
6895 .write_frame(Timestamp::from_millis(60_000).unwrap(), bytes::Bytes::from_static(b"x"))
6896 .unwrap();
6897 straggler.finish().unwrap();
6898
6899 let mut rewound = producer.create_group(1u64.into()).unwrap();
6900 rewound
6901 .write_frame(Timestamp::from_millis(0).unwrap(), bytes::Bytes::from_static(b"x"))
6902 .unwrap();
6903 rewound.finish().unwrap();
6904
6905 let mut subscriber = producer.subscribe(replay());
6906 assert_eq!(drain(&mut subscriber), vec![0, 1]);
6907 }
6908
6909 #[test]
6910 fn a_requested_end_does_not_cap_the_live_edge() {
6911 let mut producer = track_producer("test", None);
6912 for second in 0..4 {
6913 append_at(&mut producer, second * 1000);
6914 }
6915
6916 let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(1)));
6921 assert_eq!(drain(&mut subscriber), vec![3]);
6922 }
6923
6924 #[test]
6925 fn a_capped_subscriber_measures_drift_against_its_cap() {
6926 let mut producer = track_producer("test", None);
6927 append_at(&mut producer, 0);
6928 append_at(&mut producer, 1000);
6929
6930 let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(0)));
6935 subscriber.set_groups(..1);
6936 assert_eq!(drain(&mut subscriber), vec![0]);
6937
6938 subscriber.set_groups(..);
6941 assert_eq!(drain(&mut subscriber), vec![1]);
6942 }
6943
6944 #[test]
6945 fn subscriber_control_updates_while_read_future_is_pending() {
6946 let producer = track_producer("test", None);
6947 let mut subscriber = producer.subscribe(None);
6948 let control = subscriber.control();
6949
6950 let mut recv = Box::pin(subscriber.recv_group());
6951 assert!(recv.as_mut().now_or_never().is_none());
6952
6953 control.update(Subscription::default().with_priority(7)).unwrap();
6954
6955 let aggregate = producer.subscription().expect("expected an active subscription");
6956 assert_eq!(aggregate.priority, 7);
6957 }
6958
6959 #[test]
6960 fn dropped_subscriber_leaves_no_ghost_in_aggregate() {
6961 let mut producer = track_producer("test", None);
6966 let a = producer.subscribe(Subscription::default().with_priority(5));
6967
6968 let waiter = kio::Waiter::noop();
6970 assert!(
6971 matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(Some(_)))),
6972 "one live subscriber should aggregate to Some",
6973 );
6974
6975 drop(a);
6977
6978 assert!(
6980 matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(None))),
6981 "a dropped subscriber must not linger in the aggregate",
6982 );
6983
6984 assert!(
6986 producer.subscription().is_none(),
6987 "snapshot must exclude a dropped subscriber",
6988 );
6989 }
6990
6991 #[test]
6992 fn dropped_subscriber_wakes_the_aggregate() {
6993 use std::sync::atomic::{AtomicBool, Ordering};
7000
7001 let mut producer = track_producer("test", None);
7002 let a = producer.subscribe(Subscription::default().with_priority(5));
7003
7004 let woken = Arc::new(AtomicBool::new(false));
7005 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
7006
7007 assert!(matches!(
7009 producer.poll_subscription_changed(&waiter),
7010 Poll::Ready(Ok(Some(_)))
7011 ));
7012 assert!(
7013 producer.poll_subscription_changed(&waiter).is_pending(),
7014 "the aggregate is unchanged, so this poll must park",
7015 );
7016 assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");
7017
7018 drop(a);
7019 assert!(
7020 woken.load(Ordering::SeqCst),
7021 "the last subscriber leaving must wake the aggregate watcher",
7022 );
7023 }
7024
7025 #[test]
7026 fn widest_subscriber_update_wakes_the_aggregate() {
7027 use std::sync::atomic::{AtomicBool, Ordering};
7033
7034 let mut producer = track_producer("test", None);
7035 let _narrow = producer.subscribe(Subscription::default().with_end(Position::after_group(3)));
7036 let mut wide = producer.subscribe(Subscription::default());
7037
7038 let woken = Arc::new(AtomicBool::new(false));
7039 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
7040
7041 assert!(matches!(
7042 producer.poll_subscription_changed(&waiter),
7043 Poll::Ready(Ok(Some(_)))
7044 ));
7045 assert!(producer.poll_subscription_changed(&waiter).is_pending());
7046 assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");
7047
7048 wide.update(Subscription::default().with_end(Position::after_group(5)))
7049 .unwrap();
7050 assert!(
7051 woken.load(Ordering::SeqCst),
7052 "the widest subscriber changing must wake the aggregate watcher",
7053 );
7054 match producer.poll_subscription_changed(&waiter) {
7055 Poll::Ready(Ok(Some(sub))) => assert_eq!(sub.end, Position::after_group(5)),
7056 other => panic!("expected the narrowed aggregate, got {other:?}"),
7057 }
7058 }
7059
7060 struct FlagWake(Arc<std::sync::atomic::AtomicBool>);
7062
7063 impl futures::task::ArcWake for FlagWake {
7064 fn wake_by_ref(arc_self: &Arc<Self>) {
7065 arc_self.0.store(true, std::sync::atomic::Ordering::SeqCst);
7066 }
7067 }
7068
7069 #[tokio::test]
7070 async fn out_of_order_max_sequence_at_front() {
7071 let producer = track_producer("test", None);
7072
7073 producer.create_group(group::Info { sequence: 5 }).unwrap();
7075 producer.create_group(group::Info { sequence: 3 }).unwrap();
7076 producer.create_group(group::Info { sequence: 4 }).unwrap();
7077
7078 {
7080 let state = producer.state.read();
7081 assert_eq!(state.max_sequence, Some(5));
7082 }
7083
7084 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7086
7087 producer.append_group().unwrap(); {
7093 let state = producer.state.read();
7094 assert_eq!(live_groups(&state), 1);
7095 assert_eq!(first_live_sequence(&state), 6);
7096 assert!(!state.lookup.contains_key(&3));
7097 assert!(!state.lookup.contains_key(&4));
7098 assert!(!state.lookup.contains_key(&5));
7099 assert!(state.lookup.contains_key(&6));
7100 }
7101 }
7102
7103 #[tokio::test]
7104 async fn max_sequence_at_front_blocks_trim() {
7105 let producer = track_producer("test", None);
7106
7107 producer.create_group(group::Info { sequence: 5 }).unwrap();
7109
7110 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7111
7112 producer.create_group(group::Info { sequence: 3 }).unwrap();
7114
7115 {
7118 let state = producer.state.read();
7119 assert_eq!(live_groups(&state), 2);
7120 assert_eq!(state.offset, 0);
7121 }
7122
7123 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7125
7126 producer.create_group(group::Info { sequence: 2 }).unwrap();
7128
7129 {
7134 let state = producer.state.read();
7135 assert_eq!(live_groups(&state), 2);
7136 assert_eq!(state.offset, 0);
7137 assert!(state.lookup.contains_key(&5));
7138 assert!(!state.lookup.contains_key(&3));
7139 assert!(state.lookup.contains_key(&2));
7140 }
7141
7142 let mut consumer = producer.subscribe(None);
7144 let group = consumer.assert_group();
7145 assert_eq!(group.sequence, 5);
7147 }
7148
7149 #[tokio::test]
7150 async fn abort_drops_open_groups() {
7151 let producer = track_producer("test", None);
7152 producer.append_group().unwrap();
7153 producer.append_group().unwrap();
7154
7155 let mut consumer = producer.subscribe(None);
7156 assert_eq!(live_groups(&producer.state.read()), 2);
7157
7158 producer.clone().abort(Error::Cancel).unwrap();
7159
7160 assert!(
7162 producer.state.read().lookup.is_empty(),
7163 "open groups are dropped on abort"
7164 );
7165 let result = consumer.recv_group().now_or_never().expect("should not block");
7166 assert!(matches!(result, Err(Error::Cancel)));
7167 }
7168
7169 #[tokio::test]
7170 async fn drop_unfinished_clears_cached_groups() {
7171 let producer = track_producer("test", None);
7172 let writer = producer.clone();
7173 writer.append_group().unwrap();
7174
7175 let mut consumer = producer.subscribe(None);
7177 assert_eq!(live_groups(&producer.state.read()), 1);
7178
7179 drop(writer);
7181 drop(producer);
7182
7183 let result = consumer.recv_group().now_or_never().expect("should not block");
7184 assert!(matches!(result, Err(Error::Dropped)));
7185 }
7186
7187 #[tokio::test]
7188 async fn drop_after_abort_does_not_warn() {
7189 let warns = count_drop_warnings("track::Producer dropped without finish", || {
7192 let producer = track_producer("test", None);
7193 let keep = producer.clone();
7194 let writer = producer.clone();
7195 let group = writer.append_group().unwrap();
7196 group.finish().unwrap();
7197 let _consumer = producer.subscribe(None);
7198 writer.abort(Error::Cancel).unwrap();
7199 drop(keep);
7200 });
7201 assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN");
7202 }
7203
7204 #[tokio::test]
7205 async fn drop_unfinished_warns() {
7206 let warns = count_drop_warnings("track::Producer dropped without finish", || {
7207 let producer = track_producer("test", None);
7208 let writer = producer.clone();
7209 writer.append_group().unwrap();
7210 let _consumer = producer.subscribe(None);
7211 drop(writer);
7212 drop(producer);
7213 });
7214 assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN");
7215 }
7216
7217 #[tokio::test]
7218 async fn drop_finished_keeps_cached_groups() {
7219 let producer = track_producer("test", None);
7220 producer.append_group().unwrap();
7221 producer.finish().unwrap();
7222
7223 let mut consumer = producer.subscribe(None);
7224 drop(producer);
7225
7226 assert_eq!(consumer.assert_group().sequence, 0);
7228 let done = consumer.recv_group().now_or_never().expect("should not block").unwrap();
7229 assert!(done.is_none(), "consumer should drain then see clean finish");
7230 }
7231
7232 #[tokio::test]
7236 async fn drop_short_of_the_boundary_ends_cleanly() {
7237 let mut producer = track_producer("test", None);
7238 producer.append_group().unwrap();
7239 producer.finish_at(3).unwrap();
7240
7241 let mut consumer = producer.subscribe(None);
7242 assert_eq!(consumer.assert_group().sequence, 0);
7243 assert!(
7244 consumer.recv_group().now_or_never().is_none(),
7245 "groups 1 and 2 are owed"
7246 );
7247
7248 drop(producer);
7249 let done = consumer.recv_group().now_or_never().expect("should not block").unwrap();
7250 assert!(
7251 done.is_none(),
7252 "the track ends at its boundary without the missing groups"
7253 );
7254 }
7255
7256 #[tokio::test]
7259 async fn ordered_ends_cleanly_when_sealed_short_of_the_boundary() {
7260 let mut producer = track_producer("test", None);
7261 producer.create_group(group::Info { sequence: 0 }).unwrap();
7262 producer.create_group(group::Info { sequence: 2 }).unwrap();
7263 producer.finish_at(4).unwrap();
7264
7265 let mut ordered = producer.subscribe(None).ordered();
7266 drop(producer);
7267
7268 let first = ordered.next_group().now_or_never().expect("group 0").unwrap().unwrap();
7269 assert_eq!(first.sequence, 0);
7270 let second = ordered.next_group().now_or_never().expect("group 2").unwrap().unwrap();
7271 assert_eq!(second.sequence, 2);
7272 let done = ordered.next_group().now_or_never().expect("end").unwrap();
7273 assert!(done.is_none(), "missing groups below the boundary are not an error");
7274 }
7275
7276 #[tokio::test]
7277 async fn cached_groups_preserve_arrival_order() {
7278 let producer = track_producer("test", None);
7279 producer.create_group(group::Info { sequence: 5 }).unwrap();
7280 producer.create_group(group::Info { sequence: 3 }).unwrap();
7281
7282 let groups = producer.consume().cached_groups();
7283 let sequences: Vec<u64> = groups.iter().map(|(group, _)| group.sequence).collect();
7284 assert_eq!(
7285 sequences,
7286 vec![5, 3],
7287 "warm snapshot must follow arrival, not sequence order"
7288 );
7289 }
7290
7291 #[test]
7292 fn append_finish_cannot_be_rewritten() {
7293 let producer = track_producer("test", None);
7294
7295 assert!(producer.finish().is_ok());
7297 assert!(producer.finish().is_err());
7298 assert!(producer.append_group().is_err());
7299 }
7300
7301 #[test]
7302 fn finish_after_groups() {
7303 let producer = track_producer("test", None);
7304
7305 producer.append_group().unwrap();
7306 assert!(producer.finish().is_ok());
7307 assert!(producer.finish().is_err());
7308 assert!(producer.append_group().is_err());
7309 }
7310
7311 #[test]
7312 fn finish_at_rejects_a_boundary_at_or_below_the_live_edge() {
7313 let mut producer = track_producer("test", None);
7314 producer.create_group(group::Info { sequence: 5 }).unwrap();
7315
7316 assert!(producer.finish_at(4).is_err());
7319 assert!(producer.finish_at(5).is_err());
7320 assert!(producer.finish_at(6).is_ok());
7321
7322 {
7323 let state = producer.state.read();
7324 assert_eq!(state.final_sequence, Some(6));
7325 }
7326
7327 assert!(producer.finish_at(6).is_err());
7329 assert!(producer.create_group(group::Info { sequence: 4 }).is_ok());
7330 assert!(producer.create_group(group::Info { sequence: 6 }).is_err());
7331 }
7332
7333 #[test]
7334 fn final_sequence_reports_the_declared_boundary() {
7335 let mut producer = track_producer("test", None);
7336 assert_eq!(producer.final_sequence(), None);
7337
7338 producer.create_group(group::Info { sequence: 5 }).unwrap();
7339 assert_eq!(producer.final_sequence(), None, "a group does not declare a boundary");
7340
7341 producer.finish_at(9).unwrap();
7342 assert_eq!(producer.final_sequence(), Some(9));
7343
7344 assert!(producer.finish().is_err());
7346 }
7347
7348 #[test]
7349 fn final_sequence_reports_the_live_edge_after_finish() {
7350 let producer = track_producer("test", None);
7351 producer.create_group(group::Info { sequence: 5 }).unwrap();
7352 producer.finish().unwrap();
7353 assert_eq!(producer.final_sequence(), Some(6));
7354 }
7355
7356 #[tokio::test]
7357 async fn finish_at_declares_a_future_boundary() {
7358 let mut producer = track_producer("test", None);
7359 producer.create_group(group::Info { sequence: 5 }).unwrap();
7360
7361 producer.finish_at(7).unwrap();
7363
7364 let mut consumer = producer.subscribe(None);
7365 assert_eq!(consumer.assert_group().sequence, 5);
7366
7367 let boundary = consumer
7370 .finished()
7371 .now_or_never()
7372 .expect("boundary is known immediately")
7373 .expect("would have errored");
7374 assert_eq!(boundary, 7);
7375 assert!(
7376 consumer.recv_group().now_or_never().is_none(),
7377 "should wait for the outstanding group"
7378 );
7379
7380 producer.create_group(group::Info { sequence: 6 }).unwrap();
7382 assert_eq!(consumer.assert_group().sequence, 6);
7383 let done = consumer
7384 .recv_group()
7385 .now_or_never()
7386 .expect("should not block")
7387 .expect("would have errored");
7388 assert!(done.is_none(), "track completes once the boundary is reached");
7389 }
7390
7391 #[tokio::test]
7394 async fn abort_before_the_end_settles_wins() {
7395 let mut producer = track_producer("test", None);
7396 let mut consumer = producer.subscribe(None);
7397
7398 let head = producer.create_group(group::Info { sequence: 0 }).unwrap();
7399 head.finish().unwrap();
7400 let _tail = producer.create_group(group::Info { sequence: 1 }).unwrap();
7401 producer.finish_at(2).unwrap();
7402 assert_eq!(consumer.assert_group().sequence, 0);
7403
7404 producer.abort(Error::Timeout).unwrap();
7405 let res = consumer.recv_group().now_or_never().expect("should not block");
7406 assert!(matches!(res, Err(Error::Timeout)));
7407 }
7408
7409 #[tokio::test]
7410 async fn local_close_keeps_finished_groups_without_declaring_an_end() {
7411 let producer = track_producer("local", None);
7412 let mut consumer = producer.consume().subscribe(None).await.unwrap();
7413 producer.append_group().unwrap().finish().unwrap();
7414 producer.clone().close().unwrap();
7415 assert!(producer.final_sequence().is_none());
7416 assert!(matches!(consumer.finished().await, Err(Error::Closed)));
7417 assert!(consumer.recv_group().await.unwrap().is_some());
7418 assert!(consumer.recv_group().await.unwrap().is_none());
7419 }
7420
7421 #[tokio::test]
7422 async fn local_close_cannot_mask_an_abort_before_declared_end_settles() {
7423 let mut producer = track_producer("local", None);
7424 let mut consumer = producer.consume().subscribe(None).await.unwrap();
7425 let _group = producer.append_group().unwrap();
7426 producer.finish_at(2).unwrap();
7427 producer.clone().abort(Error::Timeout).unwrap();
7428 assert!(producer.close().is_err());
7429 assert!(matches!(consumer.recv_group().await, Err(Error::Timeout)));
7430 }
7431
7432 #[test]
7435 fn local_close_is_not_completion_and_keeps_the_resume_position() {
7436 let producer = track_producer("local", None);
7437 let consumer = producer.consume();
7438 producer.append_group().unwrap().finish().unwrap();
7439 let mut group = producer.append_group().unwrap();
7440 group
7441 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
7442 .unwrap();
7443 group
7444 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"b"))
7445 .unwrap();
7446 producer.close().unwrap();
7447 assert!(matches!(
7448 consumer.poll_complete(&kio::Waiter::noop()),
7449 Poll::Ready(Err(_))
7450 ));
7451 assert_eq!(consumer.resume_position(), Some(Position { group: 1, frame: 2 }));
7452 }
7453
7454 #[tokio::test]
7458 async fn abort_keeps_finished_groups_for_a_slow_reader() {
7459 let producer = track_producer("test", None);
7460 let mut arrival = producer.subscribe(None);
7461 let mut ordered = producer.subscribe(None).ordered();
7462
7463 for sequence in 0..2 {
7464 producer
7465 .create_group(group::Info { sequence })
7466 .unwrap()
7467 .finish()
7468 .unwrap();
7469 }
7470 let _open = producer.create_group(group::Info { sequence: 2 }).unwrap();
7471 producer.abort(Error::Timeout).unwrap();
7472
7473 assert_eq!(arrival.assert_group().sequence, 0);
7474 assert_eq!(arrival.assert_group().sequence, 1);
7475 let res = arrival.recv_group().now_or_never().expect("should not block");
7476 assert!(matches!(res, Err(Error::Timeout)), "expected the abort");
7477
7478 assert_eq!(drain_ordered(&mut ordered), [0, 1]);
7479 let res = ordered
7480 .next_group()
7481 .now_or_never()
7482 .expect("should not block")
7483 .map(|group| group.map(|group| group.sequence));
7484 assert!(matches!(res, Err(Error::Timeout)), "expected the abort, got {res:?}");
7485 }
7486
7487 #[tokio::test]
7489 async fn dropped_producer_keeps_finished_groups_for_a_slow_reader() {
7490 let producer = track_producer("test", None);
7491 let mut consumer = producer.subscribe(None);
7492 producer
7493 .create_group(group::Info { sequence: 0 })
7494 .unwrap()
7495 .finish()
7496 .unwrap();
7497 drop(producer);
7498
7499 assert_eq!(consumer.assert_group().sequence, 0);
7500 let res = consumer.recv_group().now_or_never().expect("should not block");
7501 assert!(matches!(res, Err(Error::Dropped)), "expected the drop");
7502 }
7503
7504 #[tokio::test]
7507 async fn closed_track_expires_its_latest_group() {
7508 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
7509
7510 let live = track_producer_pooled("live", pool.clone());
7511 live.append_group().unwrap().finish().unwrap();
7512
7513 let aborted = track_producer_pooled("aborted", pool.clone());
7514 aborted.append_group().unwrap().finish().unwrap();
7515 let stale_aborted = aborted.consume();
7516 aborted.abort(Error::Timeout).unwrap();
7517
7518 let finished = track_producer_pooled("finished", pool.clone());
7519 finished.append_group().unwrap().finish().unwrap();
7520 finished.finish().unwrap();
7521 let stale_finished = finished.consume();
7522 drop(finished);
7523
7524 for _ in 0..2 {
7526 crate::model::clock::advance(Duration::from_secs(2));
7527 pool.sweep();
7528 }
7529
7530 assert!(live.consume().peek_group(0).is_some(), "a live track keeps its latest");
7531 assert!(
7532 stale_aborted.peek_group(0).is_none(),
7533 "an aborted track's latest expired"
7534 );
7535 assert!(
7536 stale_finished.peek_group(0).is_none(),
7537 "a sealed track's latest expired"
7538 );
7539 }
7540
7541 #[tokio::test]
7545 async fn closed_track_expires_a_shared_group_unless_live_edge() {
7546 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
7547 let sweep = || {
7549 for _ in 0..2 {
7550 crate::model::clock::advance(Duration::from_secs(2));
7551 pool.sweep();
7552 }
7553 };
7554 let adopt = |name: &str, group: &group::Producer| {
7555 let mut track = track_producer_pooled(name, pool.clone());
7556 track.adopt_group(group.clone(), true).unwrap();
7557 track.finish().unwrap();
7558 let stale = track.consume();
7559 drop(track);
7560 stale
7561 };
7562
7563 let live = track_producer_pooled("live", pool.clone());
7564 let group = live.append_group().unwrap();
7565 group.finish().unwrap();
7566 let stale = adopt("ended", &group);
7567 drop(group);
7568
7569 sweep();
7570 assert!(live.consume().peek_group(0).is_some(), "the live edge stays");
7571 assert!(
7572 stale.peek_group(0).is_some(),
7573 "and so does the ended track's share of it"
7574 );
7575
7576 live.append_group().unwrap().finish().unwrap();
7577 sweep();
7578 assert!(stale.peek_group(0).is_none(), "once demoted, it expires");
7579
7580 let group = live.append_group().unwrap();
7581 group.finish().unwrap();
7582 let (first, second) = (adopt("first", &group), adopt("second", &group));
7583 drop(group);
7584 live.finish().unwrap();
7585 drop(live);
7586
7587 sweep();
7588 assert!(first.peek_group(2).is_none(), "ended tracks alone don't keep it");
7589 assert!(second.peek_group(2).is_none(), "for each other");
7590 }
7591
7592 #[tokio::test]
7594 async fn abort_after_the_end_settles_ends_clean() {
7595 let mut producer = track_producer("test", None);
7596 let mut consumer = producer.subscribe(None);
7597
7598 for sequence in 0..2 {
7599 let group = producer.create_group(group::Info { sequence }).unwrap();
7600 group.finish().unwrap();
7601 }
7602 producer.finish_at(2).unwrap();
7603 assert_eq!(consumer.assert_group().sequence, 0);
7604 assert_eq!(consumer.assert_group().sequence, 1);
7605
7606 producer.abort(Error::Timeout).unwrap();
7607 let res = consumer.recv_group().now_or_never().expect("should not block");
7608 assert!(matches!(res, Ok(None)));
7609 }
7610
7611 #[tokio::test]
7614 async fn abort_after_the_end_settles_ends_clean_for_an_ordered_reader() {
7615 let mut producer = track_producer("test", None);
7616 let mut consumer = producer.subscribe(None).ordered();
7617
7618 for sequence in 0..2 {
7619 let group = producer.create_group(group::Info { sequence }).unwrap();
7620 group.finish().unwrap();
7621 }
7622 producer.finish_at(2).unwrap();
7623 producer.abort(Error::Timeout).unwrap();
7624
7625 assert_eq!(drain_ordered(&mut consumer), [0, 1]);
7626 let end = consumer
7627 .next_group()
7628 .now_or_never()
7629 .expect("should not block")
7630 .map(|group| group.map(|group| group.sequence));
7631 assert!(matches!(end, Ok(None)), "expected the clean end, got {end:?}");
7632 }
7633
7634 #[tokio::test]
7638 async fn abort_after_the_end_settles_ends_clean_below_the_cap() {
7639 let mut producer = track_producer("test", None);
7640 let mut consumer = producer.subscribe(None).ordered();
7641 for sequence in 0..2 {
7642 producer
7643 .create_group(group::Info { sequence })
7644 .unwrap()
7645 .finish()
7646 .unwrap();
7647 }
7648 producer.finish_at(2).unwrap();
7649 producer.abort(Error::Timeout).unwrap();
7650
7651 consumer.set_groups(..1);
7652 assert_eq!(drain_ordered(&mut consumer), [0]);
7653 let end = consumer
7654 .next_group()
7655 .now_or_never()
7656 .expect("should not block")
7657 .map(|group| group.map(|group| group.sequence));
7658 assert!(
7659 matches!(end, Ok(None)),
7660 "capped at the end: expected the clean end, got {end:?}"
7661 );
7662
7663 let mut producer = track_producer("test", None);
7664 let mut consumer = producer.subscribe(None).ordered();
7665 for sequence in [0, 2] {
7666 producer
7667 .create_group(group::Info { sequence })
7668 .unwrap()
7669 .finish()
7670 .unwrap();
7671 }
7672 producer.finish_at(3).unwrap();
7673 producer.abort(Error::Timeout).unwrap();
7674
7675 consumer.set_groups(..2);
7676 assert_eq!(drain_ordered(&mut consumer), [0]);
7677 let end = consumer
7678 .next_group()
7679 .now_or_never()
7680 .expect("should not block")
7681 .map(|group| group.map(|group| group.sequence));
7682 assert!(
7683 matches!(end, Ok(None)),
7684 "gap below the cap: expected the clean end, got {end:?}"
7685 );
7686 }
7687
7688 #[tokio::test]
7689 async fn recv_group_finishes_without_waiting_for_gaps() {
7690 let producer = track_producer("test", None);
7691 producer.create_group(group::Info { sequence: 1 }).unwrap();
7692 producer.finish().unwrap();
7693
7694 let mut consumer = producer.subscribe(None);
7695 assert_eq!(consumer.assert_group().sequence, 1);
7696
7697 let done = consumer
7698 .recv_group()
7699 .now_or_never()
7700 .expect("should not block")
7701 .expect("would have errored");
7702 assert!(done.is_none(), "track should finish without waiting for gaps");
7703 }
7704
7705 #[tokio::test]
7706 async fn next_group_skips_late_arrivals() {
7707 let producer = track_producer("test", None);
7708 let mut consumer = producer.subscribe(None).ordered();
7709
7710 producer.create_group(group::Info { sequence: 5 }).unwrap();
7712 let group = consumer
7713 .next_group()
7714 .now_or_never()
7715 .expect("should not block")
7716 .expect("would have errored")
7717 .expect("track should not be closed");
7718 assert_eq!(group.sequence, 5);
7719
7720 producer.create_group(group::Info { sequence: 3 }).unwrap();
7722 producer.create_group(group::Info { sequence: 4 }).unwrap();
7724 producer.create_group(group::Info { sequence: 7 }).unwrap();
7726
7727 let group = consumer
7728 .next_group()
7729 .now_or_never()
7730 .expect("should not block")
7731 .expect("would have errored")
7732 .expect("track should not be closed");
7733 assert_eq!(group.sequence, 7);
7734
7735 assert!(
7737 consumer.next_group().now_or_never().is_none(),
7738 "should block waiting for a higher sequence"
7739 );
7740 }
7741
7742 #[tokio::test]
7743 async fn next_group_returns_arrivals_in_order() {
7744 let producer = track_producer("test", None);
7745 let mut consumer = producer.subscribe(replay()).ordered();
7746
7747 producer.create_group(group::Info { sequence: 3 }).unwrap();
7749 producer.create_group(group::Info { sequence: 5 }).unwrap();
7750
7751 let group = consumer
7752 .next_group()
7753 .now_or_never()
7754 .expect("should not block")
7755 .expect("would have errored")
7756 .expect("track should not be closed");
7757 assert_eq!(group.sequence, 3);
7758
7759 let group = consumer
7760 .next_group()
7761 .now_or_never()
7762 .expect("should not block")
7763 .expect("would have errored")
7764 .expect("track should not be closed");
7765 assert_eq!(group.sequence, 5);
7766 }
7767
7768 #[tokio::test]
7769 async fn ordered_and_arrival_cursors_are_independent() {
7770 let producer = track_producer("test", None);
7771 let mut ordered = producer.subscribe(replay()).ordered();
7772 let mut arrival = producer.subscribe(replay());
7773
7774 producer.create_group(group::Info { sequence: 5 }).unwrap();
7776 producer.create_group(group::Info { sequence: 3 }).unwrap();
7777
7778 let group = ordered
7781 .next_group()
7782 .now_or_never()
7783 .expect("should not block")
7784 .expect("would have errored")
7785 .expect("track should not be closed");
7786 assert_eq!(group.sequence, 3);
7787
7788 assert_eq!(arrival.assert_group().sequence, 5);
7790 }
7791
7792 #[tokio::test]
7793 async fn end_at_caps_next_group() {
7794 let producer = track_producer("test", None);
7795 let mut consumer = producer.subscribe(replay()).ordered();
7796
7797 for s in 0..6 {
7798 producer.create_group(group::Info { sequence: s }).unwrap();
7799 }
7800
7801 consumer.set_groups(..3);
7802
7803 assert_eq!(
7805 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7806 0
7807 );
7808 assert_eq!(
7809 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7810 1
7811 );
7812 assert_eq!(
7813 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7814 2
7815 );
7816
7817 assert!(
7819 consumer.next_group().now_or_never().is_none(),
7820 "capped consumer must block instead of returning out-of-range groups"
7821 );
7822 }
7823
7824 #[tokio::test]
7825 async fn end_at_release_drains_cached_groups() {
7826 let producer = track_producer("test", None);
7827 let mut consumer = producer.subscribe(replay()).ordered();
7828
7829 for s in 0..6 {
7830 producer.create_group(group::Info { sequence: s }).unwrap();
7831 }
7832
7833 consumer.set_groups(..2);
7834 assert_eq!(
7835 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7836 0
7837 );
7838 assert_eq!(
7839 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7840 1
7841 );
7842 assert!(consumer.next_group().now_or_never().is_none(), "capped at 2");
7843
7844 consumer.set_groups(..5);
7846 assert_eq!(
7847 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7848 2
7849 );
7850 assert_eq!(
7851 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7852 3
7853 );
7854 assert_eq!(
7855 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7856 4
7857 );
7858 assert!(consumer.next_group().now_or_never().is_none(), "capped at 5");
7859
7860 consumer.set_groups(..);
7862 assert_eq!(
7863 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7864 5
7865 );
7866 assert!(consumer.next_group().now_or_never().is_none(), "no more groups");
7867 }
7868
7869 #[tokio::test]
7870 async fn end_at_lower_than_cursor_parks_consumer() {
7871 let producer = track_producer("test", None);
7872 let mut consumer = producer.subscribe(replay()).ordered();
7873
7874 for s in 0..3 {
7875 producer.create_group(group::Info { sequence: s }).unwrap();
7876 }
7877
7878 assert_eq!(
7880 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7881 0
7882 );
7883 assert_eq!(
7884 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7885 1
7886 );
7887 assert_eq!(
7888 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7889 2
7890 );
7891
7892 consumer.set_groups(..2);
7894 producer.create_group(group::Info { sequence: 3 }).unwrap();
7895 producer.create_group(group::Info { sequence: 4 }).unwrap();
7896 assert!(
7897 consumer.next_group().now_or_never().is_none(),
7898 "cap is below cursor; nothing returnable until cap rises"
7899 );
7900
7901 consumer.set_groups(..);
7903 assert_eq!(
7904 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7905 3
7906 );
7907 assert_eq!(
7908 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7909 4
7910 );
7911 }
7912
7913 #[tokio::test]
7914 async fn end_at_toggling_around_late_arrivals() {
7915 let producer = track_producer("test", None);
7916 let mut consumer = producer.subscribe(replay()).ordered();
7917
7918 consumer.set_groups(..6);
7919
7920 producer.create_group(group::Info { sequence: 2 }).unwrap();
7922 producer.create_group(group::Info { sequence: 5 }).unwrap();
7923 producer.create_group(group::Info { sequence: 3 }).unwrap();
7924 producer.create_group(group::Info { sequence: 8 }).unwrap();
7926 producer.create_group(group::Info { sequence: 4 }).unwrap();
7927
7928 assert_eq!(
7930 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7931 2
7932 );
7933 assert_eq!(
7934 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7935 3
7936 );
7937 assert_eq!(
7938 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7939 4
7940 );
7941 assert_eq!(
7942 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7943 5
7944 );
7945 assert!(consumer.next_group().now_or_never().is_none());
7947
7948 consumer.set_groups(..11);
7950 assert_eq!(
7951 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7952 8
7953 );
7954 }
7955
7956 #[tokio::test]
7960 async fn end_at_parks_recv_group() {
7961 let producer = track_producer("test", None);
7962 let mut consumer = producer.subscribe(replay());
7963
7964 for s in 0..3 {
7965 producer.create_group(group::Info { sequence: s }).unwrap();
7966 }
7967
7968 consumer.set_groups(..2);
7969 assert_eq!(
7970 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7971 0
7972 );
7973 assert_eq!(
7974 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7975 1
7976 );
7977 assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");
7978
7979 producer.finish().unwrap();
7981 assert!(
7982 consumer.recv_group().now_or_never().is_none(),
7983 "still parked after finish"
7984 );
7985
7986 consumer.set_groups(..);
7987 assert_eq!(
7988 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7989 2
7990 );
7991 assert!(
7992 matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
7993 "finished once the parked group drains"
7994 );
7995 }
7996
7997 #[tokio::test]
8000 async fn recv_group_serves_arrivals_behind_the_cap() {
8001 let producer = track_producer("test", None);
8002 let mut consumer = producer.subscribe(replay());
8003
8004 consumer.set_groups(..2);
8005
8006 producer.create_group(group::Info { sequence: 2 }).unwrap();
8008 producer.create_group(group::Info { sequence: 0 }).unwrap();
8009 producer.create_group(group::Info { sequence: 1 }).unwrap();
8010
8011 assert_eq!(
8012 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8013 0
8014 );
8015 assert_eq!(
8016 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8017 1
8018 );
8019 assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");
8020
8021 consumer.set_groups(..3);
8022 assert_eq!(
8023 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8024 2
8025 );
8026 }
8027
8028 #[tokio::test]
8029 async fn group_ranges_preserve_the_floor_when_the_cap_changes() {
8030 let producer = track_producer("test", None);
8031 let mut consumer = producer.subscribe(None);
8032 consumer.set_groups(2..=2);
8033 producer.create_group(group::Info { sequence: 2 }).unwrap();
8034 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 2);
8035 consumer.set_groups(..4);
8036 producer.create_group(group::Info { sequence: 1 }).unwrap();
8037 producer.create_group(group::Info { sequence: 3 }).unwrap();
8038 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 3);
8039 consumer.set_groups(0..=4);
8040 producer.create_group(group::Info { sequence: 0 }).unwrap();
8041 producer.create_group(group::Info { sequence: 4 }).unwrap();
8042 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 4);
8043 }
8044
8045 #[tokio::test]
8048 async fn start_at_drops_parked_recv_groups() {
8049 let producer = track_producer("test", None);
8050 let mut consumer = producer.subscribe(None);
8051
8052 consumer.set_groups(..1);
8053 producer.create_group(group::Info { sequence: 1 }).unwrap();
8054 assert!(
8055 consumer.recv_group().now_or_never().is_none(),
8056 "group 1 parked at the cap"
8057 );
8058
8059 consumer.start_at(2);
8060 consumer.set_groups(..);
8061 producer.create_group(group::Info { sequence: 2 }).unwrap();
8062 assert_eq!(
8063 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8064 2,
8065 "the overtaken parked group is dropped, not re-offered"
8066 );
8067 }
8068
8069 #[tokio::test]
8073 async fn evicted_parked_recv_groups_are_dropped() {
8074 let producer = track_producer("test", None);
8075 let mut consumer = producer.subscribe(None);
8076
8077 producer.create_group(group::Info { sequence: 0 }).unwrap();
8078 assert_eq!(
8079 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8080 0
8081 );
8082
8083 consumer.set_groups(..1);
8084 let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
8085 assert!(
8086 consumer.recv_group().now_or_never().is_none(),
8087 "group 1 parked at the cap"
8088 );
8089
8090 straggler.abort(Error::Old).unwrap();
8092 producer.finish().unwrap();
8093
8094 consumer.set_groups(..);
8095 assert!(
8096 matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
8097 "a dead parked group must not be delivered or hold the stream open"
8098 );
8099 }
8100
8101 #[tokio::test]
8105 async fn evicted_parked_group_wakes_the_clean_end() {
8106 use std::sync::atomic::{AtomicUsize, Ordering};
8107 use std::task::{Context, Wake};
8108
8109 struct CountWaker(AtomicUsize);
8112 impl Wake for CountWaker {
8113 fn wake(self: std::sync::Arc<Self>) {
8114 self.0.fetch_add(1, Ordering::SeqCst);
8115 }
8116 }
8117
8118 let producer = track_producer("test", None);
8119 let mut consumer = producer.subscribe(None);
8120
8121 producer.create_group(group::Info { sequence: 0 }).unwrap();
8122 assert_eq!(
8123 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8124 0
8125 );
8126
8127 consumer.set_groups(..1);
8128 let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
8129 assert!(consumer.recv_group().now_or_never().is_none(), "parked at the cap");
8130 producer.finish().unwrap();
8131
8132 let counter = std::sync::Arc::new(CountWaker(AtomicUsize::new(0)));
8133 let waker = std::task::Waker::from(counter.clone());
8134 let mut cx = Context::from_waker(&waker);
8135 let mut fut = std::pin::pin!(consumer.recv_group());
8136 assert!(
8137 fut.as_mut().poll(&mut cx).is_pending(),
8138 "the parked group holds it open"
8139 );
8140
8141 straggler.abort(Error::Old).unwrap();
8142 assert!(counter.0.load(Ordering::SeqCst) > 0, "the eviction wakeup was lost");
8143 assert!(matches!(fut.as_mut().poll(&mut cx), Poll::Ready(Ok(None))));
8144 }
8145
8146 #[tokio::test]
8148 async fn end_at_zero_is_the_empty_range() {
8149 let producer = track_producer("test", None);
8150 let mut consumer = producer.subscribe(replay());
8151 producer.create_group(group::Info { sequence: 0 }).unwrap();
8152 producer.create_group(group::Info { sequence: 1 }).unwrap();
8153
8154 consumer.set_groups(..0);
8155 assert!(
8156 consumer.recv_group().now_or_never().is_none(),
8157 "empty cap delivers nothing"
8158 );
8159
8160 consumer.set_groups(..1);
8161 assert_eq!(
8162 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8163 0
8164 );
8165 assert!(consumer.recv_group().now_or_never().is_none(), "group 1 stays parked");
8166 }
8167
8168 #[tokio::test]
8171 async fn empty_local_cap_holds_while_another_subscriber_requests_everything() {
8172 let producer = track_producer("test", None);
8173 let mut everything = producer.subscribe(replay());
8174 let mut empty = producer.subscribe(Subscription::default().with_end(Position::group(0)));
8175 empty.set_groups((Bound::Unbounded, Position::group(0).group_end()));
8176
8177 for s in 0..3 {
8178 producer.create_group(group::Info { sequence: s }).unwrap();
8179 }
8180
8181 assert_eq!(
8182 everything
8183 .recv_group()
8184 .now_or_never()
8185 .unwrap()
8186 .unwrap()
8187 .unwrap()
8188 .sequence,
8189 0
8190 );
8191 assert!(
8192 empty.recv_group().now_or_never().is_none(),
8193 "local empty cap must not ride the unbounded aggregate"
8194 );
8195
8196 empty.set_groups(..2);
8197 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 0);
8198 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 1);
8199 assert!(empty.recv_group().now_or_never().is_none(), "still capped at 2");
8200
8201 empty.set_groups(..);
8202 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 2);
8203 }
8204
8205 #[tokio::test]
8207 async fn end_at_frame_limited_last_group() {
8208 let producer = track_producer("test", None);
8209 let mut consumer = producer.subscribe(replay()).ordered();
8210 let end = Position::after(1, 1).unwrap();
8211 consumer.set_groups((Bound::Unbounded, end.group_end()));
8212
8213 for s in 0..3u64 {
8214 let mut group = producer.create_group(group::Info { sequence: s }).unwrap();
8215 for i in 0..3u8 {
8216 group.write_frame(Timestamp::ZERO, vec![i]).unwrap();
8217 }
8218 group.finish().unwrap();
8219 }
8220
8221 assert_eq!(
8222 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8223 0
8224 );
8225 let mut last = consumer.next_group().now_or_never().unwrap().unwrap().unwrap();
8226 assert_eq!(last.sequence, 1);
8227 last.set_frames(..end.frame);
8228 assert_eq!(
8229 last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
8230 0
8231 );
8232 assert_eq!(
8233 last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
8234 1
8235 );
8236 assert!(
8237 last.read_frame().now_or_never().unwrap().unwrap().is_none(),
8238 "frame cap is exclusive"
8239 );
8240 assert!(
8241 consumer.next_group().now_or_never().is_none(),
8242 "group 2 is past the exclusive group cap"
8243 );
8244 }
8245
8246 #[tokio::test]
8248 async fn end_at_maximum_group_is_unbounded() {
8249 let producer = track_producer("test", None);
8250 let mut consumer = producer.subscribe(replay()).ordered();
8251 consumer.set_groups(..=u64::MAX);
8252
8253 producer.create_group(group::Info { sequence: 0 }).unwrap();
8254 producer.create_group(group::Info { sequence: u64::MAX }).unwrap();
8255
8256 assert_eq!(
8257 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8258 0
8259 );
8260 assert_eq!(
8261 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8262 u64::MAX
8263 );
8264 }
8265
8266 #[test]
8267 fn write_frame_rejects_an_oversized_frame_before_appending_its_group() {
8268 let mut producer = track_producer("test", None);
8269 let frame = bytes::Bytes::from(vec![0; group::MAX_CACHE_BYTES as usize + 1]);
8270
8271 assert!(matches!(
8272 producer.write_frame(Timestamp::ZERO, frame),
8273 Err(Error::FrameTooLarge)
8274 ));
8275 assert_eq!(producer.latest(), None, "the rejected frame did not publish a group");
8276 }
8277
8278 #[test]
8279 fn append_group_returns_bounds_exceeded_on_sequence_overflow() {
8280 let producer = track_producer("test", None);
8281 {
8282 let mut state = producer.state.write().ok().unwrap();
8283 state.max_sequence = Some(u64::MAX);
8284 }
8285
8286 assert!(matches!(producer.append_group(), Err(Error::BoundsExceeded(_))));
8287 }
8288
8289 #[tokio::test]
8290 async fn fetch_cache_hit() {
8291 let producer = track_producer("test", None);
8292
8293 let mut group = producer.append_group().unwrap(); group
8296 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hello"))
8297 .unwrap();
8298 group.finish().unwrap();
8299
8300 let dynamic = producer.dynamic();
8303 let consumer = producer.consume();
8304 assert!(consumer.peek_group(0).is_some());
8305 let mut g = consumer.fetch_group(0, None).await.unwrap();
8306 assert_eq!(g.sequence, 0);
8307 assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hello");
8308
8309 assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
8311 }
8312
8313 #[tokio::test]
8314 async fn fetch_miss_signals_dynamic() {
8315 let producer = track_producer("test", None);
8316 let dynamic = producer.dynamic();
8317 let consumer = producer.consume();
8318
8319 assert!(consumer.peek_group(5).is_none());
8323 let pending = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
8324 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
8325
8326 let req = dynamic
8327 .requested_group()
8328 .now_or_never()
8329 .expect("should not block")
8330 .unwrap();
8331 assert_eq!(req.sequence(), 5);
8332 assert_eq!(req.priority(), 7);
8333
8334 let mut group = req.accept(None).unwrap();
8336 group
8337 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
8338 .unwrap();
8339 group.finish().unwrap();
8340
8341 let mut g = pending.await.unwrap();
8342 assert_eq!(g.sequence, 5);
8343 assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hi");
8344 }
8345
8346 #[tokio::test]
8347 async fn fetch_miss_rejects() {
8348 let producer = track_producer("test", None);
8349 let dynamic = producer.dynamic();
8350 let consumer = producer.consume();
8351
8352 let pending = consumer.fetch_group(5, None);
8353 let req = dynamic
8354 .requested_group()
8355 .now_or_never()
8356 .expect("should not block")
8357 .unwrap();
8358
8359 req.reject(Error::Cancel);
8360 assert!(matches!(pending.await, Err(Error::Cancel)));
8361 let fetch = producer.state.read().fetch.clone();
8362 assert!(fetch.read().is_empty());
8363 }
8364
8365 #[tokio::test]
8366 async fn fetch_miss_drop_rejects() {
8367 let producer = track_producer("test", None);
8368 let dynamic = producer.dynamic();
8369 let consumer = producer.consume();
8370
8371 let pending = consumer.fetch_group(5, None);
8372 let req = dynamic
8373 .requested_group()
8374 .now_or_never()
8375 .expect("should not block")
8376 .unwrap();
8377
8378 drop(req);
8379 assert!(matches!(pending.await, Err(Error::Dropped)));
8380 }
8381
8382 #[tokio::test]
8385 async fn fetch_request_unused_once_every_fetch_leaves() {
8386 let producer = track_producer("test", None);
8387 let dynamic = producer.dynamic();
8388 let consumer = producer.consume();
8389
8390 let first = consumer.fetch_group(5, None);
8391 let req = dynamic
8392 .requested_group()
8393 .now_or_never()
8394 .expect("should not block")
8395 .unwrap();
8396 let second = consumer.fetch_group(5, None);
8397
8398 drop(first);
8399 assert!(req.poll_unused(&kio::Waiter::noop()).is_pending());
8400 drop(second);
8401 assert!(req.poll_unused(&kio::Waiter::noop()).is_ready());
8402
8403 let retry = consumer.fetch_group(5, None);
8404 let fresh = dynamic
8405 .requested_group()
8406 .now_or_never()
8407 .expect("the retry queues a fresh request")
8408 .unwrap();
8409 drop(req);
8410 assert!(fresh.poll_unused(&kio::Waiter::noop()).is_pending());
8411 fresh.accept(None).unwrap().finish().unwrap();
8412 assert_eq!(retry.await.unwrap().sequence, 5);
8413 }
8414
8415 #[test]
8418 fn fetching_is_unwind_safe() {
8419 fn assert_unwind_safe<T: std::panic::UnwindSafe + std::panic::RefUnwindSafe>() {}
8420 assert_unwind_safe::<Fetching>();
8421 assert_unwind_safe::<group::Consumer>();
8422 }
8423
8424 #[tokio::test]
8427 async fn fetch_hit_keeps_the_group_wanted() {
8428 let producer = track_producer("test", None);
8429 let dynamic = producer.dynamic();
8430 let consumer = producer.consume();
8431
8432 let first = consumer.fetch_group(5, None);
8433 let req = dynamic
8434 .requested_group()
8435 .now_or_never()
8436 .expect("should not block")
8437 .unwrap();
8438 let mut group = req.accept(None).unwrap();
8439 group
8440 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
8441 .unwrap();
8442 drop(first.await.unwrap());
8443
8444 let hit = consumer.fetch_group(5, None);
8445 assert!(!group.abort_unused(Error::Cancel), "the pending hit wants the group");
8446 let mut fetched = hit.await.unwrap();
8447 assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head");
8448 }
8449
8450 #[tokio::test]
8454 async fn accepted_group_starts_at_the_request() {
8455 let producer = track_producer("test", None);
8456 let dynamic = producer.dynamic();
8457 let consumer = producer.consume();
8458
8459 let _tail = consumer.fetch_group(5, group::Fetch::default().with_frame_start(3));
8460 let req = dynamic
8461 .requested_group()
8462 .now_or_never()
8463 .expect("should not block")
8464 .unwrap();
8465 let _group = req.accept(None).unwrap();
8466
8467 let whole = consumer.fetch_group(5, None);
8468 let req = dynamic
8469 .requested_group()
8470 .now_or_never()
8471 .expect("the wider fetch misses and queues its own request")
8472 .unwrap();
8473 assert_eq!(req.frame_start(), 0);
8474 assert!(kio::Pollable::poll(&*whole, &kio::Waiter::noop()).is_pending());
8475 }
8476
8477 #[tokio::test]
8480 async fn fetch_misses_an_abandoned_group() {
8481 let producer = track_producer("test", None);
8482 let dynamic = producer.dynamic();
8483 let consumer = producer.consume();
8484
8485 let first = consumer.fetch_group(5, None);
8486 let req = dynamic
8487 .requested_group()
8488 .now_or_never()
8489 .expect("should not block")
8490 .unwrap();
8491 let group = req.accept(None).unwrap();
8492 drop(first.await.unwrap());
8493 assert!(group.abort_unused(Error::Cancel));
8494
8495 let retry = consumer.fetch_group(5, None);
8496 let req = dynamic
8497 .requested_group()
8498 .now_or_never()
8499 .expect("the retry queues a fresh request")
8500 .unwrap();
8501 req.accept(None).unwrap().finish().unwrap();
8502 assert!(retry.await.unwrap().read_frame().await.unwrap().is_none());
8503 }
8504
8505 #[tokio::test]
8506 async fn fetch_reject_does_not_poison_retry() {
8507 let producer = track_producer("test", None);
8508 let dynamic = producer.dynamic();
8509 let consumer = producer.consume();
8510
8511 let pending = consumer.fetch_group(5, None);
8512 let req = dynamic
8513 .requested_group()
8514 .now_or_never()
8515 .expect("should not block")
8516 .unwrap();
8517 req.reject(Error::Cancel);
8518 assert!(matches!(pending.await, Err(Error::Cancel)));
8519
8520 let retry = consumer.fetch_group(5, None);
8521 let req = dynamic
8522 .requested_group()
8523 .now_or_never()
8524 .expect("should not block")
8525 .unwrap();
8526 let mut group = req.accept(None).unwrap();
8527 group
8528 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"retry"))
8529 .unwrap();
8530 group.finish().unwrap();
8531
8532 let mut group = retry.await.unwrap();
8533 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"retry");
8534 }
8535
8536 #[tokio::test]
8540 async fn fetch_ignores_a_group_that_starts_too_late() {
8541 let producer = track_producer("test", None);
8542 let dynamic = producer.dynamic();
8543 let consumer = producer.consume();
8544
8545 let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
8547 group.start_at(3).unwrap();
8548 group
8549 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"tail"))
8550 .unwrap();
8551 group.finish().unwrap();
8552
8553 let fetch = consumer.fetch_group(0, group::Fetch::default().with_frame_start(3));
8555 let cached = fetch.now_or_never().expect("covered by the cache").unwrap();
8556 assert_eq!(cached.index(), 3);
8557
8558 let mut fetch = std::pin::pin!(consumer.fetch_group(0, None));
8561 assert!(
8562 futures::poll!(fetch.as_mut()).is_pending(),
8563 "must not answer from the tail"
8564 );
8565
8566 let request = dynamic.requested_group().await.unwrap();
8567 assert_eq!((request.sequence(), request.frame_start()), (0, 0));
8568
8569 let mut whole = request.accept(None).unwrap();
8571 whole
8572 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
8573 .unwrap();
8574 whole.finish().unwrap();
8575
8576 let mut served = fetch.await.unwrap();
8577 assert_eq!(served.index(), 0);
8578 assert_eq!(
8579 served.read_frame().await.unwrap().unwrap().payload,
8580 bytes::Bytes::from_static(b"head")
8581 );
8582 }
8583
8584 #[tokio::test]
8587 async fn fetch_widens_or_fails_cleanly() {
8588 let producer = track_producer("test", None);
8589 let dynamic = producer.dynamic();
8590 let consumer = producer.consume();
8591
8592 let _narrow = consumer.fetch_group(0, group::Fetch::default().with_frame_start(5));
8593 let mut narrow = std::pin::pin!(_narrow);
8594 assert!(futures::poll!(narrow.as_mut()).is_pending());
8595
8596 let _wider = consumer.fetch_group(0, group::Fetch::default().with_frame_start(2));
8598 let mut wider = std::pin::pin!(_wider);
8599 assert!(futures::poll!(wider.as_mut()).is_pending());
8600
8601 let request = dynamic.requested_group().await.unwrap();
8602 assert_eq!(request.frame_start(), 2, "widened while queued");
8603
8604 let _widest = consumer.fetch_group(0, group::Fetch::default().with_frame_start(0));
8607 let mut widest = std::pin::pin!(_widest);
8608 assert!(futures::poll!(widest.as_mut()).is_pending());
8609 assert_eq!(request.frame_start(), 2, "the in-flight range is already on the wire");
8610
8611 let _joined = request.result.clone();
8614 let mut group = request.accept(None).unwrap();
8615 group.start_at(2).unwrap();
8618 group
8619 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"from2"))
8620 .unwrap();
8621
8622 assert_eq!(narrow.await.unwrap().index(), 5);
8626 assert_eq!(wider.await.unwrap().index(), 2);
8627 assert!(matches!(widest.await, Err(Error::NotFound)));
8628 group.finish().unwrap();
8629 }
8630
8631 #[tokio::test]
8632 async fn fetch_coalesces_concurrent() {
8633 let producer = track_producer("test", None);
8634 let dynamic = producer.dynamic();
8635 let consumer = producer.consume();
8636
8637 let first = consumer.fetch_group(5, group::Fetch::default().with_priority(1));
8640 let second = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
8641 assert!(kio::Pollable::poll(&*first, &kio::Waiter::noop()).is_pending());
8642
8643 let req = dynamic
8644 .requested_group()
8645 .now_or_never()
8646 .expect("should not block")
8647 .unwrap();
8648 assert_eq!(req.sequence(), 5);
8649 assert_eq!(req.priority(), 7);
8650 assert!(
8651 dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending(),
8652 "the second fetch queued a duplicate request"
8653 );
8654
8655 let third = consumer.fetch_group(5, None);
8657
8658 let mut group = req.accept(None).unwrap();
8660 group
8661 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
8662 .unwrap();
8663 group.finish().unwrap();
8664
8665 assert_eq!(first.await.unwrap().sequence, 5);
8666 assert_eq!(second.await.unwrap().sequence, 5);
8667 assert_eq!(third.await.unwrap().sequence, 5);
8668 }
8669
8670 #[tokio::test]
8671 async fn fetch_coalesced_reject_fails_all() {
8672 let producer = track_producer("test", None);
8673 let dynamic = producer.dynamic();
8674 let consumer = producer.consume();
8675
8676 let first = consumer.fetch_group(5, None);
8677 let second = consumer.fetch_group(5, None);
8678 let req = dynamic
8679 .requested_group()
8680 .now_or_never()
8681 .expect("should not block")
8682 .unwrap();
8683 req.reject(Error::Cancel);
8684
8685 assert!(matches!(first.await, Err(Error::Cancel)));
8686 assert!(matches!(second.await, Err(Error::Cancel)));
8687
8688 let retry = consumer.fetch_group(5, None);
8690 assert!(kio::Pollable::poll(&*retry, &kio::Waiter::noop()).is_pending());
8691 let req = dynamic
8692 .requested_group()
8693 .now_or_never()
8694 .expect("should not block")
8695 .unwrap();
8696 assert_eq!(req.sequence(), 5);
8697 }
8698
8699 #[tokio::test]
8700 async fn fetch_queued_fails_when_handlers_leave() {
8701 let producer = track_producer("test", None);
8702 let dynamic = producer.dynamic();
8703 let consumer = producer.consume();
8704
8705 let pending = consumer.fetch_group(5, None);
8707 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
8708 drop(dynamic);
8709 assert!(matches!(pending.await, Err(Error::NotFound)));
8710
8711 let fetch = producer.state.read().fetch.clone();
8713 assert!(fetch.read().is_empty());
8714 }
8715
8716 #[tokio::test]
8717 async fn fetch_miss_no_dynamic_not_found() {
8718 let producer = track_producer("test", None);
8721 producer.append_group().unwrap(); let consumer = producer.consume();
8723 assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));
8724 }
8725
8726 #[tokio::test]
8727 async fn fetch_past_final_not_found() {
8728 let producer = track_producer("test", None);
8729 producer.append_group().unwrap(); producer.finish().unwrap(); let dynamic = producer.dynamic();
8735 let consumer = producer.consume();
8736 assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));
8737
8738 assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
8740 }
8741
8742 fn pooled_producer(capacity: u64) -> (Producer, cache::Pool) {
8744 let config = cache::Config::default()
8745 .with_capacity(capacity)
8746 .with_expiry(cache::DEFAULT_EXPIRY);
8747 let pool = cache::Pool::new(config);
8748 let broadcast = broadcast::Info {
8749 pool: pool.clone(),
8750 ..Default::default()
8751 };
8752 let producer = Producer::new(Arc::new(broadcast), "test", None);
8753 (producer, pool)
8754 }
8755
8756 fn finished_group(producer: &mut Producer, size: usize) -> u64 {
8757 let mut group = producer.append_group().unwrap();
8758 group
8759 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; size]))
8760 .unwrap();
8761 group.finish().unwrap();
8762 group.sequence
8763 }
8764
8765 #[tokio::test]
8768 async fn debt_evicts_oldest_group() {
8769 let (mut producer, pool) = pooled_producer(10_000);
8771
8772 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let consumer = producer.consume();
8777 assert!(consumer.peek_group(0).is_none(), "oldest group is evicted");
8778 assert!(consumer.peek_group(2).is_some(), "latest group survives");
8779 assert!(
8782 pool.used() <= 2 * (10_000 + cache::ENTRY_OVERHEAD),
8783 "usage hovers near capacity: {}",
8784 pool.used()
8785 );
8786
8787 let mut subscriber = producer.subscribe(replay());
8789 assert!(subscriber.assert_group().sequence > 0, "evicted group is not delivered");
8790 }
8791
8792 #[tokio::test]
8794 async fn latest_group_never_evicted() {
8795 let (mut producer, pool) = pooled_producer(100);
8797 finished_group(&mut producer, 1000); assert!(pool.used() > 100, "the latest may exceed the budget");
8799
8800 finished_group(&mut producer, 1000); finished_group(&mut producer, 1000); let consumer = producer.consume();
8805 assert!(consumer.peek_group(0).is_none());
8806 let mut group = consumer.peek_group(2).expect("latest survives");
8807 assert_eq!(group.read_frame().await.unwrap().unwrap().payload.len(), 1000);
8808 }
8809
8810 #[tokio::test]
8814 async fn fetch_refresh_survives_eviction() {
8815 let (mut producer, _pool) = pooled_producer(10_000);
8816 let consumer = producer.consume();
8817
8818 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8820 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8822 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_millis(500));
8824
8825 let mut fetched = consumer.fetch_group(0, None).await.unwrap();
8827 assert_eq!(fetched.read_frame().await.unwrap().unwrap().payload.len(), 3_000);
8828 crate::model::clock::advance(Duration::from_millis(500));
8829
8830 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8834 finished_group(&mut producer, 3_000); assert!(consumer.peek_group(0).is_some(), "refreshed group survives");
8837 assert!(consumer.peek_group(1).is_none(), "unread group is evicted instead");
8838 }
8839
8840 #[tokio::test]
8843 async fn eviction_aborts_readers() {
8844 let (mut producer, _pool) = pooled_producer(10_000);
8845 let mut subscriber = producer.subscribe(None);
8846
8847 finished_group(&mut producer, 10_000); let mut group0 = subscriber.assert_group();
8849
8850 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let read = group0.read_frame().await;
8854 assert!(matches!(read, Err(Error::Evicted)), "expected Evicted, got {read:?}");
8855 }
8856
8857 #[tokio::test]
8861 async fn small_writes_carry_debt() {
8862 let unit = 100 * cache::ENTRY_OVERHEAD;
8865 let (mut producer, pool) = pooled_producer(22 * unit);
8866 let consumer = producer.consume();
8867
8868 finished_group(&mut producer, 20 * unit as usize); for _ in 0..3 {
8873 finished_group(&mut producer, unit as usize);
8874 }
8875 assert!(consumer.peek_group(0).is_some(), "debt smaller than the victim carries");
8876
8877 for _ in 0..20 {
8879 finished_group(&mut producer, unit as usize);
8880 }
8881 assert!(
8882 consumer.peek_group(0).is_none(),
8883 "accumulated debt evicts the large group"
8884 );
8885 assert!(pool.used() <= 24 * unit, "usage hovers near capacity: {}", pool.used());
8888 }
8889
8890 #[tokio::test]
8894 async fn payment_capped_per_write() {
8895 let (mut producer, pool) = pooled_producer(1 << 40);
8896 for _ in 0..10 {
8897 finished_group(&mut producer, 1_000);
8898 }
8899
8900 pool.resize(100);
8902 let before = pool.used();
8903
8904 finished_group(&mut producer, 1_000);
8906
8907 let consumer = producer.consume();
8908 assert!(consumer.peek_group(0).is_none(), "the oldest groups are evicted");
8909 assert!(consumer.peek_group(1).is_none());
8910 assert!(consumer.peek_group(2).is_some(), "the backlog drains gradually");
8911 assert!(pool.used() > before - 4_000, "one write must not dump the backlog");
8912 }
8913
8914 #[tokio::test]
8918 async fn accept_preserves_write_accounting() {
8919 let config = cache::Config::default()
8920 .with_capacity(12_000)
8921 .with_expiry(cache::DEFAULT_EXPIRY);
8922 let pool = cache::Pool::new(config);
8923 let broadcast = broadcast::Info {
8924 pool: pool.clone(),
8925 ..Default::default()
8926 };
8927 let request = Request::new(Arc::new(broadcast), "test");
8928 let dynamic = request.dynamic();
8929 let consumer = request.consume();
8930
8931 let pending = consumer.fetch_group(0, None);
8933 let req = dynamic
8934 .requested_group()
8935 .now_or_never()
8936 .expect("should not block")
8937 .unwrap();
8938 let mut backfill = req.accept(None).unwrap();
8939 pending.await.unwrap();
8940 backfill
8941 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 30_000]))
8942 .unwrap();
8943
8944 let producer = request.accept(None);
8947 producer.append_group().unwrap().finish().unwrap();
8948 producer.append_group().unwrap().finish().unwrap();
8949
8950 assert!(
8951 producer.consume().peek_group(0).is_none(),
8952 "pre-accept backfill growth is reclaimed after accept"
8953 );
8954 assert!(pool.used() <= 13_000, "usage converges: {}", pool.used());
8955 }
8956
8957 #[tokio::test]
8960 async fn a_fetch_only_cache_is_not_live() {
8961 let request = Request::new(Arc::new(broadcast::Info::default()), "test");
8962 let dynamic = request.dynamic();
8963 let consumer = request.consume();
8964
8965 let pending = consumer.fetch_group(2, None);
8966 let req = dynamic
8967 .requested_group()
8968 .now_or_never()
8969 .expect("should not block")
8970 .unwrap();
8971 let mut fetched = req.accept(None).unwrap();
8972 fetched
8973 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"2-0"))
8974 .unwrap();
8975 fetched.finish().unwrap();
8976 pending.await.unwrap();
8977 assert!(!consumer.is_live(), "a fetch-only cache is not live");
8978
8979 let producer = request.accept(None);
8980 assert!(consumer.is_live(), "an accepted track is live");
8981
8982 drop(dynamic);
8984 drop(producer);
8985 assert!(!consumer.is_live(), "a track whose producer went is not live");
8986 }
8987
8988 #[tokio::test]
8991 async fn recreated_sequence_bounds_eviction_hints() {
8992 let (producer, _pool) = pooled_producer(1 << 40);
8993 producer.create_group(5u64.into()).unwrap().finish().unwrap();
8994
8995 for _ in 0..200 {
8996 let group = producer.create_group(1u64.into()).unwrap();
8997 group.abort(Error::Cancel).unwrap();
8998 }
8999
9000 let state = producer.state.read();
9001 assert!(
9002 state.evict.len() <= 2 * state.lookup.len() + EVICT_SLACK,
9003 "stale hints are compacted: {} entries for {} slots",
9004 state.evict.len(),
9005 state.lookup.len()
9006 );
9007 }
9008
9009 #[tokio::test]
9012 async fn same_tick_write_outranks_inserted() {
9013 let unit = 100 * cache::ENTRY_OVERHEAD;
9016 let (mut producer, _pool) = pooled_producer(10 * unit);
9018
9019 producer.append_group().unwrap().finish().unwrap(); finished_group(&mut producer, 3 * unit as usize); finished_group(&mut producer, 3 * unit as usize); finished_group(&mut producer, 3 * unit as usize); finished_group(&mut producer, 3 * unit as usize); let consumer = producer.consume();
9026 assert!(consumer.peek_group(0).is_none(), "insert-only content pays first");
9027 assert!(consumer.peek_group(1).is_some(), "same-tick written content survives");
9028 }
9029
9030 #[tokio::test]
9033 async fn frame_only_writer_pays() {
9034 let (producer, pool) = pooled_producer(2_000);
9035 let mut demoted = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); demoted
9041 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
9042 .unwrap();
9043
9044 assert!(
9045 pool.used() <= 5_000,
9046 "the frame write settled the debt: {}",
9047 pool.used()
9048 );
9049 assert!(matches!(demoted.finish(), Err(Error::Evicted)));
9050 }
9051
9052 #[tokio::test]
9055 async fn each_track_owns_its_account() {
9056 let broadcast = Arc::new(broadcast::Info::default());
9057 let info = Info::default();
9058 let a = Producer::new(broadcast.clone(), "a", info.clone());
9059 let b = Producer::new(broadcast, "b", info);
9060
9061 let a = a.state.read().cache.clone();
9062 let b = b.state.read().cache.clone();
9063 assert!(!Arc::ptr_eq(&a, &b), "each track owns its account");
9064 }
9065
9066 #[tokio::test]
9069 async fn a_dynamic_defers_teardown() {
9070 let (mut producer, pool) = pooled_producer(1 << 40);
9071 let dynamic = producer.dynamic();
9072 finished_group(&mut producer, 100);
9073
9074 drop(producer);
9075 assert!(pool.used() > 0, "the handler still serves the cache");
9076
9077 drop(dynamic);
9078 assert_eq!(pool.used(), 0, "the last handle tears it down");
9079 }
9080
9081 #[tokio::test]
9087 async fn finished_track_frees_its_cache() {
9088 let (mut producer, pool) = pooled_producer(1 << 40);
9089 finished_group(&mut producer, 100);
9090 producer.finish().unwrap();
9091
9092 let state = producer.state.downgrade();
9093 drop(producer);
9094
9095 assert!(state.upgrade().is_none(), "the track state is freed");
9096 assert_eq!(pool.used(), 0, "so are its cached bytes");
9097 }
9098
9099 #[tokio::test]
9103 async fn teardown_ignores_a_settling_group() {
9104 let (producer, pool) = pooled_producer(1 << 40);
9105 let mut group = producer.append_group().unwrap();
9107 group
9108 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9109 .unwrap();
9110 drop(group);
9111
9112 let settling = producer.state.downgrade().upgrade().expect("open");
9114 drop(producer);
9115
9116 assert_eq!(pool.used(), 0, "the abrupt teardown still released the cache");
9117 drop(settling);
9118 }
9119
9120 #[tokio::test]
9123 async fn cached_group_outlives_its_track() {
9124 let (mut producer, pool) = pooled_producer(1 << 40);
9125 let sequence = finished_group(&mut producer, 100);
9126 let group = producer.consume().peek_group(sequence).expect("cached");
9127 producer.finish().unwrap();
9128
9129 let state = producer.state.downgrade();
9130 drop(producer);
9131 assert!(state.upgrade().is_none(), "the track state is freed");
9132 assert!(pool.used() > 0, "the retained group keeps its own bytes");
9133
9134 drop(group);
9135 assert_eq!(pool.used(), 0, "which it releases when dropped");
9136 }
9137
9138 #[tokio::test]
9142 async fn pre_accept_backfill_settles_late_writes() {
9143 let config = cache::Config::default()
9144 .with_capacity(2_000)
9145 .with_expiry(cache::DEFAULT_EXPIRY);
9146 let pool = cache::Pool::new(config);
9147 let broadcast = broadcast::Info {
9148 pool: pool.clone(),
9149 ..Default::default()
9150 };
9151 let request = Request::new(Arc::new(broadcast), "test");
9152 let dynamic = request.dynamic();
9153 let consumer = request.consume();
9154
9155 let pending = consumer.fetch_group(0, None);
9157 let req = dynamic
9158 .requested_group()
9159 .now_or_never()
9160 .expect("should not block")
9161 .unwrap();
9162 let mut backfill = req.accept(None).unwrap();
9163 pending.await.unwrap();
9164
9165 let producer = request.accept(None);
9167 producer.append_group().unwrap().finish().unwrap();
9168
9169 backfill
9172 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
9173 .unwrap();
9174
9175 assert!(
9176 pool.used() <= 5_000,
9177 "the frame write settled the debt: {}",
9178 pool.used()
9179 );
9180 }
9181
9182 #[tokio::test]
9186 async fn write_restarts_retention_clock() {
9187 let (producer, _pool) = pooled_producer(1 << 40);
9188 let mut straggler = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9193 straggler
9194 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9195 .unwrap();
9196 producer.append_group().unwrap().finish().unwrap(); let consumer = producer.consume();
9199 assert!(consumer.peek_group(0).is_some(), "the write restarted the clock");
9200
9201 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9203 producer.append_group().unwrap().finish().unwrap(); assert!(consumer.peek_group(0).is_none(), "idle content still expires");
9205 }
9206
9207 #[tokio::test]
9210 async fn refreshed_front_does_not_starve_expiry() {
9211 let (producer, _pool) = pooled_producer(1 << 40);
9212 let dynamic = producer.dynamic();
9213 let consumer = producer.consume();
9214
9215 producer.create_group(10u64.into()).unwrap().finish().unwrap();
9216 for sequence in 1..=5u64 {
9217 let pending = consumer.fetch_group(sequence, None);
9218 let req = dynamic
9219 .requested_group()
9220 .now_or_never()
9221 .expect("should not block")
9222 .unwrap();
9223 let mut group = req.accept(None).unwrap();
9224 group
9225 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9226 .unwrap();
9227 group.finish().unwrap();
9228 pending.await.unwrap();
9229 }
9230
9231 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9234 for sequence in 1..=4u64 {
9235 consumer.fetch_group(sequence, None).await.unwrap();
9236 }
9237
9238 for _ in 0..3 {
9240 producer.append_group().unwrap().finish().unwrap();
9241 }
9242 assert!(consumer.peek_group(5).is_none(), "expired backfill is reclaimed");
9243 assert!(consumer.peek_group(1).is_some(), "refreshed backfill survives");
9244 }
9245
9246 #[tokio::test]
9249 async fn recreated_sequence_delivered_once() {
9250 let (producer, _pool) = pooled_producer(1 << 40);
9251
9252 producer.create_group(0u64.into()).unwrap().finish().unwrap();
9253 let aborted = producer.create_group(1u64.into()).unwrap();
9254 aborted.abort(Error::Cancel).unwrap();
9255 producer.create_group(2u64.into()).unwrap().finish().unwrap();
9256 producer.create_group(1u64.into()).unwrap().finish().unwrap();
9257
9258 let mut subscriber = producer.subscribe(replay());
9259 assert_eq!(subscriber.assert_group().sequence, 0);
9260 assert_eq!(subscriber.assert_group().sequence, 2);
9261 assert_eq!(
9262 subscriber.assert_group().sequence,
9263 1,
9264 "replacement arrives at its own position"
9265 );
9266 subscriber.assert_no_group();
9267 }
9268
9269 #[tokio::test]
9273 async fn datagrams_do_not_block_eviction() {
9274 let (mut producer, pool) = pooled_producer(1_000);
9275 for _ in 0..10 {
9276 finished_group(&mut producer, 1_000);
9277 producer.append_datagram(Timestamp::ZERO, &b"beat"[..]).unwrap();
9278 }
9279
9280 let consumer = producer.consume();
9281 assert!(consumer.peek_group(0).is_none(), "old groups still evict");
9282 assert!(
9283 pool.used() < 4 * 1_256,
9284 "interleaved datagrams must not bypass the budget: {}",
9285 pool.used()
9286 );
9287 }
9288
9289 #[tokio::test]
9293 async fn aborted_group_leaves_no_ghost_sample() {
9294 let (producer, pool) = pooled_producer(1 << 40);
9295 let group0 = producer.append_group().unwrap();
9296 producer.append_group().unwrap(); assert!(pool.average().is_some(), "demoted group is sampled");
9299 group0.abort(Error::Cancel).unwrap();
9300 assert_eq!(pool.average(), None, "the abort must remove the sample");
9301 }
9302
9303 #[tokio::test]
9306 async fn empty_groups_repay_overhead() {
9307 let (producer, pool) = pooled_producer(1_000);
9308 for _ in 0..100 {
9309 let group = producer.append_group().unwrap();
9310 group.finish().unwrap();
9311 }
9312
9313 assert!(
9314 pool.used() <= 3_000,
9315 "empty-group overhead must stay near the budget: {}",
9316 pool.used()
9317 );
9318 }
9319
9320 #[tokio::test]
9323 async fn growth_on_demoted_group_is_billed() {
9324 let (producer, pool) = pooled_producer(2_000);
9325 let mut straggler = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); straggler
9330 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000]))
9331 .unwrap();
9332
9333 producer.append_group().unwrap().finish().unwrap(); let consumer = producer.consume();
9337 assert!(consumer.peek_group(0).is_none(), "the ballooned group is evicted");
9338 assert!(pool.used() <= 3_000, "growth is reclaimed: {}", pool.used());
9339 }
9340
9341 #[tokio::test]
9344 async fn refilled_sequence_stays_out_of_subscriptions() {
9345 let (producer, _pool) = pooled_producer(1 << 40);
9346 let dynamic = producer.dynamic();
9347 let consumer = producer.consume();
9348
9349 producer.create_group(0u64.into()).unwrap().finish().unwrap();
9350 let aborted = producer.create_group(1u64.into()).unwrap();
9351 aborted.abort(Error::Cancel).unwrap();
9352 producer.create_group(2u64.into()).unwrap().finish().unwrap();
9353
9354 let pending = consumer.fetch_group(1, None);
9357 let req = dynamic
9358 .requested_group()
9359 .now_or_never()
9360 .expect("should not block")
9361 .unwrap();
9362 let mut group = req.accept(None).unwrap();
9363 group
9364 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
9365 .unwrap();
9366 group.finish().unwrap();
9367 pending.await.unwrap();
9368
9369 assert!(consumer.peek_group(1).is_some());
9371 let mut subscriber = producer.subscribe(replay());
9372 assert_eq!(subscriber.assert_group().sequence, 0);
9373 assert_eq!(subscriber.assert_group().sequence, 2);
9374 subscriber.assert_no_group();
9375 }
9376
9377 #[tokio::test]
9380 async fn expired_backfill_behind_refreshed_reclaimed() {
9381 let (producer, _pool) = pooled_producer(1 << 40);
9382 let dynamic = producer.dynamic();
9383 let consumer = producer.consume();
9384
9385 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9386 for sequence in [2u64, 3u64] {
9387 let pending = consumer.fetch_group(sequence, None);
9388 let req = dynamic
9389 .requested_group()
9390 .now_or_never()
9391 .expect("should not block")
9392 .unwrap();
9393 let mut group = req.accept(None).unwrap();
9394 group
9395 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9396 .unwrap();
9397 group.finish().unwrap();
9398 pending.await.unwrap();
9399 }
9400
9401 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
9403 consumer.fetch_group(2, None).await.unwrap();
9404 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
9405 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9406
9407 let consumer = producer.consume();
9408 assert!(consumer.peek_group(2).is_some(), "refreshed backfill survives");
9409 assert!(consumer.peek_group(3).is_none(), "expired backfill is reclaimed");
9410 }
9411
9412 #[tokio::test]
9415 async fn same_tick_fetch_protects() {
9416 let (mut producer, _pool) = pooled_producer(10_000);
9418 let consumer = producer.consume();
9419
9420 finished_group(&mut producer, 3_000); finished_group(&mut producer, 3_000); finished_group(&mut producer, 3_000); consumer.fetch_group(0, None).await.unwrap();
9425
9426 finished_group(&mut producer, 3_000); finished_group(&mut producer, 3_000); assert!(consumer.peek_group(0).is_some(), "same-tick refresh protects");
9430 assert!(consumer.peek_group(1).is_none(), "the unread group dies instead");
9431 }
9432
9433 #[tokio::test]
9437 async fn refetched_latest_stays_protected() {
9438 let (producer, _pool) = pooled_producer(10_000);
9439 let dynamic = producer.dynamic();
9440 let consumer = producer.consume();
9441
9442 let straggler = producer.append_group().unwrap(); let latest = producer.append_group().unwrap(); latest.abort(Error::Cancel).unwrap();
9447
9448 let pending = consumer.fetch_group(1, None);
9450 let req = dynamic
9451 .requested_group()
9452 .now_or_never()
9453 .expect("should not block")
9454 .unwrap();
9455 let mut group = req.accept(None).unwrap();
9456 group
9457 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
9458 .unwrap();
9459 group.finish().unwrap();
9460 pending.await.unwrap();
9461
9462 {
9465 let state = producer.state.read();
9466 assert!(state.lookup.contains_key(&1), "refetched group is cached");
9467 assert!(
9468 state.evict.iter().all(|(sequence, _)| *sequence != 1),
9469 "the live edge must not be an eviction candidate"
9470 );
9471 }
9472 drop(straggler);
9473 }
9474
9475 #[tokio::test]
9478 async fn eviction_allows_refetch() {
9479 let (mut producer, _pool) = pooled_producer(10_000);
9480 let dynamic = producer.dynamic();
9481
9482 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let consumer = producer.consume();
9487 assert!(consumer.peek_group(0).is_none());
9488 let pending = consumer.fetch_group(0, None);
9489
9490 let req = dynamic
9491 .requested_group()
9492 .now_or_never()
9493 .expect("should not block")
9494 .unwrap();
9495 assert_eq!(req.sequence(), 0);
9496
9497 let mut group = req.accept(None).unwrap();
9498 group
9499 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"refetched"))
9500 .unwrap();
9501 group.finish().unwrap();
9502
9503 let mut group = pending.await.unwrap();
9504 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"refetched");
9505 }
9506
9507 #[test]
9517 fn an_aborted_group_releases_its_sequence() {
9518 let producer = track_producer("test", None);
9519 let consumer = producer.consume();
9520
9521 let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
9522 group
9523 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
9524 .unwrap();
9525
9526 assert!(matches!(
9528 producer.create_group(group::Info { sequence: 3 }),
9529 Err(Error::Duplicate)
9530 ));
9531 assert!(consumer.peek_group(3).is_some());
9532
9533 group.abort(Error::Cancel).unwrap();
9534
9535 assert!(consumer.peek_group(3).is_none(), "an aborted slot is a cache miss");
9536 producer
9537 .create_group(group::Info { sequence: 3 })
9538 .expect("an aborted slot releases its sequence")
9539 .finish()
9540 .unwrap();
9541 }
9542
9543 #[tokio::test]
9546 async fn fetched_backfill_not_subscribed() {
9547 let (producer, _pool) = pooled_producer(1 << 40);
9548 let dynamic = producer.dynamic();
9549 let consumer = producer.consume();
9550
9551 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9553 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9554
9555 let pending = consumer.fetch_group(2, None);
9557 let req = dynamic
9558 .requested_group()
9559 .now_or_never()
9560 .expect("should not block")
9561 .unwrap();
9562 let mut group = req.accept(None).unwrap();
9563 group
9564 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
9565 .unwrap();
9566 group.finish().unwrap();
9567 let mut fetched = pending.await.unwrap();
9568 assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"backfill");
9569 assert!(consumer.peek_group(2).is_some(), "backfill is cached for later fetches");
9570
9571 let mut subscriber = producer.subscribe(replay());
9573 assert_eq!(subscriber.assert_group().sequence, 5);
9574 assert_eq!(subscriber.assert_group().sequence, 6);
9575 subscriber.assert_no_group();
9576 }
9577
9578 #[tokio::test]
9581 async fn expired_backfill_reclaimed() {
9582 let (producer, pool) = pooled_producer(1 << 40);
9583 let dynamic = producer.dynamic();
9584 let consumer = producer.consume();
9585
9586 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9587
9588 let pending = consumer.fetch_group(2, None);
9590 let req = dynamic
9591 .requested_group()
9592 .now_or_never()
9593 .expect("should not block")
9594 .unwrap();
9595 let mut group = req.accept(None).unwrap();
9596 group
9597 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
9598 .unwrap();
9599 group.finish().unwrap();
9600 pending.await.unwrap();
9601 let used = pool.used();
9602
9603 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9605 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9606
9607 assert!(consumer.peek_group(2).is_none(), "expired backfill is reclaimed");
9608 assert!(pool.used() < used, "its bytes are released");
9609 }
9610
9611 #[tokio::test]
9612 async fn fetch_aborts_with_track() {
9613 let producer = track_producer("test", None);
9614 let dynamic = producer.dynamic();
9615 let consumer = producer.consume();
9616
9617 let pending = consumer.fetch_group(3, None);
9618 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
9619
9620 producer.abort(Error::Cancel).unwrap();
9621 assert!(pending.await.is_err());
9622 drop(dynamic);
9623 }
9624}