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(crate) fn poll_resolved(&mut self, waiter: &kio::Waiter) -> Poll<()> {
3215 let _ = ready!(self.state.poll_ref(waiter, |state| {
3216 state.poll_fetch_cached(self.sequence, self.frame_start)
3217 }));
3218 self.done = true;
3221 self.remove();
3222 Poll::Ready(())
3223 }
3224
3225 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
3231 loop {
3232 ready!(self.result.poll_unused(waiter));
3233 let mut fetch = self.fetch.lock();
3236 if self.result.is_used() {
3237 continue;
3238 }
3239 fetch.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
3240 return Poll::Ready(());
3241 }
3242 }
3243
3244 fn remove(&self) {
3247 self.fetch
3248 .lock()
3249 .remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result));
3250 }
3251}
3252
3253impl Drop for group::Request {
3254 fn drop(&mut self) {
3255 if self.done {
3256 return;
3257 }
3258 self.remove();
3259 if let Ok(mut outcome) = self.result.write() {
3260 outcome.rejected = Some(Error::Dropped);
3261 }
3262 }
3263}
3264
3265pub struct Fetching {
3271 inner: FetchingKind,
3272 stats: stats::Scope,
3275}
3276
3277enum FetchingKind {
3278 Plain {
3279 state: kio::Consumer<TrackState>,
3280 fetch: kio::Shared<FetchState>,
3281 sequence: u64,
3282 frame_start: u64,
3285 result: Option<kio::Consumer<FetchOutcome>>,
3288 hit: Option<Box<group::Consumer>>,
3291 },
3292 Spliced(kio::Pending<super::resume::Fetching>),
3294}
3295
3296impl kio::Pollable for Fetching {
3297 type Output = Result<group::Consumer>;
3298
3299 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3300 let (state, fetch, sequence, frame_start, result, hit) = match &self.inner {
3301 FetchingKind::Plain {
3302 state,
3303 fetch,
3304 sequence,
3305 frame_start,
3306 result,
3307 hit,
3308 } => (state, fetch, *sequence, *frame_start, result.as_ref(), hit),
3309 FetchingKind::Spliced(spliced) => {
3310 return kio::Pollable::poll(&**spliced, waiter)
3313 .map(|res| res.map(|group| group.with_meter(self.stats.meter())));
3314 }
3315 };
3316
3317 let resolve = |mut group: group::Consumer| {
3320 group.start_at(frame_start);
3321 group.with_meter(self.stats.meter())
3322 };
3323 if let Some(group) = hit {
3324 return Poll::Ready(Ok(resolve((**group).clone())));
3325 }
3326
3327 let cached =
3330 |waiter: &kio::Waiter| match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) {
3331 Poll::Ready(Ok(res)) => Poll::Ready(res.map(resolve)),
3332 Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))),
3333 Poll::Pending => Poll::Pending,
3334 };
3335 if let Poll::Ready(res) = cached(waiter) {
3336 return Poll::Ready(res);
3337 }
3338
3339 let Some(result) = result else {
3341 return match fetch.poll(waiter, |fetch| match fetch.has_handlers() {
3344 false => Poll::Ready(()),
3345 true => Poll::Pending,
3346 }) {
3347 Poll::Ready(_guard) => Poll::Ready(Err(Error::NotFound)),
3348 Poll::Pending => Poll::Pending,
3349 };
3350 };
3351
3352 let err = match result.poll(waiter, |outcome| match &outcome.rejected {
3355 Some(err) => Poll::Ready(err.clone()),
3356 None => Poll::Pending,
3357 }) {
3358 Poll::Ready(Ok(err)) => err,
3359 Poll::Ready(Err(_closed)) => Error::NotFound,
3360 Poll::Pending => return Poll::Pending,
3361 };
3362 match cached(waiter) {
3365 Poll::Ready(res) => Poll::Ready(res),
3366 Poll::Pending => Poll::Ready(Err(err)),
3367 }
3368 }
3369}
3370
3371pub struct Subscriber {
3399 name: Arc<str>,
3400 broadcast: Arc<broadcast::Info>,
3402 info: Info,
3403 inner: SubscriberKind,
3404 stats: stats::Scope,
3407 _stats_sub: stats::Subscription,
3410}
3411
3412enum SubscriberKind {
3413 Plain(PlainSubscriber),
3414 Spliced(Box<super::resume::Subscriber>),
3416}
3417
3418#[derive(Clone)]
3423struct Drift {
3424 budget: Duration,
3425 edge: Edge,
3426 outer: Option<LiveEdge>,
3427 successor: Option<Successor>,
3428}
3429
3430struct GroupExpiry {
3432 state: kio::ConsumerWeak<TrackState>,
3438 subscription: kio::Consumer<Subscription>,
3439 anchor: kio::Consumer<Anchor>,
3440 bound: Option<u64>,
3441 sequence: u64,
3442}
3443
3444impl group::Expiry for GroupExpiry {
3445 fn is_expired(&self, waiter: &kio::Waiter) -> bool {
3446 let mut max_age = Duration::default();
3447 let _ = self.subscription.poll(waiter, |subscription| {
3448 max_age = subscription.max_age;
3449 Poll::<()>::Pending
3450 });
3451
3452 let mut anchor = Anchor::default();
3453 let _ = self.anchor.poll(waiter, |current| {
3454 anchor = (**current).clone();
3455 Poll::<()>::Pending
3456 });
3457 let anchor = anchor.capped(self.bound);
3458 let cap = anchor.cap;
3459 let outer = anchor
3461 .edge
3462 .filter(LiveEdge::is_live)
3463 .map(|live| (live.sequence, live.timestamp));
3464 let successor = anchor
3465 .successor
3466 .as_ref()
3467 .and_then(|successor| successor.poll_start(waiter));
3468
3469 let mut expired = false;
3470 let _ = self.state.poll(waiter, |state| {
3471 let budget = clamp_max_age(max_age, state.max_age_bound());
3472 loop {
3473 let next = loop {
3478 let next = state.first_servable(self.sequence.saturating_add(1), cap);
3479 match next {
3480 Some(group) if group.poll_closed(waiter).is_ready() && group.is_aborted() => continue,
3481 next => break next,
3482 }
3483 };
3484 let edge = state.drift_edge(cap, outer, successor);
3485 if let Some(live) = edge.presentation
3488 && let Some(slot) = state.lookup.get(&live.sequence)
3489 && slot.group.poll_closed(waiter).is_ready()
3490 && slot.group.is_aborted()
3491 {
3492 continue;
3493 }
3494 expired = state.is_stale(self.sequence, &edge, budget);
3495 if expired {
3496 break;
3497 }
3498
3499 let mut timestamp_raced = false;
3509 if let Some(slot) = state.lookup.get(&self.sequence) {
3510 let group = &slot.group;
3511 if group.timestamp().is_none()
3512 && group.poll_timestamp(waiter).is_ready()
3513 && group.timestamp().is_some()
3514 {
3515 timestamp_raced = true;
3516 }
3517 }
3518 let past = edge
3519 .presentation
3520 .map_or(self.sequence, |live| live.sequence.max(self.sequence));
3521 let beyond = state
3522 .lookup
3523 .range((std::ops::Bound::Excluded(past), std::ops::Bound::Unbounded))
3524 .map(|(_, slot)| slot)
3525 .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap))
3526 .filter(|slot| slot.visible && !slot.group.is_aborted())
3527 .map(|slot| &slot.group);
3528 for group in next.into_iter().chain(beyond) {
3529 if group.timestamp().is_none()
3530 && group.poll_timestamp(waiter).is_ready()
3531 && group.timestamp().is_some()
3532 {
3533 timestamp_raced = true;
3534 break;
3535 }
3536 }
3537 if !timestamp_raced {
3538 break;
3539 }
3540 }
3541
3542 Poll::<()>::Pending
3545 });
3546
3547 expired
3548 }
3549}
3550
3551#[derive(Clone, Copy)]
3554struct Edge {
3555 presentation: Option<PresentationEdge>,
3557 outer: Option<(u64, Timestamp)>,
3560 cap: Option<u64>,
3563 successor: Option<Timestamp>,
3565}
3566
3567#[derive(Clone, Default, PartialEq)]
3570pub(crate) struct Anchor {
3571 pub cap: Option<u64>,
3573 pub edge: Option<LiveEdge>,
3578 pub successor: Option<Successor>,
3584}
3585
3586impl Anchor {
3587 pub fn capped(mut self, cap: Option<u64>) -> Self {
3590 let capped = servable_cap(cap, self.cap);
3591 if capped != self.cap {
3592 self.cap = capped;
3593 self.successor = None;
3594 }
3595 self
3596 }
3597}
3598
3599#[derive(Clone)]
3602pub(crate) struct LiveEdge {
3603 pub sequence: u64,
3604 pub timestamp: Timestamp,
3605 stamp: u32,
3606 track: kio::ConsumerWeak<TrackState>,
3607}
3608
3609impl LiveEdge {
3610 fn is_live(&self) -> bool {
3615 self.track.read().holds(self.sequence, self.stamp)
3616 }
3617}
3618
3619impl PartialEq for LiveEdge {
3620 fn eq(&self, other: &Self) -> bool {
3621 self.sequence == other.sequence
3622 && self.timestamp == other.timestamp
3623 && self.stamp == other.stamp
3624 && self.track.same_channel(&other.track)
3625 }
3626}
3627
3628#[derive(Clone, Copy)]
3630struct PresentationEdge {
3631 sequence: u64,
3632 stamp: u32,
3635 timestamp: Timestamp,
3640}
3641
3642struct PlainSubscriber {
3644 state: kio::Consumer<TrackState>,
3645
3646 subscription: kio::Producer<Subscription>,
3647 index: usize,
3649 datagram_index: usize,
3651 min_sequence: u64,
3653 next_sequence: u64,
3656 end_sequence: Option<u64>,
3662 parked: BTreeMap<u64, group::Consumer>,
3667 stale_cap: Option<u64>,
3671 drift_anchor: kio::Producer<Anchor>,
3674 stale: stats::Content,
3680 seek_pending: BTreeMap<u64, stats::Content>,
3683}
3684
3685impl PlainSubscriber {
3686 fn anchor(&self, end: Option<u64>) -> Anchor {
3689 self.drift_anchor.read().clone().capped(end)
3690 }
3691
3692 fn update_drift_anchor(&mut self, outer: Anchor) {
3694 let anchor = outer.capped(self.end_sequence);
3695 if *self.drift_anchor.read() != anchor
3697 && let Ok(mut current) = self.drift_anchor.write()
3698 {
3699 *current = anchor;
3700 }
3701 }
3702
3703 fn poll<F, R>(&self, waiter: &kio::Waiter, f: F) -> Poll<Result<R>>
3705 where
3706 F: Fn(&kio::Ref<'_, TrackState>) -> Poll<Result<R>>,
3707 {
3708 Poll::Ready(match ready!(self.state.poll(waiter, f)) {
3709 Ok(res) => res,
3710 Err(state) => Err(state.abort.clone().unwrap_or(Error::Dropped)),
3712 })
3713 }
3714
3715 fn take_stale(&mut self) -> stats::Content {
3717 std::mem::take(&mut self.stale)
3718 }
3719
3720 fn note_stale(&mut self, group: &group::Consumer) {
3723 self.stale.add(group.content());
3724 }
3725
3726 fn poll_drift(&self, anchor: Anchor, waiter: &kio::Waiter) -> Poll<Result<Drift>> {
3737 let mut max_age = Duration::default();
3738 let _ = self.subscription.poll(waiter, |subscription| {
3739 max_age = subscription.max_age;
3740 Poll::<()>::Pending
3741 });
3742 let cap = anchor.cap;
3743 let outer = anchor.edge;
3744 let successor = anchor.successor;
3745 self.poll(waiter, |state| {
3746 Poll::Ready(Ok(Drift {
3749 budget: clamp_max_age(max_age, state.max_age_bound()),
3750 edge: state.drift_edge(cap, None, None),
3751 outer: outer.clone(),
3752 successor: successor.clone(),
3753 }))
3754 })
3755 }
3756
3757 fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll<Result<bool>> {
3760 let outer = drift
3763 .outer
3764 .as_ref()
3765 .filter(|live| live.is_live())
3766 .map(|live| (live.sequence, live.timestamp));
3767 let successor = drift
3768 .successor
3769 .as_ref()
3770 .and_then(|successor| successor.poll_start(waiter));
3771 let presentation = drift.edge.presentation;
3772 let cap = drift.edge.cap;
3773 let budget = drift.budget;
3774 self.poll(waiter, move |state| {
3775 let edge = Edge {
3776 presentation,
3777 outer,
3778 cap,
3779 successor,
3780 };
3781 Poll::Ready(Ok(state.is_stale(group.sequence, &edge, budget)))
3782 })
3783 }
3784
3785 fn with_expiry(&self, group: group::Consumer) -> group::Consumer {
3786 let sequence = group.sequence;
3787 group.with_expiry(Arc::new(GroupExpiry {
3788 state: self.state.weak(),
3789 subscription: self.subscription.consume(),
3790 anchor: self.drift_anchor.consume(),
3791 bound: None,
3792 sequence,
3793 }))
3794 }
3795
3796 fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
3797 let watch = |group: &group::Consumer| match group.poll_closed(waiter) {
3805 Poll::Pending => true,
3806 Poll::Ready(()) => !group.is_aborted(),
3807 };
3808
3809 let min_sequence = self.min_sequence;
3814 self.parked
3815 .retain(|sequence, group| *sequence >= min_sequence && watch(group));
3816
3817 let drift = ready!(self.poll_drift(self.anchor(self.end_sequence), waiter))?;
3819
3820 loop {
3821 let consumer = match self.parked.keys().next().copied() {
3824 Some(sequence) if super::subscription::before_end(sequence, self.end_sequence) => {
3825 let group = self.parked.remove(&sequence).expect("just looked it up");
3826 group.cache_refresh();
3828 group
3829 }
3830 _ => {
3831 let Some((producer, found_index)) =
3832 ready!(self.poll(waiter, |state| state.poll_recv_group(self.index, self.min_sequence))?)
3833 else {
3834 if self.parked.is_empty() {
3837 return Poll::Ready(Ok(None));
3838 }
3839 return Poll::Pending;
3840 };
3841 let consumer = producer.consume();
3842 consumer.cache_refresh();
3845 self.index = found_index + 1;
3846
3847 if !super::subscription::before_end(consumer.sequence, self.end_sequence) {
3850 if watch(&consumer) {
3854 self.parked.insert(consumer.sequence, consumer);
3855 }
3856 continue;
3857 }
3858 consumer
3859 }
3860 };
3861
3862 if ready!(self.poll_stale(&consumer, &drift, waiter))? {
3865 self.stale.add(consumer.content());
3866 continue;
3867 }
3868 return Poll::Ready(Ok(Some(self.with_expiry(consumer))));
3869 }
3870 }
3871
3872 fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
3873 let Some((datagram, found_index)) =
3874 ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?)
3875 else {
3876 return Poll::Ready(Ok(None));
3877 };
3878
3879 self.datagram_index = found_index + 1;
3880 Poll::Ready(Ok(Some(datagram)))
3881 }
3882
3883 fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
3884 let floor = self.next_sequence.max(self.min_sequence);
3885 let Some(group) = ready!(self.poll_seek_group(floor, self.end_sequence, waiter))? else {
3886 return Poll::Ready(Ok(None));
3887 };
3888 self.next_sequence = group.sequence.saturating_add(1);
3889 self.commit_seek_stale(self.next_sequence);
3891 group.cache_refresh();
3893 Poll::Ready(Ok(Some(group)))
3894 }
3895
3896 fn poll_seek_group(
3909 &mut self,
3910 floor: u64,
3911 end: Option<u64>,
3912 waiter: &kio::Waiter,
3913 ) -> Poll<Result<Option<group::Consumer>>> {
3914 let mut floor = floor.max(self.min_sequence);
3915 let end = super::subscription::min_some(end, self.end_sequence);
3916 let drift = ready!(self.poll_drift(self.anchor(end), waiter))?;
3918
3919 loop {
3920 let Some(producer) = ready!(self.poll(waiter, |state| state.poll_next_in_range(floor, end))?) else {
3921 return Poll::Ready(Ok(None));
3928 };
3929 let group = producer.consume();
3930
3931 if ready!(self.poll_stale(&group, &drift, waiter))? {
3934 self.seek_pending.insert(group.sequence, group.content());
3939 floor = group.sequence.saturating_add(1);
3940 continue;
3941 }
3942
3943 self.seek_pending.remove(&group.sequence);
3946 return Poll::Ready(Ok(Some(self.with_expiry(group))));
3947 }
3948 }
3949
3950 fn commit_seek_stale(&mut self, committed: u64) {
3961 let keep = self.seek_pending.split_off(&committed);
3962 for (_, content) in std::mem::replace(&mut self.seek_pending, keep) {
3963 self.stale.add(content);
3964 }
3965 }
3966
3967 fn discard_seek_conviction(&mut self, sequence: u64) {
3971 self.seek_pending.remove(&sequence);
3972 }
3973}
3974
3975#[derive(Clone)]
3981pub struct Control {
3982 subscription: kio::Producer<Subscription>,
3983}
3984
3985impl Control {
3986 pub fn subscription(&self) -> Subscription {
3988 self.subscription.read().clone()
3989 }
3990
3991 pub fn update(&self, subscription: Subscription) -> Result<()> {
3996 let mut state = self.subscription.write().map_err(|_| Error::Closed)?;
3997 *state = subscription;
3998 Ok(())
3999 }
4000}
4001
4002impl Subscriber {
4003 pub fn info(&self) -> &Info {
4008 &self.info
4009 }
4010
4011 pub fn name(&self) -> &str {
4013 &self.name
4014 }
4015
4016 pub fn broadcast(&self) -> &broadcast::Info {
4020 &self.broadcast
4021 }
4022
4023 fn count_stale(&mut self, meter: &stats::Meter) {
4029 if meter.is_tracked() {
4034 meter.stale(self.take_stale());
4035 }
4036 }
4037
4038 pub(crate) fn take_stale(&mut self) -> stats::Content {
4041 match &mut self.inner {
4042 SubscriberKind::Plain(plain) => plain.take_stale(),
4043 SubscriberKind::Spliced(spliced) => spliced.take_stale(),
4044 }
4045 }
4046
4047 pub(crate) fn set_anchor(&mut self, anchor: Anchor) {
4058 match &mut self.inner {
4059 SubscriberKind::Plain(plain) => {
4060 plain.stale_cap = anchor.cap;
4061 plain.update_drift_anchor(anchor);
4062 }
4063 SubscriberKind::Spliced(spliced) => spliced.set_anchor(anchor),
4064 }
4065 }
4066
4067 pub(crate) fn commit_seek_stale(&mut self, committed: u64) {
4074 match &mut self.inner {
4075 SubscriberKind::Plain(plain) => plain.commit_seek_stale(committed),
4076 SubscriberKind::Spliced(spliced) => spliced.commit_seek_stale(committed),
4077 }
4078 }
4079
4080 pub(crate) fn discard_seek_conviction(&mut self, sequence: u64) {
4085 match &mut self.inner {
4086 SubscriberKind::Plain(plain) => plain.discard_seek_conviction(sequence),
4087 SubscriberKind::Spliced(spliced) => spliced.discard_seek_conviction(sequence),
4088 }
4089 }
4090
4091 pub(crate) fn poll_stale(&mut self, group: &group::Consumer, waiter: &kio::Waiter) -> Poll<Result<bool>> {
4097 let plain = match &mut self.inner {
4098 SubscriberKind::Plain(plain) => plain,
4099 SubscriberKind::Spliced(spliced) => return spliced.poll_stale(group, waiter),
4100 };
4101 let drift = ready!(plain.poll_drift(plain.anchor(plain.end_sequence), waiter))?;
4102 let stale = ready!(plain.poll_stale(group, &drift, waiter))?;
4103 if stale {
4104 plain.note_stale(group);
4105 }
4106 Poll::Ready(Ok(stale))
4107 }
4108
4109 pub fn control(&self) -> Control {
4111 Control {
4112 subscription: match &self.inner {
4113 SubscriberKind::Plain(plain) => plain.subscription.clone(),
4114 SubscriberKind::Spliced(spliced) => spliced.prefs(),
4115 },
4116 }
4117 }
4118
4119 pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4148 let meter = self.stats.meter();
4149 let res = match &mut self.inner {
4150 SubscriberKind::Plain(plain) => plain.poll_recv_group(waiter),
4151 SubscriberKind::Spliced(spliced) => spliced.poll_recv_group(waiter),
4152 };
4153 self.count_stale(&meter);
4154 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4155 }
4156
4157 pub async fn recv_group(&mut self) -> Result<Option<group::Consumer>> {
4164 kio::wait(|waiter| self.poll_recv_group(waiter)).await
4165 }
4166
4167 pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
4178 let meter = self.stats.meter();
4179 let res = match &mut self.inner {
4180 SubscriberKind::Plain(plain) => plain.poll_recv_datagram(waiter),
4181 SubscriberKind::Spliced(spliced) => spliced.poll_recv_datagram(waiter),
4182 };
4183 if let Poll::Ready(Ok(Some(datagram))) = &res {
4186 meter.datagram(datagram.payload.len() as u64);
4187 }
4188 res
4189 }
4190
4191 pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
4198 kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
4199 }
4200
4201 fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4203 let meter = self.stats.meter();
4204 let res = match &mut self.inner {
4205 SubscriberKind::Plain(plain) => plain.poll_next_group(waiter),
4206 SubscriberKind::Spliced(spliced) => spliced.poll_next_group(waiter),
4207 };
4208 self.count_stale(&meter);
4209 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4210 }
4211
4212 pub(crate) fn poll_seek_group(
4217 &mut self,
4218 floor: u64,
4219 end: Option<u64>,
4220 waiter: &kio::Waiter,
4221 ) -> Poll<Result<Option<group::Consumer>>> {
4222 let meter = self.stats.meter();
4223 let res = match &mut self.inner {
4224 SubscriberKind::Plain(plain) => plain.poll_seek_group(floor, end, waiter),
4225 SubscriberKind::Spliced(spliced) => spliced.poll_seek_group(floor, end, waiter),
4226 };
4227 self.count_stale(&meter);
4228 res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter))))
4229 }
4230
4231 pub fn ordered(self) -> Ordered {
4241 Ordered { inner: self }
4242 }
4243
4244 pub fn is_clone(&self, other: &Self) -> bool {
4246 match (&self.inner, &other.inner) {
4247 (SubscriberKind::Plain(a), SubscriberKind::Plain(b)) => a.state.same_channel(&b.state),
4248 (SubscriberKind::Spliced(a), SubscriberKind::Spliced(b)) => a.is_clone(b),
4249 _ => false,
4250 }
4251 }
4252
4253 pub(crate) fn poll_start(&mut self, waiter: &kio::Waiter) -> Poll<Option<u64>> {
4257 match &mut self.inner {
4258 SubscriberKind::Plain(plain) => {
4259 let start = ready!(Consumer::poll_state_start(&plain.state, waiter));
4260 Poll::Ready(start.map(|start| start.max(plain.min_sequence)))
4261 }
4262 SubscriberKind::Spliced(spliced) => spliced.poll_start(waiter),
4263 }
4264 }
4265
4266 pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
4268 match &mut self.inner {
4269 SubscriberKind::Plain(plain) => plain.poll(waiter, |state| state.poll_finished()),
4270 SubscriberKind::Spliced(spliced) => spliced.poll_finished(waiter),
4271 }
4272 }
4273
4274 pub(crate) fn poll_end(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<u64>>> {
4276 match &mut self.inner {
4277 SubscriberKind::Plain(plain) => plain.poll(waiter, |state| state.poll_end()),
4278 SubscriberKind::Spliced(spliced) => spliced.poll_end(waiter),
4279 }
4280 }
4281
4282 pub async fn finished(&mut self) -> Result<u64> {
4291 kio::wait(|waiter| self.poll_finished(waiter)).await
4292 }
4293
4294 pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
4302 let (start, end) = super::subscription::sequence_bounds(groups);
4303 self.raise_start_to(start);
4304 self.end_at(end.map_or(Bound::Unbounded, Bound::Excluded));
4305 }
4306
4307 pub(crate) fn start_at(&mut self, sequence: u64) {
4314 match &mut self.inner {
4315 SubscriberKind::Plain(plain) => plain.min_sequence = sequence,
4316 SubscriberKind::Spliced(spliced) => spliced.start_at(sequence),
4317 }
4318 }
4319
4320 pub(crate) fn raise_start_to(&mut self, sequence: u64) {
4326 match &mut self.inner {
4327 SubscriberKind::Plain(plain) => plain.min_sequence = plain.min_sequence.max(sequence),
4328 SubscriberKind::Spliced(spliced) => spliced.raise_start_to(sequence),
4329 }
4330 }
4331
4332 pub(crate) fn end_at(&mut self, end: impl Into<Cap>) {
4346 let end = end.into();
4347 match &mut self.inner {
4348 SubscriberKind::Plain(plain) => {
4349 plain.end_sequence = end.exclusive();
4350 let outer = Anchor {
4353 cap: plain.stale_cap,
4354 ..plain.drift_anchor.read().clone()
4355 };
4356 plain.update_drift_anchor(outer);
4357 }
4358 SubscriberKind::Spliced(spliced) => spliced.end_at(end),
4359 }
4360 }
4361
4362 pub fn subscription(&self) -> Subscription {
4364 self.control().subscription()
4365 }
4366
4367 pub fn update(&mut self, subscription: Subscription) -> Result<()> {
4373 match &mut self.inner {
4374 SubscriberKind::Plain(plain) => {
4375 let mut state = plain.subscription.write().map_err(|_| Error::Closed)?;
4376 *state = subscription;
4377 }
4378 SubscriberKind::Spliced(spliced) => spliced.update(subscription),
4379 }
4380 Ok(())
4381 }
4382
4383 pub fn latest(&self) -> Option<u64> {
4385 match &self.inner {
4386 SubscriberKind::Plain(plain) => plain.state.read().max_sequence,
4387 SubscriberKind::Spliced(spliced) => spliced.latest(),
4388 }
4389 }
4390}
4391
4392pub struct Ordered {
4419 inner: Subscriber,
4420}
4421
4422impl Ordered {
4423 pub fn info(&self) -> &Info {
4425 self.inner.info()
4426 }
4427
4428 pub fn name(&self) -> &str {
4430 self.inner.name()
4431 }
4432
4433 pub fn broadcast(&self) -> &broadcast::Info {
4435 self.inner.broadcast()
4436 }
4437
4438 pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
4451 self.inner.poll_next_group(waiter)
4452 }
4453
4454 pub async fn next_group(&mut self) -> Result<Option<group::Consumer>> {
4456 kio::wait(|waiter| self.poll_next_group(waiter)).await
4457 }
4458
4459 pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Datagram>>> {
4467 self.inner.poll_recv_datagram(waiter)
4468 }
4469
4470 pub async fn recv_datagram(&mut self) -> Result<Option<Datagram>> {
4477 kio::wait(|waiter| self.poll_recv_datagram(waiter)).await
4478 }
4479
4480 pub fn set_groups(&mut self, groups: impl RangeBounds<u64>) {
4484 self.inner.set_groups(groups);
4485 }
4486
4487 pub fn control(&self) -> Control {
4490 self.inner.control()
4491 }
4492
4493 pub fn subscription(&self) -> Subscription {
4495 self.inner.subscription()
4496 }
4497
4498 pub fn update(&mut self, subscription: Subscription) -> Result<()> {
4502 self.inner.update(subscription)
4503 }
4504
4505 pub fn poll_finished(&mut self, waiter: &kio::Waiter) -> Poll<Result<u64>> {
4507 self.inner.poll_finished(waiter)
4508 }
4509
4510 pub async fn finished(&mut self) -> Result<u64> {
4515 kio::wait(|waiter| self.poll_finished(waiter)).await
4516 }
4517
4518 pub fn latest(&self) -> Option<u64> {
4520 self.inner.latest()
4521 }
4522
4523 pub fn is_clone(&self, other: &Self) -> bool {
4525 self.inner.is_clone(&other.inner)
4526 }
4527}
4528
4529pub struct Request {
4541 name: Arc<str>,
4542 broadcast: Arc<broadcast::Info>,
4544 state: kio::Producer<TrackState>,
4545
4546 prev_subscription: Option<Subscription>,
4548
4549 alive: Arc<Alive>,
4552
4553 _dynamic: Dynamic,
4558
4559 stats: stats::Scope,
4562
4563 resolving_start: bool,
4566}
4567
4568impl Request {
4569 pub(crate) fn new(broadcast: Arc<broadcast::Info>, name: impl Into<Arc<str>>) -> Self {
4570 let name = name.into();
4571 let state = TrackState::spawn(broadcast.clone());
4572 let alive = Alive::new(name.clone(), state.clone());
4573 let dynamic = Dynamic::new(name.clone(), state.clone(), alive.clone());
4574 Self {
4575 name,
4576 broadcast,
4577 state,
4578 prev_subscription: None,
4579 alive,
4580 _dynamic: dynamic,
4581 stats: stats::Scope::default(),
4582 resolving_start: false,
4583 }
4584 }
4585
4586 pub(crate) fn resolving_start(mut self) -> Self {
4591 self.resolving_start = true;
4592 self
4593 }
4594
4595 pub(crate) fn with_stats(mut self, scope: stats::Scope) -> Self {
4598 self.stats = scope;
4599 self
4600 }
4601
4602 pub fn name(&self) -> &str {
4604 &self.name
4605 }
4606
4607 pub fn consume(&self) -> Consumer {
4609 Consumer::plain(self.name.clone(), self.state.consume())
4610 }
4611
4612 pub fn dynamic(&self) -> Dynamic {
4616 Dynamic::new(self.name.clone(), self.state.clone(), self.alive.clone())
4617 }
4618
4619 pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> {
4622 self.state.poll_unused(waiter).map(|_| ())
4623 }
4624
4625 pub(crate) fn claim(self) -> Self {
4627 if let Ok(mut state) = self.state.write() {
4628 state.claimed = true;
4629 }
4630 self
4631 }
4632
4633 pub(crate) fn reject_unused(&self, err: Error) -> bool {
4637 match self.state.write_unused() {
4638 kio::Unused::Idle(guard) => {
4639 commit_abort(guard, err);
4640 true
4641 }
4642 kio::Unused::Closed => true,
4643 kio::Unused::Used => false,
4644 }
4645 }
4646
4647 pub fn accept(self, info: impl Into<Option<Info>>) -> Producer {
4654 let info = TrackState::normalize_info(&self.broadcast, info.into().unwrap_or_default());
4655 if let Ok(mut state) = self.state.write() {
4658 state.accept(info.clone());
4659 state.start_pending = self.resolving_start;
4660 }
4661 self.alive.publish(Some(&self.stats));
4664 Producer {
4665 name: self.name,
4666 info,
4667 broadcast: self.broadcast,
4668 state: self.state,
4669 prev_subscription: None,
4670 alive: self.alive,
4671 stats: self.stats,
4672 }
4673 }
4674
4675 pub fn reject(self, err: Error) {
4677 if let Ok(mut state) = self.state.write() {
4678 state.abort = Some(err);
4679 }
4680 }
4681
4682 pub fn subscription(&self) -> Option<Subscription> {
4685 let state = self.state.read();
4686 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
4687 drop(state);
4688 snapshot_subscription(&subs, bound)
4689 }
4690
4691 pub async fn subscription_changed(&mut self) -> Option<Subscription> {
4694 kio::wait(|waiter| self.poll_subscription_changed(waiter)).await
4695 }
4696
4697 pub fn poll_subscription_changed(&mut self, waiter: &kio::Waiter) -> Poll<Option<Subscription>> {
4699 let state = self.state.read();
4700 let (subs, bound) = (state.subscriptions.clone(), state.max_age_bound());
4701 drop(state);
4702
4703 poll_combined_changed(&subs, bound, &mut self.prev_subscription, waiter)
4704 }
4705
4706 pub(super) fn weak(&self) -> TrackWeak {
4707 TrackWeak {
4708 name: self.name.clone(),
4709 state: self.state.weak(),
4710 }
4711 }
4712}
4713
4714#[cfg(test)]
4715use futures::FutureExt;
4716
4717#[cfg(test)]
4718#[allow(missing_docs)] impl Subscriber {
4720 pub fn assert_group(&mut self) -> group::Consumer {
4721 self.recv_group()
4722 .now_or_never()
4723 .expect("group would have blocked")
4724 .expect("would have errored")
4725 .expect("track was closed")
4726 }
4727
4728 pub fn assert_no_group(&mut self) {
4729 assert!(
4730 self.recv_group().now_or_never().is_none(),
4731 "recv_group would not have blocked"
4732 );
4733 }
4734
4735 pub fn assert_not_closed(&mut self) {
4736 assert!(self.finished().now_or_never().is_none(), "should not be closed");
4737 }
4738
4739 pub fn assert_closed(&mut self) {
4740 assert!(self.finished().now_or_never().is_some(), "should be closed");
4741 }
4742
4743 pub fn assert_error(&mut self) {
4745 assert!(
4746 self.finished().now_or_never().expect("should not block").is_err(),
4747 "should be error"
4748 );
4749 }
4750
4751 pub fn assert_is_clone(&self, other: &Self) {
4752 assert!(self.is_clone(other), "should be clone");
4753 }
4754
4755 pub fn assert_not_clone(&self, other: &Self) {
4756 assert!(!self.is_clone(other), "should not be clone");
4757 }
4758}
4759
4760#[cfg(test)]
4761mod test {
4762 use super::*;
4763 use crate::frame;
4764 use crate::model::test_tracing::count_drop_warnings;
4765 use std::time::Duration;
4766
4767 fn track_producer(name: impl Into<Arc<str>>, info: impl Into<Option<Info>>) -> Producer {
4770 Producer::new(Arc::new(broadcast::Info::default()), name, info)
4771 }
4772
4773 fn replay() -> Subscription {
4775 Subscription::default().with_max_age(Duration::from_secs(30))
4776 }
4777
4778 fn live_groups(state: &TrackState) -> usize {
4780 state.lookup.len()
4781 }
4782
4783 fn first_live_sequence(state: &TrackState) -> u64 {
4785 state
4786 .arrival
4787 .iter()
4788 .find(|(sequence, stamp)| state.lookup.get(sequence).is_some_and(|slot| slot.stamp == *stamp))
4789 .map(|(sequence, _)| *sequence)
4790 .unwrap()
4791 }
4792
4793 fn recv_datagram(dg: &mut Subscriber) -> Datagram {
4795 dg.recv_datagram()
4796 .now_or_never()
4797 .expect("datagram would have blocked")
4798 .expect("would have errored")
4799 .expect("track was closed")
4800 }
4801
4802 #[tokio::test]
4806 async fn peek_resolves_below_the_declared_start() {
4807 let mut producer = track_producer("test", None);
4808 let consumer = producer.consume();
4809
4810 let mut cached = producer.create_group(group::Info { sequence: 1 }).unwrap();
4811 cached.write_frame(Timestamp::ZERO, b"backfill".to_vec()).unwrap();
4812 cached.finish().unwrap();
4813
4814 let waiter = kio::Waiter::noop();
4816 assert!(consumer.poll_peek_group(0, &waiter).is_pending());
4817
4818 producer.start_at(3).unwrap();
4821 assert!(matches!(consumer.poll_peek_group(0, &waiter), Poll::Ready(None)));
4822 assert!(matches!(consumer.poll_peek_group(1, &waiter), Poll::Ready(Some(_))));
4823 assert!(consumer.poll_peek_group(3, &waiter).is_pending());
4824
4825 producer.start_at(4).unwrap();
4828 assert!(matches!(consumer.poll_peek_group(3, &waiter), Poll::Ready(None)));
4829 producer.start_at(0).unwrap();
4830 assert!(consumer.poll_peek_group(0, &waiter).is_pending());
4831 }
4832
4833 #[tokio::test]
4834 async fn append_datagram_shares_group_sequence() {
4835 let mut producer = track_producer("test", None);
4836 let ts = Timestamp::from_millis(10).unwrap();
4837
4838 assert_eq!(producer.append_group().unwrap().sequence, 0);
4840 assert_eq!(producer.append_datagram(ts, &b"a"[..]).unwrap(), 1);
4841 assert_eq!(producer.append_group().unwrap().sequence, 2);
4842 assert_eq!(producer.append_datagram(ts, &b"b"[..]).unwrap(), 3);
4843 assert_eq!(producer.latest(), Some(3));
4844 }
4845
4846 #[tokio::test]
4847 async fn append_datagram_roundtrip() {
4848 let mut producer = track_producer("test", None);
4849 let mut dg = producer.subscribe(None);
4850
4851 let ts = Timestamp::from_millis(42).unwrap();
4852 let seq = producer.append_datagram(ts, &b"hello"[..]).unwrap();
4853
4854 let got = recv_datagram(&mut dg);
4855 assert_eq!(got.sequence, seq);
4856 assert_eq!(got.timestamp, ts);
4857 assert_eq!(&got.payload[..], b"hello");
4858 }
4859
4860 #[tokio::test]
4861 async fn insert_datagram_preserves_sequence() {
4862 let mut producer = track_producer("test", None);
4863 let mut dg = producer.subscribe(None);
4864
4865 let ts = Timestamp::from_millis(5).unwrap();
4866 producer
4868 .insert_datagram(100, ts, bytes::Bytes::from_static(b"x"))
4869 .unwrap();
4870
4871 assert_eq!(recv_datagram(&mut dg).sequence, 100);
4872 assert_eq!(producer.append_group().unwrap().sequence, 101);
4874 }
4875
4876 #[tokio::test]
4877 async fn insert_datagram_leaves_a_gap() {
4878 let mut producer = track_producer("test", None);
4879 let mut dg = producer.subscribe(None);
4880 let ts = Timestamp::from_millis(0).unwrap();
4881
4882 producer
4883 .insert_datagram(10, ts, bytes::Bytes::from_static(b"gap"))
4884 .unwrap();
4885 assert_eq!(recv_datagram(&mut dg).sequence, 10);
4886 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
4887 assert_eq!(producer.append_group().unwrap().sequence, 12);
4888 }
4889
4890 #[tokio::test]
4891 async fn insert_datagram_out_of_order_does_not_rewind() {
4892 let mut producer = track_producer("test", None);
4893 let mut dg = producer.subscribe(None);
4894 let ts = Timestamp::from_millis(0).unwrap();
4895
4896 producer
4897 .insert_datagram(10, ts, bytes::Bytes::from_static(b"high"))
4898 .unwrap();
4899 producer
4900 .insert_datagram(5, ts, bytes::Bytes::from_static(b"low"))
4901 .unwrap();
4902
4903 assert_eq!(recv_datagram(&mut dg).sequence, 10);
4904 assert_eq!(recv_datagram(&mut dg).sequence, 5);
4905 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 11);
4906 }
4907
4908 #[tokio::test]
4909 async fn insert_datagram_duplicate_is_best_effort() {
4910 let mut producer = track_producer("test", None);
4911 let mut dg = producer.subscribe(None);
4912 let ts = Timestamp::from_millis(0).unwrap();
4913
4914 producer
4915 .insert_datagram(3, ts, bytes::Bytes::from_static(b"first"))
4916 .unwrap();
4917 producer
4918 .insert_datagram(3, ts, bytes::Bytes::from_static(b"again"))
4919 .unwrap();
4920
4921 assert_eq!(&recv_datagram(&mut dg).payload[..], b"first");
4922 assert_eq!(&recv_datagram(&mut dg).payload[..], b"again");
4923 assert_eq!(producer.append_datagram(ts, &b"next"[..]).unwrap(), 4);
4924 }
4925
4926 #[tokio::test]
4927 async fn insert_datagram_stale_does_not_rewind_after_append() {
4928 let mut producer = track_producer("test", None);
4929 let mut dg = producer.subscribe(None);
4930 let ts = Timestamp::from_millis(0).unwrap();
4931
4932 assert_eq!(producer.append_datagram(ts, &b"0"[..]).unwrap(), 0);
4933 assert_eq!(producer.append_datagram(ts, &b"1"[..]).unwrap(), 1);
4934 producer
4935 .insert_datagram(0, ts, bytes::Bytes::from_static(b"stale"))
4936 .unwrap();
4937
4938 assert_eq!(recv_datagram(&mut dg).sequence, 0);
4939 assert_eq!(recv_datagram(&mut dg).sequence, 1);
4940 assert_eq!(recv_datagram(&mut dg).sequence, 0);
4941 assert_eq!(producer.append_datagram(ts, &b"2"[..]).unwrap(), 2);
4942 assert_eq!(producer.append_group().unwrap().sequence, 3);
4943 }
4944
4945 #[tokio::test]
4946 async fn insert_datagram_cloned_producers_share_counter() {
4947 let mut producer = track_producer("test", None);
4948 let mut other = producer.clone();
4949 let mut dg = producer.subscribe(None);
4950 let ts = Timestamp::from_millis(0).unwrap();
4951
4952 producer
4953 .insert_datagram(4, ts, bytes::Bytes::from_static(b"a"))
4954 .unwrap();
4955 assert_eq!(other.append_datagram(ts, &b"b"[..]).unwrap(), 5);
4956 other.insert_datagram(8, ts, bytes::Bytes::from_static(b"c")).unwrap();
4957 assert_eq!(producer.append_group().unwrap().sequence, 9);
4958
4959 assert_eq!(recv_datagram(&mut dg).sequence, 4);
4960 assert_eq!(recv_datagram(&mut dg).sequence, 5);
4961 assert_eq!(recv_datagram(&mut dg).sequence, 8);
4962 }
4963
4964 #[test]
4965 fn insert_datagram_after_finish_is_closed() {
4966 let mut producer = track_producer("test", None);
4967 let ts = Timestamp::from_millis(0).unwrap();
4968 producer.finish().unwrap();
4969 assert!(matches!(
4970 producer.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")),
4971 Err(Error::Closed)
4972 ));
4973 assert!(matches!(producer.append_datagram(ts, &b"x"[..]), Err(Error::Closed)));
4974 }
4975
4976 #[test]
4977 fn insert_datagram_after_abort_fails() {
4978 let producer = track_producer("test", None);
4979 let mut other = producer.clone();
4980 let ts = Timestamp::from_millis(0).unwrap();
4981 producer.abort(Error::Cancel).unwrap();
4982 assert!(other.insert_datagram(0, ts, bytes::Bytes::from_static(b"x")).is_err());
4983 }
4984
4985 #[test]
4986 fn insert_datagram_respects_finish_at() {
4987 let mut producer = track_producer("test", None);
4988 let ts = Timestamp::from_millis(0).unwrap();
4989 producer.finish_at(10).unwrap();
4990 producer
4991 .insert_datagram(5, ts, bytes::Bytes::from_static(b"ok"))
4992 .unwrap();
4993 assert!(matches!(
4994 producer.insert_datagram(10, ts, bytes::Bytes::from_static(b"late")),
4995 Err(Error::Closed)
4996 ));
4997 assert_eq!(producer.append_group().unwrap().sequence, 6);
4998 }
4999
5000 #[test]
5002 fn resume_position_uses_the_latest_group() {
5003 let mut datagram_only = track_producer("datagram-only", None);
5004 let datagram_only_consumer = datagram_only.consume();
5005 datagram_only
5006 .insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
5007 .unwrap();
5008 assert_eq!(
5009 datagram_only_consumer.resume_position(),
5010 None,
5011 "a datagram creates no group position to resume"
5012 );
5013
5014 let mut producer = track_producer("mixed", None);
5015 let consumer = producer.consume();
5016 let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
5017 group
5018 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
5019 .unwrap();
5020 producer
5021 .insert_datagram(8, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
5022 .unwrap();
5023
5024 assert_eq!(
5025 consumer.resume_position(),
5026 Some(Position { group: 3, frame: 1 }),
5027 "the replacement must continue the open group"
5028 );
5029 group.abort(Error::Cancel).unwrap();
5030 }
5031
5032 #[tokio::test]
5035 async fn recv_datagram_leaves_the_ordered_cursor_alone() {
5036 let mut producer = track_producer("test", None);
5037 let mut datagrams = producer.subscribe(None);
5038 let mut subscriber = producer.subscribe(None).ordered();
5039 let ts = Timestamp::from_millis(5).unwrap();
5040
5041 producer
5042 .insert_datagram(5, ts, bytes::Bytes::from_static(b"x"))
5043 .unwrap();
5044 assert_eq!(recv_datagram(&mut datagrams).sequence, 5);
5045
5046 producer.create_group(group::Info { sequence: 3 }).unwrap();
5047 producer.create_group(group::Info { sequence: 6 }).unwrap();
5048
5049 let mut next = || {
5050 subscriber
5051 .next_group()
5052 .now_or_never()
5053 .expect("group would have blocked")
5054 .expect("would have errored")
5055 .expect("track was closed")
5056 .sequence
5057 };
5058 assert_eq!(next(), 3, "the datagram at sequence 5 did not consume group 3");
5059 assert_eq!(next(), 6);
5060 }
5061
5062 #[tokio::test]
5063 async fn datagram_normalized_to_track_timescale() {
5064 let info = Info::default().with_timescale(Timescale::MICRO);
5065 let mut producer = track_producer("test", info);
5066 let mut dg = producer.subscribe(None);
5067
5068 producer
5070 .append_datagram(Timestamp::from_millis(2).unwrap(), &b"z"[..])
5071 .unwrap();
5072 let got = recv_datagram(&mut dg);
5073 assert_eq!(got.timestamp.scale(), Timescale::MICRO);
5074 assert_eq!(got.timestamp.value(), 2_000);
5075 }
5076
5077 #[tokio::test]
5078 async fn datagram_rejects_oversized() {
5079 let mut producer = track_producer("test", None);
5080 let big = bytes::Bytes::from(vec![0u8; crate::model::datagram::MAX_DATAGRAM_PAYLOAD + 1]);
5081 let ts = Timestamp::from_millis(0).unwrap();
5082 assert!(matches!(
5083 producer.append_datagram(ts, big.clone()),
5084 Err(Error::FrameTooLarge)
5085 ));
5086 assert!(matches!(
5087 producer.insert_datagram(0, ts, big),
5088 Err(Error::FrameTooLarge)
5089 ));
5090 }
5091
5092 #[tokio::test]
5093 async fn datagram_fanout_to_subscribers() {
5094 let mut producer = track_producer("test", None);
5095 let mut a = producer.subscribe(None);
5097 let mut b = producer.subscribe(None);
5098 let ts = Timestamp::from_millis(1).unwrap();
5099
5100 producer.append_datagram(ts, &b"first"[..]).unwrap();
5101 producer.append_datagram(ts, &b"second"[..]).unwrap();
5102
5103 assert_eq!(&recv_datagram(&mut a).payload[..], b"first");
5105 assert_eq!(&recv_datagram(&mut a).payload[..], b"second");
5106 assert_eq!(&recv_datagram(&mut b).payload[..], b"first");
5107 assert_eq!(&recv_datagram(&mut b).payload[..], b"second");
5108 }
5109
5110 #[test]
5111 fn datagram_buffer_drops_oldest_at_capacity() {
5112 let mut producer = track_producer("test", None);
5113 let mut slow = producer.subscribe(None);
5114 let mut fast = producer.subscribe(None);
5115 let count = MAX_DATAGRAMS * 3;
5116 for sequence in 0..count {
5117 producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();
5118 assert_eq!(recv_datagram(&mut fast).sequence, sequence as u64);
5119 }
5120 assert_eq!(producer.state.read().datagrams.len(), MAX_DATAGRAMS);
5121 for sequence in count - MAX_DATAGRAMS..count {
5122 assert_eq!(recv_datagram(&mut slow).sequence, sequence as u64);
5123 }
5124 assert!(slow.poll_recv_datagram(&kio::Waiter::noop()).is_pending());
5125 }
5126
5127 #[tokio::test]
5128 async fn datagram_recv_pends_until_written() {
5129 let mut producer = track_producer("test", None);
5130 let mut dg = producer.subscribe(None);
5131
5132 assert!(
5133 dg.recv_datagram().now_or_never().is_none(),
5134 "should block with no datagrams"
5135 );
5136
5137 producer
5138 .append_datagram(Timestamp::from_millis(0).unwrap(), &b"go"[..])
5139 .unwrap();
5140 assert_eq!(&recv_datagram(&mut dg).payload[..], b"go");
5141 }
5142
5143 #[tokio::test]
5147 async fn datagram_wire_roundtrip_between_tracks() {
5148 use crate::coding::{Decode, Encode};
5149 use crate::lite;
5150
5151 let version = lite::Version::Lite05;
5152
5153 let mut origin = track_producer("test", None);
5155 let mut origin_dg = origin.subscribe(None);
5156 let ts = Timestamp::from_millis(7).unwrap();
5157 let seq = origin.append_datagram(ts, &b"payload"[..]).unwrap();
5158
5159 let d = recv_datagram(&mut origin_dg);
5160 let body = lite::Datagram {
5161 subscribe: 5,
5162 sequence: d.sequence,
5163 timestamp: d.timestamp.value(),
5164 payload: d.payload.clone(),
5165 }
5166 .encode_bytes(version)
5167 .unwrap();
5168
5169 let mut slice = &body[..];
5171 let wire = lite::Datagram::decode(&mut slice, version).unwrap();
5172 let mut downstream = track_producer("test", None);
5173 let mut downstream_dg = downstream.subscribe(None);
5174 downstream
5175 .insert_datagram(
5176 wire.sequence,
5177 Timestamp::new(wire.timestamp, Timescale::MILLI).unwrap(),
5178 wire.payload,
5179 )
5180 .unwrap();
5181
5182 let got = recv_datagram(&mut downstream_dg);
5183 assert_eq!(got.sequence, seq);
5184 assert_eq!(got.timestamp, ts);
5185 assert_eq!(&got.payload[..], b"payload");
5186 }
5187
5188 #[tokio::test]
5189 async fn evict_expired_groups() {
5190 let producer = track_producer("test", None);
5191
5192 producer.append_group().unwrap(); producer.append_group().unwrap(); producer.append_group().unwrap(); {
5198 let state = producer.state.read();
5199 assert_eq!(live_groups(&state), 3);
5200 assert_eq!(state.offset, 0);
5201 }
5202
5203 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5205
5206 producer.append_group().unwrap(); {
5213 let state = producer.state.read();
5214 assert_eq!(live_groups(&state), 1);
5215 assert_eq!(first_live_sequence(&state), 3);
5216 assert_eq!(state.offset, 3);
5217 assert!(!state.lookup.contains_key(&0));
5218 assert!(!state.lookup.contains_key(&1));
5219 assert!(!state.lookup.contains_key(&2));
5220 assert!(state.lookup.contains_key(&3));
5221 }
5222 }
5223
5224 #[tokio::test]
5228 async fn aging_out_a_finished_group_keeps_the_clean_end() {
5229 let producer = track_producer("test", None);
5230 let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
5231 let mut consumer = group.consume();
5232
5233 group
5234 .write_frame(Timestamp::from_millis(0).unwrap(), b"hello".as_slice())
5235 .unwrap();
5236 assert_eq!(consumer.next_frame().await.unwrap().unwrap().size, 5);
5237
5238 crate::model::clock::advance(cache::DEFAULT_EXPIRY * 2);
5240 group.finish().unwrap();
5241 let _next = producer.create_group(group::Info { sequence: 1 }).unwrap();
5242
5243 assert!(consumer.next_frame().await.unwrap().is_none());
5244 }
5245
5246 #[tokio::test]
5250 async fn active_reader_survives_expiry() {
5251 let producer = track_producer("test", None);
5252 let mut subscriber = producer.subscribe(None);
5253
5254 let mut group = producer.create_group(0u64.into()).unwrap();
5256 for _ in 0..10 {
5257 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5258 }
5259 group.finish().unwrap();
5260 let mut reading = subscriber.assert_group();
5261
5262 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5264
5265 for seq in 2..12u64 {
5268 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5269 let frame = reading.next_frame().await;
5270 assert!(
5271 matches!(frame, Ok(Some(_))),
5272 "an actively-read group must not expire mid-read (step {seq})"
5273 );
5274 producer.create_group(seq.into()).unwrap().finish().unwrap();
5275 }
5276
5277 let state = producer.state.read();
5278 assert!(state.lookup.contains_key(&0), "the read group survived");
5279 assert!(!state.lookup.contains_key(&1), "the unread group still expired");
5280 }
5281
5282 #[tokio::test]
5287 async fn slow_prefetch_reader_survives_expiry() {
5288 let producer = track_producer("test", None);
5289 let mut subscriber = producer.subscribe(None);
5290
5291 let mut group = producer.create_group(0u64.into()).unwrap();
5292 for _ in 0..20 {
5293 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5294 }
5295 group.finish().unwrap();
5296 let mut reading = subscriber.assert_group();
5297
5298 for seq in 1..20u64 {
5301 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5302 let frame = reading.read_frame().await;
5303 assert!(
5304 matches!(frame, Ok(Some(_))),
5305 "a slow prefetch reader must not expire mid-read (step {seq})"
5306 );
5307 producer.create_group(seq.into()).unwrap().finish().unwrap();
5308 }
5309 }
5310
5311 #[tokio::test]
5315 async fn delivery_restarts_the_expiry_clock() {
5316 let producer = track_producer("test", None);
5317 let mut subscriber = producer.subscribe(replay());
5318
5319 let mut group = producer.create_group(0u64.into()).unwrap();
5320 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5321 group.finish().unwrap();
5322 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5324
5325 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5327 let mut reading = subscriber.assert_group();
5328
5329 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5332 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5333
5334 let frame = reading.read_frame().await.unwrap();
5335 assert!(frame.is_some(), "a just-delivered group must not expire unread");
5336 }
5337
5338 #[tokio::test]
5342 async fn streaming_frame_writes_keep_the_group_alive() {
5343 let producer = track_producer("test", None);
5344 let mut straggler = producer.create_group(0u64.into()).unwrap();
5345 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5347
5348 let mut frame = straggler
5349 .create_frame(frame::Info {
5350 size: 10,
5351 timestamp: Timestamp::ZERO,
5352 })
5353 .unwrap();
5354 for seq in 2..12u64 {
5357 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2);
5358 frame.write(bytes::Bytes::from_static(b"x")).unwrap();
5359 producer.create_group(seq.into()).unwrap().finish().unwrap();
5360 }
5361 frame.finish().unwrap();
5362 straggler.finish().unwrap();
5363
5364 let state = producer.state.read();
5365 assert!(
5366 state.lookup.contains_key(&0),
5367 "a group streaming a frame survives expiry"
5368 );
5369 }
5370
5371 #[tokio::test]
5376 async fn coalesced_frame_completion_keeps_the_group_alive() {
5377 let producer = track_producer("test", None);
5378 let mut straggler = producer.create_group(0u64.into()).unwrap();
5379 producer.create_group(1u64.into()).unwrap().finish().unwrap();
5381
5382 let mut frame = straggler
5383 .create_frame_owned(
5384 frame::Info {
5385 size: 3,
5386 timestamp: Timestamp::ZERO,
5387 },
5388 &Default::default(),
5389 )
5390 .unwrap();
5391
5392 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5396 frame.write(bytes::Bytes::from_static(b"abc")).unwrap();
5397 frame.finish().unwrap();
5398 straggler.finish().unwrap();
5399
5400 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5402
5403 let state = producer.state.read();
5404 assert!(
5405 state.lookup.contains_key(&0),
5406 "a group whose frame just completed must not expire"
5407 );
5408 }
5409
5410 #[tokio::test]
5413 async fn parked_reoffer_restarts_the_expiry_clock() {
5414 let producer = track_producer("test", None);
5415 let mut subscriber = producer.subscribe(None);
5416 subscriber.set_groups(..1);
5417
5418 for seq in 0..2u64 {
5419 let mut group = producer.create_group(seq.into()).unwrap();
5420 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5421 group.finish().unwrap();
5422 }
5423
5424 assert_eq!(subscriber.assert_group().sequence, 0);
5426 subscriber.assert_no_group();
5427
5428 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5430 subscriber.set_groups(..2);
5431 let mut reading = subscriber.assert_group();
5432 assert_eq!(reading.sequence, 1);
5433
5434 crate::model::clock::advance(cache::DEFAULT_EXPIRY - Duration::from_secs(1));
5437 producer.create_group(2u64.into()).unwrap().finish().unwrap();
5438
5439 let frame = reading.read_frame().await.unwrap();
5440 assert!(frame.is_some(), "a just-re-offered group must not expire unread");
5441 }
5442
5443 #[tokio::test]
5444 async fn evict_keeps_max_sequence() {
5445 let producer = track_producer("test", None);
5446 producer.append_group().unwrap(); crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5450
5451 producer.append_group().unwrap(); {
5455 let state = producer.state.read();
5456 assert_eq!(live_groups(&state), 1);
5457 assert_eq!(first_live_sequence(&state), 1);
5458 assert_eq!(state.offset, 1);
5459 }
5460 }
5461
5462 #[tokio::test]
5463 async fn no_eviction_when_fresh() {
5464 let producer = track_producer("test", None);
5465 producer.append_group().unwrap(); producer.append_group().unwrap(); producer.append_group().unwrap(); {
5470 let state = producer.state.read();
5471 assert_eq!(live_groups(&state), 3);
5472 assert_eq!(state.offset, 0);
5473 }
5474 }
5475
5476 #[tokio::test]
5477 async fn consumer_skips_evicted_groups() {
5478 let producer = track_producer("test", None);
5479 producer.append_group().unwrap(); let mut consumer = producer.subscribe(None);
5482
5483 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
5484 producer.append_group().unwrap(); let group = consumer.assert_group();
5488 assert_eq!(group.sequence, 1);
5489 }
5490
5491 fn track_producer_expiring(name: impl Into<Arc<str>>, expiry: impl Into<Option<Duration>>) -> Producer {
5493 track_producer_pooled(name, cache::Pool::new(cache::Config::default().with_expiry(expiry)))
5494 }
5495
5496 fn track_producer_pooled(name: impl Into<Arc<str>>, pool: cache::Pool) -> Producer {
5498 Producer::new(
5499 Arc::new(broadcast::Info {
5500 pool,
5501 ..Default::default()
5502 }),
5503 name,
5504 None,
5505 )
5506 }
5507
5508 #[tokio::test]
5512 async fn pool_sweep_expires_without_a_write() {
5513 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5514 let producer = track_producer_pooled("test", pool.clone());
5515 let mut stalled = producer.append_group().unwrap(); stalled.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5517 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5520 pool.sweep();
5521
5522 assert!(
5523 !producer.state.read().lookup.contains_key(&0),
5524 "the sweep reclaimed an idle open group with no write behind it"
5525 );
5526 }
5527
5528 #[test]
5529 fn cache_gc_dates_activity_before_expiring_it() {
5530 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5531 let now = crate::model::clock::now();
5532 assert_eq!(pool.gc(now), Some(now + Duration::from_millis(500)));
5533 let producer = track_producer_pooled("test", pool.clone());
5534 let mut group = producer.append_group().unwrap();
5535 group.write_frame(Timestamp::ZERO, b"first".as_slice()).unwrap();
5536 producer.append_group().unwrap();
5537 let later = now + Duration::from_secs(60);
5539 pool.gc(later);
5540 assert!(producer.state.read().lookup.contains_key(&0));
5541 pool.gc(later + Duration::from_secs(2));
5542 assert!(!producer.state.read().lookup.contains_key(&0));
5543 }
5544
5545 #[test]
5546 fn cache_gc_reaches_old_entries_behind_a_fresh_front() {
5547 let expiry = Duration::from_secs(1);
5548 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
5549 let producer = track_producer_pooled("test", pool.clone());
5550 let now = crate::model::clock::now();
5551 let groups: Vec<_> = (0..EVICT_SCAN * 3).map(|_| producer.append_group().unwrap()).collect();
5552 producer.append_group().unwrap();
5553 pool.gc(now);
5554 for group in &groups[..EVICT_SCAN * 2] {
5556 group.cache_refresh();
5557 }
5558 pool.gc(now + expiry * 2);
5559 let state = producer.state.read();
5560 for sequence in 0..EVICT_SCAN * 2 {
5561 assert!(state.lookup.contains_key(&(sequence as u64)), "fresh front survives");
5562 }
5563 for sequence in EVICT_SCAN * 2..EVICT_SCAN * 3 {
5564 assert!(!state.lookup.contains_key(&(sequence as u64)), "old tail is reclaimed");
5565 }
5566 }
5567
5568 #[tokio::test]
5572 async fn pool_sweep_drains_a_deep_backlog() {
5573 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
5574 let producer = track_producer_pooled("test", pool.clone());
5575
5576 let backlog = 4 * EVICT_SCAN;
5578 for _ in 0..backlog {
5579 let mut group = producer.append_group().unwrap();
5580 group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5581 }
5582 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5585 pool.sweep();
5586
5587 let state = producer.state.read();
5588 let stale = (0..backlog as u64).filter(|seq| state.lookup.contains_key(seq)).count();
5589 assert_eq!(stale, 0, "one sweep reclaimed the whole idle backlog");
5590 }
5591
5592 #[tokio::test]
5593 async fn pool_expiry_controls_eviction() {
5594 let producer = track_producer_expiring("test", Duration::from_secs(1));
5596 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5600 producer.append_group().unwrap(); let state = producer.state.read();
5604 assert_eq!(live_groups(&state), 1);
5605 assert_eq!(first_live_sequence(&state), 1);
5606 }
5607
5608 #[tokio::test]
5609 async fn small_frame_write_expires_idle_siblings() {
5610 let producer = track_producer_expiring("test", Duration::from_secs(1));
5611 producer.append_group().unwrap().finish().unwrap(); let mut live = producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5615 live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5616
5617 let expired = !producer.state.read().lookup.contains_key(&0);
5618 assert!(expired, "a small frame write runs expiry");
5619 }
5620
5621 #[tokio::test]
5622 async fn fresh_expiry_scan_does_not_wake_track_consumers() {
5623 use std::sync::atomic::{AtomicBool, Ordering};
5624
5625 let producer = track_producer_expiring("test", cache::DEFAULT_EXPIRY);
5626 producer.append_group().unwrap().finish().unwrap();
5627 let mut live = producer.append_group().unwrap();
5628 let mut consumer = producer.subscribe(None);
5629 assert_eq!(consumer.assert_group().sequence, 0);
5630 assert_eq!(consumer.assert_group().sequence, 1);
5631
5632 let woken = Arc::new(AtomicBool::new(false));
5633 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
5634 assert!(consumer.poll_recv_group(&waiter).is_pending());
5635
5636 live.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
5637 assert!(
5638 !woken.load(Ordering::SeqCst),
5639 "a no-op expiry scan must not wake track consumers"
5640 );
5641 }
5642
5643 #[tokio::test]
5644 async fn streaming_frame_write_expires_idle_siblings() {
5645 let producer = track_producer_expiring("test", Duration::from_secs(1));
5646 producer.append_group().unwrap().finish().unwrap(); let mut live = producer.append_group().unwrap(); let mut frame = live
5649 .create_frame(frame::Info {
5650 size: 1,
5651 timestamp: Timestamp::ZERO,
5652 })
5653 .unwrap();
5654
5655 crate::model::clock::advance(Duration::from_secs(2));
5656 frame.write(b"x".as_slice()).unwrap();
5657
5658 let expired = !producer.state.read().lookup.contains_key(&0);
5659 assert!(expired, "a streamed chunk runs expiry");
5660 }
5661
5662 #[tokio::test]
5663 async fn appended_datagram_expires_idle_groups() {
5664 let mut producer = track_producer_expiring("test", Duration::from_secs(1));
5665 producer.append_group().unwrap().finish().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5669 producer.append_datagram(Timestamp::ZERO, b"x".as_slice()).unwrap();
5670
5671 let expired = !producer.state.read().lookup.contains_key(&0);
5672 assert!(expired, "an appended datagram runs expiry");
5673 }
5674
5675 #[tokio::test]
5676 async fn forwarded_datagram_expires_idle_groups() {
5677 let mut producer = track_producer_expiring("test", Duration::from_secs(1));
5678 producer.append_group().unwrap().finish().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5682 producer
5683 .insert_datagram(2, Timestamp::ZERO, bytes::Bytes::from_static(b"x"))
5684 .unwrap();
5685
5686 let expired = !producer.state.read().lookup.contains_key(&0);
5687 assert!(expired, "a forwarded datagram runs expiry");
5688 }
5689
5690 #[tokio::test]
5694 async fn max_age_does_not_drive_wall_eviction() {
5695 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(1)));
5696 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(10));
5700 producer.append_group().unwrap(); let state = producer.state.read();
5703 assert_eq!(live_groups(&state), 2, "max_age is media time, not a wall clock");
5704 }
5705
5706 #[tokio::test]
5708 async fn disabled_pool_expiry_never_reclaims() {
5709 let producer = track_producer_expiring("test", None);
5710 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(3600));
5713 producer.append_group().unwrap(); let state = producer.state.read();
5716 assert_eq!(live_groups(&state), 2);
5717 }
5718
5719 #[test]
5720 fn max_age_clamped_to_cache() {
5721 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5722
5723 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
5727 assert_eq!(subscriber.subscription().max_age, Duration::from_secs(10));
5728 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5729
5730 subscriber
5732 .update(Subscription::default().with_max_age(Duration::from_millis(500)))
5733 .unwrap();
5734 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_millis(500));
5735
5736 subscriber
5737 .update(Subscription::default().with_max_age(Duration::ZERO))
5738 .unwrap();
5739 assert_eq!(producer.subscription().unwrap().max_age, Duration::ZERO);
5740 }
5741
5742 fn track_producer_capped(name: impl Into<Arc<str>>, info: Info, cap: Duration) -> Producer {
5745 Producer::new(
5746 Arc::new(broadcast::Info {
5747 cache_duration: cap,
5748 ..Default::default()
5749 }),
5750 name,
5751 info,
5752 )
5753 }
5754
5755 #[test]
5756 fn origin_cache_duration_clamps_max_age() {
5757 let capped = track_producer_capped(
5760 "test",
5761 Info::default().with_max_age(Duration::from_secs(60)),
5762 Duration::from_secs(1),
5763 );
5764 assert_eq!(capped.state.read().max_age_bound(), Some(Duration::from_secs(1)));
5765 assert_eq!(capped.subscribe(None).info().max_age, Duration::from_secs(1));
5766
5767 let under = track_producer_capped(
5768 "test",
5769 Info::default().with_max_age(Duration::from_millis(500)),
5770 Duration::from_secs(1),
5771 );
5772 assert_eq!(under.state.read().max_age_bound(), Some(Duration::from_millis(500)));
5773 }
5774
5775 #[tokio::test]
5778 async fn origin_cache_duration_does_not_wall_evict() {
5779 let producer = track_producer_capped(
5780 "test",
5781 Info::default().with_max_age(Duration::from_secs(60)),
5782 Duration::from_secs(1),
5783 );
5784 producer.append_group().unwrap(); crate::model::clock::advance(Duration::from_secs(2));
5788 producer.append_group().unwrap(); let state = producer.state.read();
5791 assert_eq!(live_groups(&state), 2);
5792 }
5793
5794 #[test]
5795 fn max_age_clamped_via_every_update_path() {
5796 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5797 let over = Subscription::default().with_max_age(Duration::from_secs(10));
5798
5799 let mut subscriber = producer.subscribe(over.clone());
5802 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5803
5804 subscriber.control().update(over.clone()).unwrap();
5805 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5806
5807 subscriber.update(over).unwrap();
5808 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5809 }
5810
5811 #[test]
5812 fn max_age_aggregate_clamps_across_subscribers() {
5813 let producer = track_producer("test", Info::default().with_max_age(Duration::from_secs(2)));
5814
5815 let _a = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
5818 let _b = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
5819
5820 assert_eq!(producer.subscription().unwrap().max_age, Duration::from_secs(2));
5821 }
5822
5823 #[test]
5824 fn churned_subscribers_do_not_accumulate() {
5825 let producer = track_producer("test", None);
5826 let consumer = producer.consume();
5827 let _steady = producer.subscribe(None);
5828
5829 for _ in 0..100 {
5831 drop(producer.subscribe(None));
5832 drop(consumer.subscribe(None));
5833 }
5834
5835 let subs = producer.state.read().subscriptions.clone();
5838 let len = subs.read().len();
5839 assert!(len <= 8, "departed subscribers accumulated: {len}");
5840 }
5841
5842 #[test]
5843 fn aggregate_poll_prunes_after_a_peak() {
5844 let mut producer = track_producer("test", None);
5845 let waiter = kio::Waiter::noop();
5846 let _steady = producer.subscribe(None);
5847 assert!(producer.poll_subscription_changed(&waiter).is_ready());
5848
5849 drop((0..1000).map(|_| producer.subscribe(None)).collect::<Vec<_>>());
5852 for _ in 0..100 {
5853 drop(producer.subscribe(None));
5854 assert!(producer.poll_subscription_changed(&waiter).is_pending());
5855
5856 let subs = producer.state.read().subscriptions.clone();
5857 assert_eq!(subs.read().len(), 1, "the poll walked past departed subscribers");
5858 }
5859 }
5860
5861 fn append_at(producer: &mut Producer, millis: u64) -> u64 {
5864 let mut group = producer.append_group().unwrap();
5865 group
5866 .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
5867 .unwrap();
5868 group.finish().unwrap();
5869 group.sequence
5870 }
5871
5872 fn drain(subscriber: &mut Subscriber) -> Vec<u64> {
5874 let mut sequences = Vec::new();
5875 while let Some(Ok(Some(group))) = subscriber.recv_group().now_or_never() {
5876 sequences.push(group.sequence);
5877 }
5878 sequences
5879 }
5880
5881 #[test]
5882 fn real_time_skips_a_backlog_to_the_live_edge() {
5883 let mut producer = track_producer("test", None);
5884 for second in 0..5 {
5885 append_at(&mut producer, second * 1000);
5886 }
5887
5888 let mut subscriber = producer.subscribe(None);
5892 assert_eq!(drain(&mut subscriber), vec![4]);
5893
5894 append_at(&mut producer, 5000);
5896 assert_eq!(drain(&mut subscriber), vec![5]);
5897 }
5898
5899 #[test]
5900 fn real_time_skips_a_backlog_after_catching_up() {
5901 let mut producer = track_producer("test", None);
5902 append_at(&mut producer, 0);
5903 let mut subscriber = producer.subscribe(None);
5904 assert_eq!(drain(&mut subscriber), vec![0]);
5905
5906 for second in 1..6 {
5909 append_at(&mut producer, second * 1000);
5910 }
5911 assert_eq!(drain(&mut subscriber), vec![5]);
5912 }
5913
5914 #[test]
5915 fn a_newer_edge_changes_an_active_catch_up() {
5916 let mut producer = track_producer("test", None);
5917 for second in 0..5 {
5918 append_at(&mut producer, second * 1000);
5919 }
5920 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
5921 assert_eq!(
5922 subscriber
5923 .recv_group()
5924 .now_or_never()
5925 .unwrap()
5926 .unwrap()
5927 .unwrap()
5928 .sequence,
5929 3
5930 );
5931
5932 append_at(&mut producer, 5000);
5935 append_at(&mut producer, 10000);
5936 assert_eq!(drain(&mut subscriber), vec![5, 6]);
5937 }
5938
5939 #[test]
5940 fn a_growing_edge_changes_an_active_catch_up() {
5941 let mut producer = track_producer("test", None);
5942 append_at(&mut producer, 0);
5943 append_at(&mut producer, 1000);
5944 let mut edge = producer.append_group().unwrap();
5945 edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"a"))
5946 .unwrap();
5947
5948 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
5949 assert_eq!(
5950 subscriber
5951 .recv_group()
5952 .now_or_never()
5953 .unwrap()
5954 .unwrap()
5955 .unwrap()
5956 .sequence,
5957 0
5958 );
5959
5960 edge.write_frame(Timestamp::from_millis(5000).unwrap(), bytes::Bytes::from_static(b"b"))
5963 .unwrap();
5964 assert_eq!(drain(&mut subscriber), vec![2]);
5965 }
5966
5967 #[test]
5968 fn a_budget_admits_groups_within_it() {
5969 let mut producer = track_producer("test", None);
5970 for second in 0..5 {
5971 append_at(&mut producer, second * 1000);
5972 }
5973
5974 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(2)));
5979 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
5980 }
5981
5982 #[test]
5983 fn a_budget_reaches_back_over_the_cache() {
5984 let mut producer = track_producer("test", None);
5985 for second in 0..5 {
5986 append_at(&mut producer, second * 1000);
5987 }
5988
5989 let mut live = producer.subscribe(None);
5992 assert_eq!(drain(&mut live), vec![4]);
5993
5994 let budget = Subscription::default().with_max_age(Duration::from_secs(2));
5999 let mut subscriber = producer.subscribe(budget);
6000 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
6001 }
6002
6003 #[test]
6004 fn a_named_start_is_a_floor_not_a_request() {
6005 let mut producer = track_producer("test", None);
6006 for second in 0..5 {
6007 append_at(&mut producer, second * 1000);
6008 }
6009
6010 let named = Subscription::default().with_start(Position::group(1));
6014 let mut subscriber = producer.subscribe(named);
6015 assert_eq!(drain(&mut subscriber), vec![4]);
6016
6017 let floored = Subscription::default()
6019 .with_start(Position::group(3))
6020 .with_max_age(Duration::from_secs(10));
6021 let mut subscriber = producer.subscribe(floored);
6022 assert_eq!(drain(&mut subscriber), vec![3, 4]);
6023
6024 let slack = Subscription::default()
6026 .with_start(Position::group(1))
6027 .with_max_age(Duration::from_secs(2));
6028 let mut subscriber = producer.subscribe(slack);
6029 assert_eq!(drain(&mut subscriber), vec![2, 3, 4]);
6030 }
6031
6032 #[test]
6033 fn a_floor_above_the_live_edge_waits_there() {
6034 let mut producer = track_producer("test", None);
6035 for second in 0..3 {
6036 append_at(&mut producer, second * 1000);
6037 }
6038
6039 let resumed = Subscription::default()
6042 .with_start(Position::group(7))
6043 .with_max_age(Duration::from_secs(10));
6044 let mut subscriber = producer.subscribe(resumed);
6045 assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
6046 append_at(&mut producer, 3000); assert_eq!(drain(&mut subscriber), Vec::<u64>::new());
6048 for second in 4..8 {
6049 append_at(&mut producer, second * 1000);
6050 }
6051 assert_eq!(drain(&mut subscriber), vec![7]);
6052 }
6053
6054 #[test]
6055 fn a_late_lower_group_within_the_budget_is_delivered() {
6056 let producer = track_producer("test", None);
6057 for (sequence, millis) in [(5, 0), (6, 1000), (7, 2000)] {
6058 let mut group = producer.create_group(group::Info { sequence }).unwrap();
6059 group
6060 .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x"))
6061 .unwrap();
6062 group.finish().unwrap();
6063 }
6064
6065 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5)));
6066 assert_eq!(drain(&mut subscriber), vec![5, 6, 7]);
6067
6068 let mut late = producer.create_group(group::Info { sequence: 4 }).unwrap();
6072 late.write_frame(Timestamp::from_millis(500).unwrap(), bytes::Bytes::from_static(b"late"))
6073 .unwrap();
6074 late.finish().unwrap();
6075 assert_eq!(drain(&mut subscriber), vec![4]);
6076 }
6077
6078 #[test]
6079 fn drift_is_measured_in_presentation_time_not_arrival_time() {
6080 let mut producer = track_producer("test", None);
6081 for second in 0..4 {
6085 append_at(&mut producer, second * 1000);
6086 }
6087
6088 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)));
6089 assert_eq!(drain(&mut subscriber), vec![1, 2, 3]);
6090 }
6091
6092 #[tokio::test]
6093 async fn a_stamped_successor_expires_an_unstamped_group() {
6094 let mut producer = track_producer("test", None);
6095 let mut subscriber = producer.subscribe(None);
6096 producer.append_group().unwrap(); append_at(&mut producer, 1000); assert_eq!(drain(&mut subscriber), vec![1]);
6103 }
6104
6105 #[tokio::test]
6106 async fn a_handed_out_group_expires_while_its_first_frame_is_stalled() {
6107 let mut producer = track_producer("test", None);
6108 let mut subscriber = producer.subscribe(None);
6109 producer.append_group().unwrap();
6110
6111 let mut stalled = subscriber.recv_group().await.unwrap().expect("stalled group");
6112 let pending = tokio::spawn(async move { stalled.read_frame().await });
6113 tokio::task::yield_now().await;
6114 assert!(
6115 !pending.is_finished(),
6116 "the empty live edge still waits for its first frame"
6117 );
6118
6119 crate::model::clock::advance(Duration::from_secs(1));
6120 append_at(&mut producer, 1000);
6121
6122 let result = pending.await.unwrap();
6127 assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
6128 }
6129
6130 #[tokio::test]
6133 async fn aborted_stamped_successor_wakes_a_parked_read() {
6134 tokio::time::pause();
6135 let mut producer = track_producer("test", None);
6136 let mut head = producer.append_group().unwrap();
6137 head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap();
6138 let mut successor = producer.append_group().unwrap();
6139 successor
6140 .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice())
6141 .unwrap();
6142 append_at(&mut producer, 1000);
6143 append_at(&mut producer, 20_000);
6144 let mut sub = producer.subscribe(None);
6145 let mut reading = sub.recv_group().await.unwrap().unwrap();
6146 assert_eq!(reading.sequence, 0, "the rewound successor extends the reach");
6147 assert!(reading.read_frame().await.unwrap().is_some());
6148
6149 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6150 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6151 let mut cx = std::task::Context::from_waker(&waker);
6152 let mut next = std::pin::pin!(reading.read_frame());
6153 assert!(next.as_mut().poll(&mut cx).is_pending());
6154 successor.abort(Error::Cancel).unwrap();
6155 assert!(
6156 woken.load(Ordering::SeqCst),
6157 "the stamped successor's abort lost its wakeup"
6158 );
6159 let result = next.as_mut().poll(&mut cx);
6160 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6161 }
6162
6163 #[tokio::test]
6168 async fn a_handed_out_group_wakes_when_a_newer_group_gets_its_first_timestamp() {
6169 let mut producer = track_producer("test", None);
6170 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6171 let mut old = producer.append_group().unwrap();
6172 old.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"old"))
6173 .unwrap();
6174
6175 append_at(&mut producer, 1000);
6177
6178 let mut held = subscriber.recv_group().await.unwrap().expect("old group");
6179 assert!(held.read_frame().await.unwrap().is_some());
6180
6181 let mut live = producer.append_group().unwrap();
6182 let pending = tokio::spawn(async move { held.read_frame().await });
6183 tokio::task::yield_now().await;
6184 assert!(!pending.is_finished(), "the newer group has no timestamp yet");
6185
6186 live.write_frame(
6188 Timestamp::from_millis(2000).unwrap(),
6189 bytes::Bytes::from_static(b"live"),
6190 )
6191 .unwrap();
6192 tokio::task::yield_now().await;
6193
6194 assert!(pending.is_finished(), "the new presentation edge wakes the held reader");
6195 let result = pending.await.unwrap();
6197 assert!(matches!(result, Ok(None)), "the held group ends: {result:?}");
6198 }
6199
6200 #[test]
6205 fn a_parked_read_ignores_first_frames_between_its_successor_and_the_edge() {
6206 let mut producer = track_producer("test", None);
6207 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
6208 let mut head = producer.append_group().unwrap();
6209 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6210 .unwrap();
6211 append_at(&mut producer, 1000); let mut between = producer.append_group().unwrap();
6213 append_at(&mut producer, 3000); let mut held = subscriber
6216 .recv_group()
6217 .now_or_never()
6218 .unwrap()
6219 .unwrap()
6220 .expect("head group");
6221 assert_eq!(held.sequence, 0);
6222 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6223
6224 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6225 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6226 let mut cx = std::task::Context::from_waker(&waker);
6227 let mut next = std::pin::pin!(held.read_frame());
6228 assert!(next.as_mut().poll(&mut cx).is_pending());
6229
6230 between
6231 .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x"))
6232 .unwrap();
6233 assert!(
6234 !woken.load(Ordering::SeqCst),
6235 "a first frame below the edge moves neither the reach nor the edge"
6236 );
6237
6238 let mut beyond = producer.append_group().unwrap();
6240 assert!(next.as_mut().poll(&mut cx).is_pending());
6241 woken.store(false, Ordering::SeqCst);
6242 beyond
6243 .write_frame(Timestamp::from_millis(4000).unwrap(), bytes::Bytes::from_static(b"x"))
6244 .unwrap();
6245 assert!(woken.load(Ordering::SeqCst), "a new edge wakes the parked read");
6246 }
6247
6248 #[test]
6251 fn a_parked_read_watches_the_replacement_for_an_aborted_successor() {
6252 let mut producer = track_producer("test", None);
6253 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6254 let mut head = producer.append_group().unwrap();
6255 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6256 .unwrap();
6257 let successor = producer.append_group().unwrap();
6258 let mut replacement = producer.append_group().unwrap();
6259 append_at(&mut producer, 20_000); let mut held = subscriber
6262 .recv_group()
6263 .now_or_never()
6264 .unwrap()
6265 .unwrap()
6266 .expect("head group");
6267 assert_eq!(held.sequence, 0, "an unstamped successor leaves the reach unbounded");
6268 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6269
6270 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6271 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6272 let mut cx = std::task::Context::from_waker(&waker);
6273 let mut next = std::pin::pin!(held.read_frame());
6274 assert!(next.as_mut().poll(&mut cx).is_pending());
6275
6276 successor.abort(Error::Cancel).unwrap();
6277 assert!(
6278 woken.load(Ordering::SeqCst),
6279 "the successor's abort wakes the parked read"
6280 );
6281 woken.store(false, Ordering::SeqCst);
6282 assert!(
6283 next.as_mut().poll(&mut cx).is_pending(),
6284 "the unstamped replacement leaves the reach unbounded"
6285 );
6286
6287 replacement
6288 .write_frame(Timestamp::from_millis(1000).unwrap(), bytes::Bytes::from_static(b"x"))
6289 .unwrap();
6290 assert!(
6291 woken.load(Ordering::SeqCst),
6292 "the replacement's first frame bounds the reach"
6293 );
6294 let result = next.as_mut().poll(&mut cx);
6295 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6296 }
6297
6298 #[test]
6302 fn a_parked_read_watches_its_edge_abort() {
6303 let mut producer = track_producer("test", None);
6304 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(5)));
6305 let mut head = producer.append_group().unwrap();
6306 head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
6307 .unwrap();
6308 append_at(&mut producer, 1000); let mut between = producer.append_group().unwrap();
6310 let mut edge = producer.append_group().unwrap();
6311 edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x"))
6312 .unwrap();
6313 edge.finish().unwrap();
6314
6315 let mut held = subscriber
6316 .recv_group()
6317 .now_or_never()
6318 .unwrap()
6319 .unwrap()
6320 .expect("head group");
6321 assert_eq!(held.sequence, 0);
6322 assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some());
6323
6324 let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
6325 let waker = futures::task::waker(Arc::new(FlagWake(woken.clone())));
6326 let mut cx = std::task::Context::from_waker(&waker);
6327 let mut next = std::pin::pin!(held.read_frame());
6328 assert!(next.as_mut().poll(&mut cx).is_pending());
6329
6330 edge.abort(Error::Cancel).unwrap();
6331 between
6332 .write_frame(Timestamp::from_millis(10_000).unwrap(), bytes::Bytes::from_static(b"x"))
6333 .unwrap();
6334 assert!(
6335 woken.load(Ordering::SeqCst),
6336 "the edge's abort, then a new edge below it, lost the wakeup"
6337 );
6338 let result = next.as_mut().poll(&mut cx);
6339 assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}");
6340 }
6341
6342 #[tokio::test]
6353 async fn real_time_reads_a_live_stream_without_truncating_it() {
6354 let producer = track_producer("test", None);
6355 let mut subscriber = producer.subscribe(None);
6356
6357 let gop = |n: u64| {
6358 [
6359 Timestamp::from_millis(n * 2000).unwrap(),
6360 Timestamp::from_millis(n * 2000 + 1900).unwrap(),
6361 ]
6362 };
6363 let write = |group: &mut group::Producer, timestamp| {
6364 group.write_frame(timestamp, bytes::Bytes::from_static(b"x")).unwrap();
6365 };
6366
6367 let mut open = producer.append_group().unwrap();
6368 write(&mut open, gop(0)[0]);
6369 write(&mut open, gop(0)[1]);
6370 let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");
6371
6372 let mut read = Vec::new();
6373 for n in 1..5u64 {
6374 let sequence = reading.sequence;
6375 let mut frames = 0;
6376 while let Some(res) = reading.read_frame().now_or_never() {
6377 match res.expect("no truncation while draining") {
6378 Some(_) => frames += 1,
6379 None => panic!("group {sequence} ended early"),
6380 }
6381 }
6382 read.push((sequence, frames));
6383
6384 let next = {
6385 let mut end = std::pin::pin!(reading.read_frame());
6388 assert!(futures::poll!(end.as_mut()).is_pending(), "parked on the FIN");
6389
6390 let mut opened = producer.append_group().unwrap();
6393 write(&mut opened, gop(n)[0]);
6394 let verdict = futures::poll!(end.as_mut());
6395
6396 open.finish().unwrap();
6397 let res = match verdict {
6398 Poll::Ready(res) => res,
6399 Poll::Pending => end.await,
6400 };
6401 assert!(
6402 matches!(res, Ok(None)),
6403 "group {sequence} ends at the boundary rather than failing: {res:?}"
6404 );
6405 opened
6406 };
6407 let mut next = next;
6408 write(&mut next, gop(n)[1]);
6409
6410 reading = subscriber.recv_group().await.unwrap().expect("the next live group");
6411 open = next;
6412 }
6413
6414 assert_eq!(read, vec![(0, 2), (1, 2), (2, 2), (3, 2)], "every frame of every group");
6415 }
6416
6417 #[tokio::test]
6427 async fn a_budget_is_measured_from_the_readers_position() {
6428 let producer = track_producer("test", None);
6429 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
6430
6431 let mut open = producer.append_group().unwrap();
6432 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"key"))
6433 .unwrap();
6434 open.write_frame(
6435 Timestamp::from_millis(1900).unwrap(),
6436 bytes::Bytes::from_static(b"tail"),
6437 )
6438 .unwrap();
6439
6440 let mut reading = subscriber.recv_group().await.unwrap().expect("the live group");
6441 assert!(reading.read_frame().await.unwrap().is_some());
6442 assert!(reading.read_frame().await.unwrap().is_some());
6443
6444 let mut end = std::pin::pin!(reading.read_frame());
6445 assert!(futures::poll!(end.as_mut()).is_pending(), "parked at 1900ms");
6446
6447 let mut next = producer.append_group().unwrap();
6449 next.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"key"))
6450 .unwrap();
6451 assert!(
6452 futures::poll!(end.as_mut()).is_pending(),
6453 "a reader inside its budget is not expired by the next group opening"
6454 );
6455
6456 open.write_frame(
6458 Timestamp::from_millis(1950).unwrap(),
6459 bytes::Bytes::from_static(b"late"),
6460 )
6461 .unwrap();
6462 let late = end.await.expect("the straggler is not truncated");
6463 assert_eq!(
6464 late.map(|frame| frame.timestamp),
6465 Some(Timestamp::from_millis(1950).unwrap())
6466 );
6467 }
6468
6469 #[tokio::test]
6473 async fn an_ended_group_stays_ended_when_probed_again() {
6474 let mut producer = track_producer("test", None);
6475 let mut subscriber = producer.subscribe(None);
6476 let mut open = producer.append_group().unwrap();
6477 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
6478 .unwrap();
6479
6480 let mut group = subscriber.recv_group().await.unwrap().expect("group");
6481 assert!(group.read_frame().await.unwrap().is_some());
6482
6483 let probes = tokio::spawn(async move {
6484 let first = group.read_frame().await;
6485 let second = group.read_frame().await;
6486 let finished = group.finished().await;
6487 (first, second, finished)
6488 });
6489 tokio::task::yield_now().await;
6490
6491 append_at(&mut producer, 1000);
6492
6493 let (first, second, finished) = probes.await.unwrap();
6494 assert!(matches!(first, Ok(None)), "the group ends: {first:?}");
6495 assert!(matches!(second, Ok(None)), "and stays ended: {second:?}");
6496 assert!(matches!(finished, Ok(1)), "reporting what it delivered: {finished:?}");
6497 }
6498
6499 #[tokio::test]
6500 async fn a_drained_group_finishes_cleanly_after_the_live_edge_advances() {
6501 let mut producer = track_producer("test", None);
6502 let mut subscriber = producer.subscribe(None);
6503 append_at(&mut producer, 0);
6504
6505 let mut group = subscriber.recv_group().await.unwrap().expect("first group");
6506 assert!(group.read_frame().await.unwrap().is_some());
6507
6508 append_at(&mut producer, 1000);
6509
6510 assert!(group.read_frame().await.unwrap().is_none());
6511 assert!(!group.latency_expired());
6512 }
6513
6514 #[tokio::test]
6515 async fn a_handed_out_partial_frame_expires_while_its_payload_is_stalled() {
6516 let mut producer = track_producer("test", None);
6517 let mut subscriber = producer.subscribe(None);
6518 let mut source = producer.append_group().unwrap();
6519 let mut writing = source
6520 .create_frame(frame::Info {
6521 size: 6,
6522 timestamp: Timestamp::ZERO,
6523 })
6524 .unwrap();
6525 writing.write(bytes::Bytes::from_static(b"old")).unwrap();
6526
6527 let mut group = subscriber.recv_group().await.unwrap().expect("partial group");
6528 let mut frame = group.next_frame().await.unwrap().expect("partial frame");
6529 assert_eq!(
6530 frame.read_chunk().await.unwrap(),
6531 Some(bytes::Bytes::from_static(b"old"))
6532 );
6533 let pending = tokio::spawn(async move { frame.read_chunk().await });
6534 tokio::task::yield_now().await;
6535 assert!(!pending.is_finished(), "the partial payload is still stalled");
6536
6537 crate::model::clock::advance(Duration::from_secs(1));
6538 append_at(&mut producer, 1000);
6539
6540 let result = pending.await.unwrap();
6541 assert!(
6542 matches!(result, Err(Error::Old)),
6543 "the in-flight frame expires: {result:?}"
6544 );
6545 writing.abort(Error::Cancel).unwrap();
6546 }
6547
6548 #[test]
6549 fn max_age_bounds_the_budget() {
6550 let mut producer = track_producer("test", Info::default().with_max_age(Duration::from_millis(500)));
6554 append_at(&mut producer, 0);
6555 append_at(&mut producer, 1000);
6556 append_at(&mut producer, 2000);
6557
6558 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(10)));
6561 assert_eq!(drain(&mut subscriber), vec![1, 2]);
6562 }
6563
6564 #[test]
6565 fn an_explicit_start_gets_no_exemption_from_the_budget() {
6566 let mut producer = track_producer("test", None);
6567 for second in 0..4 {
6568 append_at(&mut producer, second * 1000);
6569 }
6570
6571 let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(0)));
6574 subscriber.start_at(0);
6575 assert_eq!(drain(&mut subscriber), vec![3]);
6576
6577 let mut patient = producer.subscribe(
6578 Subscription::default()
6579 .with_start(Position::group(0))
6580 .with_max_age(Duration::from_secs(10)),
6581 );
6582 patient.start_at(0);
6583 assert_eq!(drain(&mut patient), vec![0, 1, 2, 3]);
6584 }
6585
6586 #[test]
6590 fn a_long_group_is_not_stale_while_its_tail_reaches_the_edge() {
6591 let mut producer = track_producer("test", None);
6592
6593 let mut long = producer.append_group().unwrap();
6595 for ms in [0u64, 500, 1000, 1500, 2000] {
6596 long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
6597 .unwrap();
6598 }
6599 long.finish().unwrap();
6600 append_at(&mut producer, 2000);
6601
6602 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6604 assert_eq!(drain(&mut sub), vec![0, 1]);
6605 }
6606
6607 #[test]
6613 fn a_long_group_is_stale_once_its_successor_falls_behind() {
6614 let mut producer = track_producer("test", None);
6615
6616 let mut long = producer.append_group().unwrap();
6617 for ms in [0u64, 500, 1000] {
6618 long.write_frame(Timestamp::from_millis(ms).unwrap(), bytes::Bytes::from_static(b"x"))
6619 .unwrap();
6620 }
6621 long.finish().unwrap();
6622 append_at(&mut producer, 3000);
6625 append_at(&mut producer, 4000);
6626
6627 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6628 assert_eq!(drain(&mut sub), vec![1, 2]);
6629 }
6630
6631 #[tokio::test]
6635 async fn an_unstamped_immediate_successor_leaves_reach_unbounded() {
6636 let mut producer = track_producer("test", None);
6637 let mut subscriber = producer.subscribe(None);
6638
6639 append_at(&mut producer, 0); producer.append_group().unwrap(); append_at(&mut producer, 10_000); assert_eq!(drain(&mut subscriber), vec![0, 2]);
6647 }
6648
6649 #[test]
6655 fn reach_follows_the_immediate_successor_not_a_later_rewind() {
6656 let mut producer = track_producer("test", None);
6657
6658 append_at(&mut producer, 0);
6660 append_at(&mut producer, 10_000);
6661 append_at(&mut producer, 1_000);
6662 append_at(&mut producer, 2_000);
6663
6664 let mut sub = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(500)));
6668 assert!(
6669 drain(&mut sub).contains(&0),
6670 "group 0 is bounded by its successor at 10s, not by a later rewind"
6671 );
6672 }
6673
6674 #[tokio::test]
6677 async fn ordered_carries_datagrams() {
6678 let mut producer = track_producer("test", None);
6679 let mut sub = producer.subscribe(None).ordered();
6680
6681 producer
6682 .insert_datagram(5, Timestamp::from_millis(5).unwrap(), bytes::Bytes::from_static(b"x"))
6683 .unwrap();
6684 producer.create_group(group::Info { sequence: 3 }).unwrap();
6685
6686 let datagram = sub
6687 .recv_datagram()
6688 .now_or_never()
6689 .expect("datagram would have blocked")
6690 .expect("would have errored")
6691 .expect("track was closed");
6692 assert_eq!(datagram.sequence, 5);
6693
6694 let group = sub
6696 .next_group()
6697 .now_or_never()
6698 .expect("group would have blocked")
6699 .expect("would have errored")
6700 .expect("track was closed");
6701 assert_eq!(group.sequence, 3);
6702 }
6703
6704 fn drain_ordered(subscriber: &mut Ordered) -> Vec<u64> {
6706 let mut sequences = Vec::new();
6707 while let Some(Ok(Some(group))) = subscriber.next_group().now_or_never() {
6708 sequences.push(group.sequence);
6709 }
6710 sequences
6711 }
6712
6713 #[test]
6717 fn next_group_sheds_a_stale_backlog() {
6718 let mut producer = track_producer("test", None);
6719 for second in 0..4 {
6720 append_at(&mut producer, second * 1000);
6721 }
6722
6723 let mut subscriber = producer.subscribe(None).ordered();
6724 assert_eq!(drain_ordered(&mut subscriber), vec![3]);
6725
6726 let mut arrival = producer.subscribe(None);
6728 assert_eq!(drain(&mut arrival), vec![3]);
6729 }
6730
6731 #[test]
6735 fn next_group_keeps_a_backlog_inside_the_budget() {
6736 let mut producer = track_producer("test", None);
6737 for second in 0..4 {
6738 append_at(&mut producer, second * 1000);
6739 }
6740
6741 let mut subscriber = producer
6744 .subscribe(Subscription::default().with_max_age(Duration::from_millis(1500)))
6745 .ordered();
6746 assert_eq!(drain_ordered(&mut subscriber), vec![1, 2, 3]);
6747
6748 let mut replay = producer.subscribe(replay()).ordered();
6750 assert_eq!(drain_ordered(&mut replay), vec![0, 1, 2, 3]);
6751 }
6752
6753 #[test]
6756 fn next_group_keeps_a_group_with_no_proven_reach() {
6757 let mut producer = track_producer("test", None);
6758 append_at(&mut producer, 0); producer.append_group().unwrap(); append_at(&mut producer, 10_000); let mut subscriber = producer.subscribe(None).ordered();
6765 assert_eq!(drain_ordered(&mut subscriber), vec![0, 2]);
6766 }
6767
6768 #[tokio::test]
6769 async fn real_time_skips_older_sequences_with_equal_ages() {
6770 let mut producer = track_producer("test", None);
6771 append_at(&mut producer, 0);
6772 append_at(&mut producer, 0);
6773
6774 let mut subscriber = producer.subscribe(None);
6775 assert_eq!(drain(&mut subscriber), vec![1]);
6776 }
6777
6778 #[test]
6779 fn fetch_ignores_the_budget() {
6780 let mut producer = track_producer("test", None);
6781 for second in 0..4 {
6782 append_at(&mut producer, second * 1000);
6783 }
6784
6785 let consumer = producer.consume();
6788 let group = consumer.fetch_group(0, None).now_or_never().unwrap().unwrap();
6789 assert_eq!(group.sequence, 0);
6790 }
6791
6792 #[tokio::test]
6795 async fn fetched_group_is_not_a_live_drift_edge() {
6796 let mut producer = track_producer("test", None);
6797 let dynamic = producer.dynamic();
6798 let consumer = producer.consume();
6799 append_at(&mut producer, 0);
6800
6801 let pending = consumer.fetch_group(100, None);
6802 let req = dynamic
6803 .requested_group()
6804 .now_or_never()
6805 .expect("fetch request is ready")
6806 .unwrap();
6807 let mut fetched = req.accept(None).unwrap();
6808 fetched
6809 .write_frame(
6810 Timestamp::from_millis(100_000).unwrap(),
6811 bytes::Bytes::from_static(b"fetched"),
6812 )
6813 .unwrap();
6814 fetched.finish().unwrap();
6815 pending.await.unwrap();
6816
6817 let mut groups = producer.subscribe(None);
6818 assert_eq!(groups.assert_group().sequence, 0);
6819 groups.assert_no_group();
6820 }
6821
6822 #[test]
6827 fn an_evicted_live_edge_convicts_nothing() {
6828 let mut producer = track_producer("test", None);
6829 append_at(&mut producer, 0);
6830 let edge = append_at(&mut producer, 30_000);
6831
6832 let state = producer.state.read();
6833 let drift = Drift {
6834 budget: Duration::ZERO,
6835 edge: state.drift_edge(None, None, None),
6836 outer: None,
6837 successor: None,
6838 };
6839 assert!(
6840 state.is_stale(0, &drift.edge, drift.budget),
6841 "stale against a live edge"
6842 );
6843 drop(state);
6844
6845 let slot = producer.modify().unwrap().lookup.remove(&edge).unwrap();
6847 let _ = slot.group.abort(Error::Evicted);
6848
6849 let state = producer.state.read();
6850 assert!(
6851 !state.is_stale(0, &drift.edge, drift.budget),
6852 "a vanished edge is no reason to drop what is left"
6853 );
6854 }
6855
6856 #[tokio::test]
6860 async fn an_evicted_outer_edge_convicts_nothing() {
6861 let mut producer = track_producer("a", None);
6862 let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1)));
6863 let mut open = producer.append_group().unwrap();
6864 open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
6865 .unwrap();
6866 let mut group = subscriber.recv_group().await.unwrap().expect("group");
6867 assert!(group.read_frame().await.unwrap().is_some());
6868 append_at(&mut producer, 10);
6870
6871 let mut next = track_producer("b", None);
6872 append_at(&mut next, 30_000);
6873 let edge = append_at(&mut next, 30_010);
6874 subscriber.set_anchor(Anchor {
6875 cap: None,
6876 edge: next.consume().live_edge(None),
6877 successor: None,
6878 });
6879
6880 let mut control = group.clone();
6881 assert!(
6882 matches!(control.read_frame().now_or_never(), Some(Ok(None))),
6883 "the open group ends against the outer edge"
6884 );
6885
6886 let slot = next.modify().unwrap().lookup.remove(&edge).unwrap();
6888 let _ = slot.group.abort(Error::Evicted);
6889
6890 assert!(
6891 group.read_frame().now_or_never().is_none(),
6892 "a vanished outer edge is no reason to drop what is left"
6893 );
6894 assert!(!group.latency_expired());
6895 open.finish().unwrap();
6896 }
6897
6898 #[tokio::test]
6899 async fn a_lower_sequence_is_never_the_live_edge() {
6900 let producer = track_producer("test", None);
6901 let mut straggler = producer.create_group(0u64.into()).unwrap();
6906 straggler
6907 .write_frame(Timestamp::from_millis(60_000).unwrap(), bytes::Bytes::from_static(b"x"))
6908 .unwrap();
6909 straggler.finish().unwrap();
6910
6911 let mut rewound = producer.create_group(1u64.into()).unwrap();
6912 rewound
6913 .write_frame(Timestamp::from_millis(0).unwrap(), bytes::Bytes::from_static(b"x"))
6914 .unwrap();
6915 rewound.finish().unwrap();
6916
6917 let mut subscriber = producer.subscribe(replay());
6918 assert_eq!(drain(&mut subscriber), vec![0, 1]);
6919 }
6920
6921 #[test]
6922 fn a_requested_end_does_not_cap_the_live_edge() {
6923 let mut producer = track_producer("test", None);
6924 for second in 0..4 {
6925 append_at(&mut producer, second * 1000);
6926 }
6927
6928 let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(1)));
6933 assert_eq!(drain(&mut subscriber), vec![3]);
6934 }
6935
6936 #[test]
6937 fn a_capped_subscriber_measures_drift_against_its_cap() {
6938 let mut producer = track_producer("test", None);
6939 append_at(&mut producer, 0);
6940 append_at(&mut producer, 1000);
6941
6942 let mut subscriber = producer.subscribe(Subscription::default().with_end(Position::after_group(0)));
6947 subscriber.set_groups(..1);
6948 assert_eq!(drain(&mut subscriber), vec![0]);
6949
6950 subscriber.set_groups(..);
6953 assert_eq!(drain(&mut subscriber), vec![1]);
6954 }
6955
6956 #[test]
6957 fn subscriber_control_updates_while_read_future_is_pending() {
6958 let producer = track_producer("test", None);
6959 let mut subscriber = producer.subscribe(None);
6960 let control = subscriber.control();
6961
6962 let mut recv = Box::pin(subscriber.recv_group());
6963 assert!(recv.as_mut().now_or_never().is_none());
6964
6965 control.update(Subscription::default().with_priority(7)).unwrap();
6966
6967 let aggregate = producer.subscription().expect("expected an active subscription");
6968 assert_eq!(aggregate.priority, 7);
6969 }
6970
6971 #[test]
6972 fn dropped_subscriber_leaves_no_ghost_in_aggregate() {
6973 let mut producer = track_producer("test", None);
6978 let a = producer.subscribe(Subscription::default().with_priority(5));
6979
6980 let waiter = kio::Waiter::noop();
6982 assert!(
6983 matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(Some(_)))),
6984 "one live subscriber should aggregate to Some",
6985 );
6986
6987 drop(a);
6989
6990 assert!(
6992 matches!(producer.poll_subscription_changed(&waiter), Poll::Ready(Ok(None))),
6993 "a dropped subscriber must not linger in the aggregate",
6994 );
6995
6996 assert!(
6998 producer.subscription().is_none(),
6999 "snapshot must exclude a dropped subscriber",
7000 );
7001 }
7002
7003 #[test]
7004 fn dropped_subscriber_wakes_the_aggregate() {
7005 use std::sync::atomic::{AtomicBool, Ordering};
7012
7013 let mut producer = track_producer("test", None);
7014 let a = producer.subscribe(Subscription::default().with_priority(5));
7015
7016 let woken = Arc::new(AtomicBool::new(false));
7017 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
7018
7019 assert!(matches!(
7021 producer.poll_subscription_changed(&waiter),
7022 Poll::Ready(Ok(Some(_)))
7023 ));
7024 assert!(
7025 producer.poll_subscription_changed(&waiter).is_pending(),
7026 "the aggregate is unchanged, so this poll must park",
7027 );
7028 assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");
7029
7030 drop(a);
7031 assert!(
7032 woken.load(Ordering::SeqCst),
7033 "the last subscriber leaving must wake the aggregate watcher",
7034 );
7035 }
7036
7037 #[test]
7038 fn widest_subscriber_update_wakes_the_aggregate() {
7039 use std::sync::atomic::{AtomicBool, Ordering};
7045
7046 let mut producer = track_producer("test", None);
7047 let _narrow = producer.subscribe(Subscription::default().with_end(Position::after_group(3)));
7048 let mut wide = producer.subscribe(Subscription::default());
7049
7050 let woken = Arc::new(AtomicBool::new(false));
7051 let waiter = kio::Waiter::new(futures::task::waker(Arc::new(FlagWake(woken.clone()))));
7052
7053 assert!(matches!(
7054 producer.poll_subscription_changed(&waiter),
7055 Poll::Ready(Ok(Some(_)))
7056 ));
7057 assert!(producer.poll_subscription_changed(&waiter).is_pending());
7058 assert!(!woken.load(Ordering::SeqCst), "nothing happened yet");
7059
7060 wide.update(Subscription::default().with_end(Position::after_group(5)))
7061 .unwrap();
7062 assert!(
7063 woken.load(Ordering::SeqCst),
7064 "the widest subscriber changing must wake the aggregate watcher",
7065 );
7066 match producer.poll_subscription_changed(&waiter) {
7067 Poll::Ready(Ok(Some(sub))) => assert_eq!(sub.end, Position::after_group(5)),
7068 other => panic!("expected the narrowed aggregate, got {other:?}"),
7069 }
7070 }
7071
7072 struct FlagWake(Arc<std::sync::atomic::AtomicBool>);
7074
7075 impl futures::task::ArcWake for FlagWake {
7076 fn wake_by_ref(arc_self: &Arc<Self>) {
7077 arc_self.0.store(true, std::sync::atomic::Ordering::SeqCst);
7078 }
7079 }
7080
7081 #[tokio::test]
7082 async fn out_of_order_max_sequence_at_front() {
7083 let producer = track_producer("test", None);
7084
7085 producer.create_group(group::Info { sequence: 5 }).unwrap();
7087 producer.create_group(group::Info { sequence: 3 }).unwrap();
7088 producer.create_group(group::Info { sequence: 4 }).unwrap();
7089
7090 {
7092 let state = producer.state.read();
7093 assert_eq!(state.max_sequence, Some(5));
7094 }
7095
7096 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7098
7099 producer.append_group().unwrap(); {
7105 let state = producer.state.read();
7106 assert_eq!(live_groups(&state), 1);
7107 assert_eq!(first_live_sequence(&state), 6);
7108 assert!(!state.lookup.contains_key(&3));
7109 assert!(!state.lookup.contains_key(&4));
7110 assert!(!state.lookup.contains_key(&5));
7111 assert!(state.lookup.contains_key(&6));
7112 }
7113 }
7114
7115 #[tokio::test]
7116 async fn max_sequence_at_front_blocks_trim() {
7117 let producer = track_producer("test", None);
7118
7119 producer.create_group(group::Info { sequence: 5 }).unwrap();
7121
7122 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7123
7124 producer.create_group(group::Info { sequence: 3 }).unwrap();
7126
7127 {
7130 let state = producer.state.read();
7131 assert_eq!(live_groups(&state), 2);
7132 assert_eq!(state.offset, 0);
7133 }
7134
7135 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
7137
7138 producer.create_group(group::Info { sequence: 2 }).unwrap();
7140
7141 {
7146 let state = producer.state.read();
7147 assert_eq!(live_groups(&state), 2);
7148 assert_eq!(state.offset, 0);
7149 assert!(state.lookup.contains_key(&5));
7150 assert!(!state.lookup.contains_key(&3));
7151 assert!(state.lookup.contains_key(&2));
7152 }
7153
7154 let mut consumer = producer.subscribe(None);
7156 let group = consumer.assert_group();
7157 assert_eq!(group.sequence, 5);
7159 }
7160
7161 #[tokio::test]
7162 async fn abort_drops_open_groups() {
7163 let producer = track_producer("test", None);
7164 producer.append_group().unwrap();
7165 producer.append_group().unwrap();
7166
7167 let mut consumer = producer.subscribe(None);
7168 assert_eq!(live_groups(&producer.state.read()), 2);
7169
7170 producer.clone().abort(Error::Cancel).unwrap();
7171
7172 assert!(
7174 producer.state.read().lookup.is_empty(),
7175 "open groups are dropped on abort"
7176 );
7177 let result = consumer.recv_group().now_or_never().expect("should not block");
7178 assert!(matches!(result, Err(Error::Cancel)));
7179 }
7180
7181 #[tokio::test]
7182 async fn drop_unfinished_clears_cached_groups() {
7183 let producer = track_producer("test", None);
7184 let writer = producer.clone();
7185 writer.append_group().unwrap();
7186
7187 let mut consumer = producer.subscribe(None);
7189 assert_eq!(live_groups(&producer.state.read()), 1);
7190
7191 drop(writer);
7193 drop(producer);
7194
7195 let result = consumer.recv_group().now_or_never().expect("should not block");
7196 assert!(matches!(result, Err(Error::Dropped)));
7197 }
7198
7199 #[tokio::test]
7200 async fn drop_after_abort_does_not_warn() {
7201 let warns = count_drop_warnings("track::Producer dropped without finish", || {
7204 let producer = track_producer("test", None);
7205 let keep = producer.clone();
7206 let writer = producer.clone();
7207 let group = writer.append_group().unwrap();
7208 group.finish().unwrap();
7209 let _consumer = producer.subscribe(None);
7210 writer.abort(Error::Cancel).unwrap();
7211 drop(keep);
7212 });
7213 assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN");
7214 }
7215
7216 #[tokio::test]
7217 async fn drop_unfinished_warns() {
7218 let warns = count_drop_warnings("track::Producer dropped without finish", || {
7219 let producer = track_producer("test", None);
7220 let writer = producer.clone();
7221 writer.append_group().unwrap();
7222 let _consumer = producer.subscribe(None);
7223 drop(writer);
7224 drop(producer);
7225 });
7226 assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN");
7227 }
7228
7229 #[tokio::test]
7230 async fn drop_finished_keeps_cached_groups() {
7231 let producer = track_producer("test", None);
7232 producer.append_group().unwrap();
7233 producer.finish().unwrap();
7234
7235 let mut consumer = producer.subscribe(None);
7236 drop(producer);
7237
7238 assert_eq!(consumer.assert_group().sequence, 0);
7240 let done = consumer.recv_group().now_or_never().expect("should not block").unwrap();
7241 assert!(done.is_none(), "consumer should drain then see clean finish");
7242 }
7243
7244 #[tokio::test]
7248 async fn drop_short_of_the_boundary_ends_cleanly() {
7249 let mut producer = track_producer("test", None);
7250 producer.append_group().unwrap();
7251 producer.finish_at(3).unwrap();
7252
7253 let mut consumer = producer.subscribe(None);
7254 assert_eq!(consumer.assert_group().sequence, 0);
7255 assert!(
7256 consumer.recv_group().now_or_never().is_none(),
7257 "groups 1 and 2 are owed"
7258 );
7259
7260 drop(producer);
7261 let done = consumer.recv_group().now_or_never().expect("should not block").unwrap();
7262 assert!(
7263 done.is_none(),
7264 "the track ends at its boundary without the missing groups"
7265 );
7266 }
7267
7268 #[tokio::test]
7271 async fn ordered_ends_cleanly_when_sealed_short_of_the_boundary() {
7272 let mut producer = track_producer("test", None);
7273 producer.create_group(group::Info { sequence: 0 }).unwrap();
7274 producer.create_group(group::Info { sequence: 2 }).unwrap();
7275 producer.finish_at(4).unwrap();
7276
7277 let mut ordered = producer.subscribe(None).ordered();
7278 drop(producer);
7279
7280 let first = ordered.next_group().now_or_never().expect("group 0").unwrap().unwrap();
7281 assert_eq!(first.sequence, 0);
7282 let second = ordered.next_group().now_or_never().expect("group 2").unwrap().unwrap();
7283 assert_eq!(second.sequence, 2);
7284 let done = ordered.next_group().now_or_never().expect("end").unwrap();
7285 assert!(done.is_none(), "missing groups below the boundary are not an error");
7286 }
7287
7288 #[tokio::test]
7289 async fn cached_groups_preserve_arrival_order() {
7290 let producer = track_producer("test", None);
7291 producer.create_group(group::Info { sequence: 5 }).unwrap();
7292 producer.create_group(group::Info { sequence: 3 }).unwrap();
7293
7294 let groups = producer.consume().cached_groups();
7295 let sequences: Vec<u64> = groups.iter().map(|(group, _)| group.sequence).collect();
7296 assert_eq!(
7297 sequences,
7298 vec![5, 3],
7299 "warm snapshot must follow arrival, not sequence order"
7300 );
7301 }
7302
7303 #[test]
7304 fn append_finish_cannot_be_rewritten() {
7305 let producer = track_producer("test", None);
7306
7307 assert!(producer.finish().is_ok());
7309 assert!(producer.finish().is_err());
7310 assert!(producer.append_group().is_err());
7311 }
7312
7313 #[test]
7314 fn finish_after_groups() {
7315 let producer = track_producer("test", None);
7316
7317 producer.append_group().unwrap();
7318 assert!(producer.finish().is_ok());
7319 assert!(producer.finish().is_err());
7320 assert!(producer.append_group().is_err());
7321 }
7322
7323 #[test]
7324 fn finish_at_rejects_a_boundary_at_or_below_the_live_edge() {
7325 let mut producer = track_producer("test", None);
7326 producer.create_group(group::Info { sequence: 5 }).unwrap();
7327
7328 assert!(producer.finish_at(4).is_err());
7331 assert!(producer.finish_at(5).is_err());
7332 assert!(producer.finish_at(6).is_ok());
7333
7334 {
7335 let state = producer.state.read();
7336 assert_eq!(state.final_sequence, Some(6));
7337 }
7338
7339 assert!(producer.finish_at(6).is_err());
7341 assert!(producer.create_group(group::Info { sequence: 4 }).is_ok());
7342 assert!(producer.create_group(group::Info { sequence: 6 }).is_err());
7343 }
7344
7345 #[test]
7346 fn final_sequence_reports_the_declared_boundary() {
7347 let mut producer = track_producer("test", None);
7348 assert_eq!(producer.final_sequence(), None);
7349
7350 producer.create_group(group::Info { sequence: 5 }).unwrap();
7351 assert_eq!(producer.final_sequence(), None, "a group does not declare a boundary");
7352
7353 producer.finish_at(9).unwrap();
7354 assert_eq!(producer.final_sequence(), Some(9));
7355
7356 assert!(producer.finish().is_err());
7358 }
7359
7360 #[test]
7361 fn final_sequence_reports_the_live_edge_after_finish() {
7362 let producer = track_producer("test", None);
7363 producer.create_group(group::Info { sequence: 5 }).unwrap();
7364 producer.finish().unwrap();
7365 assert_eq!(producer.final_sequence(), Some(6));
7366 }
7367
7368 #[tokio::test]
7369 async fn finish_at_declares_a_future_boundary() {
7370 let mut producer = track_producer("test", None);
7371 producer.create_group(group::Info { sequence: 5 }).unwrap();
7372
7373 producer.finish_at(7).unwrap();
7375
7376 let mut consumer = producer.subscribe(None);
7377 assert_eq!(consumer.assert_group().sequence, 5);
7378
7379 let boundary = consumer
7382 .finished()
7383 .now_or_never()
7384 .expect("boundary is known immediately")
7385 .expect("would have errored");
7386 assert_eq!(boundary, 7);
7387 assert!(
7388 consumer.recv_group().now_or_never().is_none(),
7389 "should wait for the outstanding group"
7390 );
7391
7392 producer.create_group(group::Info { sequence: 6 }).unwrap();
7394 assert_eq!(consumer.assert_group().sequence, 6);
7395 let done = consumer
7396 .recv_group()
7397 .now_or_never()
7398 .expect("should not block")
7399 .expect("would have errored");
7400 assert!(done.is_none(), "track completes once the boundary is reached");
7401 }
7402
7403 #[tokio::test]
7406 async fn abort_before_the_end_settles_wins() {
7407 let mut producer = track_producer("test", None);
7408 let mut consumer = producer.subscribe(None);
7409
7410 let head = producer.create_group(group::Info { sequence: 0 }).unwrap();
7411 head.finish().unwrap();
7412 let _tail = producer.create_group(group::Info { sequence: 1 }).unwrap();
7413 producer.finish_at(2).unwrap();
7414 assert_eq!(consumer.assert_group().sequence, 0);
7415
7416 producer.abort(Error::Timeout).unwrap();
7417 let res = consumer.recv_group().now_or_never().expect("should not block");
7418 assert!(matches!(res, Err(Error::Timeout)));
7419 }
7420
7421 #[tokio::test]
7422 async fn local_close_keeps_finished_groups_without_declaring_an_end() {
7423 let producer = track_producer("local", None);
7424 let mut consumer = producer.consume().subscribe(None).await.unwrap();
7425 producer.append_group().unwrap().finish().unwrap();
7426 producer.clone().close().unwrap();
7427 assert!(producer.final_sequence().is_none());
7428 assert!(matches!(consumer.finished().await, Err(Error::Closed)));
7429 assert!(consumer.recv_group().await.unwrap().is_some());
7430 assert!(consumer.recv_group().await.unwrap().is_none());
7431 }
7432
7433 #[tokio::test]
7434 async fn local_close_cannot_mask_an_abort_before_declared_end_settles() {
7435 let mut producer = track_producer("local", None);
7436 let mut consumer = producer.consume().subscribe(None).await.unwrap();
7437 let _group = producer.append_group().unwrap();
7438 producer.finish_at(2).unwrap();
7439 producer.clone().abort(Error::Timeout).unwrap();
7440 assert!(producer.close().is_err());
7441 assert!(matches!(consumer.recv_group().await, Err(Error::Timeout)));
7442 }
7443
7444 #[test]
7447 fn local_close_is_not_completion_and_keeps_the_resume_position() {
7448 let producer = track_producer("local", None);
7449 let consumer = producer.consume();
7450 producer.append_group().unwrap().finish().unwrap();
7451 let mut group = producer.append_group().unwrap();
7452 group
7453 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a"))
7454 .unwrap();
7455 group
7456 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"b"))
7457 .unwrap();
7458 producer.close().unwrap();
7459 assert!(matches!(
7460 consumer.poll_complete(&kio::Waiter::noop()),
7461 Poll::Ready(Err(_))
7462 ));
7463 assert_eq!(consumer.resume_position(), Some(Position { group: 1, frame: 2 }));
7464 }
7465
7466 #[tokio::test]
7470 async fn abort_keeps_finished_groups_for_a_slow_reader() {
7471 let producer = track_producer("test", None);
7472 let mut arrival = producer.subscribe(None);
7473 let mut ordered = producer.subscribe(None).ordered();
7474
7475 for sequence in 0..2 {
7476 producer
7477 .create_group(group::Info { sequence })
7478 .unwrap()
7479 .finish()
7480 .unwrap();
7481 }
7482 let _open = producer.create_group(group::Info { sequence: 2 }).unwrap();
7483 producer.abort(Error::Timeout).unwrap();
7484
7485 assert_eq!(arrival.assert_group().sequence, 0);
7486 assert_eq!(arrival.assert_group().sequence, 1);
7487 let res = arrival.recv_group().now_or_never().expect("should not block");
7488 assert!(matches!(res, Err(Error::Timeout)), "expected the abort");
7489
7490 assert_eq!(drain_ordered(&mut ordered), [0, 1]);
7491 let res = ordered
7492 .next_group()
7493 .now_or_never()
7494 .expect("should not block")
7495 .map(|group| group.map(|group| group.sequence));
7496 assert!(matches!(res, Err(Error::Timeout)), "expected the abort, got {res:?}");
7497 }
7498
7499 #[tokio::test]
7501 async fn dropped_producer_keeps_finished_groups_for_a_slow_reader() {
7502 let producer = track_producer("test", None);
7503 let mut consumer = producer.subscribe(None);
7504 producer
7505 .create_group(group::Info { sequence: 0 })
7506 .unwrap()
7507 .finish()
7508 .unwrap();
7509 drop(producer);
7510
7511 assert_eq!(consumer.assert_group().sequence, 0);
7512 let res = consumer.recv_group().now_or_never().expect("should not block");
7513 assert!(matches!(res, Err(Error::Dropped)), "expected the drop");
7514 }
7515
7516 #[tokio::test]
7519 async fn closed_track_expires_its_latest_group() {
7520 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
7521
7522 let live = track_producer_pooled("live", pool.clone());
7523 live.append_group().unwrap().finish().unwrap();
7524
7525 let aborted = track_producer_pooled("aborted", pool.clone());
7526 aborted.append_group().unwrap().finish().unwrap();
7527 let stale_aborted = aborted.consume();
7528 aborted.abort(Error::Timeout).unwrap();
7529
7530 let finished = track_producer_pooled("finished", pool.clone());
7531 finished.append_group().unwrap().finish().unwrap();
7532 finished.finish().unwrap();
7533 let stale_finished = finished.consume();
7534 drop(finished);
7535
7536 for _ in 0..2 {
7538 crate::model::clock::advance(Duration::from_secs(2));
7539 pool.sweep();
7540 }
7541
7542 assert!(live.consume().peek_group(0).is_some(), "a live track keeps its latest");
7543 assert!(
7544 stale_aborted.peek_group(0).is_none(),
7545 "an aborted track's latest expired"
7546 );
7547 assert!(
7548 stale_finished.peek_group(0).is_none(),
7549 "a sealed track's latest expired"
7550 );
7551 }
7552
7553 #[tokio::test]
7557 async fn closed_track_expires_a_shared_group_unless_live_edge() {
7558 let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1)));
7559 let sweep = || {
7561 for _ in 0..2 {
7562 crate::model::clock::advance(Duration::from_secs(2));
7563 pool.sweep();
7564 }
7565 };
7566 let adopt = |name: &str, group: &group::Producer| {
7567 let mut track = track_producer_pooled(name, pool.clone());
7568 track.adopt_group(group.clone(), true).unwrap();
7569 track.finish().unwrap();
7570 let stale = track.consume();
7571 drop(track);
7572 stale
7573 };
7574
7575 let live = track_producer_pooled("live", pool.clone());
7576 let group = live.append_group().unwrap();
7577 group.finish().unwrap();
7578 let stale = adopt("ended", &group);
7579 drop(group);
7580
7581 sweep();
7582 assert!(live.consume().peek_group(0).is_some(), "the live edge stays");
7583 assert!(
7584 stale.peek_group(0).is_some(),
7585 "and so does the ended track's share of it"
7586 );
7587
7588 live.append_group().unwrap().finish().unwrap();
7589 sweep();
7590 assert!(stale.peek_group(0).is_none(), "once demoted, it expires");
7591
7592 let group = live.append_group().unwrap();
7593 group.finish().unwrap();
7594 let (first, second) = (adopt("first", &group), adopt("second", &group));
7595 drop(group);
7596 live.finish().unwrap();
7597 drop(live);
7598
7599 sweep();
7600 assert!(first.peek_group(2).is_none(), "ended tracks alone don't keep it");
7601 assert!(second.peek_group(2).is_none(), "for each other");
7602 }
7603
7604 #[tokio::test]
7606 async fn abort_after_the_end_settles_ends_clean() {
7607 let mut producer = track_producer("test", None);
7608 let mut consumer = producer.subscribe(None);
7609
7610 for sequence in 0..2 {
7611 let group = producer.create_group(group::Info { sequence }).unwrap();
7612 group.finish().unwrap();
7613 }
7614 producer.finish_at(2).unwrap();
7615 assert_eq!(consumer.assert_group().sequence, 0);
7616 assert_eq!(consumer.assert_group().sequence, 1);
7617
7618 producer.abort(Error::Timeout).unwrap();
7619 let res = consumer.recv_group().now_or_never().expect("should not block");
7620 assert!(matches!(res, Ok(None)));
7621 }
7622
7623 #[tokio::test]
7626 async fn abort_after_the_end_settles_ends_clean_for_an_ordered_reader() {
7627 let mut producer = track_producer("test", None);
7628 let mut consumer = producer.subscribe(None).ordered();
7629
7630 for sequence in 0..2 {
7631 let group = producer.create_group(group::Info { sequence }).unwrap();
7632 group.finish().unwrap();
7633 }
7634 producer.finish_at(2).unwrap();
7635 producer.abort(Error::Timeout).unwrap();
7636
7637 assert_eq!(drain_ordered(&mut consumer), [0, 1]);
7638 let end = consumer
7639 .next_group()
7640 .now_or_never()
7641 .expect("should not block")
7642 .map(|group| group.map(|group| group.sequence));
7643 assert!(matches!(end, Ok(None)), "expected the clean end, got {end:?}");
7644 }
7645
7646 #[tokio::test]
7650 async fn abort_after_the_end_settles_ends_clean_below_the_cap() {
7651 let mut producer = track_producer("test", None);
7652 let mut consumer = producer.subscribe(None).ordered();
7653 for sequence in 0..2 {
7654 producer
7655 .create_group(group::Info { sequence })
7656 .unwrap()
7657 .finish()
7658 .unwrap();
7659 }
7660 producer.finish_at(2).unwrap();
7661 producer.abort(Error::Timeout).unwrap();
7662
7663 consumer.set_groups(..1);
7664 assert_eq!(drain_ordered(&mut consumer), [0]);
7665 let end = consumer
7666 .next_group()
7667 .now_or_never()
7668 .expect("should not block")
7669 .map(|group| group.map(|group| group.sequence));
7670 assert!(
7671 matches!(end, Ok(None)),
7672 "capped at the end: expected the clean end, got {end:?}"
7673 );
7674
7675 let mut producer = track_producer("test", None);
7676 let mut consumer = producer.subscribe(None).ordered();
7677 for sequence in [0, 2] {
7678 producer
7679 .create_group(group::Info { sequence })
7680 .unwrap()
7681 .finish()
7682 .unwrap();
7683 }
7684 producer.finish_at(3).unwrap();
7685 producer.abort(Error::Timeout).unwrap();
7686
7687 consumer.set_groups(..2);
7688 assert_eq!(drain_ordered(&mut consumer), [0]);
7689 let end = consumer
7690 .next_group()
7691 .now_or_never()
7692 .expect("should not block")
7693 .map(|group| group.map(|group| group.sequence));
7694 assert!(
7695 matches!(end, Ok(None)),
7696 "gap below the cap: expected the clean end, got {end:?}"
7697 );
7698 }
7699
7700 #[tokio::test]
7701 async fn recv_group_finishes_without_waiting_for_gaps() {
7702 let producer = track_producer("test", None);
7703 producer.create_group(group::Info { sequence: 1 }).unwrap();
7704 producer.finish().unwrap();
7705
7706 let mut consumer = producer.subscribe(None);
7707 assert_eq!(consumer.assert_group().sequence, 1);
7708
7709 let done = consumer
7710 .recv_group()
7711 .now_or_never()
7712 .expect("should not block")
7713 .expect("would have errored");
7714 assert!(done.is_none(), "track should finish without waiting for gaps");
7715 }
7716
7717 #[tokio::test]
7718 async fn next_group_skips_late_arrivals() {
7719 let producer = track_producer("test", None);
7720 let mut consumer = producer.subscribe(None).ordered();
7721
7722 producer.create_group(group::Info { sequence: 5 }).unwrap();
7724 let group = consumer
7725 .next_group()
7726 .now_or_never()
7727 .expect("should not block")
7728 .expect("would have errored")
7729 .expect("track should not be closed");
7730 assert_eq!(group.sequence, 5);
7731
7732 producer.create_group(group::Info { sequence: 3 }).unwrap();
7734 producer.create_group(group::Info { sequence: 4 }).unwrap();
7736 producer.create_group(group::Info { sequence: 7 }).unwrap();
7738
7739 let group = consumer
7740 .next_group()
7741 .now_or_never()
7742 .expect("should not block")
7743 .expect("would have errored")
7744 .expect("track should not be closed");
7745 assert_eq!(group.sequence, 7);
7746
7747 assert!(
7749 consumer.next_group().now_or_never().is_none(),
7750 "should block waiting for a higher sequence"
7751 );
7752 }
7753
7754 #[tokio::test]
7755 async fn next_group_returns_arrivals_in_order() {
7756 let producer = track_producer("test", None);
7757 let mut consumer = producer.subscribe(replay()).ordered();
7758
7759 producer.create_group(group::Info { sequence: 3 }).unwrap();
7761 producer.create_group(group::Info { sequence: 5 }).unwrap();
7762
7763 let group = consumer
7764 .next_group()
7765 .now_or_never()
7766 .expect("should not block")
7767 .expect("would have errored")
7768 .expect("track should not be closed");
7769 assert_eq!(group.sequence, 3);
7770
7771 let group = consumer
7772 .next_group()
7773 .now_or_never()
7774 .expect("should not block")
7775 .expect("would have errored")
7776 .expect("track should not be closed");
7777 assert_eq!(group.sequence, 5);
7778 }
7779
7780 #[tokio::test]
7781 async fn ordered_and_arrival_cursors_are_independent() {
7782 let producer = track_producer("test", None);
7783 let mut ordered = producer.subscribe(replay()).ordered();
7784 let mut arrival = producer.subscribe(replay());
7785
7786 producer.create_group(group::Info { sequence: 5 }).unwrap();
7788 producer.create_group(group::Info { sequence: 3 }).unwrap();
7789
7790 let group = ordered
7793 .next_group()
7794 .now_or_never()
7795 .expect("should not block")
7796 .expect("would have errored")
7797 .expect("track should not be closed");
7798 assert_eq!(group.sequence, 3);
7799
7800 assert_eq!(arrival.assert_group().sequence, 5);
7802 }
7803
7804 #[tokio::test]
7805 async fn end_at_caps_next_group() {
7806 let producer = track_producer("test", None);
7807 let mut consumer = producer.subscribe(replay()).ordered();
7808
7809 for s in 0..6 {
7810 producer.create_group(group::Info { sequence: s }).unwrap();
7811 }
7812
7813 consumer.set_groups(..3);
7814
7815 assert_eq!(
7817 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7818 0
7819 );
7820 assert_eq!(
7821 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7822 1
7823 );
7824 assert_eq!(
7825 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7826 2
7827 );
7828
7829 assert!(
7831 consumer.next_group().now_or_never().is_none(),
7832 "capped consumer must block instead of returning out-of-range groups"
7833 );
7834 }
7835
7836 #[tokio::test]
7837 async fn end_at_release_drains_cached_groups() {
7838 let producer = track_producer("test", None);
7839 let mut consumer = producer.subscribe(replay()).ordered();
7840
7841 for s in 0..6 {
7842 producer.create_group(group::Info { sequence: s }).unwrap();
7843 }
7844
7845 consumer.set_groups(..2);
7846 assert_eq!(
7847 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7848 0
7849 );
7850 assert_eq!(
7851 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7852 1
7853 );
7854 assert!(consumer.next_group().now_or_never().is_none(), "capped at 2");
7855
7856 consumer.set_groups(..5);
7858 assert_eq!(
7859 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7860 2
7861 );
7862 assert_eq!(
7863 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7864 3
7865 );
7866 assert_eq!(
7867 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7868 4
7869 );
7870 assert!(consumer.next_group().now_or_never().is_none(), "capped at 5");
7871
7872 consumer.set_groups(..);
7874 assert_eq!(
7875 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7876 5
7877 );
7878 assert!(consumer.next_group().now_or_never().is_none(), "no more groups");
7879 }
7880
7881 #[tokio::test]
7882 async fn end_at_lower_than_cursor_parks_consumer() {
7883 let producer = track_producer("test", None);
7884 let mut consumer = producer.subscribe(replay()).ordered();
7885
7886 for s in 0..3 {
7887 producer.create_group(group::Info { sequence: s }).unwrap();
7888 }
7889
7890 assert_eq!(
7892 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7893 0
7894 );
7895 assert_eq!(
7896 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7897 1
7898 );
7899 assert_eq!(
7900 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7901 2
7902 );
7903
7904 consumer.set_groups(..2);
7906 producer.create_group(group::Info { sequence: 3 }).unwrap();
7907 producer.create_group(group::Info { sequence: 4 }).unwrap();
7908 assert!(
7909 consumer.next_group().now_or_never().is_none(),
7910 "cap is below cursor; nothing returnable until cap rises"
7911 );
7912
7913 consumer.set_groups(..);
7915 assert_eq!(
7916 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7917 3
7918 );
7919 assert_eq!(
7920 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7921 4
7922 );
7923 }
7924
7925 #[tokio::test]
7926 async fn end_at_toggling_around_late_arrivals() {
7927 let producer = track_producer("test", None);
7928 let mut consumer = producer.subscribe(replay()).ordered();
7929
7930 consumer.set_groups(..6);
7931
7932 producer.create_group(group::Info { sequence: 2 }).unwrap();
7934 producer.create_group(group::Info { sequence: 5 }).unwrap();
7935 producer.create_group(group::Info { sequence: 3 }).unwrap();
7936 producer.create_group(group::Info { sequence: 8 }).unwrap();
7938 producer.create_group(group::Info { sequence: 4 }).unwrap();
7939
7940 assert_eq!(
7942 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7943 2
7944 );
7945 assert_eq!(
7946 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7947 3
7948 );
7949 assert_eq!(
7950 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7951 4
7952 );
7953 assert_eq!(
7954 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7955 5
7956 );
7957 assert!(consumer.next_group().now_or_never().is_none());
7959
7960 consumer.set_groups(..11);
7962 assert_eq!(
7963 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7964 8
7965 );
7966 }
7967
7968 #[tokio::test]
7972 async fn end_at_parks_recv_group() {
7973 let producer = track_producer("test", None);
7974 let mut consumer = producer.subscribe(replay());
7975
7976 for s in 0..3 {
7977 producer.create_group(group::Info { sequence: s }).unwrap();
7978 }
7979
7980 consumer.set_groups(..2);
7981 assert_eq!(
7982 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7983 0
7984 );
7985 assert_eq!(
7986 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
7987 1
7988 );
7989 assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");
7990
7991 producer.finish().unwrap();
7993 assert!(
7994 consumer.recv_group().now_or_never().is_none(),
7995 "still parked after finish"
7996 );
7997
7998 consumer.set_groups(..);
7999 assert_eq!(
8000 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8001 2
8002 );
8003 assert!(
8004 matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
8005 "finished once the parked group drains"
8006 );
8007 }
8008
8009 #[tokio::test]
8012 async fn recv_group_serves_arrivals_behind_the_cap() {
8013 let producer = track_producer("test", None);
8014 let mut consumer = producer.subscribe(replay());
8015
8016 consumer.set_groups(..2);
8017
8018 producer.create_group(group::Info { sequence: 2 }).unwrap();
8020 producer.create_group(group::Info { sequence: 0 }).unwrap();
8021 producer.create_group(group::Info { sequence: 1 }).unwrap();
8022
8023 assert_eq!(
8024 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8025 0
8026 );
8027 assert_eq!(
8028 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8029 1
8030 );
8031 assert!(consumer.recv_group().now_or_never().is_none(), "capped at 2");
8032
8033 consumer.set_groups(..3);
8034 assert_eq!(
8035 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8036 2
8037 );
8038 }
8039
8040 #[tokio::test]
8041 async fn group_ranges_preserve_the_floor_when_the_cap_changes() {
8042 let producer = track_producer("test", None);
8043 let mut consumer = producer.subscribe(None);
8044 consumer.set_groups(2..=2);
8045 producer.create_group(group::Info { sequence: 2 }).unwrap();
8046 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 2);
8047 consumer.set_groups(..4);
8048 producer.create_group(group::Info { sequence: 1 }).unwrap();
8049 producer.create_group(group::Info { sequence: 3 }).unwrap();
8050 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 3);
8051 consumer.set_groups(0..=4);
8052 producer.create_group(group::Info { sequence: 0 }).unwrap();
8053 producer.create_group(group::Info { sequence: 4 }).unwrap();
8054 assert_eq!(consumer.recv_group().await.unwrap().unwrap().sequence, 4);
8055 }
8056
8057 #[tokio::test]
8060 async fn start_at_drops_parked_recv_groups() {
8061 let producer = track_producer("test", None);
8062 let mut consumer = producer.subscribe(None);
8063
8064 consumer.set_groups(..1);
8065 producer.create_group(group::Info { sequence: 1 }).unwrap();
8066 assert!(
8067 consumer.recv_group().now_or_never().is_none(),
8068 "group 1 parked at the cap"
8069 );
8070
8071 consumer.start_at(2);
8072 consumer.set_groups(..);
8073 producer.create_group(group::Info { sequence: 2 }).unwrap();
8074 assert_eq!(
8075 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8076 2,
8077 "the overtaken parked group is dropped, not re-offered"
8078 );
8079 }
8080
8081 #[tokio::test]
8085 async fn evicted_parked_recv_groups_are_dropped() {
8086 let producer = track_producer("test", None);
8087 let mut consumer = producer.subscribe(None);
8088
8089 producer.create_group(group::Info { sequence: 0 }).unwrap();
8090 assert_eq!(
8091 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8092 0
8093 );
8094
8095 consumer.set_groups(..1);
8096 let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
8097 assert!(
8098 consumer.recv_group().now_or_never().is_none(),
8099 "group 1 parked at the cap"
8100 );
8101
8102 straggler.abort(Error::Old).unwrap();
8104 producer.finish().unwrap();
8105
8106 consumer.set_groups(..);
8107 assert!(
8108 matches!(consumer.recv_group().now_or_never(), Some(Ok(None))),
8109 "a dead parked group must not be delivered or hold the stream open"
8110 );
8111 }
8112
8113 #[tokio::test]
8117 async fn evicted_parked_group_wakes_the_clean_end() {
8118 use std::sync::atomic::{AtomicUsize, Ordering};
8119 use std::task::{Context, Wake};
8120
8121 struct CountWaker(AtomicUsize);
8124 impl Wake for CountWaker {
8125 fn wake(self: std::sync::Arc<Self>) {
8126 self.0.fetch_add(1, Ordering::SeqCst);
8127 }
8128 }
8129
8130 let producer = track_producer("test", None);
8131 let mut consumer = producer.subscribe(None);
8132
8133 producer.create_group(group::Info { sequence: 0 }).unwrap();
8134 assert_eq!(
8135 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8136 0
8137 );
8138
8139 consumer.set_groups(..1);
8140 let straggler = producer.create_group(group::Info { sequence: 1 }).unwrap();
8141 assert!(consumer.recv_group().now_or_never().is_none(), "parked at the cap");
8142 producer.finish().unwrap();
8143
8144 let counter = std::sync::Arc::new(CountWaker(AtomicUsize::new(0)));
8145 let waker = std::task::Waker::from(counter.clone());
8146 let mut cx = Context::from_waker(&waker);
8147 let mut fut = std::pin::pin!(consumer.recv_group());
8148 assert!(
8149 fut.as_mut().poll(&mut cx).is_pending(),
8150 "the parked group holds it open"
8151 );
8152
8153 straggler.abort(Error::Old).unwrap();
8154 assert!(counter.0.load(Ordering::SeqCst) > 0, "the eviction wakeup was lost");
8155 assert!(matches!(fut.as_mut().poll(&mut cx), Poll::Ready(Ok(None))));
8156 }
8157
8158 #[tokio::test]
8160 async fn end_at_zero_is_the_empty_range() {
8161 let producer = track_producer("test", None);
8162 let mut consumer = producer.subscribe(replay());
8163 producer.create_group(group::Info { sequence: 0 }).unwrap();
8164 producer.create_group(group::Info { sequence: 1 }).unwrap();
8165
8166 consumer.set_groups(..0);
8167 assert!(
8168 consumer.recv_group().now_or_never().is_none(),
8169 "empty cap delivers nothing"
8170 );
8171
8172 consumer.set_groups(..1);
8173 assert_eq!(
8174 consumer.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8175 0
8176 );
8177 assert!(consumer.recv_group().now_or_never().is_none(), "group 1 stays parked");
8178 }
8179
8180 #[tokio::test]
8183 async fn empty_local_cap_holds_while_another_subscriber_requests_everything() {
8184 let producer = track_producer("test", None);
8185 let mut everything = producer.subscribe(replay());
8186 let mut empty = producer.subscribe(Subscription::default().with_end(Position::group(0)));
8187 empty.set_groups((Bound::Unbounded, Position::group(0).group_end()));
8188
8189 for s in 0..3 {
8190 producer.create_group(group::Info { sequence: s }).unwrap();
8191 }
8192
8193 assert_eq!(
8194 everything
8195 .recv_group()
8196 .now_or_never()
8197 .unwrap()
8198 .unwrap()
8199 .unwrap()
8200 .sequence,
8201 0
8202 );
8203 assert!(
8204 empty.recv_group().now_or_never().is_none(),
8205 "local empty cap must not ride the unbounded aggregate"
8206 );
8207
8208 empty.set_groups(..2);
8209 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 0);
8210 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 1);
8211 assert!(empty.recv_group().now_or_never().is_none(), "still capped at 2");
8212
8213 empty.set_groups(..);
8214 assert_eq!(empty.recv_group().now_or_never().unwrap().unwrap().unwrap().sequence, 2);
8215 }
8216
8217 #[tokio::test]
8219 async fn end_at_frame_limited_last_group() {
8220 let producer = track_producer("test", None);
8221 let mut consumer = producer.subscribe(replay()).ordered();
8222 let end = Position::after(1, 1).unwrap();
8223 consumer.set_groups((Bound::Unbounded, end.group_end()));
8224
8225 for s in 0..3u64 {
8226 let mut group = producer.create_group(group::Info { sequence: s }).unwrap();
8227 for i in 0..3u8 {
8228 group.write_frame(Timestamp::ZERO, vec![i]).unwrap();
8229 }
8230 group.finish().unwrap();
8231 }
8232
8233 assert_eq!(
8234 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8235 0
8236 );
8237 let mut last = consumer.next_group().now_or_never().unwrap().unwrap().unwrap();
8238 assert_eq!(last.sequence, 1);
8239 last.set_frames(..end.frame);
8240 assert_eq!(
8241 last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
8242 0
8243 );
8244 assert_eq!(
8245 last.read_frame().now_or_never().unwrap().unwrap().unwrap().payload[0],
8246 1
8247 );
8248 assert!(
8249 last.read_frame().now_or_never().unwrap().unwrap().is_none(),
8250 "frame cap is exclusive"
8251 );
8252 assert!(
8253 consumer.next_group().now_or_never().is_none(),
8254 "group 2 is past the exclusive group cap"
8255 );
8256 }
8257
8258 #[tokio::test]
8260 async fn end_at_maximum_group_is_unbounded() {
8261 let producer = track_producer("test", None);
8262 let mut consumer = producer.subscribe(replay()).ordered();
8263 consumer.set_groups(..=u64::MAX);
8264
8265 producer.create_group(group::Info { sequence: 0 }).unwrap();
8266 producer.create_group(group::Info { sequence: u64::MAX }).unwrap();
8267
8268 assert_eq!(
8269 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8270 0
8271 );
8272 assert_eq!(
8273 consumer.next_group().now_or_never().unwrap().unwrap().unwrap().sequence,
8274 u64::MAX
8275 );
8276 }
8277
8278 #[test]
8279 fn write_frame_rejects_an_oversized_frame_before_appending_its_group() {
8280 let mut producer = track_producer("test", None);
8281 let frame = bytes::Bytes::from(vec![0; group::MAX_CACHE_BYTES as usize + 1]);
8282
8283 assert!(matches!(
8284 producer.write_frame(Timestamp::ZERO, frame),
8285 Err(Error::FrameTooLarge)
8286 ));
8287 assert_eq!(producer.latest(), None, "the rejected frame did not publish a group");
8288 }
8289
8290 #[test]
8291 fn append_group_returns_bounds_exceeded_on_sequence_overflow() {
8292 let producer = track_producer("test", None);
8293 {
8294 let mut state = producer.state.write().ok().unwrap();
8295 state.max_sequence = Some(u64::MAX);
8296 }
8297
8298 assert!(matches!(producer.append_group(), Err(Error::BoundsExceeded(_))));
8299 }
8300
8301 #[tokio::test]
8302 async fn fetch_cache_hit() {
8303 let producer = track_producer("test", None);
8304
8305 let mut group = producer.append_group().unwrap(); group
8308 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hello"))
8309 .unwrap();
8310 group.finish().unwrap();
8311
8312 let dynamic = producer.dynamic();
8315 let consumer = producer.consume();
8316 assert!(consumer.peek_group(0).is_some());
8317 let mut g = consumer.fetch_group(0, None).await.unwrap();
8318 assert_eq!(g.sequence, 0);
8319 assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hello");
8320
8321 assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
8323 }
8324
8325 #[tokio::test]
8326 async fn fetch_miss_signals_dynamic() {
8327 let producer = track_producer("test", None);
8328 let dynamic = producer.dynamic();
8329 let consumer = producer.consume();
8330
8331 assert!(consumer.peek_group(5).is_none());
8335 let pending = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
8336 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
8337
8338 let req = dynamic
8339 .requested_group()
8340 .now_or_never()
8341 .expect("should not block")
8342 .unwrap();
8343 assert_eq!(req.sequence(), 5);
8344 assert_eq!(req.priority(), 7);
8345
8346 let mut group = req.accept(None).unwrap();
8348 group
8349 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
8350 .unwrap();
8351 group.finish().unwrap();
8352
8353 let mut g = pending.await.unwrap();
8354 assert_eq!(g.sequence, 5);
8355 assert_eq!(&g.read_frame().await.unwrap().unwrap().payload[..], b"hi");
8356 }
8357
8358 #[tokio::test]
8359 async fn fetch_miss_rejects() {
8360 let producer = track_producer("test", None);
8361 let dynamic = producer.dynamic();
8362 let consumer = producer.consume();
8363
8364 let pending = consumer.fetch_group(5, None);
8365 let req = dynamic
8366 .requested_group()
8367 .now_or_never()
8368 .expect("should not block")
8369 .unwrap();
8370
8371 req.reject(Error::Cancel);
8372 assert!(matches!(pending.await, Err(Error::Cancel)));
8373 let fetch = producer.state.read().fetch.clone();
8374 assert!(fetch.read().is_empty());
8375 }
8376
8377 #[tokio::test]
8378 async fn fetch_miss_drop_rejects() {
8379 let producer = track_producer("test", None);
8380 let dynamic = producer.dynamic();
8381 let consumer = producer.consume();
8382
8383 let pending = consumer.fetch_group(5, None);
8384 let req = dynamic
8385 .requested_group()
8386 .now_or_never()
8387 .expect("should not block")
8388 .unwrap();
8389
8390 drop(req);
8391 assert!(matches!(pending.await, Err(Error::Dropped)));
8392 }
8393
8394 #[tokio::test]
8397 async fn fetch_request_unused_once_every_fetch_leaves() {
8398 let producer = track_producer("test", None);
8399 let dynamic = producer.dynamic();
8400 let consumer = producer.consume();
8401
8402 let first = consumer.fetch_group(5, None);
8403 let req = dynamic
8404 .requested_group()
8405 .now_or_never()
8406 .expect("should not block")
8407 .unwrap();
8408 let second = consumer.fetch_group(5, None);
8409
8410 drop(first);
8411 assert!(req.poll_unused(&kio::Waiter::noop()).is_pending());
8412 drop(second);
8413 assert!(req.poll_unused(&kio::Waiter::noop()).is_ready());
8414
8415 let retry = consumer.fetch_group(5, None);
8416 let fresh = dynamic
8417 .requested_group()
8418 .now_or_never()
8419 .expect("the retry queues a fresh request")
8420 .unwrap();
8421 drop(req);
8422 assert!(fresh.poll_unused(&kio::Waiter::noop()).is_pending());
8423 fresh.accept(None).unwrap().finish().unwrap();
8424 assert_eq!(retry.await.unwrap().sequence, 5);
8425 }
8426
8427 #[test]
8430 fn fetching_is_unwind_safe() {
8431 fn assert_unwind_safe<T: std::panic::UnwindSafe + std::panic::RefUnwindSafe>() {}
8432 assert_unwind_safe::<Fetching>();
8433 assert_unwind_safe::<group::Consumer>();
8434 }
8435
8436 #[tokio::test]
8439 async fn fetch_hit_keeps_the_group_wanted() {
8440 let producer = track_producer("test", None);
8441 let dynamic = producer.dynamic();
8442 let consumer = producer.consume();
8443
8444 let first = consumer.fetch_group(5, None);
8445 let req = dynamic
8446 .requested_group()
8447 .now_or_never()
8448 .expect("should not block")
8449 .unwrap();
8450 let mut group = req.accept(None).unwrap();
8451 group
8452 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
8453 .unwrap();
8454 drop(first.await.unwrap());
8455
8456 let hit = consumer.fetch_group(5, None);
8457 assert!(!group.abort_unused(Error::Cancel), "the pending hit wants the group");
8458 let mut fetched = hit.await.unwrap();
8459 assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head");
8460 }
8461
8462 #[tokio::test]
8466 async fn accepted_group_starts_at_the_request() {
8467 let producer = track_producer("test", None);
8468 let dynamic = producer.dynamic();
8469 let consumer = producer.consume();
8470
8471 let _tail = consumer.fetch_group(5, group::Fetch::default().with_frame_start(3));
8472 let req = dynamic
8473 .requested_group()
8474 .now_or_never()
8475 .expect("should not block")
8476 .unwrap();
8477 let _group = req.accept(None).unwrap();
8478
8479 let whole = consumer.fetch_group(5, None);
8480 let req = dynamic
8481 .requested_group()
8482 .now_or_never()
8483 .expect("the wider fetch misses and queues its own request")
8484 .unwrap();
8485 assert_eq!(req.frame_start(), 0);
8486 assert!(kio::Pollable::poll(&*whole, &kio::Waiter::noop()).is_pending());
8487 }
8488
8489 #[tokio::test]
8492 async fn fetch_misses_an_abandoned_group() {
8493 let producer = track_producer("test", None);
8494 let dynamic = producer.dynamic();
8495 let consumer = producer.consume();
8496
8497 let first = consumer.fetch_group(5, None);
8498 let req = dynamic
8499 .requested_group()
8500 .now_or_never()
8501 .expect("should not block")
8502 .unwrap();
8503 let group = req.accept(None).unwrap();
8504 drop(first.await.unwrap());
8505 assert!(group.abort_unused(Error::Cancel));
8506
8507 let retry = consumer.fetch_group(5, None);
8508 let req = dynamic
8509 .requested_group()
8510 .now_or_never()
8511 .expect("the retry queues a fresh request")
8512 .unwrap();
8513 req.accept(None).unwrap().finish().unwrap();
8514 assert!(retry.await.unwrap().read_frame().await.unwrap().is_none());
8515 }
8516
8517 #[tokio::test]
8518 async fn fetch_reject_does_not_poison_retry() {
8519 let producer = track_producer("test", None);
8520 let dynamic = producer.dynamic();
8521 let consumer = producer.consume();
8522
8523 let pending = consumer.fetch_group(5, None);
8524 let req = dynamic
8525 .requested_group()
8526 .now_or_never()
8527 .expect("should not block")
8528 .unwrap();
8529 req.reject(Error::Cancel);
8530 assert!(matches!(pending.await, Err(Error::Cancel)));
8531
8532 let retry = consumer.fetch_group(5, None);
8533 let req = dynamic
8534 .requested_group()
8535 .now_or_never()
8536 .expect("should not block")
8537 .unwrap();
8538 let mut group = req.accept(None).unwrap();
8539 group
8540 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"retry"))
8541 .unwrap();
8542 group.finish().unwrap();
8543
8544 let mut group = retry.await.unwrap();
8545 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"retry");
8546 }
8547
8548 #[tokio::test]
8552 async fn fetch_ignores_a_group_that_starts_too_late() {
8553 let producer = track_producer("test", None);
8554 let dynamic = producer.dynamic();
8555 let consumer = producer.consume();
8556
8557 let mut group = producer.create_group(group::Info { sequence: 0 }).unwrap();
8559 group.start_at(3).unwrap();
8560 group
8561 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"tail"))
8562 .unwrap();
8563 group.finish().unwrap();
8564
8565 let fetch = consumer.fetch_group(0, group::Fetch::default().with_frame_start(3));
8567 let cached = fetch.now_or_never().expect("covered by the cache").unwrap();
8568 assert_eq!(cached.index(), 3);
8569
8570 let mut fetch = std::pin::pin!(consumer.fetch_group(0, None));
8573 assert!(
8574 futures::poll!(fetch.as_mut()).is_pending(),
8575 "must not answer from the tail"
8576 );
8577
8578 let request = dynamic.requested_group().await.unwrap();
8579 assert_eq!((request.sequence(), request.frame_start()), (0, 0));
8580
8581 let mut whole = request.accept(None).unwrap();
8583 whole
8584 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
8585 .unwrap();
8586 whole.finish().unwrap();
8587
8588 let mut served = fetch.await.unwrap();
8589 assert_eq!(served.index(), 0);
8590 assert_eq!(
8591 served.read_frame().await.unwrap().unwrap().payload,
8592 bytes::Bytes::from_static(b"head")
8593 );
8594 }
8595
8596 #[tokio::test]
8599 async fn fetch_widens_or_fails_cleanly() {
8600 let producer = track_producer("test", None);
8601 let dynamic = producer.dynamic();
8602 let consumer = producer.consume();
8603
8604 let _narrow = consumer.fetch_group(0, group::Fetch::default().with_frame_start(5));
8605 let mut narrow = std::pin::pin!(_narrow);
8606 assert!(futures::poll!(narrow.as_mut()).is_pending());
8607
8608 let _wider = consumer.fetch_group(0, group::Fetch::default().with_frame_start(2));
8610 let mut wider = std::pin::pin!(_wider);
8611 assert!(futures::poll!(wider.as_mut()).is_pending());
8612
8613 let request = dynamic.requested_group().await.unwrap();
8614 assert_eq!(request.frame_start(), 2, "widened while queued");
8615
8616 let _widest = consumer.fetch_group(0, group::Fetch::default().with_frame_start(0));
8619 let mut widest = std::pin::pin!(_widest);
8620 assert!(futures::poll!(widest.as_mut()).is_pending());
8621 assert_eq!(request.frame_start(), 2, "the in-flight range is already on the wire");
8622
8623 let _joined = request.result.clone();
8626 let mut group = request.accept(None).unwrap();
8627 group.start_at(2).unwrap();
8630 group
8631 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"from2"))
8632 .unwrap();
8633
8634 assert_eq!(narrow.await.unwrap().index(), 5);
8638 assert_eq!(wider.await.unwrap().index(), 2);
8639 assert!(matches!(widest.await, Err(Error::NotFound)));
8640 group.finish().unwrap();
8641 }
8642
8643 #[tokio::test]
8644 async fn fetch_coalesces_concurrent() {
8645 let producer = track_producer("test", None);
8646 let dynamic = producer.dynamic();
8647 let consumer = producer.consume();
8648
8649 let first = consumer.fetch_group(5, group::Fetch::default().with_priority(1));
8652 let second = consumer.fetch_group(5, group::Fetch::default().with_priority(7));
8653 assert!(kio::Pollable::poll(&*first, &kio::Waiter::noop()).is_pending());
8654
8655 let req = dynamic
8656 .requested_group()
8657 .now_or_never()
8658 .expect("should not block")
8659 .unwrap();
8660 assert_eq!(req.sequence(), 5);
8661 assert_eq!(req.priority(), 7);
8662 assert!(
8663 dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending(),
8664 "the second fetch queued a duplicate request"
8665 );
8666
8667 let third = consumer.fetch_group(5, None);
8669
8670 let mut group = req.accept(None).unwrap();
8672 group
8673 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"hi"))
8674 .unwrap();
8675 group.finish().unwrap();
8676
8677 assert_eq!(first.await.unwrap().sequence, 5);
8678 assert_eq!(second.await.unwrap().sequence, 5);
8679 assert_eq!(third.await.unwrap().sequence, 5);
8680 }
8681
8682 #[tokio::test]
8683 async fn fetch_coalesced_reject_fails_all() {
8684 let producer = track_producer("test", None);
8685 let dynamic = producer.dynamic();
8686 let consumer = producer.consume();
8687
8688 let first = consumer.fetch_group(5, None);
8689 let second = consumer.fetch_group(5, None);
8690 let req = dynamic
8691 .requested_group()
8692 .now_or_never()
8693 .expect("should not block")
8694 .unwrap();
8695 req.reject(Error::Cancel);
8696
8697 assert!(matches!(first.await, Err(Error::Cancel)));
8698 assert!(matches!(second.await, Err(Error::Cancel)));
8699
8700 let retry = consumer.fetch_group(5, None);
8702 assert!(kio::Pollable::poll(&*retry, &kio::Waiter::noop()).is_pending());
8703 let req = dynamic
8704 .requested_group()
8705 .now_or_never()
8706 .expect("should not block")
8707 .unwrap();
8708 assert_eq!(req.sequence(), 5);
8709 }
8710
8711 #[tokio::test]
8712 async fn fetch_queued_fails_when_handlers_leave() {
8713 let producer = track_producer("test", None);
8714 let dynamic = producer.dynamic();
8715 let consumer = producer.consume();
8716
8717 let pending = consumer.fetch_group(5, None);
8719 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
8720 drop(dynamic);
8721 assert!(matches!(pending.await, Err(Error::NotFound)));
8722
8723 let fetch = producer.state.read().fetch.clone();
8725 assert!(fetch.read().is_empty());
8726 }
8727
8728 #[tokio::test]
8729 async fn fetch_miss_no_dynamic_not_found() {
8730 let producer = track_producer("test", None);
8733 producer.append_group().unwrap(); let consumer = producer.consume();
8735 assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));
8736 }
8737
8738 #[tokio::test]
8739 async fn fetch_past_final_not_found() {
8740 let producer = track_producer("test", None);
8741 producer.append_group().unwrap(); producer.finish().unwrap(); let dynamic = producer.dynamic();
8747 let consumer = producer.consume();
8748 assert!(matches!(consumer.fetch_group(5, None).await, Err(Error::NotFound)));
8749
8750 assert!(dynamic.poll_requested_group(&kio::Waiter::noop()).is_pending());
8752 }
8753
8754 fn pooled_producer(capacity: u64) -> (Producer, cache::Pool) {
8756 let config = cache::Config::default()
8757 .with_capacity(capacity)
8758 .with_expiry(cache::DEFAULT_EXPIRY);
8759 let pool = cache::Pool::new(config);
8760 let broadcast = broadcast::Info {
8761 pool: pool.clone(),
8762 ..Default::default()
8763 };
8764 let producer = Producer::new(Arc::new(broadcast), "test", None);
8765 (producer, pool)
8766 }
8767
8768 fn finished_group(producer: &mut Producer, size: usize) -> u64 {
8769 let mut group = producer.append_group().unwrap();
8770 group
8771 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; size]))
8772 .unwrap();
8773 group.finish().unwrap();
8774 group.sequence
8775 }
8776
8777 #[tokio::test]
8780 async fn debt_evicts_oldest_group() {
8781 let (mut producer, pool) = pooled_producer(10_000);
8783
8784 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let consumer = producer.consume();
8789 assert!(consumer.peek_group(0).is_none(), "oldest group is evicted");
8790 assert!(consumer.peek_group(2).is_some(), "latest group survives");
8791 assert!(
8794 pool.used() <= 2 * (10_000 + cache::ENTRY_OVERHEAD),
8795 "usage hovers near capacity: {}",
8796 pool.used()
8797 );
8798
8799 let mut subscriber = producer.subscribe(replay());
8801 assert!(subscriber.assert_group().sequence > 0, "evicted group is not delivered");
8802 }
8803
8804 #[tokio::test]
8806 async fn latest_group_never_evicted() {
8807 let (mut producer, pool) = pooled_producer(100);
8809 finished_group(&mut producer, 1000); assert!(pool.used() > 100, "the latest may exceed the budget");
8811
8812 finished_group(&mut producer, 1000); finished_group(&mut producer, 1000); let consumer = producer.consume();
8817 assert!(consumer.peek_group(0).is_none());
8818 let mut group = consumer.peek_group(2).expect("latest survives");
8819 assert_eq!(group.read_frame().await.unwrap().unwrap().payload.len(), 1000);
8820 }
8821
8822 #[tokio::test]
8826 async fn fetch_refresh_survives_eviction() {
8827 let (mut producer, _pool) = pooled_producer(10_000);
8828 let consumer = producer.consume();
8829
8830 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8832 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8834 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_millis(500));
8836
8837 let mut fetched = consumer.fetch_group(0, None).await.unwrap();
8839 assert_eq!(fetched.read_frame().await.unwrap().unwrap().payload.len(), 3_000);
8840 crate::model::clock::advance(Duration::from_millis(500));
8841
8842 finished_group(&mut producer, 3_000); crate::model::clock::advance(Duration::from_secs(1));
8846 finished_group(&mut producer, 3_000); assert!(consumer.peek_group(0).is_some(), "refreshed group survives");
8849 assert!(consumer.peek_group(1).is_none(), "unread group is evicted instead");
8850 }
8851
8852 #[tokio::test]
8855 async fn eviction_aborts_readers() {
8856 let (mut producer, _pool) = pooled_producer(10_000);
8857 let mut subscriber = producer.subscribe(None);
8858
8859 finished_group(&mut producer, 10_000); let mut group0 = subscriber.assert_group();
8861
8862 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let read = group0.read_frame().await;
8866 assert!(matches!(read, Err(Error::Evicted)), "expected Evicted, got {read:?}");
8867 }
8868
8869 #[tokio::test]
8873 async fn small_writes_carry_debt() {
8874 let unit = 100 * cache::ENTRY_OVERHEAD;
8877 let (mut producer, pool) = pooled_producer(22 * unit);
8878 let consumer = producer.consume();
8879
8880 finished_group(&mut producer, 20 * unit as usize); for _ in 0..3 {
8885 finished_group(&mut producer, unit as usize);
8886 }
8887 assert!(consumer.peek_group(0).is_some(), "debt smaller than the victim carries");
8888
8889 for _ in 0..20 {
8891 finished_group(&mut producer, unit as usize);
8892 }
8893 assert!(
8894 consumer.peek_group(0).is_none(),
8895 "accumulated debt evicts the large group"
8896 );
8897 assert!(pool.used() <= 24 * unit, "usage hovers near capacity: {}", pool.used());
8900 }
8901
8902 #[tokio::test]
8906 async fn payment_capped_per_write() {
8907 let (mut producer, pool) = pooled_producer(1 << 40);
8908 for _ in 0..10 {
8909 finished_group(&mut producer, 1_000);
8910 }
8911
8912 pool.resize(100);
8914 let before = pool.used();
8915
8916 finished_group(&mut producer, 1_000);
8918
8919 let consumer = producer.consume();
8920 assert!(consumer.peek_group(0).is_none(), "the oldest groups are evicted");
8921 assert!(consumer.peek_group(1).is_none());
8922 assert!(consumer.peek_group(2).is_some(), "the backlog drains gradually");
8923 assert!(pool.used() > before - 4_000, "one write must not dump the backlog");
8924 }
8925
8926 #[tokio::test]
8930 async fn accept_preserves_write_accounting() {
8931 let config = cache::Config::default()
8932 .with_capacity(12_000)
8933 .with_expiry(cache::DEFAULT_EXPIRY);
8934 let pool = cache::Pool::new(config);
8935 let broadcast = broadcast::Info {
8936 pool: pool.clone(),
8937 ..Default::default()
8938 };
8939 let request = Request::new(Arc::new(broadcast), "test");
8940 let dynamic = request.dynamic();
8941 let consumer = request.consume();
8942
8943 let pending = consumer.fetch_group(0, None);
8945 let req = dynamic
8946 .requested_group()
8947 .now_or_never()
8948 .expect("should not block")
8949 .unwrap();
8950 let mut backfill = req.accept(None).unwrap();
8951 pending.await.unwrap();
8952 backfill
8953 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 30_000]))
8954 .unwrap();
8955
8956 let producer = request.accept(None);
8959 producer.append_group().unwrap().finish().unwrap();
8960 producer.append_group().unwrap().finish().unwrap();
8961
8962 assert!(
8963 producer.consume().peek_group(0).is_none(),
8964 "pre-accept backfill growth is reclaimed after accept"
8965 );
8966 assert!(pool.used() <= 13_000, "usage converges: {}", pool.used());
8967 }
8968
8969 #[tokio::test]
8972 async fn a_fetch_only_cache_is_not_live() {
8973 let request = Request::new(Arc::new(broadcast::Info::default()), "test");
8974 let dynamic = request.dynamic();
8975 let consumer = request.consume();
8976
8977 let pending = consumer.fetch_group(2, None);
8978 let req = dynamic
8979 .requested_group()
8980 .now_or_never()
8981 .expect("should not block")
8982 .unwrap();
8983 let mut fetched = req.accept(None).unwrap();
8984 fetched
8985 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"2-0"))
8986 .unwrap();
8987 fetched.finish().unwrap();
8988 pending.await.unwrap();
8989 assert!(!consumer.is_live(), "a fetch-only cache is not live");
8990
8991 let producer = request.accept(None);
8992 assert!(consumer.is_live(), "an accepted track is live");
8993
8994 drop(dynamic);
8996 drop(producer);
8997 assert!(!consumer.is_live(), "a track whose producer went is not live");
8998 }
8999
9000 #[tokio::test]
9003 async fn recreated_sequence_bounds_eviction_hints() {
9004 let (producer, _pool) = pooled_producer(1 << 40);
9005 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9006
9007 for _ in 0..200 {
9008 let group = producer.create_group(1u64.into()).unwrap();
9009 group.abort(Error::Cancel).unwrap();
9010 }
9011
9012 let state = producer.state.read();
9013 assert!(
9014 state.evict.len() <= 2 * state.lookup.len() + EVICT_SLACK,
9015 "stale hints are compacted: {} entries for {} slots",
9016 state.evict.len(),
9017 state.lookup.len()
9018 );
9019 }
9020
9021 #[tokio::test]
9024 async fn same_tick_write_outranks_inserted() {
9025 let unit = 100 * cache::ENTRY_OVERHEAD;
9028 let (mut producer, _pool) = pooled_producer(10 * unit);
9030
9031 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();
9038 assert!(consumer.peek_group(0).is_none(), "insert-only content pays first");
9039 assert!(consumer.peek_group(1).is_some(), "same-tick written content survives");
9040 }
9041
9042 #[tokio::test]
9045 async fn frame_only_writer_pays() {
9046 let (producer, pool) = pooled_producer(2_000);
9047 let mut demoted = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); demoted
9053 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
9054 .unwrap();
9055
9056 assert!(
9057 pool.used() <= 5_000,
9058 "the frame write settled the debt: {}",
9059 pool.used()
9060 );
9061 assert!(matches!(demoted.finish(), Err(Error::Evicted)));
9062 }
9063
9064 #[tokio::test]
9067 async fn each_track_owns_its_account() {
9068 let broadcast = Arc::new(broadcast::Info::default());
9069 let info = Info::default();
9070 let a = Producer::new(broadcast.clone(), "a", info.clone());
9071 let b = Producer::new(broadcast, "b", info);
9072
9073 let a = a.state.read().cache.clone();
9074 let b = b.state.read().cache.clone();
9075 assert!(!Arc::ptr_eq(&a, &b), "each track owns its account");
9076 }
9077
9078 #[tokio::test]
9081 async fn a_dynamic_defers_teardown() {
9082 let (mut producer, pool) = pooled_producer(1 << 40);
9083 let dynamic = producer.dynamic();
9084 finished_group(&mut producer, 100);
9085
9086 drop(producer);
9087 assert!(pool.used() > 0, "the handler still serves the cache");
9088
9089 drop(dynamic);
9090 assert_eq!(pool.used(), 0, "the last handle tears it down");
9091 }
9092
9093 #[tokio::test]
9099 async fn finished_track_frees_its_cache() {
9100 let (mut producer, pool) = pooled_producer(1 << 40);
9101 finished_group(&mut producer, 100);
9102 producer.finish().unwrap();
9103
9104 let state = producer.state.downgrade();
9105 drop(producer);
9106
9107 assert!(state.upgrade().is_none(), "the track state is freed");
9108 assert_eq!(pool.used(), 0, "so are its cached bytes");
9109 }
9110
9111 #[tokio::test]
9115 async fn teardown_ignores_a_settling_group() {
9116 let (producer, pool) = pooled_producer(1 << 40);
9117 let mut group = producer.append_group().unwrap();
9119 group
9120 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9121 .unwrap();
9122 drop(group);
9123
9124 let settling = producer.state.downgrade().upgrade().expect("open");
9126 drop(producer);
9127
9128 assert_eq!(pool.used(), 0, "the abrupt teardown still released the cache");
9129 drop(settling);
9130 }
9131
9132 #[tokio::test]
9135 async fn cached_group_outlives_its_track() {
9136 let (mut producer, pool) = pooled_producer(1 << 40);
9137 let sequence = finished_group(&mut producer, 100);
9138 let group = producer.consume().peek_group(sequence).expect("cached");
9139 producer.finish().unwrap();
9140
9141 let state = producer.state.downgrade();
9142 drop(producer);
9143 assert!(state.upgrade().is_none(), "the track state is freed");
9144 assert!(pool.used() > 0, "the retained group keeps its own bytes");
9145
9146 drop(group);
9147 assert_eq!(pool.used(), 0, "which it releases when dropped");
9148 }
9149
9150 #[tokio::test]
9154 async fn pre_accept_backfill_settles_late_writes() {
9155 let config = cache::Config::default()
9156 .with_capacity(2_000)
9157 .with_expiry(cache::DEFAULT_EXPIRY);
9158 let pool = cache::Pool::new(config);
9159 let broadcast = broadcast::Info {
9160 pool: pool.clone(),
9161 ..Default::default()
9162 };
9163 let request = Request::new(Arc::new(broadcast), "test");
9164 let dynamic = request.dynamic();
9165 let consumer = request.consume();
9166
9167 let pending = consumer.fetch_group(0, None);
9169 let req = dynamic
9170 .requested_group()
9171 .now_or_never()
9172 .expect("should not block")
9173 .unwrap();
9174 let mut backfill = req.accept(None).unwrap();
9175 pending.await.unwrap();
9176
9177 let producer = request.accept(None);
9179 producer.append_group().unwrap().finish().unwrap();
9180
9181 backfill
9184 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 300_000]))
9185 .unwrap();
9186
9187 assert!(
9188 pool.used() <= 5_000,
9189 "the frame write settled the debt: {}",
9190 pool.used()
9191 );
9192 }
9193
9194 #[tokio::test]
9198 async fn write_restarts_retention_clock() {
9199 let (producer, _pool) = pooled_producer(1 << 40);
9200 let mut straggler = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9205 straggler
9206 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9207 .unwrap();
9208 producer.append_group().unwrap().finish().unwrap(); let consumer = producer.consume();
9211 assert!(consumer.peek_group(0).is_some(), "the write restarted the clock");
9212
9213 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9215 producer.append_group().unwrap().finish().unwrap(); assert!(consumer.peek_group(0).is_none(), "idle content still expires");
9217 }
9218
9219 #[tokio::test]
9222 async fn refreshed_front_does_not_starve_expiry() {
9223 let (producer, _pool) = pooled_producer(1 << 40);
9224 let dynamic = producer.dynamic();
9225 let consumer = producer.consume();
9226
9227 producer.create_group(10u64.into()).unwrap().finish().unwrap();
9228 for sequence in 1..=5u64 {
9229 let pending = consumer.fetch_group(sequence, None);
9230 let req = dynamic
9231 .requested_group()
9232 .now_or_never()
9233 .expect("should not block")
9234 .unwrap();
9235 let mut group = req.accept(None).unwrap();
9236 group
9237 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9238 .unwrap();
9239 group.finish().unwrap();
9240 pending.await.unwrap();
9241 }
9242
9243 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9246 for sequence in 1..=4u64 {
9247 consumer.fetch_group(sequence, None).await.unwrap();
9248 }
9249
9250 for _ in 0..3 {
9252 producer.append_group().unwrap().finish().unwrap();
9253 }
9254 assert!(consumer.peek_group(5).is_none(), "expired backfill is reclaimed");
9255 assert!(consumer.peek_group(1).is_some(), "refreshed backfill survives");
9256 }
9257
9258 #[tokio::test]
9261 async fn recreated_sequence_delivered_once() {
9262 let (producer, _pool) = pooled_producer(1 << 40);
9263
9264 producer.create_group(0u64.into()).unwrap().finish().unwrap();
9265 let aborted = producer.create_group(1u64.into()).unwrap();
9266 aborted.abort(Error::Cancel).unwrap();
9267 producer.create_group(2u64.into()).unwrap().finish().unwrap();
9268 producer.create_group(1u64.into()).unwrap().finish().unwrap();
9269
9270 let mut subscriber = producer.subscribe(replay());
9271 assert_eq!(subscriber.assert_group().sequence, 0);
9272 assert_eq!(subscriber.assert_group().sequence, 2);
9273 assert_eq!(
9274 subscriber.assert_group().sequence,
9275 1,
9276 "replacement arrives at its own position"
9277 );
9278 subscriber.assert_no_group();
9279 }
9280
9281 #[tokio::test]
9285 async fn datagrams_do_not_block_eviction() {
9286 let (mut producer, pool) = pooled_producer(1_000);
9287 for _ in 0..10 {
9288 finished_group(&mut producer, 1_000);
9289 producer.append_datagram(Timestamp::ZERO, &b"beat"[..]).unwrap();
9290 }
9291
9292 let consumer = producer.consume();
9293 assert!(consumer.peek_group(0).is_none(), "old groups still evict");
9294 assert!(
9295 pool.used() < 4 * 1_256,
9296 "interleaved datagrams must not bypass the budget: {}",
9297 pool.used()
9298 );
9299 }
9300
9301 #[tokio::test]
9305 async fn aborted_group_leaves_no_ghost_sample() {
9306 let (producer, pool) = pooled_producer(1 << 40);
9307 let group0 = producer.append_group().unwrap();
9308 producer.append_group().unwrap(); assert!(pool.average().is_some(), "demoted group is sampled");
9311 group0.abort(Error::Cancel).unwrap();
9312 assert_eq!(pool.average(), None, "the abort must remove the sample");
9313 }
9314
9315 #[tokio::test]
9318 async fn empty_groups_repay_overhead() {
9319 let (producer, pool) = pooled_producer(1_000);
9320 for _ in 0..100 {
9321 let group = producer.append_group().unwrap();
9322 group.finish().unwrap();
9323 }
9324
9325 assert!(
9326 pool.used() <= 3_000,
9327 "empty-group overhead must stay near the budget: {}",
9328 pool.used()
9329 );
9330 }
9331
9332 #[tokio::test]
9335 async fn growth_on_demoted_group_is_billed() {
9336 let (producer, pool) = pooled_producer(2_000);
9337 let mut straggler = producer.append_group().unwrap(); producer.append_group().unwrap().finish().unwrap(); straggler
9342 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000]))
9343 .unwrap();
9344
9345 producer.append_group().unwrap().finish().unwrap(); let consumer = producer.consume();
9349 assert!(consumer.peek_group(0).is_none(), "the ballooned group is evicted");
9350 assert!(pool.used() <= 3_000, "growth is reclaimed: {}", pool.used());
9351 }
9352
9353 #[tokio::test]
9356 async fn refilled_sequence_stays_out_of_subscriptions() {
9357 let (producer, _pool) = pooled_producer(1 << 40);
9358 let dynamic = producer.dynamic();
9359 let consumer = producer.consume();
9360
9361 producer.create_group(0u64.into()).unwrap().finish().unwrap();
9362 let aborted = producer.create_group(1u64.into()).unwrap();
9363 aborted.abort(Error::Cancel).unwrap();
9364 producer.create_group(2u64.into()).unwrap().finish().unwrap();
9365
9366 let pending = consumer.fetch_group(1, None);
9369 let req = dynamic
9370 .requested_group()
9371 .now_or_never()
9372 .expect("should not block")
9373 .unwrap();
9374 let mut group = req.accept(None).unwrap();
9375 group
9376 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
9377 .unwrap();
9378 group.finish().unwrap();
9379 pending.await.unwrap();
9380
9381 assert!(consumer.peek_group(1).is_some());
9383 let mut subscriber = producer.subscribe(replay());
9384 assert_eq!(subscriber.assert_group().sequence, 0);
9385 assert_eq!(subscriber.assert_group().sequence, 2);
9386 subscriber.assert_no_group();
9387 }
9388
9389 #[tokio::test]
9392 async fn expired_backfill_behind_refreshed_reclaimed() {
9393 let (producer, _pool) = pooled_producer(1 << 40);
9394 let dynamic = producer.dynamic();
9395 let consumer = producer.consume();
9396
9397 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9398 for sequence in [2u64, 3u64] {
9399 let pending = consumer.fetch_group(sequence, None);
9400 let req = dynamic
9401 .requested_group()
9402 .now_or_never()
9403 .expect("should not block")
9404 .unwrap();
9405 let mut group = req.accept(None).unwrap();
9406 group
9407 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 100]))
9408 .unwrap();
9409 group.finish().unwrap();
9410 pending.await.unwrap();
9411 }
9412
9413 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
9415 consumer.fetch_group(2, None).await.unwrap();
9416 crate::model::clock::advance(cache::DEFAULT_EXPIRY / 2 + Duration::from_secs(1));
9417 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9418
9419 let consumer = producer.consume();
9420 assert!(consumer.peek_group(2).is_some(), "refreshed backfill survives");
9421 assert!(consumer.peek_group(3).is_none(), "expired backfill is reclaimed");
9422 }
9423
9424 #[tokio::test]
9427 async fn same_tick_fetch_protects() {
9428 let (mut producer, _pool) = pooled_producer(10_000);
9430 let consumer = producer.consume();
9431
9432 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();
9437
9438 finished_group(&mut producer, 3_000); finished_group(&mut producer, 3_000); assert!(consumer.peek_group(0).is_some(), "same-tick refresh protects");
9442 assert!(consumer.peek_group(1).is_none(), "the unread group dies instead");
9443 }
9444
9445 #[tokio::test]
9449 async fn refetched_latest_stays_protected() {
9450 let (producer, _pool) = pooled_producer(10_000);
9451 let dynamic = producer.dynamic();
9452 let consumer = producer.consume();
9453
9454 let straggler = producer.append_group().unwrap(); let latest = producer.append_group().unwrap(); latest.abort(Error::Cancel).unwrap();
9459
9460 let pending = consumer.fetch_group(1, None);
9462 let req = dynamic
9463 .requested_group()
9464 .now_or_never()
9465 .expect("should not block")
9466 .unwrap();
9467 let mut group = req.accept(None).unwrap();
9468 group
9469 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
9470 .unwrap();
9471 group.finish().unwrap();
9472 pending.await.unwrap();
9473
9474 {
9477 let state = producer.state.read();
9478 assert!(state.lookup.contains_key(&1), "refetched group is cached");
9479 assert!(
9480 state.evict.iter().all(|(sequence, _)| *sequence != 1),
9481 "the live edge must not be an eviction candidate"
9482 );
9483 }
9484 drop(straggler);
9485 }
9486
9487 #[tokio::test]
9490 async fn eviction_allows_refetch() {
9491 let (mut producer, _pool) = pooled_producer(10_000);
9492 let dynamic = producer.dynamic();
9493
9494 finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); finished_group(&mut producer, 10_000); let consumer = producer.consume();
9499 assert!(consumer.peek_group(0).is_none());
9500 let pending = consumer.fetch_group(0, None);
9501
9502 let req = dynamic
9503 .requested_group()
9504 .now_or_never()
9505 .expect("should not block")
9506 .unwrap();
9507 assert_eq!(req.sequence(), 0);
9508
9509 let mut group = req.accept(None).unwrap();
9510 group
9511 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"refetched"))
9512 .unwrap();
9513 group.finish().unwrap();
9514
9515 let mut group = pending.await.unwrap();
9516 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"refetched");
9517 }
9518
9519 #[test]
9529 fn an_aborted_group_releases_its_sequence() {
9530 let producer = track_producer("test", None);
9531 let consumer = producer.consume();
9532
9533 let mut group = producer.create_group(group::Info { sequence: 3 }).unwrap();
9534 group
9535 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head"))
9536 .unwrap();
9537
9538 assert!(matches!(
9540 producer.create_group(group::Info { sequence: 3 }),
9541 Err(Error::Duplicate)
9542 ));
9543 assert!(consumer.peek_group(3).is_some());
9544
9545 group.abort(Error::Cancel).unwrap();
9546
9547 assert!(consumer.peek_group(3).is_none(), "an aborted slot is a cache miss");
9548 producer
9549 .create_group(group::Info { sequence: 3 })
9550 .expect("an aborted slot releases its sequence")
9551 .finish()
9552 .unwrap();
9553 }
9554
9555 #[tokio::test]
9558 async fn fetched_backfill_not_subscribed() {
9559 let (producer, _pool) = pooled_producer(1 << 40);
9560 let dynamic = producer.dynamic();
9561 let consumer = producer.consume();
9562
9563 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9565 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9566
9567 let pending = consumer.fetch_group(2, None);
9569 let req = dynamic
9570 .requested_group()
9571 .now_or_never()
9572 .expect("should not block")
9573 .unwrap();
9574 let mut group = req.accept(None).unwrap();
9575 group
9576 .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"backfill"))
9577 .unwrap();
9578 group.finish().unwrap();
9579 let mut fetched = pending.await.unwrap();
9580 assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"backfill");
9581 assert!(consumer.peek_group(2).is_some(), "backfill is cached for later fetches");
9582
9583 let mut subscriber = producer.subscribe(replay());
9585 assert_eq!(subscriber.assert_group().sequence, 5);
9586 assert_eq!(subscriber.assert_group().sequence, 6);
9587 subscriber.assert_no_group();
9588 }
9589
9590 #[tokio::test]
9593 async fn expired_backfill_reclaimed() {
9594 let (producer, pool) = pooled_producer(1 << 40);
9595 let dynamic = producer.dynamic();
9596 let consumer = producer.consume();
9597
9598 producer.create_group(5u64.into()).unwrap().finish().unwrap();
9599
9600 let pending = consumer.fetch_group(2, None);
9602 let req = dynamic
9603 .requested_group()
9604 .now_or_never()
9605 .expect("should not block")
9606 .unwrap();
9607 let mut group = req.accept(None).unwrap();
9608 group
9609 .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 1000]))
9610 .unwrap();
9611 group.finish().unwrap();
9612 pending.await.unwrap();
9613 let used = pool.used();
9614
9615 crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
9617 producer.create_group(6u64.into()).unwrap().finish().unwrap();
9618
9619 assert!(consumer.peek_group(2).is_none(), "expired backfill is reclaimed");
9620 assert!(pool.used() < used, "its bytes are released");
9621 }
9622
9623 #[tokio::test]
9624 async fn fetch_aborts_with_track() {
9625 let producer = track_producer("test", None);
9626 let dynamic = producer.dynamic();
9627 let consumer = producer.consume();
9628
9629 let pending = consumer.fetch_group(3, None);
9630 assert!(kio::Pollable::poll(&*pending, &kio::Waiter::noop()).is_pending());
9631
9632 producer.abort(Error::Cancel).unwrap();
9633 assert!(pending.await.is_err());
9634 drop(dynamic);
9635 }
9636}