1use crate::{broadcast, cache, group, stats, track};
2use kio::Pollable;
3use std::{
4 cmp::Reverse,
5 collections::{BTreeSet, HashMap, HashSet, VecDeque},
6 fmt,
7 sync::Arc,
8 sync::atomic::{AtomicU64, Ordering},
9 task::{Poll, ready},
10 time::Duration,
11};
12
13use rand::RngExt;
14
15use super::{
16 Requests, WeakCache, WeakEntry,
17 front::{Action, Candidate, Event, Front, Pin, Refusal},
18};
19use crate::{
20 AsPath, Error, InvalidPattern, Path, PathOwned, Pattern, Patterns,
21 coding::{BoundsExceeded, Decode, DecodeError, Encode, EncodeError},
22 path::Segment,
23 runtime::{Instant, Timers},
24 time::Clock,
25 util::{Keepalive, TaskSet, Tasks, TasksWeak},
26};
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
38pub struct Hop {
39 id: u64,
41}
42
43impl Hop {
44 pub const UNKNOWN: Self = Self { id: 0 };
51
52 pub fn new(id: u64) -> Result<Self, InvalidHop> {
58 if id == 0 || id >= 1u64 << 62 {
59 return Err(InvalidHop::Range);
60 }
61 Ok(Self { id })
62 }
63
64 pub fn random() -> Self {
71 let mut rng = rand::rng();
72 let id = rng.random_range(1..(1u64 << 53));
73 Self { id }
74 }
75
76 pub fn id(self) -> u64 {
78 self.id
79 }
80
81 pub(crate) fn from_wire(id: u64) -> Result<Self, DecodeError> {
83 if id >= 1u64 << 62 {
84 return Err(DecodeError::InvalidValue);
85 }
86 Ok(Self { id })
87 }
88}
89
90#[derive(Clone, Debug)]
96#[non_exhaustive]
97pub struct Config {
98 pub hop: Hop,
101
102 pub max_hops: HopLimit,
109
110 pub pool: cache::Pool,
117
118 pub cache_duration: Duration,
126
127 pub default_max_age: Duration,
137
138 pub update_hold: Duration,
155}
156
157pub const DEFAULT_UPDATE_HOLD: Duration = Duration::from_millis(300);
160
161const RETRACT_QUARANTINE: Duration = Duration::from_secs(2);
165
166impl Default for Config {
167 fn default() -> Self {
169 let pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY));
170 Self {
171 hop: Hop::random(),
172 max_hops: HopLimit::default(),
173 pool,
174 cache_duration: Duration::MAX,
175 default_max_age: track::DEFAULT_MAX_AGE,
176 update_hold: DEFAULT_UPDATE_HOLD,
177 }
178 }
179}
180
181impl Config {
182 pub fn new(hop: Hop) -> Self {
184 Self { hop, ..Self::default() }
185 }
186}
187
188impl From<Hop> for Config {
189 fn from(hop: Hop) -> Self {
191 Self::new(hop)
192 }
193}
194
195impl TryFrom<u64> for Hop {
196 type Error = InvalidHop;
197
198 fn try_from(id: u64) -> Result<Self, Self::Error> {
199 Self::new(id)
200 }
201}
202
203impl fmt::Display for Hop {
204 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
205 self.id.fmt(f)
206 }
207}
208
209impl<V: Copy> Encode<V> for Hop
210where
211 u64: Encode<V>,
212{
213 fn encode<W: bytes::BufMut>(&self, w: &mut W, version: V) -> Result<(), EncodeError> {
214 self.id.encode(w, version)
215 }
216}
217
218impl<V: Copy> Decode<V> for Hop
219where
220 u64: Decode<V>,
221{
222 fn decode<R: bytes::Buf>(r: &mut R, version: V) -> Result<Self, DecodeError> {
223 Self::from_wire(u64::decode(r, version)?)
224 }
225}
226
227pub(crate) const MAX_HOPS: usize = 32;
233
234#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
236pub struct HopLimit(usize);
237
238impl HopLimit {
239 pub const fn new(max: usize) -> Option<Self> {
241 if max == 0 || max > MAX_HOPS {
242 None
243 } else {
244 Some(Self(max))
245 }
246 }
247
248 pub const fn get(self) -> usize {
250 self.0
251 }
252}
253
254impl Default for HopLimit {
255 fn default() -> Self {
256 Self(MAX_HOPS)
257 }
258}
259
260#[derive(Debug, Clone, Default, PartialEq, Eq)]
268pub struct Hops(Vec<Hop>);
269
270#[derive(Debug, Clone, Copy, PartialEq, Eq)]
272#[non_exhaustive]
273pub enum InvalidHop {
274 Range,
277
278 TooMany,
281
282 Duplicate,
286}
287
288impl fmt::Display for InvalidHop {
289 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
290 match self {
291 Self::Range => write!(f, "local hop id must be non-zero and below 2^62"),
292 Self::TooMany => write!(f, "too many hops (max {MAX_HOPS})"),
293 Self::Duplicate => write!(f, "hop already in the chain"),
294 }
295 }
296}
297
298impl std::error::Error for InvalidHop {}
299
300impl From<InvalidHop> for DecodeError {
301 fn from(err: InvalidHop) -> Self {
302 match err {
303 InvalidHop::TooMany => DecodeError::BoundsExceeded,
304 InvalidHop::Range | InvalidHop::Duplicate => DecodeError::InvalidValue,
305 }
306 }
307}
308
309impl Hops {
310 pub fn new() -> Self {
312 Self(Vec::new())
313 }
314
315 pub fn push(&mut self, hop: Hop) -> Result<(), InvalidHop> {
321 if self.0.len() >= MAX_HOPS {
322 return Err(InvalidHop::TooMany);
323 }
324 if hop != Hop::UNKNOWN && self.0.contains(&hop) {
325 return Err(InvalidHop::Duplicate);
326 }
327 self.0.push(hop);
328 Ok(())
329 }
330
331 pub(crate) fn stamp(&mut self, stamp: Hop) -> Result<(), InvalidHop> {
340 match self.0.first() {
341 None => {
342 self.push(stamp)?;
343 self.push(Hop::UNKNOWN)
344 }
345 Some(first) if *first == Hop::UNKNOWN => {
346 if self.0.len() >= MAX_HOPS {
347 return Err(InvalidHop::TooMany);
348 }
349 if self.0.contains(&stamp) {
350 return Err(InvalidHop::Duplicate);
351 }
352 self.0.insert(0, stamp);
353 Ok(())
354 }
355 Some(_) => Ok(()),
356 }
357 }
358
359 pub fn contains(&self, hop: &Hop) -> bool {
361 self.0.contains(hop)
362 }
363
364 pub fn len(&self) -> usize {
366 self.0.len()
367 }
368
369 pub fn is_empty(&self) -> bool {
371 self.0.is_empty()
372 }
373
374 pub fn iter(&self) -> std::slice::Iter<'_, Hop> {
376 self.0.iter()
377 }
378
379 pub fn as_slice(&self) -> &[Hop] {
381 &self.0
382 }
383}
384
385impl TryFrom<Vec<Hop>> for Hops {
386 type Error = InvalidHop;
387
388 fn try_from(v: Vec<Hop>) -> Result<Self, Self::Error> {
389 if v.len() > MAX_HOPS {
390 return Err(InvalidHop::TooMany);
391 }
392 for (i, hop) in v.iter().enumerate() {
394 if *hop != Hop::UNKNOWN && v[i + 1..].contains(hop) {
395 return Err(InvalidHop::Duplicate);
396 }
397 }
398 Ok(Self(v))
399 }
400}
401
402impl<'a> IntoIterator for &'a Hops {
403 type Item = &'a Hop;
404 type IntoIter = std::slice::Iter<'a, Hop>;
405
406 fn into_iter(self) -> Self::IntoIter {
407 self.iter()
408 }
409}
410
411impl<V: Copy> Encode<V> for Hops
412where
413 u64: Encode<V>,
414 Hop: Encode<V>,
415{
416 fn encode<W: bytes::BufMut>(&self, w: &mut W, version: V) -> Result<(), EncodeError> {
417 (self.0.len() as u64).encode(w, version)?;
418 for origin in &self.0 {
419 origin.encode(w, version)?;
420 }
421 Ok(())
422 }
423}
424
425impl<V: Copy> Decode<V> for Hops
426where
427 u64: Decode<V>,
428 Hop: Decode<V>,
429{
430 fn decode<R: bytes::Buf>(r: &mut R, version: V) -> Result<Self, DecodeError> {
431 let count = u64::decode(r, version)? as usize;
432 if count > MAX_HOPS {
433 return Err(DecodeError::BoundsExceeded);
434 }
435 let mut list = Self(Vec::with_capacity(count));
438 for _ in 0..count {
439 list.push(Hop::decode(r, version)?)?;
440 }
441 Ok(list)
442 }
443}
444
445const MAX_COST: u64 = (1 << 62) - 1;
453
454#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
463pub struct Cost {
464 pub warm: u64,
473
474 pub cold: u64,
481}
482
483impl Cost {
484 pub const fn new(cost: u64) -> Self {
487 Self { warm: cost, cold: cost }
488 }
489
490 pub const MAX: Self = Self::new(MAX_COST);
497
498 pub const DRAIN: Self = Self::MAX;
501
502 pub(crate) const UNKNOWN: Self = Self {
506 warm: 0,
507 cold: MAX_COST,
508 };
509
510 pub(crate) fn charged(self, link_cost: u64) -> Self {
513 Self {
514 warm: self.warm.saturating_add(link_cost).min(MAX_COST),
515 cold: self.cold.saturating_add(link_cost).min(MAX_COST),
516 }
517 }
518
519 pub(crate) fn clamped(self) -> Self {
522 Self {
523 warm: self.warm.min(MAX_COST),
524 cold: self.cold.min(MAX_COST),
525 }
526 }
527}
528
529impl From<u64> for Cost {
530 fn from(cost: u64) -> Self {
531 Self::new(cost)
532 }
533}
534
535#[derive(Clone, Debug, PartialEq, Eq)]
546#[non_exhaustive]
547pub struct Route {
548 pub hops: Hops,
554
555 pub cost: Cost,
560
561 pub(crate) via: Hop,
567
568 pub(crate) source: Source,
570}
571
572impl Default for Route {
573 fn default() -> Self {
574 Self {
575 hops: Hops::new(),
576 cost: Cost::default(),
577 via: Hop::UNKNOWN,
578 source: Source::Local,
579 }
580 }
581}
582
583#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
589pub enum Source {
590 #[default]
592 Local,
593 Peer(Hop),
596}
597
598impl Route {
599 pub fn with_hops(mut self, hops: Hops) -> Self {
601 self.hops = hops;
602 self
603 }
604
605 pub fn with_cost(mut self, cost: impl Into<Cost>) -> Self {
610 self.cost = cost.into();
611 self
612 }
613
614 pub(crate) fn with_via(mut self, via: Hop) -> Self {
619 self.via = via;
620 self
621 }
622
623 pub fn is_anonymous(&self) -> bool {
631 self.hops.iter().any(|hop| *hop == Hop::UNKNOWN)
632 }
633
634 pub fn source(&self) -> Source {
639 self.source
640 }
641}
642
643static NEXT_CONSUMER_ID: AtomicU64 = AtomicU64::new(0);
644
645#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
646struct ConsumerId(u64);
647
648impl ConsumerId {
649 fn new() -> Self {
650 Self(NEXT_CONSUMER_ID.fetch_add(1, Ordering::Relaxed))
651 }
652}
653
654fn fnv_key(name: &str, origins: impl IntoIterator<Item = Hop>) -> u64 {
663 const SEED: u64 = 0x420C0DECB00B; const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
665
666 let mut hash = SEED;
667 for &byte in name.as_bytes() {
668 hash = (hash ^ u64::from(byte)).wrapping_mul(FNV_PRIME);
669 }
670 for origin in origins {
671 for &byte in &origin.id().to_le_bytes() {
672 hash = (hash ^ u64::from(byte)).wrapping_mul(FNV_PRIME);
673 }
674 }
675
676 hash
677}
678
679fn route_order(prefix: &Path, entry: &RouteEntry) -> (bool, Cost, bool, usize, u64, Reverse<u64>) {
692 (
693 entry.is_anonymous(),
694 entry.cost,
695 !entry.local,
696 entry.hops.len(),
697 match entry.peer {
698 true => fnv_key(prefix.as_str(), entry.hops.iter().copied()),
699 false => 0,
700 },
701 Reverse(entry.id),
702 )
703}
704
705type RouteMeta = (Hops, Cost, Source);
707
708fn earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant> {
710 match (a, b) {
711 (Some(a), Some(b)) => Some(a.min(b)),
712 (a, b) => a.or(b),
713 }
714}
715
716#[derive(Clone, Copy)]
720struct Quarantine {
721 held: Held,
722 publisher: Option<Hop>,
723 rank: (Cost, usize),
724}
725
726impl Quarantine {
727 fn holds(&self, meta: &RouteMeta) -> bool {
731 self.publisher.is_some_and(|hop| meta.0.iter().next() == Some(&hop)) && (meta.1, meta.0.len()) > self.rank
732 }
733}
734
735#[derive(Clone, Copy, PartialEq)]
742enum Held {
743 Anchoring(Instant),
744 Until(Instant),
745}
746
747type AnnounceMeta = (RouteMeta, Option<Vec<Pattern>>);
754
755enum PendingUpdate {
756 Announce(AnnounceMeta),
757 Unannounce(AnnounceMeta),
758 UnannounceAnnounce { old: AnnounceMeta, new: AnnounceMeta },
759}
760
761struct Pending {
763 update: Option<PendingUpdate>,
764 wake: Option<Instant>,
767 held: Option<Held>,
769}
770
771#[derive(Default)]
779struct OriginConsumerState {
780 pending: HashMap<PathOwned, Pending>,
781 due: BTreeSet<PathOwned>,
784 parked: BTreeSet<(Instant, PathOwned)>,
786 delivered: BTreeSet<PathOwned>,
791 ended: bool,
794 #[cfg(test)]
796 steps: usize,
797}
798
799impl OriginConsumerState {
800 fn put(&mut self, prefix: PathOwned, update: PendingUpdate) {
803 let Some(pending) = self.pending.get_mut(&prefix) else {
804 self.due.insert(prefix.clone());
805 self.pending.insert(
806 prefix,
807 Pending {
808 update: Some(update),
809 wake: None,
810 held: None,
811 },
812 );
813 return;
814 };
815 pending.update = Some(update);
816 if let Some(wake) = pending.wake.take() {
817 let parked = (wake, prefix);
818 self.parked.remove(&parked);
819 self.due.insert(parked.1);
820 }
821 }
822
823 fn remove(&mut self, prefix: &PathOwned) -> Option<Pending> {
825 let (prefix, pending) = self.pending.remove_entry(prefix)?;
826 match pending.wake {
827 Some(wake) => self.parked.remove(&(wake, prefix)),
828 None => self.due.remove(&prefix),
829 };
830 Some(pending)
831 }
832
833 fn end(&mut self) {
835 self.ended = true;
836 for (_, prefix) in std::mem::take(&mut self.parked) {
837 self.pending.get_mut(&prefix).unwrap().wake = None;
838 self.due.insert(prefix);
839 }
840 }
841
842 fn apply_announce(&mut self, prefix: PathOwned, meta: RouteMeta, captures: Option<Vec<Pattern>>) {
843 let meta = (meta, captures);
844 let new = match self.pending.get_mut(&prefix).and_then(|pending| pending.update.take()) {
845 None | Some(PendingUpdate::Announce(_)) => PendingUpdate::Announce(meta),
847 Some(PendingUpdate::Unannounce(old) | PendingUpdate::UnannounceAnnounce { old, .. }) => {
849 PendingUpdate::UnannounceAnnounce { old, new: meta }
850 }
851 };
852 self.put(prefix, new);
853 }
854
855 fn apply_unannounce(&mut self, prefix: PathOwned, last: RouteMeta, captures: Option<Vec<Pattern>>) {
856 let last = (last, captures);
857 match self.pending.get_mut(&prefix).and_then(|pending| pending.update.take()) {
858 Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => {
861 self.remove(&prefix);
862 }
863 None | Some(PendingUpdate::Announce(_) | PendingUpdate::Unannounce(_)) => {
866 self.put(prefix, PendingUpdate::Unannounce(last));
867 }
868 Some(PendingUpdate::UnannounceAnnounce { old, .. }) => {
871 self.put(prefix, PendingUpdate::Unannounce(old));
872 }
873 }
874 }
875
876 fn due(&self, now: Option<Instant>) -> bool {
879 !self.due.is_empty()
880 || self.wake().is_some_and(|wake| now.is_some_and(|now| wake <= now))
881 || (self.ended && self.pending.is_empty())
882 }
883
884 fn wake(&self) -> Option<Instant> {
886 self.parked.first().map(|(wake, _)| *wake)
887 }
888
889 fn scan(
898 &mut self,
899 retracted: &HashMap<PathOwned, Quarantine>,
900 now: Option<Instant>,
901 hold: Duration,
902 ) -> Result<PathOwned, Option<Instant>> {
903 while let Some((wake, _)) = self.parked.first()
904 && now.is_some_and(|now| *wake <= now)
905 {
906 let (_, prefix) = self.parked.pop_first().unwrap();
907 self.pending.get_mut(&prefix).unwrap().wake = None;
908 self.due.insert(prefix);
909 }
910 while let Some(prefix) = self.due.first() {
911 #[cfg(test)]
912 {
913 self.steps += 1;
914 }
915 let Some(now) = now.filter(|_| !hold.is_zero() && !self.ended) else {
916 return Ok(prefix.clone());
917 };
918 let pending = self.pending.get_mut(prefix).unwrap();
919 let wake = match pending.update.as_ref().unwrap() {
920 PendingUpdate::Announce(_) if self.delivered.contains(prefix) => {
921 match pending.held {
925 None => {
926 pending.held = Some(Held::Anchoring(now));
927 Some(now + Duration::from_nanos(1))
928 }
929 Some(Held::Anchoring(seen)) if now <= seen => Some(seen + Duration::from_nanos(1)),
930 Some(Held::Anchoring(_)) => {
931 pending.held = Some(Held::Until(now + hold));
932 Some(now + hold)
933 }
934 Some(Held::Until(until)) => Some(until).filter(|until| *until > now),
935 }
936 }
937 PendingUpdate::Announce((meta, _)) => match retracted.get(prefix).filter(|q| q.holds(meta)) {
938 Some(Quarantine {
939 held: Held::Anchoring(seen),
940 ..
941 }) => Some(*seen + Duration::from_nanos(1)),
942 Some(Quarantine {
943 held: Held::Until(until),
944 ..
945 }) => Some(*until).filter(|until| *until > now),
946 None => None,
947 },
948 _ => None,
949 };
950 let Some(wake) = wake else {
951 return Ok(prefix.clone());
952 };
953 pending.wake = Some(wake);
954 let prefix = self.due.pop_first().unwrap();
955 self.parked.insert((wake, prefix));
956 }
957 Err(self.wake())
958 }
959
960 fn take_prefix(&mut self, prefix: PathOwned) -> Option<AnnounceUpdate> {
962 let ((meta, captures), kind) = match self.remove(&prefix).unwrap().update.unwrap() {
963 PendingUpdate::Announce(meta) => {
964 let kind = match self.delivered.insert(prefix.clone()) {
966 true => AnnounceKind::Announced,
967 false => AnnounceKind::Updated,
968 };
969 (meta, kind)
970 }
971 PendingUpdate::Unannounce(meta) => {
972 self.delivered.remove(&prefix);
973 (meta, AnnounceKind::Retracted)
974 }
975 PendingUpdate::UnannounceAnnounce { old, new } => {
976 self.delivered.remove(&prefix);
979 self.put(prefix.clone(), PendingUpdate::Announce(new));
980 (old, AnnounceKind::Retracted)
981 }
982 };
983 Some(AnnounceUpdate {
984 prefix,
985 captures,
986 route: Route {
987 hops: meta.0,
988 cost: meta.1,
989 via: Hop::UNKNOWN,
990 source: meta.2,
991 },
992 kind,
993 })
994 }
995}
996
997struct RouteEntry {
999 id: u64,
1000 prefix: PathOwned,
1001 scope: Patterns,
1004 hops: Hops,
1005 cost: Cost,
1006 via: Hop,
1010 local: bool,
1013 peer: bool,
1016 server: Option<kio::Shared<ServeState>>,
1020 source: Option<broadcast::Consumer>,
1024 advertised: bool,
1028 stale: bool,
1032 claim: Pattern,
1038 reach: Vec<Pattern>,
1042}
1043
1044impl RouteEntry {
1045 fn live(&self) -> bool {
1047 self.advertised && !self.stale
1048 }
1049
1050 fn is_anonymous(&self) -> bool {
1051 self.hops.iter().any(|hop| *hop == Hop::UNKNOWN)
1052 }
1053
1054 fn entered(&self) -> Source {
1056 match self.peer {
1057 true => Source::Peer(self.via),
1058 false => Source::Local,
1059 }
1060 }
1061
1062 fn meta(&self) -> RouteMeta {
1064 (self.hops.clone(), self.cost, self.entered())
1065 }
1066
1067 fn delivers(&self, meta: &RouteMeta) -> bool {
1069 let (hops, cost, source) = meta;
1070 *hops == self.hops && *cost == self.cost && *source == self.entered()
1071 }
1072
1073 fn serves(&self, path: &Path) -> bool {
1077 self.server.is_some() || (self.source.is_some() && self.prefix == *path)
1078 }
1079
1080 fn qualifies(&self, pin: Pin) -> bool {
1082 match pin {
1083 Pin::Any => true,
1084 Pin::Local => self.local,
1085 Pin::Publisher(first) => self.hops.iter().next() == Some(&first),
1086 Pin::Route(id) => self.id == id && self.hops.iter().next().is_none_or(|first| *first == Hop::UNKNOWN),
1088 }
1089 }
1090
1091 fn visible_to(&self, exclude: Option<Hop>) -> bool {
1096 match exclude {
1097 Some(peer) if peer != Hop::UNKNOWN => self.via != peer && !self.hops.contains(&peer),
1098 _ => true,
1099 }
1100 }
1101
1102 fn overlaps(&self, allowed: &Patterns) -> bool {
1104 self.reach.iter().any(|reach| allowed.overlaps(reach))
1105 }
1106}
1107
1108fn reach(scope: &Patterns, claim: &Pattern) -> Vec<Pattern> {
1110 scope
1111 .iter()
1112 .filter_map(|scope| scope.intersect(claim).ok())
1113 .flatten()
1114 .collect()
1115}
1116
1117fn prefix_claim(prefix: &Path) -> Result<Pattern, InvalidPattern> {
1119 if prefix.parts().count() == Path::MAX_PARTS {
1120 Pattern::literal(prefix.as_str())
1121 } else {
1122 Pattern::subtree(prefix.as_str())
1123 }
1124}
1125
1126#[derive(Default)]
1131struct ServeState {
1132 requests: Requests<PathOwned, kio::Producer<PendingBroadcast>>,
1135
1136 served: WeakCache<PathOwned, broadcast::WeakConsumer>,
1142
1143 closed: bool,
1146}
1147
1148type FrontKey = (PathOwned, Horizon);
1155
1156#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
1158struct Horizon {
1159 exclude: Option<Hop>,
1163 local: bool,
1165 max_hops: Option<HopLimit>,
1169}
1170
1171impl Horizon {
1172 fn admits(&self, entry: &RouteEntry) -> bool {
1174 !(self.local && entry.peer)
1175 && entry.visible_to(self.exclude)
1176 && self.max_hops.is_none_or(|max| entry.hops.len() < max.get())
1177 }
1178
1179 fn effective(self, routes: &RouteTable, path: &Path) -> Self {
1192 let horizon = Self { max_hops: None, ..self };
1193 match horizon.exclude {
1194 Some(peer) if routes.covering(path).any(|entry| !entry.visible_to(Some(peer))) => horizon,
1195 _ => Self {
1196 exclude: None,
1197 ..horizon
1198 },
1199 }
1200 }
1201}
1202
1203#[derive(Clone)]
1206struct RemoteFront {
1207 request: kio::Producer<PendingBroadcast>,
1211 broadcast: broadcast::WeakConsumer,
1214 pin: kio::Lock<Pin>,
1217}
1218
1219struct CursorRoute {
1221 meta: RouteMeta,
1223 served: bool,
1225 prefix: PathOwned,
1227 captures: Option<Vec<Pattern>>,
1229}
1230
1231impl WeakEntry for RemoteFront {
1232 fn is_closed(&self) -> bool {
1233 self.broadcast.is_closed()
1234 }
1235
1236 fn same_channel(&self, other: &Self) -> bool {
1237 self.broadcast.same_channel(&other.broadcast)
1238 }
1239}
1240
1241struct TableCursor {
1244 root: PathOwned,
1246 allowed: Patterns,
1248 heads: Vec<PathOwned>,
1252 horizon: Horizon,
1254 hidden: Hidden,
1256 state: kio::Producer<OriginConsumerState>,
1259 under: PathOwned,
1262 holes: Vec<PathOwned>,
1265 named: Option<(Mount, Patterns)>,
1268 current: HashMap<PathOwned, CursorRoute>,
1271}
1272
1273impl TableCursor {
1274 fn presented(&self, prefix: &Path, claim: &Pattern) -> Option<PathOwned> {
1280 if !self.allowed.overlaps(claim) {
1281 return None;
1282 }
1283
1284 if let Some(relative) = prefix.strip_prefix(&self.root) {
1285 return Some(relative.to_owned());
1286 }
1287 self.root.has_prefix(prefix).then(PathOwned::default)
1288 }
1289
1290 fn captures(&self, prefix: &Path) -> Option<Vec<Pattern>> {
1293 let (literal, allowed) = match &self.named {
1294 Some((mount, allowed)) => (Pattern::literal(mount.name(prefix)?.as_str()).ok()?, allowed),
1295 None => (Pattern::literal(prefix.as_str()).ok()?, &self.allowed),
1296 };
1297 allowed
1298 .iter()
1299 .filter_map(|allowed| {
1300 allowed
1301 .captures(&literal)
1302 .map(|captures| (allowed.specificity(), captures))
1303 })
1304 .max_by_key(|(specificity, _)| *specificity)
1305 .map(|(_, captures)| captures)
1306 }
1307
1308 fn visible(&self, entry: &RouteEntry) -> bool {
1312 entry.live()
1313 && self.horizon.admits(entry)
1314 && entry.overlaps(&self.allowed)
1315 && self.discovers(&entry.prefix)
1316 && !self.holes.iter().any(|hole| entry.prefix.has_prefix(hole))
1317 && self.named.as_ref().is_none_or(|(mount, _)| mount.names(&entry.prefix))
1318 }
1319
1320 fn update(&mut self, presented: &PathOwned, best: Option<&RouteEntry>) {
1322 let Some(entry) = best else {
1323 if let Some(last) = self.current.remove(presented)
1324 && let Ok(mut state) = self.state.write()
1325 {
1326 state.apply_unannounce(self.under.join(presented), last.meta, last.captures);
1327 }
1328 return;
1329 };
1330 let served = entry.server.is_some();
1331 if self.current.get(presented).is_some_and(|previous| {
1334 previous.prefix == entry.prefix && previous.served == served && entry.delivers(&previous.meta)
1335 }) {
1336 return;
1337 }
1338
1339 let meta = entry.meta();
1340 let captures = self.captures(&entry.prefix);
1341 let previous = self.current.insert(
1342 presented.clone(),
1343 CursorRoute {
1344 meta: meta.clone(),
1345 served,
1346 prefix: entry.prefix.clone(),
1347 captures: captures.clone(),
1348 },
1349 );
1350 match previous {
1351 Some(previous) if previous.meta == meta && previous.served == served && previous.captures == captures => {}
1358 Some(previous) if previous.captures != captures => {
1361 if let Ok(mut state) = self.state.write() {
1362 state.apply_unannounce(self.under.join(presented), previous.meta, previous.captures);
1363 state.apply_announce(self.under.join(presented), meta, captures);
1364 }
1365 }
1366 _ => {
1367 if let Ok(mut state) = self.state.write() {
1368 state.apply_announce(self.under.join(presented), meta, captures);
1369 }
1370 }
1371 }
1372 }
1373
1374 fn best<'a>(&self, candidates: impl Iterator<Item = &'a RouteEntry>) -> Option<&'a RouteEntry> {
1377 candidates
1378 .filter(|entry| self.visible(entry))
1379 .min_by_key(|entry| (Reverse(entry.prefix.len()), route_order(&entry.prefix, entry)))
1380 }
1381
1382 fn discovers(&self, prefix: &Path) -> bool {
1384 self.hidden.discovers(&self.heads, prefix)
1385 }
1386}
1387
1388#[derive(Clone)]
1391struct OriginScope {
1392 allowed: Patterns,
1395 mounts: Arc<[Mount]>,
1397}
1398
1399#[derive(Clone, Debug)]
1402struct Mount {
1403 at: PathOwned,
1404 target: PathOwned,
1405}
1406
1407impl Mount {
1408 fn resolve(&self, path: &Path) -> Option<PathOwned> {
1411 let resolved = self.target.join(path.strip_prefix(&self.at)?);
1412 (resolved.parts().count() <= Path::MAX_PARTS).then_some(resolved)
1413 }
1414
1415 fn name(&self, path: &Path) -> Option<PathOwned> {
1418 Some(self.at.join(path.strip_prefix(&self.target)?))
1419 }
1420
1421 fn names(&self, path: &Path) -> bool {
1425 path.strip_prefix(&self.target)
1426 .is_none_or(|rest| self.at.parts().count() + rest.parts().count() <= Path::MAX_PARTS)
1427 }
1428
1429 fn translate(&self, patterns: &Patterns) -> Patterns {
1436 let target = self.target.as_str();
1437 patterns
1438 .rebase(self.at.as_str())
1439 .iter()
1440 .filter_map(|member| {
1441 member
1442 .rooted(target)
1443 .or_else(|_| {
1444 let segments = member
1445 .segments()
1446 .iter()
1447 .filter(|segment| **segment != Segment::Globstar);
1448 Pattern::new(segments.cloned())?.rooted(target)
1449 })
1450 .ok()
1451 })
1452 .collect()
1453 }
1454
1455 fn translate_head(&self, head: &Path) -> Option<PathOwned> {
1458 match head.strip_prefix(&self.at) {
1459 Some(rest) => Some(self.target.join(rest)),
1460 None => self.at.has_prefix(head).then(|| self.target.clone()),
1461 }
1462 }
1463}
1464
1465impl OriginScope {
1466 fn empty() -> Self {
1468 Self {
1469 allowed: Patterns::new(),
1470 mounts: Arc::from([]),
1471 }
1472 }
1473
1474 fn narrow(&self, patterns: &Patterns) -> Option<Self> {
1476 let allowed = self.allowed.intersect(patterns).ok()?;
1477 if allowed.is_empty() {
1478 None
1479 } else {
1480 Some(Self {
1481 allowed,
1482 mounts: self.mounts.clone(),
1483 })
1484 }
1485 }
1486
1487 fn mount(&self, path: &Path) -> Option<&Mount> {
1489 self.mounts.iter().find(|mount| path.has_prefix(&mount.at))
1490 }
1491
1492 fn resolve<'a>(&self, path: &'a Path<'a>) -> Option<Path<'a>> {
1495 match self.mount(path) {
1496 Some(mount) => mount.resolve(path),
1497 None => Some(path.borrow()),
1498 }
1499 }
1500
1501 fn publishes(&self, prefix: &Path) -> bool {
1504 self.mount(prefix).is_none()
1505 }
1506
1507 fn permits(&self, path: &Path) -> bool {
1509 self.allowed.matches(path.as_str())
1510 }
1511
1512 fn relative(&self, root: &Path) -> Patterns {
1514 self.allowed.rebase(root.as_str())
1515 }
1516}
1517
1518impl Default for OriginScope {
1519 fn default() -> Self {
1520 Self {
1521 allowed: Patterns::from(Pattern::all()),
1522 mounts: Arc::from([]),
1523 }
1524 }
1525}
1526
1527pub(crate) fn interest_prefixes(allowed: &Patterns) -> Vec<PathOwned> {
1530 let mut heads: Vec<PathOwned> = allowed
1531 .iter()
1532 .map(|pattern| Path::new(pattern.head()).to_owned())
1533 .collect();
1534 heads.sort();
1535 heads.dedup();
1536 let covered = heads.clone();
1537 heads.retain(|head| !covered.iter().any(|other| other != head && head.has_prefix(other)));
1538 heads
1539}
1540
1541#[derive(Clone, Debug, Default, PartialEq, Eq)]
1548pub(crate) struct Hidden {
1549 include: bool,
1551 from: Option<PathOwned>,
1553 beyond: Option<Vec<PathOwned>>,
1556}
1557
1558impl Hidden {
1559 fn discovers(&self, heads: &[PathOwned], prefix: &Path) -> bool {
1561 (self.include || !hides(self.from.as_ref().map(std::slice::from_ref).unwrap_or(heads), prefix))
1562 && self.beyond.as_ref().is_none_or(|outer| hides(outer, prefix))
1563 }
1564
1565 fn translate(&self, mount: &Mount) -> Self {
1569 let beyond = self.beyond.as_ref().and_then(|outer| {
1570 (!hides(outer, &mount.at)).then(|| outer.iter().filter_map(|head| mount.translate_head(head)).collect())
1571 });
1572 Self {
1573 include: self.include,
1574 from: self.from.as_ref().and_then(|head| mount.translate_head(head)),
1575 beyond,
1576 }
1577 }
1578}
1579
1580fn hides(heads: &[PathOwned], prefix: &Path) -> bool {
1583 heads
1584 .iter()
1585 .any(|head| prefix.strip_prefix(head).is_some_and(|below| below.is_hidden()))
1586}
1587
1588#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1590pub enum AnnounceKind {
1591 Announced,
1593 Updated,
1595 Retracted,
1597}
1598
1599impl AnnounceKind {
1600 pub fn is_active(self) -> bool {
1602 !matches!(self, Self::Retracted)
1603 }
1604}
1605
1606#[derive(Clone, Debug)]
1614pub struct AnnounceUpdate {
1615 pub prefix: PathOwned,
1617 pub captures: Option<Vec<Pattern>>,
1620 pub route: Route,
1623 pub kind: AnnounceKind,
1625}
1626
1627#[derive(Clone)]
1629pub struct Producer {
1630 hop: Hop,
1633 max_hops: HopLimit,
1634
1635 scope: OriginScope,
1637
1638 root: PathOwned,
1640
1641 shared: kio::Shared<OriginState>,
1644
1645 pool: cache::Pool,
1648
1649 cache_duration: Duration,
1652
1653 default_max_age: Duration,
1656
1657 stats: stats::Session,
1661
1662 peer: bool,
1665
1666 tasks: Tasks,
1670
1671 timers: Clock,
1673}
1674
1675impl Producer {
1676 pub fn new(config: Config) -> (Self, Driver) {
1683 let (tasks, set) = TaskSet::new();
1684 let scope = OriginScope::default();
1685 let shared = kio::Shared::new(OriginState::new(config.update_hold));
1686 let timers = Clock::default();
1687 let pool = config.pool.clone();
1688 let producer = Self {
1689 hop: config.hop,
1690 max_hops: config.max_hops,
1691 scope: scope.clone(),
1692 root: PathOwned::default(),
1693 shared: shared.clone(),
1694 pool: config.pool,
1695 cache_duration: config.cache_duration,
1696 default_max_age: config.default_max_age,
1697 stats: stats::Session::default(),
1698 peer: false,
1699 tasks,
1700 timers: timers.clone(),
1701 };
1702 let driver = Driver {
1703 state: DriverState {
1704 set,
1705 shared,
1706 done: false,
1707 },
1708 timers,
1709 pool,
1710 };
1711 (producer, driver)
1712 }
1713
1714 pub fn with_stats(mut self, session: stats::Session) -> Self {
1718 self.stats = session;
1719 self
1720 }
1721
1722 pub(crate) fn stats(&self) -> &stats::Session {
1724 &self.stats
1725 }
1726
1727 pub fn peer(mut self) -> Self {
1735 self.peer = true;
1736 self
1737 }
1738
1739 pub fn config(&self) -> Config {
1741 Config {
1742 hop: self.hop,
1743 max_hops: self.max_hops,
1744 pool: self.pool.clone(),
1745 cache_duration: self.cache_duration,
1746 default_max_age: self.default_max_age,
1747 update_hold: self.shared.lock().update_hold,
1748 }
1749 }
1750
1751 pub fn hop(&self) -> Hop {
1753 self.hop
1754 }
1755
1756 pub(crate) fn default_max_age(&self) -> Duration {
1759 self.default_max_age
1760 }
1761
1762 pub(crate) fn max_hops(&self) -> HopLimit {
1764 self.max_hops
1765 }
1766
1767 pub(crate) fn empty(hop: Hop) -> Self {
1772 let (tasks, _) = TaskSet::new();
1775 Self {
1776 hop,
1777 max_hops: HopLimit::default(),
1778 scope: OriginScope::empty(),
1779 root: PathOwned::default(),
1780 shared: kio::Shared::new(OriginState::new(DEFAULT_UPDATE_HOLD)),
1781 pool: cache::Pool::default(),
1782 cache_duration: Duration::MAX,
1783 default_max_age: track::DEFAULT_MAX_AGE,
1784 stats: stats::Session::default(),
1785 peer: false,
1786 tasks,
1787 timers: Clock::default(),
1788 }
1789 }
1790
1791 pub fn create_broadcast(&self, path: impl AsPath) -> Result<broadcast::Producer, Error> {
1822 let path = path.as_path();
1823
1824 let full = self.root.join(&path).to_owned();
1825 if !self.scope.permits(&full) || !self.scope.publishes(&full) {
1826 return Err(Error::Unauthorized);
1827 }
1828 if full.parts().count() > Path::MAX_PARTS {
1832 return Err(BoundsExceeded.into());
1833 }
1834 let claim = prefix_claim(&full)?;
1837
1838 let ingress = self.stats.ingress(&full);
1840
1841 let announcing = Announcing {
1846 hop: self.hop,
1847 shared: self.shared.clone(),
1848 requested: full.clone(),
1849 prefixes: vec![(full.clone(), claim)],
1850 scope: self.scope.allowed.clone(),
1851 local: true,
1852 peer: self.peer,
1853 stats: self.stats.clone(),
1854 };
1855 let info = broadcast::Info {
1856 pool: self.pool.clone(),
1857 cache_duration: self.cache_duration,
1858 path: full,
1859 };
1860 let source = info.produce().with_stats(ingress.clone());
1861 let entry = announcing.announce(
1862 Route::default(),
1863 Serving {
1864 server: None,
1865 source: Some(source.consume()),
1866 advertised: false,
1867 },
1868 )?;
1869 Ok(source.with_announcer(Announcer {
1870 entry,
1871 ingress,
1872 _keepalive: self.tasks.keepalive(),
1873 }))
1874 }
1875
1876 pub fn publish(&self, path: impl AsPath, route: Route) -> Result<broadcast::Producer, Error> {
1878 let broadcast = self.create_broadcast(path)?;
1879 broadcast.announce(route)?;
1880 Ok(broadcast)
1881 }
1882
1883 pub(crate) fn create_source(&self, path: impl AsPath) -> broadcast::Producer {
1889 let path = path.as_path();
1890 let full = self.root.join(&path).to_owned();
1891 let ingress = self.stats.ingress(&full);
1892 broadcast::Info {
1893 pool: self.pool.clone(),
1894 cache_duration: self.cache_duration,
1895 path: full,
1896 }
1897 .produce()
1898 .with_stats(ingress)
1899 }
1900
1901 #[cfg(test)]
1910 pub(crate) fn announce(&self, prefix: impl AsPath, route: Route) -> Result<AnnounceProducer, Error> {
1911 Announcing::new(self, prefix)?.announce(
1912 route,
1913 Serving {
1914 server: None,
1915 source: None,
1916 advertised: true,
1917 },
1918 )
1919 }
1920
1921 pub fn dynamic(&self, prefix: impl AsPath, route: Route) -> Result<Dynamic, Error> {
1943 let announcing = Announcing::new(self, prefix)?;
1944 let serve = kio::Shared::<ServeState>::default();
1945 serve.lock().requests.add_handler();
1946 let announcement = announcing.announce(
1947 route,
1948 Serving {
1949 server: Some(serve.clone()),
1950 source: None,
1951 advertised: true,
1952 },
1953 )?;
1954 Ok(Dynamic {
1955 announcement,
1956 state: serve,
1957 _keepalive: self.tasks.keepalive(),
1958 })
1959 }
1960
1961 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Producer, Error> {
1968 let root = self.root.join(root).to_owned();
1969 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
1970 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
1971 Ok(Producer {
1972 hop: self.hop,
1973 max_hops: self.max_hops,
1974 scope,
1975 root,
1976 shared: self.shared.clone(),
1977 pool: self.pool.clone(),
1978 cache_duration: self.cache_duration,
1979 default_max_age: self.default_max_age,
1980 stats: self.stats.clone(),
1981 peer: self.peer,
1982 tasks: self.tasks.clone(),
1983 timers: self.timers.clone(),
1984 })
1985 }
1986
1987 pub fn mount(&self, at: impl AsPath, target: impl AsPath) -> Result<Producer, Error> {
2006 let at = self.root.join(at).to_owned();
2007 let target = self.root.join(target).to_owned();
2008 if [&at, &target]
2009 .into_iter()
2010 .any(|path| path.parts().count() > Path::MAX_PARTS)
2011 {
2012 return Err(BoundsExceeded.into());
2013 }
2014 Pattern::literal(at.as_str())?;
2016 if !self
2017 .scope
2018 .allowed
2019 .covers(&Patterns::from(Pattern::subtree(target.as_str())?))
2020 {
2021 return Err(Error::Unauthorized);
2022 }
2023 let overlaps = |a: &Path, b: &Path| a.has_prefix(b) || b.has_prefix(a);
2027 if overlaps(&at, &target)
2028 || self
2029 .scope
2030 .mounts
2031 .iter()
2032 .any(|mount| overlaps(&at, &mount.at) || overlaps(&target, &mount.at) || overlaps(&at, &mount.target))
2033 {
2034 return Err(Error::Duplicate);
2035 }
2036 let mounts = self
2037 .scope
2038 .mounts
2039 .iter()
2040 .cloned()
2041 .chain([Mount { at, target }])
2042 .collect();
2043 Ok(Producer {
2044 scope: OriginScope {
2045 allowed: self.scope.allowed.clone(),
2046 mounts,
2047 },
2048 ..self.clone()
2049 })
2050 }
2051
2052 pub fn consume(&self) -> Consumer {
2057 Consumer::from_producer(self, stats::Session::default())
2060 }
2061
2062 pub fn root(&self) -> &Path<'_> {
2064 &self.root
2065 }
2066
2067 pub fn allowed(&self) -> Patterns {
2069 self.scope.relative(&self.root)
2070 }
2071
2072 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
2074 self.root.join(path)
2075 }
2076}
2077
2078struct Announcing {
2082 hop: Hop,
2083 shared: kio::Shared<OriginState>,
2084 requested: PathOwned,
2086 prefixes: Vec<(PathOwned, Pattern)>,
2090 scope: Patterns,
2092 local: bool,
2093 peer: bool,
2095 stats: stats::Session,
2096}
2097
2098impl Announcing {
2099 fn new(producer: &Producer, prefix: impl AsPath) -> Result<Self, Error> {
2101 let requested = producer.root.join(prefix.as_path()).to_owned();
2102 if requested.parts().count() > Path::MAX_PARTS {
2103 return Err(BoundsExceeded.into());
2104 }
2105 let claim = prefix_claim(&requested)?;
2106 if !producer.scope.allowed.overlaps(&claim) || !producer.scope.publishes(&requested) {
2107 return Err(Error::Unauthorized);
2108 }
2109 Ok(Self {
2110 hop: producer.hop,
2111 shared: producer.shared.clone(),
2112 requested: requested.clone(),
2113 prefixes: vec![(requested, claim)],
2114 scope: producer.scope.allowed.clone(),
2115 local: false,
2116 peer: producer.peer,
2117 stats: producer.stats.clone(),
2118 })
2119 }
2120
2121 fn announce(&self, route: Route, serving: Serving) -> Result<AnnounceProducer, Error> {
2122 debug_assert!(
2123 !route.hops.contains(&self.hop),
2124 "announce called with a looping hop chain",
2125 );
2126
2127 let via = route.via;
2128
2129 let mut shared = self.shared.lock();
2130 if shared.closed {
2131 return Err(Error::Closed);
2132 }
2133
2134 let mut entries = Vec::with_capacity(self.prefixes.len());
2135 for (prefix, claim) in &self.prefixes {
2136 shared.reannounced(prefix, &route.hops);
2137 let id = shared.next_route;
2138 shared.next_route += 1;
2139 let stale = shared.withdrawn_through(prefix, &route.hops);
2140 shared.routes.insert(RouteEntry {
2141 id,
2142 prefix: prefix.clone(),
2143 scope: self.scope.clone(),
2144 hops: route.hops.clone(),
2145 cost: route.cost,
2146 via,
2147 local: self.local,
2148 peer: self.peer,
2149 server: serving.server.clone(),
2150 source: serving.source.clone(),
2151 advertised: serving.advertised,
2152 stale,
2153 claim: claim.clone(),
2154 reach: reach(&self.scope, claim),
2155 });
2156 shared.sync_route(prefix, claim);
2157 entries.push((prefix.clone(), id));
2158 }
2159 drop(shared);
2160
2161 let guard = serving
2163 .advertised
2164 .then(|| self.stats.ingress(&self.requested).announce());
2165
2166 Ok(AnnounceProducer {
2167 shared: self.shared.clone(),
2168 entries,
2169 guard,
2170 })
2171 }
2172}
2173
2174struct Serving {
2176 server: Option<kio::Shared<ServeState>>,
2177 source: Option<broadcast::Consumer>,
2178 advertised: bool,
2179}
2180
2181pub(crate) struct Announcer {
2188 entry: AnnounceProducer,
2189 ingress: stats::Scope,
2191 _keepalive: Keepalive,
2195}
2196
2197impl Announcer {
2198 pub(crate) fn announce(&mut self, route: Route) -> Result<(), Error> {
2200 self.entry.update(route)?;
2201 if self.entry.guard.is_none() {
2202 self.entry.guard = Some(self.ingress.announce());
2203 }
2204 Ok(())
2205 }
2206
2207 pub(crate) fn withdraw(&mut self) {
2209 self.entry.withdraw();
2210 self.entry.guard = None;
2211 }
2212}
2213
2214#[must_use = "dropping an announcement retracts the route"]
2221pub(crate) struct AnnounceProducer {
2222 shared: kio::Shared<OriginState>,
2223 entries: Vec<(PathOwned, u64)>,
2226 guard: Option<stats::Announce>,
2228}
2229
2230impl AnnounceProducer {
2231 pub fn update(&self, route: Route) -> Result<(), Error> {
2239 let mut shared = self.shared.lock();
2240 if shared.closed {
2241 return Err(Error::Closed);
2242 }
2243 for (prefix, id) in &self.entries {
2244 shared.reannounced(prefix, &route.hops);
2245 let stale = shared.withdrawn_through(prefix, &route.hops);
2246 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2248 return Err(Error::Closed);
2249 };
2250 if entry.hops.iter().next() != route.hops.iter().next()
2253 && let Some(server) = &entry.server
2254 {
2255 drop(std::mem::take(&mut server.lock().served));
2256 }
2257 entry.hops = route.hops.clone();
2258 entry.stale = stale;
2259 entry.cost = route.cost;
2260 entry.via = route.via;
2261 entry.advertised = true;
2262 let claim = entry.claim.clone();
2263 shared.sync_route(prefix, &claim);
2264 shared.prune_withdrawn(prefix);
2265 }
2266 Ok(())
2267 }
2268
2269 fn withdraw(&self) {
2274 let mut shared = self.shared.lock();
2275 for (prefix, id) in &self.entries {
2276 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2277 continue;
2278 };
2279 if !entry.advertised {
2280 continue;
2281 }
2282 entry.advertised = false;
2283 let claim = entry.claim.clone();
2284 shared.sync_route(prefix, &claim);
2285 }
2286 }
2287
2288 fn retract(&self, withdrawn: bool) {
2291 let mut shared = self.shared.lock();
2292 for (prefix, id) in &self.entries {
2293 let Some(entry) = shared.routes.remove(prefix, *id) else {
2294 continue;
2295 };
2296 if let Some(server) = &entry.server {
2299 let mut server = server.lock();
2300 server.closed = true;
2301 for producer in server.requests.drain_all() {
2302 if let Ok(mut request) = producer.write() {
2303 request.resolved.get_or_insert(Err(Error::Unroutable));
2304 }
2305 }
2306 }
2307 if withdrawn
2312 && let Some(&peer) = entry.hops.iter().last()
2313 && peer != Hop::UNKNOWN
2314 && !shared
2315 .routes
2316 .at(prefix)
2317 .any(|other| other.live() && other.hops.iter().last() == Some(&peer))
2318 {
2319 shared.withdrawn.entry(prefix.clone()).or_default().insert(peer);
2320 shared.restale(prefix);
2321 }
2322 shared.sync_route(&entry.prefix, &entry.claim);
2323 shared.prune_withdrawn(&entry.prefix);
2324 }
2325 }
2326}
2327
2328impl Drop for AnnounceProducer {
2329 fn drop(&mut self) {
2330 self.retract(false);
2331 }
2332}
2333
2334#[must_use = "poll the driver or the origin makes no progress"]
2349pub struct Driver {
2350 state: DriverState,
2351 timers: Clock,
2353 pool: cache::Pool,
2356}
2357
2358struct DriverState {
2360 set: TaskSet,
2362 shared: kio::Shared<OriginState>,
2365 done: bool,
2367}
2368
2369impl Driver {
2370 pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2377 self.timers.advance(now);
2378 self.timers.register_driver(waiter);
2379 let result = self.state.poll(waiter);
2380 let gc = self.pool.gc(now);
2381 if result.is_ready() {
2382 return Err(Error::Closed);
2383 }
2384 Ok(self.timers.timeout().into_iter().chain(gc).min())
2385 }
2386}
2387
2388impl crate::time::Driver for Driver {
2389 fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2390 self.poll(now, waiter)
2391 }
2392}
2393
2394impl DriverState {
2395 fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
2396 if !self.done {
2400 ready!(self.set.poll(waiter));
2401 self.done = true;
2402 }
2403 Poll::Ready(())
2404 }
2405
2406 fn teardown(&mut self) {
2410 drop(std::mem::replace(&mut self.set, TaskSet::owned()));
2413
2414 let (servers, cursors, fronts) = {
2419 let mut shared = self.shared.lock();
2420 shared.closed = true;
2421 shared.routes.poke_all();
2423 let servers: Vec<_> = shared
2424 .routes
2425 .entries()
2426 .filter_map(|entry| entry.server.clone())
2427 .collect();
2428 let cursors: Vec<_> = shared.cursors.values().map(|cursor| cursor.state.clone()).collect();
2429 let fronts: Vec<_> = shared.fronts.values().map(|front| front.request.clone()).collect();
2430 (servers, cursors, fronts)
2431 };
2432
2433 for producer in fronts {
2436 if let Ok(mut request) = producer.write() {
2437 request.resolved.get_or_insert(Err(Error::Dropped));
2438 }
2439 }
2440 for server in servers {
2444 let mut server = server.lock();
2445 server.closed = true;
2446 for producer in server.requests.drain_all() {
2447 if let Ok(mut request) = producer.write() {
2448 request.resolved.get_or_insert(Err(Error::Dropped));
2449 }
2450 }
2451 }
2452
2453 for state in cursors {
2456 if let Ok(mut state) = state.write() {
2457 state.end();
2458 }
2459 }
2460 }
2461}
2462
2463impl Drop for DriverState {
2464 fn drop(&mut self) {
2465 self.teardown();
2466 }
2467}
2468
2469const TRACK_IDLE_LINGER: Duration = Duration::from_secs(30);
2483
2484struct WarmCopy {
2490 track: track::Producer,
2491 _dynamic: track::Dynamic,
2492 edge: Option<WarmGroup>,
2495}
2496
2497impl Drop for WarmCopy {
2498 fn drop(&mut self) {
2499 let _ = self.track.finish();
2500 }
2501}
2502
2503struct WarmGroup(group::Producer);
2510
2511impl Drop for WarmGroup {
2512 fn drop(&mut self) {
2513 if !self.0.is_finished() {
2514 let _ = self.0.clone().abort(Error::Cancel);
2515 }
2516 }
2517}
2518
2519struct HeldGroup(group::Producer);
2524
2525impl Drop for HeldGroup {
2526 fn drop(&mut self) {
2527 if !self.0.is_finished() {
2528 let _ = self.0.abort_if_last(Error::Cancel);
2529 }
2530 }
2531}
2532
2533fn warm_copy(source: &track::Consumer, head: Option<&WarmGroup>, held: Option<&group::Producer>) -> Option<WarmCopy> {
2541 let info = source.cached_info()?;
2542 let mut track = track::Producer::new(Arc::new(source.broadcast().clone()), source.name(), info);
2543 let head = head.map(|head| &head.0);
2544 let mut groups = source.cached_groups();
2545 if let Some(held) = held
2546 && !groups.iter().any(|(group, _)| group.sequence == held.sequence)
2547 {
2548 groups.push((held.clone(), true));
2549 }
2550 let latest = groups
2552 .iter()
2553 .filter(|(_, visible)| *visible)
2554 .map(|(group, _)| group.sequence)
2555 .max();
2556 let mut edge = None;
2557 let mut first = None;
2558
2559 if let Some(head) = head
2563 && !groups.iter().any(|(group, _)| group.sequence == head.sequence)
2564 {
2565 let is_latest = latest.is_none_or(|latest| head.sequence >= latest);
2566 if head.is_finished() {
2567 let _ = track.adopt_group(head.clone(), true);
2568 first = Some(head.sequence);
2569 if is_latest {
2570 edge = Some(WarmGroup(head.clone()));
2571 }
2572 } else if is_latest {
2573 edge = warm_rebuild(&track, head, None);
2574 first = edge.as_ref().map(|_| head.sequence);
2575 }
2576 }
2577
2578 for (group, visible) in groups {
2579 let finished = group.is_finished();
2580 let whole = group.live_first_frame() == Some(0);
2581 let is_latest = visible && Some(group.sequence) == latest;
2582 let warm = if finished && whole {
2583 let _ = track.adopt_group(group.clone(), visible);
2584 Some(WarmGroup(group))
2585 } else if (finished || is_latest) && (whole || head.is_some_and(|head| head.sequence == group.sequence)) {
2586 warm_rebuild(&track, &group, head)
2592 } else {
2593 None
2595 };
2596 if visible && let Some(warm) = &warm {
2597 first = Some(first.map_or(warm.0.sequence, |first: u64| first.min(warm.0.sequence)));
2598 }
2599 if is_latest {
2600 edge = warm;
2601 }
2602 }
2603 track.start_at(first).ok()?;
2607 let dynamic = track.dynamic();
2608 Some(WarmCopy {
2609 track,
2610 _dynamic: dynamic,
2611 edge,
2612 })
2613}
2614
2615fn warm_rebuild(track: &track::Producer, live: &group::Producer, head: Option<&group::Producer>) -> Option<WarmGroup> {
2618 let mut start = live.live_first_frame()? as u64;
2620 let mut tail = live.consume();
2621 tail.start_at(start);
2622 let mut frames = Vec::new();
2623 if start > 0
2624 && let Some(head) = head.filter(|head| head.sequence == live.sequence)
2625 && let Some(head_start) = head.live_first_frame()
2626 {
2627 let mut head = head.consume();
2628 head.start_at(head_start as u64);
2629 while head.index() < start {
2630 match head.poll_read_frame(&kio::Waiter::noop()) {
2631 Poll::Ready(Ok(Some(frame))) => frames.push(frame),
2632 _ => break,
2633 }
2634 }
2635 match head.index() == start {
2636 true => start = head_start as u64,
2637 false => frames.clear(),
2639 }
2640 }
2641 while let Poll::Ready(Ok(Some(frame))) = tail.poll_read_frame(&kio::Waiter::noop()) {
2642 frames.push(frame);
2643 }
2644 if frames.is_empty() {
2645 return None;
2646 }
2647
2648 let rebuilt = WarmGroup(
2650 track
2651 .create_group(group::Info {
2652 sequence: live.sequence,
2653 })
2654 .ok()?,
2655 );
2656 let mut writer = rebuilt.0.clone();
2657 if start > 0 {
2658 writer.start_at(start).ok()?;
2659 }
2660 for frame in frames {
2661 writer.write_frame(frame.timestamp, frame.payload).ok()?;
2662 }
2663 if live.is_finished() {
2664 writer.finish().ok()?;
2665 }
2666 Some(rebuilt)
2667}
2668
2669struct FrontTask {
2671 shared: kio::Shared<OriginState>,
2673 broadcast: broadcast::Producer,
2675 path: PathOwned,
2677 horizon: Horizon,
2679 watch: Watch,
2681 request: kio::Producer<PendingBroadcast>,
2683 pin: kio::Lock<Pin>,
2685 timers: Clock,
2686}
2687
2688struct TrackIo {
2691 resume: super::resume::Producer,
2692 staged: Option<(u64, track::Consumer)>,
2694 query: Option<(u64, track::Consumer, track::Querying)>,
2696 copy: Option<(u64, track::Consumer)>,
2698 edge: Option<track::Position>,
2703 warm: Option<WarmCopy>,
2706 head: Option<WarmGroup>,
2709 prior: Option<(u64, WarmCopy)>,
2714 held: Option<HeldGroup>,
2719 dead: Option<track::Consumer>,
2722 used: bool,
2724}
2725
2726impl TrackIo {
2727 fn end(&mut self) {
2730 self.staged = None;
2731 self.query = None;
2732 self.copy = None;
2733 self.warm = None;
2734 self.head = None;
2735 self.prior = None;
2736 self.held = None;
2737 self.dead = None;
2738 }
2739
2740 fn hold(&mut self, group: group::Producer) {
2743 if self.prior.as_ref().is_some_and(|(edge, _)| group.sequence > *edge) {
2744 self.prior = None;
2745 }
2746 self.held = Some(HeldGroup(group));
2747 }
2748
2749 fn bury(&mut self) {
2752 let Some((_, copy)) = self.copy.take() else { return };
2753 let after = self.held.as_ref().map(|held| held.0.sequence);
2754 if let Poll::Ready(group) = copy.poll_latest_group(after, &kio::Waiter::noop()) {
2755 self.hold(group);
2756 }
2757 self.dead = Some(copy);
2758 }
2759
2760 fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> {
2766 let Some(held) = self.held.take() else {
2767 return Ok(());
2768 };
2769 let Some(warm) = warm_copy(copy, self.head.as_ref(), Some(&held.0)) else {
2770 return Ok(());
2771 };
2772 self.head = None;
2773 self.resume.park(&warm.track)?;
2774 self.warm = Some(warm);
2775 Ok(())
2776 }
2777}
2778
2779async fn run_front(task: FrontTask) {
2783 let FrontTask {
2784 shared,
2785 broadcast,
2786 path,
2787 horizon,
2788 watch,
2789 request,
2790 pin,
2791 timers,
2792 } = task;
2793
2794 enum Step {
2796 Assigned(Arc<str>, super::resume::Producer),
2797 Resolved(u64, Result<broadcast::Consumer, Error>),
2798 SourceClosed(u64),
2799 Info(Arc<str>, u64, Result<track::Info, Error>),
2800 Ended(Arc<str>, u64, Result<(), Error>),
2801 Held(Arc<str>, group::Producer),
2802 Demand(Arc<str>),
2803 Deadline,
2804 Table,
2805 }
2806
2807 let mut front = Front::new(TRACK_IDLE_LINGER);
2808 let mut sources: HashMap<u64, broadcast::Consumer> = HashMap::new();
2809 let mut next_source = 0u64;
2810 let mut upstream: Option<(u64, kio::Consumer<PendingBroadcast>)> = None;
2812 let mut tracks: HashMap<Arc<str>, TrackIo> = HashMap::new();
2813 let mut deadline = crate::runtime::Deadline::new(&timers);
2814 let mut seen = 0;
2816 let mut events: VecDeque<Event> = VecDeque::new();
2817
2818 let select = |front: &mut Front, sources: &HashMap<u64, broadcast::Consumer>, seen: &mut u64| -> Event {
2821 let table = shared.read();
2822 if table.closed {
2823 return Event::Closed;
2824 }
2825 *seen = watch.seen();
2827 front.retain_routes(|route| table.routes.covers(&path.as_path(), route));
2828 let best = table
2829 .best_route(&path.as_path(), horizon, front.pin(), front.refused_routes())
2830 .map(|entry| Candidate {
2831 route: entry.id,
2832 first: entry.hops.iter().next().copied(),
2833 local: entry.local,
2834 });
2835 let serving_closing = front
2836 .serving()
2837 .and_then(|id| sources.get(&id))
2838 .is_some_and(|source| source.is_closing());
2839 Event::Selected { best, serving_closing }
2840 };
2841
2842 events.push_back(select(&mut front, &sources, &mut seen));
2843
2844 loop {
2845 while let Some(event) = events.pop_front() {
2846 for action in front.step(event) {
2847 match action {
2848 Action::Reselect => events.push_back(select(&mut front, &sources, &mut seen)),
2849 Action::Request { route } => {
2850 let found = {
2852 let table = shared.read();
2853 table
2854 .routes
2855 .covering(&path.as_path())
2856 .find(|entry| entry.id == route && entry.live())
2857 .map(|entry| {
2858 (
2859 Candidate {
2860 route,
2861 first: entry.hops.iter().next().copied(),
2862 local: entry.local,
2863 },
2864 entry.source.clone(),
2865 entry.server.clone(),
2866 )
2867 })
2868 };
2869 let Some((candidate, source, server)) = found else {
2870 events.push_back(Event::Resolved {
2871 route,
2872 result: Err(Refusal {
2873 err: Error::Unroutable,
2874 standing: false,
2875 }),
2876 });
2877 continue;
2878 };
2879 front.identify(candidate);
2880 *pin.lock() = front.pin();
2881 if let Some(source) = source {
2882 let id = next_source;
2883 next_source += 1;
2884 sources.insert(id, source);
2885 events.push_back(Event::Resolved { route, result: Ok(id) });
2886 continue;
2887 }
2888 let Some(server) = server else {
2889 events.push_back(Event::Resolved {
2890 route,
2891 result: Err(Refusal {
2892 err: Error::Unroutable,
2893 standing: true,
2894 }),
2895 });
2896 continue;
2897 };
2898 let mut serve = server.lock();
2899 if serve.closed {
2900 drop(serve);
2903 events.push_back(Event::Resolved {
2904 route,
2905 result: Err(Refusal {
2906 err: Error::Unroutable,
2907 standing: true,
2908 }),
2909 });
2910 continue;
2911 }
2912 if let Some(weak) = serve.served.get(&path) {
2915 drop(serve);
2916 let id = next_source;
2917 next_source += 1;
2918 sources.insert(id, weak.consume());
2919 events.push_back(Event::Resolved { route, result: Ok(id) });
2920 continue;
2921 }
2922 let pending = match serve.requests.join(&path) {
2923 Some(producer) => producer.consume(),
2924 None => {
2925 let producer = kio::Producer::<PendingBroadcast>::default();
2926 let consumer = producer.consume();
2927 match serve.requests.insert(path.clone(), producer) {
2928 Ok(()) => consumer,
2929 Err(_) => {
2932 drop(serve);
2933 events.push_back(Event::Resolved {
2934 route,
2935 result: Err(Refusal {
2936 err: Error::Unroutable,
2937 standing: true,
2938 }),
2939 });
2940 continue;
2941 }
2942 }
2943 }
2944 };
2945 upstream = Some((route, pending));
2946 }
2947 Action::Detach { source } => {
2948 sources.remove(&source);
2949 for io in tracks.values_mut() {
2952 if io.copy.as_ref().is_some_and(|(s, _)| *s == source) {
2953 io.bury();
2954 }
2955 if io.query.as_ref().is_some_and(|(s, ..)| *s == source) {
2956 io.query = None;
2957 }
2958 if io.staged.as_ref().is_some_and(|(s, _)| *s == source) {
2959 io.staged = None;
2960 }
2961 }
2962 }
2963 Action::Resolve => {
2964 if let Ok(mut pending) = request.write() {
2965 pending.resolved.get_or_insert(Ok(broadcast.consume()));
2966 }
2967 }
2968 Action::Query { track: name, source } => {
2969 let Some(io) = tracks.get_mut(&name) else { continue };
2970 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
2971 match sources.get(&source).map(|s| s.track(&name)) {
2972 Some(Ok(copy)) => {
2973 let query = copy.query().into_inner();
2976 io.query = Some((source, copy, query));
2977 }
2978 Some(Err(err)) => events.push_back(Event::TrackInfo {
2979 track: name,
2980 source,
2981 closing,
2982 result: Err(err),
2983 }),
2984 None => {}
2985 }
2986 }
2987 Action::Splice { track: name, source } => {
2988 let Some(io) = tracks.get_mut(&name) else { continue };
2989 let Some((staged, copy)) = io.staged.take() else {
2990 continue;
2991 };
2992 if staged != source {
2993 continue;
2994 }
2995 let outgoing = io.copy.take().map(|(_, copy)| copy).or_else(|| io.dead.take());
2996 let retired = match outgoing {
2997 Some(outgoing) => io.retire(&outgoing),
2998 None => Ok(()),
2999 };
3000 if let Err(err) = retired.and_then(|()| io.resume.takeover(©)) {
3001 let _ = io.resume.abort(err);
3005 tracks.remove(&name);
3006 continue;
3007 }
3008 if let Some(mut warm) = io.warm.take()
3009 && let Some(head) = warm.edge.take()
3010 {
3011 io.prior = Some((head.0.sequence, warm));
3012 io.head = Some(head);
3013 }
3014 io.edge = io.resume.resume_position();
3017 io.held = None;
3018 io.copy = Some((source, copy));
3019 }
3020 Action::Park { track: name } => {
3021 let Some(io) = tracks.get_mut(&name) else { continue };
3022 let Some((_, copy)) = io.copy.take() else { continue };
3023 let warm = warm_copy(©, io.head.as_ref(), io.held.as_ref().map(|held| &held.0));
3027 drop(copy);
3028 io.head = None;
3029 io.prior = None;
3030 io.held = None;
3031 io.dead = None;
3032 let parked = match &warm {
3033 Some(warm) => io.resume.park(&warm.track),
3034 None => io.resume.release(),
3035 };
3036 if parked.is_err() {
3037 tracks.remove(&name);
3038 continue;
3039 }
3040 io.warm = warm;
3041 }
3042 Action::Forget { track: name } => {
3043 if let Some(io) = tracks.get_mut(&name)
3048 && !broadcast.forget_spliced(&name, &io.resume)
3049 {
3050 if !io.used {
3051 io.used = true;
3052 events.push_back(Event::Used { track: name });
3053 }
3054 continue;
3055 }
3056 tracks.remove(&name);
3057 events.push_back(Event::Forgotten { track: name });
3058 }
3059 Action::Finish { track: name } => {
3060 if let Some(io) = tracks.get_mut(&name) {
3061 io.end();
3062 let _ = io.resume.finish();
3063 }
3064 }
3065 Action::Abort { track: name, err } => {
3066 if let Some(io) = tracks.get_mut(&name) {
3067 tracing::debug!(name = %name, %err, "aborting track");
3068 io.end();
3069 let _ = io.resume.abort(err);
3070 }
3071 }
3072 Action::Arm { at } => deadline.set(at),
3073 Action::End { err } => {
3074 if let Ok(mut pending) = request.write() {
3075 pending.resolved.get_or_insert(Err(err.clone()));
3076 }
3077 broadcast.close();
3084 broadcast.release_spliced(err.clone());
3085 for (_, mut io) in tracks.drain() {
3086 let waiting = io.staged.take().map(|(_, copy)| copy);
3090 let waiting = waiting.or_else(|| io.query.take().map(|(_, copy, _)| copy));
3091 if let Some(copy) = waiting
3092 && io.resume.is_used()
3093 {
3094 if io.resume.takeover(©).is_err() {
3095 continue;
3096 }
3097 io.warm = None;
3098 }
3099 if !io.resume.is_used() || !io.resume.is_spliced() || io.warm.is_some() {
3101 let _ = io.resume.abort(err.clone());
3102 }
3103 }
3104 return;
3105 }
3106 }
3107 }
3108 }
3109
3110 let step = kio::wait(|waiter| {
3111 if let Poll::Ready((name, resume)) = broadcast.poll_spliced_assigned(waiter) {
3112 return Poll::Ready(Step::Assigned(name, resume));
3113 }
3114 if let Some((route, pending)) = &upstream
3115 && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved {
3116 Some(result) => Poll::Ready(result.clone()),
3117 None => Poll::Pending,
3118 }) {
3119 return Poll::Ready(Step::Resolved(
3120 *route,
3121 match result {
3122 Ok(resolved) => resolved,
3123 Err(_closed) => Err(Error::Unroutable),
3126 },
3127 ));
3128 }
3129 if let Some(id) = front.serving()
3130 && let Some(source) = sources.get(&id)
3131 && source.poll_closed(waiter).is_ready()
3132 {
3133 return Poll::Ready(Step::SourceClosed(id));
3134 }
3135 for (name, io) in &tracks {
3136 if let Some((source, _, query)) = &io.query
3137 && let Poll::Ready(result) = query.poll(waiter)
3138 {
3139 return Poll::Ready(Step::Info(name.clone(), *source, result));
3140 }
3141 if let Some((source, copy)) = &io.copy
3142 && let Poll::Ready(result) = copy.poll_complete(waiter)
3143 {
3144 return Poll::Ready(Step::Ended(name.clone(), *source, result));
3145 }
3146 if let Some((_, copy)) = &io.copy
3147 && let Poll::Ready(group) =
3148 copy.poll_latest_group(io.held.as_ref().map(|held| held.0.sequence), waiter)
3149 {
3150 return Poll::Ready(Step::Held(name.clone(), group));
3151 }
3152 let edge = match io.used {
3154 true => io.resume.poll_unused(waiter),
3155 false => io.resume.poll_used(waiter),
3156 };
3157 if edge.is_ready() {
3158 return Poll::Ready(Step::Demand(name.clone()));
3159 }
3160 }
3161 if deadline.poll(waiter).is_ready() {
3162 return Poll::Ready(Step::Deadline);
3163 }
3164 watch.poll_changed(waiter, seen).map(|()| Step::Table)
3165 })
3166 .await;
3167
3168 let event = match step {
3169 Step::Assigned(name, resume) => {
3170 tracks.insert(
3171 name.clone(),
3172 TrackIo {
3173 resume,
3174 staged: None,
3175 query: None,
3176 copy: None,
3177 edge: None,
3178 warm: None,
3179 head: None,
3180 prior: None,
3181 held: None,
3182 dead: None,
3183 used: false,
3184 },
3185 );
3186 Event::TrackAssigned {
3187 track: name,
3188 now: timers.now(),
3189 }
3190 }
3191 Step::Resolved(route, result) => {
3192 upstream = None;
3193 match result {
3194 Ok(source) => {
3195 let id = next_source;
3196 next_source += 1;
3197 sources.insert(id, source);
3198 Event::Resolved { route, result: Ok(id) }
3199 }
3200 Err(err) => {
3201 let standing =
3205 !matches!(err, Error::Unroutable) || shared.read().routes.covers(&path.as_path(), route);
3206 Event::Resolved {
3207 route,
3208 result: Err(Refusal { err, standing }),
3209 }
3210 }
3211 }
3212 }
3213 Step::SourceClosed(source) => Event::SourceClosed { source },
3214 Step::Info(name, source, result) => {
3215 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3216 let Some(io) = tracks.get_mut(&name) else { continue };
3217 let Some((_, copy, _)) = io.query.take() else { continue };
3218 let result = match result {
3221 Ok(info) => match copy.poll_complete(&kio::Waiter::noop()) {
3222 Poll::Ready(Err(err)) => Err(err),
3223 _ => Ok(info),
3224 },
3225 Err(err) => Err(err),
3226 };
3227 if result.is_ok() && io.used {
3230 io.staged = Some((source, copy));
3231 }
3232 Event::TrackInfo {
3233 track: name,
3234 source,
3235 closing,
3236 result,
3237 }
3238 }
3239 Step::Ended(name, source, result) => {
3240 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3241 let Some(io) = tracks.get_mut(&name) else { continue };
3242 io.bury();
3243 let delivered = io.resume.resume_position() != io.edge;
3244 Event::TrackEnded {
3245 track: name,
3246 source,
3247 closing,
3248 result,
3249 delivered,
3250 }
3251 }
3252 Step::Held(name, group) => {
3253 if let Some(io) = tracks.get_mut(&name) {
3254 io.hold(group);
3255 }
3256 continue;
3257 }
3258 Step::Demand(name) => {
3259 let Some(io) = tracks.get_mut(&name) else { continue };
3260 io.used = io.resume.is_used();
3261 if !io.used {
3262 io.query = None;
3265 io.staged = None;
3266 }
3267 match io.used {
3268 true => Event::Used { track: name },
3269 false => Event::Unused {
3270 track: name,
3271 now: timers.now(),
3272 },
3273 }
3274 }
3275 Step::Deadline => {
3276 deadline.set(None);
3279 Event::Deadline { now: timers.now() }
3280 }
3281 Step::Table => select(&mut front, &sources, &mut seen),
3282 };
3283 events.push_back(event);
3284 }
3285}
3286
3287#[derive(Default)]
3292struct RouteTable {
3293 root: RouteNode,
3294}
3295
3296#[derive(Default)]
3299struct RouteNode {
3300 entries: Vec<RouteEntry>,
3302 cursors: Vec<ConsumerId>,
3304 cursors_below: usize,
3307 watches: Vec<(u64, kio::Producer<Watched>)>,
3310 watches_below: usize,
3313 children: HashMap<String, RouteNode>,
3314}
3315
3316#[derive(Default)]
3319struct Watched {
3320 generation: u64,
3321}
3322
3323struct Watch {
3329 shared: kio::Shared<OriginState>,
3330 path: PathOwned,
3331 id: u64,
3332 signal: kio::Consumer<Watched>,
3333}
3334
3335impl Watch {
3336 fn seen(&self) -> u64 {
3340 self.signal.read().generation
3341 }
3342
3343 fn poll_changed(&self, waiter: &kio::Waiter, seen: u64) -> Poll<()> {
3345 self.signal
3346 .poll(waiter, |watched| match watched.generation != seen {
3347 true => Poll::Ready(()),
3348 false => Poll::Pending,
3349 })
3350 .map(|_| ())
3351 }
3352}
3353
3354impl Drop for Watch {
3355 fn drop(&mut self) {
3356 self.shared.lock().routes.remove_watch(&self.path, self.id);
3357 }
3358}
3359
3360#[derive(Clone, Copy)]
3362struct Below {
3363 cursors: usize,
3364 watches: usize,
3365}
3366
3367impl Below {
3368 const NONE: Self = Self { cursors: 0, watches: 0 };
3369 const CURSOR: Self = Self { cursors: 1, watches: 0 };
3370 const WATCH: Self = Self { cursors: 0, watches: 1 };
3371}
3372
3373impl RouteNode {
3374 fn is_empty(&self) -> bool {
3376 self.entries.is_empty() && self.cursors.is_empty() && self.watches.is_empty() && self.children.is_empty()
3377 }
3378
3379 fn find<'a>(&self, mut parts: impl Iterator<Item = &'a str>) -> Option<&Self> {
3381 match parts.next() {
3382 None => Some(self),
3383 Some(part) => self.children.get(part)?.find(parts),
3384 }
3385 }
3386
3387 fn reach<'a>(&mut self, mut parts: impl Iterator<Item = &'a str>, below: Below) -> &mut Self {
3390 self.cursors_below += below.cursors;
3391 self.watches_below += below.watches;
3392 match parts.next() {
3393 None => self,
3394 Some(part) => self.children.entry(part.to_string()).or_default().reach(parts, below),
3395 }
3396 }
3397
3398 fn edit<'a, R>(
3402 &mut self,
3403 mut parts: impl Iterator<Item = &'a str>,
3404 below: Below,
3405 f: impl FnOnce(&mut Self) -> R,
3406 ) -> Option<R> {
3407 let result = match parts.next() {
3408 None => f(self),
3409 Some(part) => {
3410 let child = self.children.get_mut(part)?;
3411 let result = child.edit(parts, below, f)?;
3412 if child.is_empty() {
3413 self.children.remove(part);
3414 }
3415 result
3416 }
3417 };
3418 self.cursors_below -= below.cursors;
3419 self.watches_below -= below.watches;
3420 Some(result)
3421 }
3422
3423 fn poke(&self) {
3425 for (_, watch) in &self.watches {
3426 if let Ok(mut watched) = watch.write() {
3427 watched.generation += 1;
3428 }
3429 }
3430 }
3431
3432 fn poke_below(&self) {
3435 if self.watches_below == 0 {
3436 return;
3437 }
3438 self.poke();
3439 for child in self.children.values() {
3440 child.poke_below();
3441 }
3442 }
3443
3444 fn walk<'a>(&'a self, visit: &mut impl FnMut(&'a Self)) {
3446 visit(self);
3447 for child in self.children.values() {
3448 child.walk(visit);
3449 }
3450 }
3451
3452 fn collect_cursors(&self, out: &mut Vec<ConsumerId>) {
3454 if self.cursors_below == 0 {
3455 return;
3456 }
3457 out.extend(&self.cursors);
3458 for child in self.children.values() {
3459 child.collect_cursors(out);
3460 }
3461 }
3462}
3463
3464impl RouteTable {
3465 fn split(&self, path: &Path) -> (Vec<&RouteNode>, Option<&RouteNode>) {
3468 let mut above = Vec::new();
3469 let mut node = &self.root;
3470 for part in path.parts() {
3471 above.push(node);
3472 match node.children.get(part) {
3473 Some(child) => node = child,
3474 None => return (above, None),
3475 }
3476 }
3477 (above, Some(node))
3478 }
3479
3480 fn covering(&self, path: &Path) -> impl Iterator<Item = &RouteEntry> {
3482 let mut parts = path.parts();
3485 let mut next = Some(&self.root);
3486 std::iter::from_fn(move || {
3487 let node = next?;
3488 next = parts.next().and_then(|part| node.children.get(part));
3489 Some(node)
3490 })
3491 .flat_map(|node| node.entries.iter())
3492 }
3493
3494 fn covers(&self, path: &Path, id: u64) -> bool {
3496 self.covering(path).any(|entry| entry.id == id && entry.live())
3497 }
3498
3499 fn at(&self, prefix: &Path) -> impl Iterator<Item = &RouteEntry> {
3501 self.root
3502 .find(prefix.parts())
3503 .into_iter()
3504 .flat_map(|node| node.entries.iter())
3505 }
3506
3507 fn at_mut(&mut self, prefix: &Path) -> impl Iterator<Item = &mut RouteEntry> {
3509 let mut node = Some(&mut self.root);
3510 for part in prefix.parts() {
3511 node = node.and_then(|node| node.children.get_mut(part));
3512 }
3513 node.into_iter().flat_map(|node| node.entries.iter_mut())
3514 }
3515
3516 fn entries(&self) -> impl Iterator<Item = &RouteEntry> {
3518 let mut nodes = Vec::new();
3519 self.root.walk(&mut |node| nodes.push(node));
3520 nodes.into_iter().flat_map(|node| node.entries.iter())
3521 }
3522
3523 fn insert(&mut self, entry: RouteEntry) {
3525 let node = self.root.reach(entry.prefix.parts(), Below::NONE);
3526 node.entries.push(entry);
3527 }
3528
3529 fn entry_mut(&mut self, prefix: &Path, id: u64) -> Option<&mut RouteEntry> {
3531 let mut node = &mut self.root;
3532 for part in prefix.parts() {
3533 node = node.children.get_mut(part)?;
3534 }
3535 node.entries.iter_mut().find(|entry| entry.id == id)
3536 }
3537
3538 fn remove(&mut self, prefix: &Path, id: u64) -> Option<RouteEntry> {
3540 self.root
3541 .edit(prefix.parts(), Below::NONE, |node| {
3542 let index = node.entries.iter().position(|entry| entry.id == id)?;
3543 Some(node.entries.swap_remove(index))
3544 })
3545 .flatten()
3546 }
3547
3548 fn add_cursor(&mut self, head: &Path, id: ConsumerId) {
3550 self.root.reach(head.parts(), Below::CURSOR).cursors.push(id);
3551 }
3552
3553 fn remove_cursor(&mut self, head: &Path, id: ConsumerId) {
3556 self.root.edit(head.parts(), Below::CURSOR, |node| {
3557 node.cursors.retain(|cursor| *cursor != id)
3558 });
3559 }
3560
3561 fn add_watch(&mut self, path: &Path, id: u64) -> kio::Consumer<Watched> {
3563 let producer = kio::Producer::<Watched>::default();
3564 let consumer = producer.consume();
3565 self.root.reach(path.parts(), Below::WATCH).watches.push((id, producer));
3566 consumer
3567 }
3568
3569 fn remove_watch(&mut self, path: &Path, id: u64) {
3572 self.root.edit(path.parts(), Below::WATCH, |node| {
3573 node.watches.retain(|(watch, _)| *watch != id)
3574 });
3575 }
3576
3577 fn poke_below(&self, prefix: &Path) {
3579 if let (_, Some(node)) = self.split(prefix) {
3580 node.poke_below();
3581 }
3582 }
3583
3584 fn poke_all(&self) {
3586 self.root.walk(&mut |node| node.poke());
3587 }
3588
3589 fn cursors_touching(&self, prefix: &Path) -> Vec<ConsumerId> {
3593 let (above, at) = self.split(prefix);
3594 let mut cursors: Vec<ConsumerId> = above.iter().flat_map(|node| node.cursors.iter().copied()).collect();
3595 if let Some(node) = at {
3596 node.collect_cursors(&mut cursors);
3597 }
3598 cursors.sort_unstable();
3600 cursors.dedup();
3601 cursors
3602 }
3603}
3604
3605struct OriginState {
3612 update_hold: Duration,
3615
3616 routes: RouteTable,
3619 next_route: u64,
3620 next_watch: u64,
3621
3622 cursors: HashMap<ConsumerId, TableCursor>,
3626
3627 fronts: WeakCache<FrontKey, RemoteFront>,
3633
3634 withdrawn: HashMap<PathOwned, HashSet<Hop>>,
3637
3638 closed: bool,
3641}
3642
3643impl OriginState {
3644 fn new(update_hold: Duration) -> Self {
3645 Self {
3646 update_hold,
3647 routes: RouteTable::default(),
3648 next_route: 0,
3649 next_watch: 0,
3650 cursors: HashMap::new(),
3651 fronts: WeakCache::default(),
3652 withdrawn: HashMap::new(),
3653 closed: false,
3654 }
3655 }
3656
3657 fn withdrawn_through(&self, prefix: &Path, hops: &Hops) -> bool {
3659 if self.withdrawn.is_empty() {
3660 return false;
3661 }
3662 self.withdrawn
3663 .get(prefix)
3664 .is_some_and(|peers| hops.iter().any(|hop| peers.contains(hop)))
3665 }
3666
3667 fn restale(&mut self, prefix: &Path) {
3669 let peers = self.withdrawn.get(prefix);
3670 let mut changed = None;
3671 for entry in self.routes.at_mut(prefix) {
3672 let stale = peers.is_some_and(|peers| entry.hops.iter().any(|hop| peers.contains(hop)));
3673 if entry.stale != stale {
3674 entry.stale = stale;
3675 changed = Some(entry.claim.clone());
3676 }
3677 }
3678 if let Some(claim) = changed {
3679 self.sync_route(prefix, &claim);
3680 }
3681 }
3682
3683 fn reannounced(&mut self, prefix: &PathOwned, hops: &Hops) {
3687 if let Some(sender) = hops.iter().last()
3688 && self.withdrawn.get_mut(prefix).is_some_and(|peers| peers.remove(sender))
3689 {
3690 self.restale(prefix);
3691 self.prune_withdrawn(prefix);
3692 }
3693 }
3694
3695 fn prune_withdrawn(&mut self, prefix: &PathOwned) {
3697 if self.withdrawn.is_empty() {
3698 return;
3699 }
3700 let Some(peers) = self.withdrawn.get_mut(prefix) else {
3701 return;
3702 };
3703 if peers.len() <= 1 {
3704 peers.retain(|peer| self.routes.at(prefix).any(|entry| entry.hops.contains(peer)));
3706 } else {
3707 let mut unreferenced = peers.clone();
3708 for entry in self.routes.at(prefix) {
3709 if unreferenced.is_empty() {
3710 break;
3711 }
3712 for hop in entry.hops.iter() {
3713 unreferenced.remove(hop);
3714 }
3715 }
3716 peers.retain(|peer| !unreferenced.contains(peer));
3717 }
3718 if peers.is_empty() {
3719 self.withdrawn.remove(prefix);
3720 }
3721 }
3722
3723 fn sync_route(&mut self, prefix: &Path, claim: &Pattern) {
3728 let routes = &self.routes;
3730 let mut candidates = None;
3731 for id in routes.cursors_touching(prefix) {
3732 let Some(cursor) = self.cursors.get_mut(&id) else {
3733 continue;
3734 };
3735 if let Some(presented) = cursor.presented(prefix, claim) {
3736 if presented.is_empty() {
3737 Self::sync_cursor(routes, cursor, &presented);
3739 } else {
3740 let candidates = candidates.get_or_insert_with(|| {
3743 let mut entries: Vec<_> = routes
3744 .at(prefix)
3745 .map(|entry| (route_order(prefix, entry), entry))
3746 .collect();
3747 entries.sort_unstable_by_key(|(order, _)| *order);
3748 entries
3749 });
3750 let best = candidates
3751 .iter()
3752 .map(|(_, entry)| *entry)
3753 .find(|entry| cursor.visible(entry));
3754 cursor.update(&presented, best);
3755 }
3756 }
3757 }
3758 routes.poke_below(prefix);
3760 }
3761
3762 fn watch(&mut self, shared: &kio::Shared<OriginState>, path: &Path) -> Watch {
3764 let id = self.next_watch;
3765 self.next_watch += 1;
3766 let signal = self.routes.add_watch(path, id);
3767 Watch {
3768 shared: shared.clone(),
3769 path: path.to_owned(),
3770 id,
3771 signal,
3772 }
3773 }
3774
3775 fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned) {
3778 let best = match presented.is_empty() {
3784 true => cursor.best(routes.covering(&cursor.root)),
3785 false => {
3786 let absolute = cursor.root.join(presented);
3787 cursor.best(routes.at(&absolute))
3788 }
3789 };
3790
3791 cursor.update(presented, best);
3792 }
3793
3794 fn register_cursor(&mut self, id: ConsumerId, mut cursor: TableCursor) {
3796 let mut presented: BTreeSet<PathOwned> = BTreeSet::new();
3799 for head in &cursor.heads {
3800 let (above, at) = self.routes.split(head);
3801 let mut nodes = above;
3802 if let Some(node) = at {
3803 node.walk(&mut |node| nodes.push(node));
3804 }
3805 for entry in nodes.into_iter().flat_map(|node| node.entries.iter()) {
3806 if let Some(p) = cursor.presented(&entry.prefix, &entry.claim) {
3807 presented.insert(p);
3808 }
3809 }
3810 }
3811 for p in &presented {
3812 Self::sync_cursor(&self.routes, &mut cursor, p);
3813 }
3814 for head in &cursor.heads {
3815 self.routes.add_cursor(head, id);
3816 }
3817 self.cursors.insert(id, cursor);
3818 }
3819
3820 fn best_route(&self, path: &Path, horizon: Horizon, pin: Pin, refused: &HashSet<u64>) -> Option<&RouteEntry> {
3835 let (above, at) = self.routes.split(path);
3839 let mut best = None;
3840 for node in above.into_iter().chain(at) {
3841 let mut candidates = node
3842 .entries
3843 .iter()
3844 .filter(|entry| entry.live())
3845 .filter(|entry| entry.scope.matches(path.as_str()))
3846 .filter(|entry| horizon.admits(entry))
3847 .filter(|entry| entry.qualifies(pin))
3848 .filter(|entry| !refused.contains(&entry.id))
3849 .peekable();
3850 if candidates.peek().is_some() {
3851 best = candidates
3852 .filter(|entry| entry.serves(path))
3853 .min_by_key(|entry| route_order(&entry.prefix, entry));
3854 }
3855 }
3856 best
3857 }
3858}
3859
3860#[derive(Default)]
3867struct PendingBroadcast {
3868 resolved: Option<Result<broadcast::Consumer, Error>>,
3869}
3870
3871#[must_use = "dropping an origin::Dynamic retracts the route"]
3886pub struct Dynamic {
3887 announcement: AnnounceProducer,
3889 state: kio::Shared<ServeState>,
3890 _keepalive: Keepalive,
3894}
3895
3896impl Dynamic {
3897 pub(crate) fn withdrawn(self) {
3900 self.announcement.retract(true);
3901 }
3902
3903 pub fn update(&self, route: Route) -> Result<(), Error> {
3913 self.announcement.update(route)
3914 }
3915
3916 pub fn poll_requested_broadcast(&self, waiter: &kio::Waiter) -> Poll<Result<Request, Error>> {
3921 let mut state = ready!(self.state.poll(waiter, |state| {
3922 if state.closed || state.requests.has_queued() {
3923 Poll::Ready(())
3924 } else {
3925 Poll::Pending
3926 }
3927 }));
3928
3929 if state.closed {
3931 return Poll::Ready(Err(Error::Closed));
3932 }
3933
3934 let path = state.requests.pop().expect("predicate guaranteed a request");
3935 let producer = state.requests.get(&path).expect("popped key must be pending").clone();
3941 Poll::Ready(Ok(Request {
3942 path,
3943 producer,
3944 home: self.state.clone(),
3945 }))
3946 }
3947
3948 pub async fn requested_broadcast(&self) -> Result<Request, Error> {
3954 kio::wait(|waiter| self.poll_requested_broadcast(waiter)).await
3955 }
3956}
3957
3958impl ServeState {
3959 fn resolve(
3970 shared: &kio::Shared<Self>,
3971 path: &PathOwned,
3972 producer: &kio::Producer<PendingBroadcast>,
3973 result: Result<broadcast::Consumer, Error>,
3974 ) {
3975 let mut state = shared.lock();
3976 if state.closed {
3977 return;
3978 }
3979 let resolved = match result {
3980 Ok(broadcast) => {
3981 let existing = state.served.insert(path.clone(), broadcast.weak());
3985 Ok(existing.map(|weak| weak.consume()).unwrap_or(broadcast))
3986 }
3987 Err(err) => Err(err),
3988 };
3989 state.requests.remove_if(path, |p| p.same_channel(producer));
3990 if let Ok(mut pending) = producer.write() {
3991 pending.resolved.get_or_insert(resolved);
3992 drop(state);
3993 }
3994 }
3995
3996 fn forget(shared: &kio::Shared<Self>, path: &PathOwned, producer: &kio::Producer<PendingBroadcast>) {
3998 shared.lock().requests.remove_if(path, |p| p.same_channel(producer));
3999 }
4000}
4001
4002pub struct Request {
4009 path: PathOwned,
4011
4012 producer: kio::Producer<PendingBroadcast>,
4015
4016 home: kio::Shared<ServeState>,
4019}
4020
4021impl Request {
4022 pub fn path(&self) -> &Path<'_> {
4024 &self.path
4025 }
4026
4027 pub fn accept(self, broadcast: impl Consume<broadcast::Consumer>) {
4033 let broadcast = broadcast.consume();
4034 ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast));
4035 }
4037
4038 pub fn reject(self, err: Error) {
4040 ServeState::resolve(&self.home, &self.path, &self.producer, Err(err));
4041 }
4042}
4043
4044impl Drop for Request {
4045 fn drop(&mut self) {
4046 ServeState::forget(&self.home, &self.path, &self.producer);
4054 }
4055}
4056
4057pub struct Requesting {
4064 inner: RequestState,
4065 path: PathOwned,
4069 stats: stats::Scope,
4072}
4073
4074enum RequestState {
4075 Failed(Error),
4078 Pending(kio::Consumer<PendingBroadcast>),
4080}
4081
4082impl Requesting {
4083 fn failed(error: Error) -> Self {
4084 Self::new(RequestState::Failed(error))
4085 }
4086
4087 fn queued(consumer: kio::Consumer<PendingBroadcast>) -> Self {
4088 Self::new(RequestState::Pending(consumer))
4089 }
4090
4091 pub fn is_queued(&self) -> bool {
4100 matches!(self.inner, RequestState::Pending(_))
4101 }
4102
4103 fn new(inner: RequestState) -> Self {
4104 Self {
4105 inner,
4106 path: PathOwned::default(),
4107 stats: stats::Scope::default(),
4108 }
4109 }
4110
4111 fn with_path(mut self, path: PathOwned) -> Self {
4112 self.path = path;
4113 self
4114 }
4115
4116 fn with_stats(mut self, scope: stats::Scope) -> Self {
4118 self.stats = scope;
4119 self
4120 }
4121
4122 fn hand_out(&self, broadcast: broadcast::Consumer) -> broadcast::Consumer {
4124 broadcast.with_path(self.path.clone()).with_stats(self.stats.clone())
4125 }
4126
4127 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<broadcast::Consumer, Error>> {
4129 match &self.inner {
4130 RequestState::Failed(error) => Poll::Ready(Err(error.clone())),
4131 RequestState::Pending(consumer) => Poll::Ready(
4132 match ready!(consumer.poll(waiter, |state| match &state.resolved {
4133 Some(result) => Poll::Ready(result.clone()),
4134 None => Poll::Pending,
4135 })) {
4136 Ok(result) => result.map(|broadcast| self.hand_out(broadcast)),
4137 Err(_closed) => Err(Error::Unroutable),
4139 },
4140 ),
4141 }
4142 }
4143}
4144
4145impl kio::Pollable for Requesting {
4146 type Output = Result<broadcast::Consumer, Error>;
4147
4148 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
4149 self.poll_ok(waiter)
4150 }
4151}
4152
4153pub trait Consume<T> {
4161 fn consume(&self) -> T;
4163}
4164
4165impl<T, U: Consume<T>> Consume<T> for &U {
4166 fn consume(&self) -> T {
4167 (**self).consume()
4168 }
4169}
4170
4171impl Consume<Consumer> for Producer {
4172 fn consume(&self) -> Consumer {
4173 Consumer::from_producer(self, stats::Session::default())
4177 }
4178}
4179
4180impl Consume<Consumer> for Consumer {
4181 fn consume(&self) -> Consumer {
4182 self.clone()
4183 }
4184}
4185
4186impl Consume<broadcast::Consumer> for broadcast::Producer {
4187 fn consume(&self) -> broadcast::Consumer {
4188 self.consume()
4190 }
4191}
4192
4193impl Consume<broadcast::Consumer> for broadcast::Consumer {
4194 fn consume(&self) -> broadcast::Consumer {
4195 self.clone()
4196 }
4197}
4198
4199impl Consume<track::Consumer> for track::Producer {
4200 fn consume(&self) -> track::Consumer {
4201 self.consume()
4202 }
4203}
4204
4205impl Consume<track::Consumer> for track::Consumer {
4206 fn consume(&self) -> track::Consumer {
4207 self.clone()
4208 }
4209}
4210
4211#[derive(Clone)]
4217pub struct Consumer {
4218 hop: Hop,
4220 max_hops: HopLimit,
4221 scope: OriginScope,
4222
4223 root: PathOwned,
4225
4226 shared: kio::Shared<OriginState>,
4229
4230 stats: stats::Session,
4234
4235 horizon: Horizon,
4240
4241 hidden: Hidden,
4243
4244 pool: cache::Pool,
4247 cache_duration: Duration,
4248
4249 tasks: TasksWeak,
4253
4254 timers: Clock,
4257}
4258
4259impl Consumer {
4260 fn from_producer(producer: &Producer, stats: stats::Session) -> Self {
4261 Self {
4262 hop: producer.hop,
4263 max_hops: producer.max_hops,
4264 scope: producer.scope.clone(),
4265 root: producer.root.clone(),
4266 shared: producer.shared.clone(),
4267 stats,
4268 horizon: Horizon::default(),
4269 hidden: Hidden::default(),
4270 pool: producer.pool.clone(),
4271 cache_duration: producer.cache_duration,
4272 tasks: producer.tasks.downgrade(),
4273 timers: producer.timers.clone(),
4274 }
4275 }
4276
4277 pub fn hop(&self) -> Hop {
4279 self.hop
4280 }
4281
4282 pub(crate) fn excluding(mut self, peer: Hop) -> Self {
4290 self.horizon.exclude = Some(peer);
4291 self
4292 }
4293
4294 pub(crate) fn forwardable(mut self) -> Self {
4298 self.horizon.max_hops = Some(self.max_hops);
4299 self
4300 }
4301
4302 pub fn local(mut self) -> Self {
4309 self.horizon.local = true;
4310 self
4311 }
4312
4313 pub fn with_hidden(mut self, hidden: bool) -> Self {
4318 self.hidden.include = hidden;
4319 self
4320 }
4321
4322 pub(crate) fn beyond(mut self, outer: &Consumer) -> Self {
4325 self.hidden.beyond = Some(
4326 outer
4327 .hidden
4328 .from
4329 .clone()
4330 .map(|head| vec![head])
4331 .unwrap_or_else(|| interest_prefixes(&outer.scope.allowed)),
4332 );
4333 self
4334 }
4335
4336 pub(crate) fn discovery(mut self, hidden: bool) -> Self {
4338 self.hidden.include = hidden;
4339 self.hidden.from = Some(self.root.clone());
4340 self
4341 }
4342
4343 pub(crate) fn includes_hidden(&self) -> bool {
4345 self.hidden.include
4346 }
4347
4348 pub fn with_stats(mut self, session: stats::Session) -> Self {
4352 self.stats = session;
4353 self
4354 }
4355
4356 pub(crate) fn stats(&self) -> &stats::Session {
4358 &self.stats
4359 }
4360
4361 fn untagged(&self) -> Self {
4365 Self {
4366 stats: stats::Session::default(),
4367 ..self.clone()
4368 }
4369 }
4370
4371 pub(crate) fn empty(&self) -> Self {
4376 Self {
4377 scope: OriginScope::empty(),
4378 ..self.clone()
4379 }
4380 }
4381
4382 pub fn announced(&self) -> AnnounceConsumer {
4391 let state = kio::Producer::<OriginConsumerState>::default();
4392 let cursor = |root: PathOwned,
4393 allowed: Patterns,
4394 hidden: Hidden,
4395 mount: Option<&Mount>,
4396 under: PathOwned,
4397 holes: Vec<PathOwned>| TableCursor {
4398 root,
4399 heads: interest_prefixes(&allowed),
4400 allowed,
4401 horizon: self.horizon,
4402 hidden,
4403 state: state.clone(),
4404 under,
4405 holes,
4406 named: mount.map(|mount| (mount.clone(), self.scope.allowed.clone())),
4407 current: HashMap::new(),
4408 };
4409
4410 if let Some(mount) = self.scope.mount(&self.root) {
4413 let Some(root) = mount.resolve(&self.root) else {
4415 return AnnounceConsumer::new(
4416 self.root.clone(),
4417 Vec::new(),
4418 state,
4419 self.stats.clone(),
4420 &self.shared,
4421 self.timers.clone(),
4422 );
4423 };
4424 let cursors = vec![cursor(
4425 root,
4426 mount.translate(&self.scope.allowed),
4427 self.hidden.translate(mount),
4428 Some(mount),
4429 PathOwned::default(),
4430 Vec::new(),
4431 )];
4432 return AnnounceConsumer::new(
4433 self.root.clone(),
4434 cursors,
4435 state,
4436 self.stats.clone(),
4437 &self.shared,
4438 self.timers.clone(),
4439 );
4440 }
4441
4442 let heads = interest_prefixes(&self.scope.allowed);
4445 let mut holes = Vec::new();
4446 let mut cursors = Vec::new();
4447 for mount in self.scope.mounts.iter() {
4448 let Some(under) = mount.at.strip_prefix(&self.root) else {
4449 continue;
4450 };
4451 holes.push(mount.at.clone());
4452 let allowed = mount.translate(&self.scope.allowed);
4453 if allowed.is_empty()
4456 || !(self.hidden.include
4457 || !hides(
4458 self.hidden.from.as_ref().map(std::slice::from_ref).unwrap_or(&heads),
4459 &mount.at,
4460 )) {
4461 continue;
4462 }
4463 cursors.push(cursor(
4464 mount.target.clone(),
4465 allowed,
4466 self.hidden.translate(mount),
4467 Some(mount),
4468 under.to_owned(),
4469 Vec::new(),
4470 ));
4471 }
4472 cursors.insert(
4473 0,
4474 cursor(
4475 self.root.clone(),
4476 self.scope.allowed.clone(),
4477 self.hidden.clone(),
4478 None,
4479 PathOwned::default(),
4480 holes,
4481 ),
4482 );
4483 AnnounceConsumer::new(
4484 self.root.clone(),
4485 cursors,
4486 state,
4487 self.stats.clone(),
4488 &self.shared,
4489 self.timers.clone(),
4490 )
4491 }
4492
4493 pub fn consume(&self) -> Self {
4495 self.clone()
4496 }
4497
4498 #[cfg(test)]
4501 pub(crate) fn get_broadcast(&self, path: impl AsPath) -> Option<broadcast::Consumer> {
4502 let full = self.root.join(path).to_owned();
4503 if !self.scope.permits(&full) {
4504 return None;
4505 }
4506 let full = self.scope.resolve(&full)?;
4507 let table = self.shared.lock();
4508 table
4509 .routes
4510 .at(&full)
4511 .filter(|entry| entry.local)
4512 .min_by_key(|entry| route_order(&entry.prefix, entry))
4513 .and_then(|entry| entry.source.clone())
4514 }
4515
4516 pub async fn routed(&self, path: impl AsPath) -> Option<Route> {
4526 let path = path.as_path();
4527
4528 let consumer = match Pattern::subtree(path.as_str()) {
4533 Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?,
4534 Err(InvalidPattern::TooManySegments) => self.clone(),
4535 Err(_) => return None,
4536 };
4537
4538 if !consumer.allowed().matches(path.as_str()) {
4542 return None;
4543 }
4544
4545 let mut announced = consumer.untagged().with_hidden(true).announced();
4549 loop {
4550 let update = announced.next().await?;
4551 if update.kind.is_active() && path.has_prefix(&update.prefix) {
4552 return Some(update.route);
4553 }
4554 }
4555 }
4556
4557 pub async fn routed_broadcast(&self, path: impl AsPath) -> Result<broadcast::Consumer, Error> {
4571 let path = path.as_path();
4572
4573 if !self.allowed().matches(path.as_str()) {
4576 return Err(Error::Unauthorized);
4577 }
4578 loop {
4579 let (watch, seen) = {
4588 let mut table = self.shared.lock();
4589 if table.closed {
4590 return Err(Error::Closed);
4591 }
4592 let named = self.root.join(&path);
4593 let resolved = self.scope.resolve(&named).ok_or(BoundsExceeded)?;
4594 let watch = table.watch(&self.shared, &resolved);
4595 let seen = watch.seen();
4596 (watch, seen)
4597 };
4598 match self.request_broadcast(&path).await {
4599 Ok(broadcast) => return Ok(broadcast),
4600 Err(Error::Unroutable) => {
4601 kio::wait(|waiter| watch.poll_changed(waiter, seen)).await;
4602 }
4603 Err(Error::Dropped) if self.shared.lock().closed => return Err(Error::Closed),
4606 Err(err) => return Err(err),
4607 }
4608 }
4609 }
4610
4611 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Consumer, Error> {
4618 let root = self.root.join(root).to_owned();
4619 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
4620 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
4621 Ok(Consumer {
4622 scope,
4623 root,
4624 ..self.clone()
4625 })
4626 }
4627
4628 pub fn request_broadcast(&self, path: impl AsPath) -> kio::Pending<Requesting> {
4651 let path = path.as_path();
4652
4653 let named = self.root.join(&path).to_owned();
4657 let scope = self.stats.egress(&named);
4658 let requested = path.to_owned();
4662
4663 if !self.scope.permits(&named) {
4665 return kio::Pending::new(Requesting::failed(Error::Unauthorized));
4666 }
4667
4668 let Some(absolute) = self.scope.resolve(&named).map(|path| path.to_owned()) else {
4671 return kio::Pending::new(Requesting::failed(BoundsExceeded.into()));
4672 };
4673
4674 let mut state = self.shared.lock();
4675
4676 if state.closed {
4678 return kio::Pending::new(Requesting::failed(Error::Closed));
4679 }
4680
4681 let horizon = self.horizon.effective(&state.routes, &absolute.as_path());
4682
4683 if state
4687 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4688 .is_none()
4689 {
4690 return kio::Pending::new(Requesting::failed(Error::Unroutable));
4691 }
4692
4693 let key = (absolute.clone(), horizon);
4701 if let Some(front) = state.fronts.get(&key) {
4702 let pin = *front.pin.lock();
4703 let current = state
4704 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4705 .is_some_and(|entry| entry.qualifies(pin));
4706 if current {
4707 let pending = Requesting::queued(front.request.consume())
4708 .with_path(requested)
4709 .with_stats(scope);
4710 return kio::Pending::new(pending);
4711 }
4712 state.fronts.remove(&key);
4713 }
4714
4715 let broadcast = broadcast::Producer::new_spliced(broadcast::Info {
4720 pool: self.pool.clone(),
4721 cache_duration: self.cache_duration,
4722 path: absolute.clone(),
4723 });
4724 let request = kio::Producer::<PendingBroadcast>::default();
4725 let consumer = request.consume();
4726 let watch = state.watch(&self.shared, &absolute);
4727 let pin = kio::Lock::new(Pin::Any);
4728 state.fronts.insert(
4729 key,
4730 RemoteFront {
4731 request: request.clone(),
4732 broadcast: broadcast.consume().weak(),
4733 pin: pin.clone(),
4734 },
4735 );
4736 drop(state);
4739 self.tasks.push(run_front(FrontTask {
4740 shared: self.shared.clone(),
4741 broadcast,
4742 path: absolute,
4743 horizon,
4744 watch,
4745 request,
4746 pin,
4747 timers: self.timers.clone(),
4748 }));
4749 kio::Pending::new(Requesting::queued(consumer).with_path(requested).with_stats(scope))
4750 }
4751
4752 pub fn root(&self) -> &Path<'_> {
4754 &self.root
4755 }
4756
4757 pub fn allowed(&self) -> Patterns {
4759 self.scope.relative(&self.root)
4760 }
4761
4762 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
4764 self.root.join(path)
4765 }
4766}
4767
4768pub struct AnnounceConsumer {
4773 ids: Vec<ConsumerId>,
4776 shared: kio::Shared<OriginState>,
4777 root: PathOwned,
4778
4779 state: kio::Producer<OriginConsumerState>,
4782
4783 stats: stats::Session,
4786
4787 guards: HashMap<PathOwned, stats::Announce>,
4791
4792 park: kio::Park,
4795
4796 update_hold: Duration,
4799 retracted: HashMap<PathOwned, Quarantine>,
4802 anchoring: VecDeque<(Instant, PathOwned)>,
4803 expiring: VecDeque<(Instant, PathOwned)>,
4804 timers: Clock,
4805 hold: crate::runtime::Deadline<Clock>,
4806}
4807
4808impl AnnounceConsumer {
4809 fn new(
4810 root: PathOwned,
4811 cursors: Vec<TableCursor>,
4812 state: kio::Producer<OriginConsumerState>,
4813 stats: stats::Session,
4814 shared: &kio::Shared<OriginState>,
4815 timers: Clock,
4816 ) -> Self {
4817 let mut ids = Vec::with_capacity(cursors.len());
4818 let update_hold;
4819 {
4820 let mut table = shared.lock();
4821 update_hold = table.update_hold;
4822 if table.closed {
4823 if let Ok(mut state) = state.write() {
4825 state.ended = true;
4826 }
4827 } else {
4828 for cursor in cursors {
4829 let id = ConsumerId::new();
4830 table.register_cursor(id, cursor);
4831 ids.push(id);
4832 }
4833 }
4834 }
4835
4836 Self {
4837 ids,
4838 shared: shared.clone(),
4839 root,
4840 state,
4841 stats,
4842 guards: HashMap::new(),
4843 park: kio::Park::default(),
4844 update_hold,
4845 retracted: HashMap::new(),
4846 anchoring: VecDeque::new(),
4847 expiring: VecDeque::new(),
4848 hold: crate::runtime::Deadline::new(&timers),
4849 timers,
4850 }
4851 }
4852
4853 fn quarantine(&mut self, update: &AnnounceUpdate, now: Option<Instant>) {
4856 if update.kind.is_active() {
4857 return;
4858 }
4859 let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else {
4860 return;
4861 };
4862 let publisher = update
4864 .route
4865 .hops
4866 .iter()
4867 .next()
4868 .copied()
4869 .filter(|hop| *hop != Hop::UNKNOWN);
4870 let quarantine = Quarantine {
4872 held: Held::Anchoring(now),
4873 publisher,
4874 rank: (update.route.cost, update.route.hops.len()),
4875 };
4876 let replaced = self.retracted.insert(update.prefix.clone(), quarantine);
4877 if !replaced.is_some_and(|q| q.held == Held::Anchoring(now)) {
4880 self.anchoring.push_back((now, update.prefix.clone()));
4881 }
4882 }
4883
4884 fn settle(&mut self, now: Option<Instant>) -> Option<Instant> {
4887 let now = now?;
4888 while let Some((seen, prefix)) = self.anchoring.front() {
4891 if *seen >= now {
4892 break;
4893 }
4894 if let Some(quarantine) = self.retracted.get_mut(prefix)
4895 && quarantine.held == Held::Anchoring(*seen)
4896 {
4897 quarantine.held = Held::Until(now + RETRACT_QUARANTINE);
4898 self.expiring.push_back((now + RETRACT_QUARANTINE, prefix.clone()));
4899 }
4900 self.anchoring.pop_front();
4901 }
4902 while let Some((until, prefix)) = self.expiring.front() {
4903 if *until > now {
4904 break;
4905 }
4906 if self
4907 .retracted
4908 .get(prefix)
4909 .is_some_and(|q| q.held == Held::Until(*until))
4910 {
4911 self.retracted.remove(prefix);
4912 }
4913 self.expiring.pop_front();
4914 }
4915 let anchoring = self.anchoring.front().map(|(seen, _)| *seen + Duration::from_nanos(1));
4916 earliest(anchoring, self.expiring.front().map(|(until, _)| *until))
4917 }
4918
4919 fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate {
4921 let absolute = self.root.join(&update.prefix).to_owned();
4922 if update.kind.is_active() {
4923 let scope = self.stats.egress(&absolute);
4924 self.guards
4925 .entry(update.prefix.clone())
4926 .or_insert_with(|| scope.announce());
4927 } else {
4928 self.guards.remove(&update.prefix);
4929 }
4930 update
4931 }
4932
4933 pub async fn next(&mut self) -> Option<AnnounceUpdate> {
4941 kio::wait(|waiter| self.poll_next(waiter)).await
4942 }
4943
4944 pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<AnnounceUpdate>> {
4950 loop {
4951 let now = self.timers.try_now();
4952 let settling = self.settle(now);
4953 let mut wake = None;
4954 let mut state = match self.state.poll(waiter, |state| match state.due(now) {
4957 true => Poll::Ready(()),
4958 false => {
4959 wake = state.wake();
4960 Poll::Pending
4961 }
4962 }) {
4963 Poll::Ready(Ok(state)) => state,
4964 Poll::Ready(Err(_)) => return Poll::Ready(None),
4966 Poll::Pending => {
4967 self.hold.set(earliest(wake, settling));
4968 ready!(self.hold.poll(waiter));
4969 continue;
4970 }
4971 };
4972 if state.pending.is_empty() {
4973 state.close();
4976 return Poll::Ready(None);
4977 }
4978 let Ok(prefix) = state.scan(&self.retracted, now, self.update_hold) else {
4980 continue;
4981 };
4982 let update = state.take_prefix(prefix);
4983 drop(state);
4984 let Some(update) = update else {
4985 return Poll::Ready(None);
4986 };
4987 self.quarantine(&update, now);
4988 return Poll::Ready(Some(self.hand_out(update)));
4989 }
4990 }
4991
4992 pub fn try_next(&mut self) -> Option<AnnounceUpdate> {
4997 let now = self.timers.try_now();
4998 let settling = self.settle(now);
4999 let idle = {
5001 let state = self.state.read();
5002 (!state.due(now) || state.pending.is_empty()).then(|| state.wake())
5003 };
5004 let scanned = match idle {
5005 Some(wake) => Err(wake),
5006 None => {
5007 let mut state = self.state.write().ok()?;
5008 state
5009 .scan(&self.retracted, now, self.update_hold)
5010 .map(|prefix| state.take_prefix(prefix))
5011 }
5012 };
5013 let update = match scanned {
5014 Ok(update) => update?,
5015 Err(wake) => {
5016 self.hold.set(earliest(wake, settling));
5019 return None;
5020 }
5021 };
5022 self.quarantine(&update, now);
5023 Some(self.hand_out(update))
5024 }
5025
5026 pub fn is_closed(&self) -> bool {
5028 let state = self.state.read();
5029 state.is_closed() || state.ended
5030 }
5031
5032 pub fn root(&self) -> &Path<'_> {
5034 &self.root
5035 }
5036
5037 pub fn absolute(&self, prefix: impl AsPath) -> Path<'_> {
5039 self.root.join(prefix)
5040 }
5041}
5042
5043impl futures::Stream for AnnounceConsumer {
5044 type Item = AnnounceUpdate;
5045
5046 fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Option<Self::Item>> {
5047 let this = self.get_mut();
5048 let waiter = this.park.hold(cx).clone();
5049 this.poll_next(&waiter)
5050 }
5051}
5052
5053impl Drop for AnnounceConsumer {
5054 fn drop(&mut self) {
5055 let mut shared = self.shared.lock();
5056 for id in &self.ids {
5057 if let Some(cursor) = shared.cursors.remove(id) {
5058 for head in &cursor.heads {
5059 shared.routes.remove_cursor(head, *id);
5060 }
5061 }
5062 }
5063 }
5064}
5065
5066#[cfg(test)]
5067use futures::FutureExt;
5068
5069#[cfg(test)]
5070#[allow(missing_docs)] impl AnnounceConsumer {
5072 pub fn assert_next_active(&mut self, expected: impl AsPath) -> Route {
5074 let expected = expected.as_path();
5075 let update = self.next().now_or_never().expect("next blocked").expect("no next");
5076 assert_eq!(update.prefix, expected, "wrong prefix");
5077 assert!(update.kind.is_active(), "should be an active route");
5078 update.route
5079 }
5080
5081 pub fn assert_try_next_active(&mut self, expected: impl AsPath) -> Route {
5083 let expected = expected.as_path();
5084 let update = self.try_next().expect("no next");
5085 assert_eq!(update.prefix, expected, "wrong prefix");
5086 assert!(update.kind.is_active(), "should be an active route");
5087 update.route
5088 }
5089
5090 pub fn assert_next_ended(&mut self, expected: impl AsPath) {
5092 let expected = expected.as_path();
5093 let update = self.next().now_or_never().expect("next blocked").expect("no next");
5094 assert_eq!(update.prefix, expected, "wrong prefix");
5095 assert_eq!(update.kind, AnnounceKind::Retracted, "should be a retraction");
5096 }
5097
5098 pub fn assert_next_wait(&mut self) {
5099 if let Some(res) = self.next().now_or_never() {
5100 panic!("next should block: got {:?}", res.map(|u| u.prefix));
5101 }
5102 }
5103}
5104
5105#[cfg(test)]
5109pub(crate) trait ProduceTest {
5110 fn produce(self) -> Producer;
5111}
5112
5113#[cfg(test)]
5114impl ProduceTest for Config {
5115 fn produce(self) -> Producer {
5116 let (producer, driver) = Producer::new(self);
5117 if tokio::runtime::Handle::try_current().is_ok() {
5118 tokio::spawn(crate::time::run(driver));
5119 } else {
5120 std::mem::forget(driver);
5123 }
5124 producer
5125 }
5126}
5127
5128#[cfg(test)]
5129impl ProduceTest for Hop {
5130 fn produce(self) -> Producer {
5131 Config::new(self).produce()
5132 }
5133}
5134
5135#[cfg(test)]
5136mod tests {
5137 use super::*;
5138 use futures::FutureExt;
5139
5140 fn origin(id: u64) -> Hop {
5141 Hop::new(id).unwrap()
5142 }
5143
5144 fn hops(ids: &[u64]) -> Hops {
5145 let mut list = Hops::new();
5146 for &id in ids {
5147 list.push(if id == 0 { Hop::UNKNOWN } else { origin(id) }).unwrap();
5148 }
5149 list
5150 }
5151
5152 fn scopes(prefixes: &[&str]) -> Patterns {
5154 prefixes
5155 .iter()
5156 .map(|prefix| Pattern::subtree(prefix).unwrap())
5157 .collect()
5158 }
5159
5160 #[test]
5161 fn hop_limit_refuses_values_outside_the_wire_bound() {
5162 for invalid in [0, 33, usize::MAX] {
5163 assert_eq!(HopLimit::new(invalid), None);
5164 }
5165 for valid in [1, 6, 32] {
5166 assert_eq!(HopLimit::new(valid).unwrap().get(), valid);
5167 }
5168 assert_eq!(HopLimit::default().get(), 32);
5169 let config = Config {
5170 max_hops: HopLimit::new(6).unwrap(),
5171 ..Default::default()
5172 };
5173 let (producer, _driver) = Producer::new(config);
5174 let scoped = producer.scope("room", &Patterns::from(Pattern::all())).unwrap();
5175 assert_eq!(scoped.config().max_hops.get(), 6);
5176 assert_eq!(scoped.consume().forwardable().horizon.max_hops, HopLimit::new(6));
5177 }
5178
5179 #[tokio::test]
5182 async fn forwardable_caps_announcements_but_not_requests() {
5183 let producer = Config {
5184 max_hops: HopLimit::new(2).unwrap(),
5185 ..Config::new(origin(1))
5186 }
5187 .produce();
5188 let mut plain = producer.consume().announced();
5189 let mut forwarded = producer.consume().forwardable().announced();
5190
5191 let _full = producer
5192 .dynamic("room", Route::default().with_hops(hops(&[7, 8])).with_cost(1))
5193 .unwrap();
5194 assert_eq!(plain.assert_next_active("room").hops, hops(&[7, 8]));
5195 forwarded.assert_next_wait();
5196
5197 let pending = producer.consume().forwardable().request_broadcast("room/alice");
5198 assert!(pending.now_or_never().is_none(), "the cap must not refuse a request");
5199
5200 let _short = producer
5201 .dynamic("room", Route::default().with_hops(hops(&[9])).with_cost(50))
5202 .unwrap();
5203 assert_eq!(forwarded.assert_next_active("room").hops, hops(&[9]));
5204 plain.assert_next_wait();
5205 }
5206
5207 #[test]
5208 fn default_config_mints_a_real_hop() {
5209 let config = Config::default();
5210 assert_ne!(config.hop, Hop::UNKNOWN);
5211 let (producer, _driver) = Producer::new(config.clone());
5212 assert_eq!(producer.hop(), config.hop);
5213 assert_eq!(producer.consume().hop(), config.hop);
5214 }
5215
5216 #[test]
5217 fn random_hops_fit_legacy_lite_clients() {
5218 for _ in 0..32 {
5219 assert!(Hop::random().id() < 1u64 << 53);
5220 }
5221 }
5222
5223 async fn settle(mut check: impl FnMut() -> bool) {
5226 for _ in 0..100 {
5227 if check() {
5228 return;
5229 }
5230 tokio::task::yield_now().await;
5231 }
5232 panic!("condition never settled");
5233 }
5234
5235 async fn queued(server: &Dynamic) -> Request {
5237 let mut request = None;
5238 settle(|| match server.poll_requested_broadcast(&kio::Waiter::noop()) {
5239 Poll::Ready(Ok(popped)) => {
5240 request = Some(popped);
5241 true
5242 }
5243 _ => false,
5244 })
5245 .await;
5246 request.unwrap()
5247 }
5248
5249 async fn next_group(subscription: &mut crate::track::Subscriber) -> Result<Option<crate::group::Consumer>, Error> {
5251 let mut next = None;
5252 settle(|| match subscription.poll_recv_group(&kio::Waiter::noop()) {
5253 Poll::Ready(result) => {
5254 next = Some(result);
5255 true
5256 }
5257 Poll::Pending => false,
5258 })
5259 .await;
5260 next.unwrap()
5261 }
5262
5263 #[tokio::test]
5264 async fn announce_and_retract() {
5265 let producer = origin(1).produce();
5266 let consumer = producer.consume();
5267 let mut announced = consumer.announced();
5268 announced.assert_next_wait();
5269
5270 let announcement = producer.announce("room/alice", Route::default()).unwrap();
5271 let route = announced.assert_next_active("room/alice");
5272 assert!(route.hops.is_empty());
5273 assert_eq!(route.cost, Cost::default());
5274 announced.assert_next_wait();
5275
5276 drop(announcement);
5277 announced.assert_next_ended("room/alice");
5278 announced.assert_next_wait();
5279 }
5280
5281 #[tokio::test]
5284 async fn hidden_routes_need_an_opt_in() {
5285 let producer = origin(1).produce();
5286 let consumer = producer.consume();
5287 let _visible = producer.announce("room/alice", Route::default()).unwrap();
5288 let _stats = producer.announce(".stats/node", Route::default()).unwrap();
5289 let _nested = producer.announce("room/.internal", Route::default()).unwrap();
5290 let _suffix = producer.announce("room/catalog.pro", Route::default()).unwrap();
5292
5293 let mut announced = consumer.announced();
5294 announced.assert_next_active("room/alice");
5295 announced.assert_next_active("room/catalog.pro");
5296 announced.assert_next_wait();
5297
5298 let mut announced = consumer.clone().with_hidden(true).announced();
5299 announced.assert_next_active(".stats/node");
5300 announced.assert_next_active("room/.internal");
5301 announced.assert_next_active("room/alice");
5302 announced.assert_next_active("room/catalog.pro");
5303 announced.assert_next_wait();
5304
5305 let mut announced = consumer
5307 .scope(".stats", &Patterns::from(Pattern::all()))
5308 .unwrap()
5309 .announced();
5310 announced.assert_next_active("node");
5311 announced.assert_next_wait();
5312 let mut announced = consumer.scope("", &scopes(&["room/.internal"])).unwrap().announced();
5313 announced.assert_next_active("room/.internal");
5314 announced.assert_next_wait();
5315
5316 let mut announced = consumer.clone().with_hidden(true).beyond(&consumer).announced();
5319 announced.assert_next_active(".stats/node");
5320 announced.assert_next_active("room/.internal");
5321 announced.assert_next_wait();
5322 let room = consumer.scope("", &scopes(&["room"])).unwrap().beyond(&consumer);
5323 let mut announced = room.announced();
5324 announced.assert_next_wait();
5325 let stats = consumer.scope("", &scopes(&[".stats"])).unwrap().beyond(&consumer);
5326 let mut announced = stats.announced();
5327 announced.assert_next_active(".stats/node");
5328 announced.assert_next_wait();
5329 }
5330
5331 #[tokio::test]
5333 async fn hidden_broadcast_resolves_by_path() {
5334 let producer = origin(1).produce();
5335 let consumer = producer.consume();
5336 let broadcast = producer.create_broadcast(".stats/node").unwrap();
5337 broadcast.announce(Route::default()).unwrap();
5338
5339 consumer.announced().assert_next_wait();
5340 let resolved = consumer.request_broadcast(".stats/node").await.expect("resolves");
5341 assert_eq!(resolved.info().path.as_str(), ".stats/node");
5342 }
5343
5344 fn mounted(producer: &Producer, pid: &str, patterns: &[&str]) -> Consumer {
5346 let patterns: Patterns = patterns.iter().map(|pattern| pattern.parse().unwrap()).collect();
5347 producer
5348 .mount(format!("{pid}/.svc"), format!(".svc/{pid}"))
5349 .unwrap()
5350 .scope(pid, &patterns)
5351 .unwrap()
5352 .consume()
5353 }
5354
5355 #[tokio::test]
5358 async fn mount_resolves_on_the_target_front() {
5359 let producer = origin(1).produce();
5360 let server = producer.dynamic(".svc", Route::default()).unwrap();
5361 let project = mounted(&producer, "p1", &["**"]);
5362
5363 let through = project.request_broadcast(".svc/foo");
5364 let direct = producer.consume().request_broadcast(".svc/p1/foo");
5365
5366 let request = queued(&server).await;
5367 assert_eq!(request.path().as_str(), ".svc/p1/foo");
5368 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5369 let source = broadcast::Info::new().produce();
5370 request.accept(&source);
5371
5372 let through = through.await.expect("resolves");
5373 let direct = direct.await.expect("resolves");
5374 assert!(through.is_clone(&direct));
5375 assert_eq!(through.info().path.as_str(), ".svc/foo");
5377 }
5378
5379 #[tokio::test]
5382 async fn mount_presents_target_routes_under_the_mount() {
5383 let producer = origin(1).produce();
5384 let _claim = producer.announce(".svc", Route::default()).unwrap();
5385 let foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5386 let _other = producer.publish(".svc/p2/bar", Route::default()).unwrap();
5387 let _cam = producer.publish("p1/cam", Route::default()).unwrap();
5388 let _shadowed = producer.publish("p1/.svc/forged", Route::default()).unwrap();
5389 let project = mounted(&producer, "p1", &["**"]);
5390
5391 let mut announced = project.announced();
5393 announced.assert_next_active("cam");
5394 announced.assert_next_wait();
5395
5396 let mut announced = project.clone().with_hidden(true).announced();
5397 announced.assert_next_active(".svc");
5398 announced.assert_next_active(".svc/foo");
5399 announced.assert_next_active("cam");
5400 announced.assert_next_wait();
5401
5402 let mut inside = project
5404 .scope(".svc", &Patterns::from(Pattern::all()))
5405 .unwrap()
5406 .announced();
5407 inside.assert_next_active("");
5408 inside.assert_next_active("foo");
5409 inside.assert_next_wait();
5410
5411 drop(foo);
5412 announced.assert_next_ended(".svc/foo");
5413 inside.assert_next_ended("foo");
5414 announced.assert_next_wait();
5415
5416 let err = project.request_broadcast(".svc/forged").await.err().unwrap();
5418 assert!(matches!(err, Error::NotFound | Error::Unroutable), "{err:?}");
5419 }
5420
5421 #[tokio::test]
5424 async fn mount_authorizes_the_named_path() {
5425 let producer = origin(1).produce();
5426 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5427 let _bar = producer.publish(".svc/p1/bar", Route::default()).unwrap();
5428
5429 let granted = mounted(&producer, "p1", &["foo", ".svc/foo"]);
5430 granted.request_broadcast(".svc/foo").await.expect("granted");
5431 let refused = granted
5432 .request_broadcast(".svc/bar")
5433 .now_or_never()
5434 .expect("refused at once");
5435 assert!(matches!(refused, Err(Error::Unauthorized)));
5436 let mut announced = granted.clone().with_hidden(true).announced();
5437 announced.assert_next_active(".svc/foo");
5438 announced.assert_next_wait();
5439
5440 let narrower = mounted(&producer, "p1", &["foo"]);
5441 let refused = narrower
5442 .request_broadcast(".svc/foo")
5443 .now_or_never()
5444 .expect("refused at once");
5445 assert!(matches!(refused, Err(Error::Unauthorized)));
5446 narrower.clone().with_hidden(true).announced().assert_next_wait();
5447
5448 let other = mounted(&producer, "p2", &["**"]);
5450 other.clone().with_hidden(true).announced().assert_next_wait();
5451 let err = other.request_broadcast(".svc/foo").await.err().unwrap();
5452 assert!(matches!(err, Error::Unroutable));
5453 }
5454
5455 #[tokio::test]
5458 async fn mount_captures_the_named_path() {
5459 let producer = origin(1).produce();
5460 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5461 let project = mounted(&producer, "p1", &["**"]).with_hidden(true);
5462
5463 let update = project.announced().try_next().expect("foo");
5464 assert_eq!(update.prefix.as_str(), ".svc/foo");
5465 assert_eq!(update.captures, Some(vec![".svc/foo".parse::<Pattern>().unwrap()]));
5466
5467 let inside = project.scope(".svc", &Patterns::from(Pattern::all())).unwrap();
5468 let update = inside.announced().try_next().expect("foo");
5469 assert_eq!(update.prefix.as_str(), "foo");
5470 assert_eq!(update.captures, Some(vec!["foo".parse::<Pattern>().unwrap()]));
5471 }
5472
5473 #[tokio::test]
5476 async fn mount_keeps_a_max_depth_target() {
5477 let producer = origin(1).produce();
5478 let deep = vec!["d"; Path::MAX_PARTS].join("/");
5479 let _leaf = producer.publish(deep.as_str(), Route::default()).unwrap();
5480 let project = producer
5481 .mount("p1/.svc", deep.as_str())
5482 .unwrap()
5483 .scope("p1", &Patterns::from(Pattern::all()))
5484 .unwrap()
5485 .consume()
5486 .with_hidden(true);
5487
5488 let mut announced = project.announced();
5489 announced.assert_next_active(".svc");
5490 announced.assert_next_wait();
5491
5492 let err = project.request_broadcast(".svc/x").await.err().unwrap();
5494 assert!(matches!(err, Error::BoundsExceeded(_)), "{err:?}");
5495 let inside = project.scope(".svc/x", &Patterns::from(Pattern::all())).unwrap();
5496 inside.announced().assert_next_wait();
5497 }
5498
5499 #[tokio::test]
5502 async fn mount_bounds_the_named_path() {
5503 let producer = origin(1).produce();
5504 let _near = producer.publish("t/x", Route::default()).unwrap();
5505 let deep = format!("t/{}", vec!["d"; Path::MAX_PARTS - 1].join("/"));
5506 let _deep = producer.publish(deep.as_str(), Route::default()).unwrap();
5507 let project = producer
5508 .mount("p1/a/b", "t")
5509 .unwrap()
5510 .scope("p1", &Patterns::from(Pattern::all()))
5511 .unwrap()
5512 .consume();
5513
5514 let mut announced = project.announced();
5515 announced.assert_next_active("a/b/x");
5516 announced.assert_next_wait();
5517
5518 let over = vec!["d"; Path::MAX_PARTS + 1].join("/");
5519 assert!(matches!(
5520 producer.mount(over.as_str(), "t"),
5521 Err(Error::BoundsExceeded(_))
5522 ));
5523 }
5524
5525 #[tokio::test]
5527 async fn mount_is_read_only() {
5528 let producer = origin(1).produce();
5529 let project = producer
5530 .mount("p1/.svc", ".svc/p1")
5531 .unwrap()
5532 .scope("p1", &Patterns::from(Pattern::all()))
5533 .unwrap();
5534 assert!(matches!(project.create_broadcast(".svc/foo"), Err(Error::Unauthorized)));
5535 assert!(matches!(
5536 project.dynamic(".svc", Route::default()),
5537 Err(Error::Unauthorized)
5538 ));
5539 assert!(matches!(
5540 project.dynamic(".svc/foo", Route::default()),
5541 Err(Error::Unauthorized)
5542 ));
5543 project.publish("cam", Route::default()).unwrap();
5544 }
5545
5546 #[test]
5548 fn mount_never_widens_a_scope() {
5549 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5550 let project = producer.scope("", &scopes(&["p1"])).unwrap();
5551 assert!(matches!(project.mount("p1/.svc", ".svc/p1"), Err(Error::Unauthorized)));
5552
5553 let mounted = producer.mount("p1/.svc", ".svc/p1").unwrap();
5554 assert!(matches!(mounted.mount("p1/.svc/x", ".other"), Err(Error::Duplicate)));
5555 assert!(matches!(mounted.mount("p1", ".other"), Err(Error::Duplicate)));
5556 assert!(matches!(mounted.mount("p2", "p1/.svc/x"), Err(Error::Duplicate)));
5558 assert!(matches!(mounted.mount("p2", "p1"), Err(Error::Duplicate)));
5559 assert!(matches!(mounted.mount(".svc/p1/x", ".other"), Err(Error::Duplicate)));
5561 assert!(matches!(mounted.mount(".svc", ".other"), Err(Error::Duplicate)));
5562 mounted.mount("p1/.other", ".other/p1").unwrap();
5563 mounted.mount("p2/.svc", ".svc/p1").unwrap();
5565 for (at, target) in [("a", "a/b"), ("a/b", "a"), ("a", "a")] {
5567 assert!(
5568 matches!(producer.mount(at, target), Err(Error::Duplicate)),
5569 "{at} -> {target}"
5570 );
5571 }
5572 }
5573
5574 #[test]
5576 fn mount_refuses_a_wildcard_mount_point() {
5577 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5578 for at in ["*", "p1/*", "p1/**", "p1/a*"] {
5579 assert!(
5580 matches!(producer.mount(at, ".svc/p1"), Err(Error::InvalidPath(_))),
5581 "{at}"
5582 );
5583 }
5584 }
5585
5586 #[tokio::test]
5589 async fn mount_egress_counts_under_the_named_path() {
5590 let registry = stats::Registry::new(stats::Config::new());
5591 let producer = origin(1).produce();
5592 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5593 let project = mounted(&producer, "p1", &["**"])
5594 .with_stats(registry.tier(stats::Tier::default()).session("p1"))
5595 .with_hidden(true);
5596
5597 let mut announced = project.announced();
5598 announced.assert_next_active(".svc/foo");
5599 project.request_broadcast(".svc/foo").await.expect("resolves");
5600
5601 let mut report = stats::Report::default();
5602 registry.report(&mut report);
5603 let paths: Vec<_> = report
5604 .traffic
5605 .iter()
5606 .map(|entry| entry.path.as_str().to_string())
5607 .collect();
5608 assert_eq!(paths, ["p1/.svc/foo"]);
5609 }
5610
5611 #[tokio::test]
5613 async fn hidden_route_announced_later_stays_hidden() {
5614 let producer = origin(1).produce();
5615 let consumer = producer.consume();
5616 let mut announced = consumer.announced();
5617 let mut opted = consumer.clone().with_hidden(true).announced();
5618
5619 let hidden = producer.announce(".stats/node", Route::default()).unwrap();
5620 announced.assert_next_wait();
5621 opted.assert_next_active(".stats/node");
5622
5623 drop(hidden);
5624 announced.assert_next_wait();
5625 opted.assert_next_ended(".stats/node");
5626 }
5627
5628 #[tokio::test]
5629 async fn broadcast_announces_its_own_path() {
5630 let producer = origin(1).produce();
5631 let consumer = producer.consume();
5632 let mut announced = consumer.announced();
5633 let mut peer = consumer.clone().excluding(Hop::UNKNOWN).announced();
5634
5635 let broadcast = producer.create_broadcast("room/alice").unwrap();
5637 announced.assert_next_wait();
5638 peer.assert_next_wait();
5639
5640 broadcast.announce(Route::default().with_cost(3)).unwrap();
5641 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(3));
5642 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(3));
5643
5644 broadcast.announce(Route::default().with_cost(1)).unwrap();
5646 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(1));
5647 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(1));
5648
5649 broadcast.unannounce();
5651 announced.assert_next_ended("room/alice");
5652 peer.assert_next_ended("room/alice");
5653 broadcast.unannounce();
5654 announced.assert_next_wait();
5655 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5656 assert!(matches!(err, Error::Unroutable));
5657
5658 broadcast.announce(Route::default()).unwrap();
5660 announced.assert_next_active("room/alice");
5661 peer.assert_next_active("room/alice");
5662 broadcast.close();
5663 announced.assert_next_ended("room/alice");
5664 peer.assert_next_ended("room/alice");
5665 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5666 announced.assert_next_wait();
5667 }
5668
5669 #[tokio::test]
5670 async fn broadcast_announcement_retracts_with_the_last_producer() {
5671 let producer = origin(1).produce();
5672 let consumer = producer.consume();
5673 let mut announced = consumer.announced();
5674
5675 let broadcast = producer.create_broadcast("room/alice").unwrap();
5676 let clone = broadcast.clone();
5677 broadcast.announce(Route::default()).unwrap();
5678 announced.assert_next_active("room/alice");
5679
5680 drop(broadcast);
5682 announced.assert_next_wait();
5683 drop(clone);
5684 announced.assert_next_ended("room/alice");
5685 }
5686
5687 #[tokio::test]
5688 async fn publish_creates_and_announces_together() {
5689 let producer = origin(1).produce();
5690 let mut announced = producer.consume().announced();
5691 let _broadcast = producer.publish("room/alice", Route::default()).unwrap();
5692 announced.assert_next_active("room/alice");
5693 }
5694
5695 #[tokio::test]
5696 async fn standalone_broadcast_cannot_announce() {
5697 let broadcast = broadcast::Info::new().produce();
5698 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5699 broadcast.unannounce();
5701 }
5702
5703 #[tokio::test]
5704 async fn announce_replays_to_late_cursor() {
5705 let producer = origin(1).produce();
5706 let _a = producer.announce("room/alice", Route::default()).unwrap();
5707 let _b = producer.announce("room/bob", Route::default()).unwrap();
5708
5709 let mut announced = producer.consume().announced();
5710 announced.assert_next_active("room/alice");
5712 announced.assert_next_active("room/bob");
5713 announced.assert_next_wait();
5714 }
5715
5716 #[tokio::test]
5717 async fn announce_keeps_its_prefix_under_a_producer_scope() {
5718 let producer = origin(1).produce();
5719 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
5720
5721 let _a = scoped.announce("", Route::default()).unwrap();
5723 let mut announced = producer.consume().announced();
5724 announced.assert_next_active("");
5725
5726 assert!(matches!(
5728 scoped.announce("other", Route::default()),
5729 Err(Error::Unauthorized)
5730 ));
5731 }
5732
5733 #[tokio::test]
5734 async fn cursor_keeps_an_overlapping_prefix_above_its_scope() {
5735 let producer = origin(1).produce();
5736 let _a = producer.announce("", Route::default()).unwrap();
5737
5738 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
5739 let mut announced = consumer.announced();
5740 announced.assert_next_active("");
5741 }
5742
5743 #[tokio::test]
5744 async fn cursor_root_strips_prefix() {
5745 let producer = origin(1).produce();
5746 let _a = producer.announce("room/alice", Route::default()).unwrap();
5747
5748 let consumer = producer
5749 .consume()
5750 .scope("room", &Patterns::from(Pattern::all()))
5751 .unwrap();
5752 let mut announced = consumer.announced();
5753 announced.assert_next_active("alice");
5754 }
5755
5756 #[tokio::test]
5757 async fn best_route_wins_and_fails_over() {
5758 let producer = origin(1).produce();
5759 let mut announced = producer.consume().announced();
5760
5761 let expensive = producer
5762 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5763 .unwrap();
5764 let route = announced.assert_next_active("room");
5765 assert_eq!(route.cost, Cost::new(5));
5766
5767 let cheap = producer
5769 .announce("room", Route::default().with_hops(hops(&[20])).with_cost(1))
5770 .unwrap();
5771 let route = announced.assert_next_active("room");
5772 assert_eq!(route.cost, Cost::new(1));
5773
5774 drop(cheap);
5776 let route = announced.assert_next_active("room");
5777 assert_eq!(route.cost, Cost::new(5));
5778
5779 drop(expensive);
5781 announced.assert_next_ended("room");
5782 }
5783
5784 #[tokio::test]
5785 async fn identical_reannounce_is_invisible() {
5786 let producer = origin(1).produce();
5787 let mut announced = producer.consume().announced();
5788
5789 let old = producer
5790 .announce("room", Route::default().with_hops(hops(&[10])))
5791 .unwrap();
5792 let first = announced.assert_next_active("room");
5793 assert_eq!(first.hops.as_slice(), hops(&[10]).as_slice());
5794
5795 let _new = producer
5799 .announce("room", Route::default().with_hops(hops(&[10])))
5800 .unwrap();
5801 announced.assert_next_wait();
5802
5803 drop(old);
5805 announced.assert_next_wait();
5806 }
5807
5808 #[tokio::test]
5809 async fn exclude_hides_routes_through_the_peer() {
5810 let producer = origin(1).produce();
5811 let _a = producer
5812 .announce("room", Route::default().with_hops(hops(&[7])))
5813 .unwrap();
5814
5815 let mut hidden = producer.consume().excluding(origin(7)).announced();
5816 hidden.assert_next_wait();
5817
5818 let mut visible = producer.consume().excluding(origin(8)).announced();
5819 visible.assert_next_active("room");
5820 }
5821
5822 #[tokio::test]
5823 async fn exclude_matches_via_when_the_chain_is_anonymous() {
5824 let producer = origin(1).produce();
5825 let assigned = origin(777);
5826 let _echoed = producer
5827 .announce("echoed", Route::default().with_hops(hops(&[0])).with_via(assigned))
5828 .unwrap();
5829 let _local = producer
5830 .announce("local", Route::default().with_hops(hops(&[10])))
5831 .unwrap();
5832
5833 let mut hidden = producer.consume().excluding(assigned).announced();
5834 hidden.assert_next_active("local");
5835 hidden.assert_next_wait();
5836 }
5837
5838 #[tokio::test]
5839 async fn anonymous_route_loses_to_identified_at_any_cost() {
5840 let producer = origin(1).produce();
5841 let mut announced = producer.consume().announced();
5842
5843 let _anonymous = producer
5844 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5845 .unwrap();
5846 let route = announced.assert_next_active("room");
5847 assert!(route.is_anonymous());
5848 assert_eq!(route.cost, Cost::new(1));
5849
5850 let _identified = producer
5851 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5852 .unwrap();
5853 let route = announced.assert_next_active("room");
5854 assert!(!route.is_anonymous());
5855 assert_eq!(route.cost, Cost::new(5));
5856 }
5857
5858 #[tokio::test]
5861 async fn a_stamped_route_ranks_below_an_identified_one() {
5862 let producer = origin(1).produce();
5863 let mut announced = producer.consume().announced();
5864
5865 let mut stamped = Hops::new();
5866 stamped.stamp(origin(5)).unwrap();
5867 assert_eq!(stamped.as_slice(), &[origin(5), Hop::UNKNOWN]);
5868
5869 let _legacy = producer
5870 .announce("room", Route::default().with_hops(stamped).with_cost(0))
5871 .unwrap();
5872 let route = announced.assert_next_active("room");
5873 assert!(route.is_anonymous());
5874
5875 let _identified = producer
5876 .announce("room", Route::default().with_hops(hops(&[10, 11])).with_cost(5))
5877 .unwrap();
5878 let route = announced.assert_next_active("room");
5879 assert_eq!(route.hops.as_slice(), hops(&[10, 11]).as_slice());
5880 }
5881
5882 #[test]
5883 fn stamping_keeps_the_leading_zero_and_names_nothing_else() {
5884 let mut chain = hops(&[0, 7]);
5885 chain.stamp(origin(5)).unwrap();
5886 assert_eq!(chain.as_slice(), hops(&[5, 0, 7]).as_slice());
5887
5888 let mut named = hops(&[7, 0]);
5890 named.stamp(origin(5)).unwrap();
5891 assert_eq!(named.as_slice(), hops(&[7, 0]).as_slice());
5892
5893 let mut full = Hops::try_from(vec![Hop::UNKNOWN; MAX_HOPS]).unwrap();
5895 assert_eq!(full.stamp(origin(5)), Err(InvalidHop::TooMany));
5896 }
5897
5898 #[tokio::test]
5899 async fn anonymous_routes_order_by_cost() {
5900 let producer = origin(1).produce();
5901 let mut announced = producer.consume().announced();
5902
5903 let expensive = producer
5904 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(5))
5905 .unwrap();
5906 let route = announced.assert_next_active("room");
5907 assert_eq!(route.cost, Cost::new(5));
5908
5909 let _cheap = producer
5910 .announce("room", Route::default().with_hops(hops(&[0, 7])).with_cost(1))
5911 .unwrap();
5912 let route = announced.assert_next_active("room");
5913 assert!(route.is_anonymous());
5914 assert_eq!(route.cost, Cost::new(1));
5915
5916 drop(expensive);
5917 announced.assert_next_wait();
5918 }
5919
5920 #[tokio::test]
5921 async fn anonymous_chain_from_identified_peer_still_ranks_last() {
5922 let producer = origin(1).produce();
5923 let mut announced = producer.consume().announced();
5924
5925 let _anonymous = producer
5926 .announce(
5927 "room",
5928 Route::default()
5929 .with_hops(hops(&[0, 7]))
5930 .with_cost(1)
5931 .with_via(origin(7)),
5932 )
5933 .unwrap();
5934 announced.assert_next_active("room");
5935
5936 let _identified = producer
5937 .announce("room", Route::default().with_hops(hops(&[10, 20])).with_cost(5))
5938 .unwrap();
5939 let route = announced.assert_next_active("room");
5940 assert!(!route.is_anonymous());
5941 assert_eq!(route.cost, Cost::new(5));
5942 }
5943
5944 #[tokio::test]
5945 async fn request_prefers_identified_over_cheaper_anonymous() {
5946 let producer = origin(1).produce();
5947 let consumer = producer.consume();
5948
5949 let anonymous = producer
5950 .dynamic("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5951 .unwrap();
5952 let identified = producer
5953 .dynamic("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5954 .unwrap();
5955
5956 let _pending = consumer.request_broadcast("room/alice");
5957 let request = queued(&identified).await;
5958 assert_eq!(request.path().as_str(), "room/alice");
5959 assert!(
5960 anonymous.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
5961 "the cheaper anonymous route must not serve"
5962 );
5963 }
5964
5965 #[tokio::test]
5966 async fn update_reprices_in_place() {
5967 let producer = origin(1).produce();
5968 let mut announced = producer.consume().announced();
5969
5970 let announcement = producer.announce("room", Route::default()).unwrap();
5971 announced.assert_next_active("room");
5972
5973 announcement.update(Route::default().with_cost(9)).unwrap();
5974 let route = announced.assert_next_active("room");
5975 assert_eq!(route.cost, Cost::new(9));
5976 }
5977
5978 #[tokio::test]
5979 async fn retract_after_undelivered_reprice_still_delivered() {
5980 let producer = origin(1).produce();
5981 let mut announced = producer.consume().announced();
5982
5983 let announcement = producer.announce("room", Route::default()).unwrap();
5984 announced.assert_next_active("room");
5985
5986 announcement.update(Route::default().with_cost(9)).unwrap();
5990 drop(announcement);
5991 announced.assert_next_ended("room");
5992 announced.assert_next_wait();
5993 }
5994
5995 #[tokio::test]
5996 async fn scoped_cursor_advertises_most_specific_covering_route() {
5997 let producer = origin(1).produce();
5998 let _broad = producer.announce("room", Route::default().with_cost(1)).unwrap();
6001 let _narrow = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
6002
6003 let consumer = producer
6004 .consume()
6005 .scope("room/alice", &Patterns::from(Pattern::all()))
6006 .unwrap();
6007 let mut announced = consumer.announced();
6008 let route = announced.assert_next_active("");
6009 assert_eq!(route.cost, Cost::new(9));
6010 announced.assert_next_wait();
6011 }
6012
6013 #[tokio::test]
6014 async fn capture_change_retracts_before_reannouncing_a_presented_prefix() {
6015 let producer = origin(1).produce();
6016 let _broad = producer.announce("room", Route::default()).unwrap();
6017 let exact = producer.announce("room/alice", Route::default()).unwrap();
6018 let consumer = producer
6019 .consume()
6020 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
6021 .unwrap()
6022 .scope("room/alice", &Patterns::from(Pattern::all()))
6023 .unwrap();
6024 let mut announced = consumer.announced();
6025
6026 let first = announced.next().now_or_never().expect("next").expect("announce");
6027 assert_eq!(first.prefix.as_str(), "");
6028 assert_eq!(first.kind, AnnounceKind::Announced);
6029 assert_eq!(first.captures, Some(Vec::new()));
6030
6031 drop(exact);
6032 let retracted = announced.next().now_or_never().expect("next").expect("retract");
6033 assert_eq!(retracted.prefix.as_str(), "");
6034 assert_eq!(retracted.kind, AnnounceKind::Retracted);
6035 assert_eq!(retracted.captures, Some(Vec::new()));
6036 let replacement = announced.next().now_or_never().expect("next").expect("announce");
6037 assert_eq!(replacement.prefix.as_str(), "");
6038 assert_eq!(replacement.kind, AnnounceKind::Announced);
6039 assert_eq!(replacement.captures, None);
6040 }
6041
6042 #[tokio::test]
6043 async fn routed_broadcast_resolves_once_announced() {
6044 let producer = origin(1).produce();
6045 let consumer = producer.consume();
6046
6047 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
6049 assert!((&mut resolving).now_or_never().is_none());
6050
6051 let broadcast = producer.create_broadcast("room/alice").unwrap();
6053 for _ in 0..20 {
6054 tokio::task::yield_now().await;
6055 }
6056 assert!((&mut resolving).now_or_never().is_none());
6057
6058 broadcast.announce(Route::default()).unwrap();
6059 let resolved = resolving.await.expect("resolves once announced");
6060 assert_eq!(resolved.info().path.as_str(), "room/alice");
6061 drop(broadcast);
6062 }
6063
6064 #[tokio::test]
6067 async fn cheaper_remote_route_beats_a_local_broadcast() {
6068 let producer = origin(1).produce();
6069 let consumer = producer.consume();
6070 let mut announced = consumer.announced();
6071
6072 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
6073 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(5));
6074
6075 let server = producer
6076 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
6077 .unwrap();
6078 let route = announced.assert_next_active("room/alice");
6079 assert_eq!(route.cost, Cost::new(1));
6080 assert_eq!(route.hops, hops(&[10]));
6081
6082 let pending = consumer.request_broadcast("room/alice");
6084 let request = queued(&server).await;
6085 let upstream = broadcast::Info::new().produce();
6086 request.accept(&upstream);
6087 pending.await.expect("resolves through the cheaper route");
6088 }
6089
6090 #[tokio::test]
6094 async fn cheaper_route_after_a_front_wins_new_requests() {
6095 let producer = origin(1).produce();
6096 let consumer = producer.consume();
6097
6098 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
6099 let first = consumer
6100 .request_broadcast("room/alice")
6101 .await
6102 .expect("resolves locally");
6103
6104 let server = producer
6105 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
6106 .unwrap();
6107
6108 let pending = consumer.request_broadcast("room/alice");
6109 let request = queued(&server).await;
6110 let upstream = broadcast::Info::new().produce();
6111 request.accept(&upstream);
6112 let second = pending.await.expect("resolves through the cheaper route");
6113
6114 assert!(!first.is_closed(), "the old front must keep serving its readers");
6115 assert!(!first.is_clone(&second), "the newcomer must not join the old front");
6116 }
6117
6118 #[tokio::test]
6122 async fn local_broadcast_wins_a_tie_with_a_hopless_route() {
6123 let producer = origin(1).produce();
6124 let consumer = producer.consume();
6125
6126 let _local = producer.publish("room/alice", Route::default()).unwrap();
6127 let server = producer.dynamic("room/alice", Route::default()).unwrap();
6128
6129 let resolved = tokio::time::timeout(Duration::from_secs(1), consumer.request_broadcast("room/alice"))
6131 .await
6132 .expect("the newer hopless route won the tie")
6133 .expect("resolves");
6134 assert_eq!(resolved.info().path.as_str(), "room/alice");
6135 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
6136 }
6137
6138 #[tokio::test]
6141 async fn announce_stats_follow_the_advertisement() {
6142 let registry = stats::Registry::new(stats::Config::new());
6143 let producer = origin(1)
6144 .produce()
6145 .with_stats(registry.tier(stats::Tier::default()).session("root"));
6146 let announces = || {
6147 registry
6148 .snapshot()
6149 .traffic()
6150 .into_iter()
6151 .find(|(_, role, _)| *role == stats::Role::Subscriber)
6152 .map(|(_, _, traffic)| (traffic.announces_started, traffic.announces_ended))
6153 .unwrap_or_default()
6154 };
6155
6156 let broadcast = producer.create_broadcast("room/alice").unwrap();
6157 assert_eq!(announces(), (0, 0), "a hidden broadcast is not announced");
6158 broadcast.announce(Route::default()).unwrap();
6159 broadcast
6160 .announce(Route {
6161 cost: Cost::new(3),
6162 ..Route::default()
6163 })
6164 .unwrap();
6165 assert_eq!(announces(), (1, 0), "a re-price is not another announce");
6166 broadcast.unannounce();
6167 assert_eq!(announces(), (1, 1));
6168 broadcast.announce(Route::default()).unwrap();
6169 drop(broadcast);
6170 assert_eq!(announces(), (2, 2));
6171 }
6172
6173 #[tokio::test]
6175 async fn local_broadcast_wins_a_cost_tie() {
6176 let producer = origin(1).produce();
6177 let consumer = producer.consume();
6178 let mut announced = consumer.announced();
6179
6180 let server = producer
6181 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(2))
6182 .unwrap();
6183 announced.assert_next_active("room/alice");
6184 let _local = producer.publish("room/alice", Route::default().with_cost(2)).unwrap();
6185 assert!(announced.assert_next_active("room/alice").hops.is_empty());
6186
6187 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6188 assert_eq!(resolved.info().path.as_str(), "room/alice");
6189 for _ in 0..20 {
6190 tokio::task::yield_now().await;
6191 }
6192 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
6193 }
6194
6195 #[tokio::test]
6198 async fn unannounce_ends_the_front() {
6199 let producer = origin(1).produce();
6200 let consumer = producer.consume();
6201
6202 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6203 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6204
6205 broadcast.unannounce();
6206 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
6207 assert!(matches!(err, Error::Unroutable), "joined a retracted front: {err}");
6208 settle(|| resolved.is_closed()).await;
6209 assert!(!broadcast.consume().is_closed(), "the broadcast itself lives on");
6210
6211 broadcast.announce(Route::default()).unwrap();
6213 let again = consumer.request_broadcast("room/alice").await.expect("resolves again");
6214 assert!(!again.is_clone(&resolved));
6215 }
6216
6217 #[tokio::test]
6221 async fn unannounce_keeps_a_track_awaiting_its_info() {
6222 let producer = origin(1).produce();
6223 let consumer = producer.consume();
6224
6225 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6226 let mut dynamic = broadcast.dynamic();
6227 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6228 let track = resolved.track("video").unwrap();
6229 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6230 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6231 .await
6232 .expect("the front asked the source")
6233 .expect("request");
6234
6235 broadcast.unannounce();
6236 settle(|| resolved.is_closed()).await;
6237
6238 let source = request.accept(None);
6239 let mut group = source.append_group().unwrap();
6240 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
6241 group.finish().unwrap();
6242 source.finish().unwrap();
6243
6244 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
6245 let mut group = subscription.recv_group().await.unwrap().expect("the source's group");
6246 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late");
6247 assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly");
6248 }
6249
6250 async fn served_front() -> (Dynamic, broadcast::Producer, broadcast::Dynamic, broadcast::Consumer) {
6253 let producer = origin(1).produce();
6254 let consumer = producer.consume();
6255 let server = producer
6256 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
6257 .unwrap();
6258 let pending = consumer.request_broadcast("room/alice");
6259 let upstream = broadcast::Info::new().produce();
6260 let dynamic = upstream.dynamic();
6261 queued(&server).await.accept(&upstream);
6262 let resolved = pending.await.expect("resolves");
6263 (server, upstream, dynamic, resolved)
6264 }
6265
6266 #[tokio::test]
6271 async fn returning_reader_skips_a_warm_cache_the_copy_resolved_past() {
6272 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
6273 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6274 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
6275
6276 let track = resolved.track("audio").unwrap();
6277 let b = budget.clone();
6278 let subscribing = tokio::spawn(async move { track.subscribe(b).await });
6279 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6280 .await
6281 .expect("the front asked the source")
6282 .expect("request");
6283 let source = request.resolving_start().accept(None);
6284 for seq in 0..4u64 {
6285 let mut group = source.create_group(seq.into()).unwrap();
6286 group.write_frame(ms(seq * 20), b"old".as_ref()).unwrap();
6287 group.finish().unwrap();
6288 }
6289 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6290 subscription.recv_group().await.unwrap().expect("the live group");
6291 drop(subscription);
6292 tokio::time::timeout(Duration::from_secs(1), source.unused())
6293 .await
6294 .expect("parked")
6295 .expect("source open");
6296 drop(source);
6297
6298 let track = resolved.track("audio").unwrap();
6299 let subscribing = tokio::spawn(async move { track.subscribe(budget).await });
6300 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6301 .await
6302 .expect("the front asked the source again")
6303 .expect("request");
6304 let mut source = request.resolving_start().accept(None);
6305 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6306
6307 assert!(
6309 tokio::time::timeout(Duration::from_millis(50), subscription.recv_group())
6310 .await
6311 .is_err(),
6312 "the warm cache was served before the copy resolved its start"
6313 );
6314
6315 source.start_at(20).unwrap();
6317 let mut group = source.create_group(20u64.into()).unwrap();
6318 group.write_frame(ms(2000), b"new".as_ref()).unwrap();
6319 group.finish().unwrap();
6320 let group = subscription.recv_group().await.unwrap().expect("the live group");
6321 assert_eq!(group.sequence, 20, "a stale warm group was served");
6322 }
6323
6324 #[tokio::test]
6329 async fn returning_reader_replays_a_current_warm_cache() {
6330 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6331
6332 let track = resolved.track("catalog").unwrap();
6333 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6334 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6335 .await
6336 .expect("the front asked the source")
6337 .expect("request");
6338 let source = request.resolving_start().accept(None);
6339 let mut group = source.create_group(0u64.into()).unwrap();
6340 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6341 group.finish().unwrap();
6342 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6343 subscription.recv_group().await.unwrap().expect("the catalog");
6344 drop(subscription);
6345 tokio::time::timeout(Duration::from_secs(1), source.unused())
6346 .await
6347 .expect("parked")
6348 .expect("source open");
6349 drop(source);
6350
6351 let track = resolved.track("catalog").unwrap();
6352 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6353 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6354 .await
6355 .expect("the front asked the source again")
6356 .expect("request");
6357 let mut source = request.resolving_start().accept(None);
6358 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6359
6360 let reading = tokio::spawn(async move {
6363 let mut group = subscription.recv_group().await.unwrap().expect("the catalog");
6364 assert_eq!(group.sequence, 0);
6365 group.read_frame().await.unwrap().expect("the snapshot").payload
6366 });
6367 tokio::task::yield_now().await;
6368 assert_eq!(
6369 source.subscription().and_then(|sub| sub.start),
6370 Some(track::Position { group: 0, frame: 1 }),
6371 "the re-splice asked past the cached catalog"
6372 );
6373 source.start_at(0).unwrap();
6374 let mut tail = source.create_group(0u64.into()).unwrap();
6375 tail.start_at(1).unwrap();
6376 tail.finish().unwrap();
6377 let payload = tokio::time::timeout(Duration::from_secs(1), reading)
6378 .await
6379 .expect("the returning reader never got the catalog")
6380 .unwrap();
6381 assert_eq!(&payload[..], b"snapshot");
6382 }
6383
6384 #[tokio::test(start_paused = true)]
6388 async fn finished_track_is_forgotten_after_the_linger() {
6389 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6390
6391 let track = resolved.track("catalog").unwrap();
6392 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6393 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6394 .await
6395 .expect("the front asked the source")
6396 .expect("request");
6397 let source = request.resolving_start().accept(None);
6398 let mut group = source.create_group(0u64.into()).unwrap();
6399 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6400 group.finish().unwrap();
6401 source.finish().unwrap();
6402 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6403 assert_eq!(
6404 next_group(&mut subscription)
6405 .await
6406 .unwrap()
6407 .expect("the catalog")
6408 .sequence,
6409 0
6410 );
6411 assert!(next_group(&mut subscription).await.unwrap().is_none());
6412 drop(subscription);
6413 drop(source);
6414
6415 let mut subscription = resolved.track("catalog").unwrap().subscribe(None).await.unwrap();
6417 assert_eq!(
6418 next_group(&mut subscription)
6419 .await
6420 .unwrap()
6421 .expect("the catalog")
6422 .sequence,
6423 0
6424 );
6425 assert!(next_group(&mut subscription).await.unwrap().is_none());
6426 drop(subscription);
6427 assert!(
6428 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6429 .await
6430 .is_err(),
6431 "a finished track within the linger asked the source again"
6432 );
6433
6434 tokio::time::sleep(TRACK_IDLE_LINGER).await;
6436
6437 let track = resolved.track("catalog").unwrap();
6438 let _subscribing = tokio::spawn(async move { track.subscribe(None).await });
6439 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6440 .await
6441 .expect("the finished track outlived the linger")
6442 .expect("request");
6443 }
6444
6445 #[tokio::test]
6450 async fn returning_reader_continues_an_open_warm_group() {
6451 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6452
6453 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6454 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6455 .await
6456 .expect("frame")
6457 .unwrap()
6458 .expect("group ended");
6459 frame.payload.to_vec()
6460 }
6461
6462 let mut expect: Vec<&[u8]> = Vec::new();
6463 let mut floor: Option<track::Position> = None;
6464 for (round, payload) in [b"a".as_ref(), b"b", b"c"].into_iter().enumerate() {
6465 let track = resolved.track("log").unwrap();
6466 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6467 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6468 .await
6469 .expect("the front asked the source")
6470 .expect("request");
6471 let mut source = request.resolving_start().accept(None);
6472 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6473
6474 source.start_at(0).unwrap();
6476 let mut group = source.create_group(0u64.into()).unwrap();
6477 if let Some(floor) = floor {
6478 group.start_at(floor.frame).unwrap();
6479 }
6480 group.write_frame(crate::Timestamp::ZERO, payload).unwrap();
6481 expect.push(payload);
6482 source
6483 .insert_datagram(10, crate::Timestamp::ZERO, b"datagram".as_ref())
6484 .unwrap();
6485
6486 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6487 .await
6488 .expect("group 0")
6489 .unwrap()
6490 .expect("track ended");
6491 assert_eq!(reading.sequence, 0);
6492 for frame in &expect {
6493 assert_eq!(read(&mut reading).await, *frame, "round {round}");
6494 }
6495 assert_eq!(
6496 source.subscription().and_then(|sub| sub.start),
6497 floor,
6498 "round {round} asked for the wrong continuation"
6499 );
6500
6501 drop(reading);
6502 drop(subscription);
6503 tokio::time::timeout(Duration::from_secs(1), source.unused())
6504 .await
6505 .expect("parked")
6506 .expect("source open");
6507 drop(group);
6508 drop(source);
6509 floor = Some(track::Position {
6510 group: 0,
6511 frame: expect.len() as u64,
6512 });
6513 }
6514 }
6515
6516 #[tokio::test]
6523 async fn a_takeover_keeps_the_open_group_head_for_later_readers() {
6524 tokio::time::pause();
6525 for dies_first in [true, false] {
6526 takeover_keeps_the_open_group_head(dies_first).await;
6527 }
6528 }
6529
6530 async fn takeover_keeps_the_open_group_head(dies_first: bool) {
6531 let producer = origin(1).produce();
6532 let first_server = producer
6533 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6534 .unwrap();
6535 let pending = producer.consume().request_broadcast("room/alice");
6536 let upstream = broadcast::Info::new().produce();
6537 let mut dynamic = upstream.dynamic();
6538 queued(&first_server).await.accept(&upstream);
6539 let resolved = pending.await.unwrap();
6540
6541 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6542 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6543 .await
6544 .expect("frame")
6545 .unwrap()
6546 .expect("group ended");
6547 frame.payload.to_vec()
6548 }
6549
6550 let track = resolved.track("log").unwrap();
6552 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6553 let source = dynamic.requested_track().await.unwrap().accept(None);
6554 let mut group = source.create_group(0u64.into()).unwrap();
6555 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6556 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6557 let mut subscription = subscribing.await.unwrap().unwrap();
6558 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6559 assert_eq!(read(&mut reading).await, b"a");
6560 assert_eq!(read(&mut reading).await, b"b");
6561
6562 let mut first = Some((group, source, upstream, dynamic));
6565 let mut joining = None;
6566 if dies_first {
6567 drop(first.take());
6568 assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none());
6571 let track = resolved.track("log").unwrap();
6572 joining = Some(tokio::spawn(async move {
6573 let mut joined = track.subscribe(None).await.unwrap();
6574 let group = joined.recv_group().await.unwrap().expect("track ended");
6575 (joined, group)
6576 }));
6577 }
6578 let cost = if dies_first { 5 } else { 0 };
6579 let replacement_server = producer
6580 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(cost))
6581 .unwrap();
6582 let replacement = broadcast::Info::new().produce();
6583 let mut replacement_dynamic = replacement.dynamic();
6584 queued(&replacement_server).await.accept(&replacement);
6585 let mut next = tokio::time::timeout(Duration::from_secs(1), replacement_dynamic.requested_track())
6586 .await
6587 .expect("the front asked the replacement")
6588 .unwrap()
6589 .resolving_start()
6590 .accept(None);
6591 next.start_at(0).unwrap();
6592 let mut next_group = next.create_group(0u64.into()).unwrap();
6593 next_group.start_at(2).unwrap();
6594 next_group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6595 assert_eq!(read(&mut reading).await, b"c", "dies_first={dies_first}");
6596
6597 if let Some((group, ..)) = &mut first {
6599 let mut independent = group.consume();
6600 group.write_frame(crate::Timestamp::ZERO, b"x".as_ref()).unwrap();
6601 for expect in [b"a", b"b", b"x"] {
6602 assert_eq!(read(&mut independent).await, expect);
6603 }
6604 group.finish().unwrap();
6605 }
6606 drop(first);
6607
6608 let track = resolved.track("log").unwrap();
6609 let mut later = tokio::time::timeout(Duration::from_secs(1), track.subscribe(None))
6610 .await
6611 .expect("subscribe")
6612 .unwrap();
6613 let mut late = tokio::time::timeout(Duration::from_secs(1), later.recv_group())
6614 .await
6615 .unwrap_or_else(|_| panic!("dies_first={dies_first}: the later reader never got the group"))
6616 .unwrap()
6617 .expect("track ended");
6618 assert_eq!(late.sequence, 0);
6619 for expect in [b"a", b"b", b"c"] {
6620 assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}");
6621 }
6622 let mut joined = None;
6623 if let Some(joining) = joining {
6624 let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining)
6625 .await
6626 .expect("the reader joining mid-outage never got the group")
6627 .unwrap();
6628 assert_eq!(group.sequence, 0);
6629 for expect in [b"a", b"b", b"c"] {
6630 assert_eq!(read(&mut group).await, expect);
6631 }
6632 joined = Some((subscription, group));
6633 }
6634
6635 assert!(
6637 tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6638 .await
6639 .is_err(),
6640 "dies_first={dies_first}: the group was handed out twice"
6641 );
6642
6643 next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap();
6645 assert_eq!(read(&mut reading).await, b"d");
6646 assert_eq!(read(&mut late).await, b"d");
6647 if let Some((_, group)) = &mut joined {
6648 assert_eq!(read(group).await, b"d");
6649 }
6650 next_group.finish().unwrap();
6651 drop((
6652 joined,
6653 reading,
6654 late,
6655 subscription,
6656 later,
6657 next,
6658 replacement,
6659 replacement_server,
6660 first_server,
6661 ));
6662 }
6663
6664 #[tokio::test]
6669 async fn returning_reader_gets_a_live_open_warm_group() {
6670 use futures::FutureExt;
6671
6672 tokio::time::pause();
6673 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6674
6675 let track = resolved.track("video").unwrap();
6677 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6678 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6679 .await
6680 .expect("the front asked the source")
6681 .expect("request");
6682 let mut source = request.resolving_start().accept(None);
6683 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6684 source.start_at(0).unwrap();
6685 let mut group = source.create_group(0u64.into()).unwrap();
6686 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6687 let mut reading = subscription.recv_group().await.unwrap().expect("group 0");
6688 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6689 drop(reading);
6690 drop(subscription);
6691 tokio::time::timeout(Duration::from_secs(1), source.unused())
6692 .await
6693 .expect("parked")
6694 .expect("source open");
6695 drop(group);
6696 drop(source);
6697
6698 let track = resolved.track("video").unwrap();
6700 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6701 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6702 .await
6703 .expect("the front asked the source")
6704 .expect("request");
6705 let mut source = request.resolving_start().accept(None);
6706 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6707 source.start_at(0).unwrap();
6708 let mut group = source.create_group(0u64.into()).unwrap();
6709 group.start_at(1).unwrap();
6710 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6711
6712 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6713 .await
6714 .expect("group 0")
6715 .unwrap()
6716 .expect("track ended");
6717 assert_eq!(reading.sequence, 0);
6718 let early = reading.finished().now_or_never();
6719 assert!(
6720 early.is_none(),
6721 "the live group already resolved its end before a frame was read: {early:?}"
6722 );
6723
6724 group.finish().unwrap();
6726 assert_eq!(reading.finished().await.unwrap(), 0);
6727 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6728 assert_eq!(reading.finished().await.unwrap(), 1);
6729 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"b");
6730 assert_eq!(reading.finished().await.unwrap(), 2);
6731 }
6732
6733 #[tokio::test]
6734 async fn warm_head_survives_another_takeover_before_park() {
6735 tokio::time::pause();
6736 let producer = origin(1).produce();
6737 let _server = producer
6738 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6739 .unwrap();
6740 let pending = producer.consume().request_broadcast("room/alice");
6741 let upstream = broadcast::Info::new().produce();
6742 let mut dynamic = upstream.dynamic();
6743 queued(&_server).await.accept(&upstream);
6744 let resolved = pending.await.unwrap();
6745
6746 let track = resolved.track("log").unwrap();
6747 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6748 let source = dynamic.requested_track().await.unwrap().accept(None);
6749 let mut group = source.create_group(0u64.into()).unwrap();
6750 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6751 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6752 let mut subscription = subscribing.await.unwrap().unwrap();
6753 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6754 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6755 drop(reading);
6756 drop(subscription);
6757 source.unused().await.unwrap();
6758 drop(group);
6759 drop(source);
6760
6761 let track = resolved.track("log").unwrap();
6762 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6763 let mut resumed = dynamic.requested_track().await.unwrap().resolving_start().accept(None);
6764 resumed.start_at(0).unwrap();
6765 let mut subscription = subscribing.await.unwrap().unwrap();
6766 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6767 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6768
6769 let replacement_server = producer
6770 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
6771 .unwrap();
6772 let replacement = broadcast::Info::new().produce();
6773 let mut replacement_dynamic = replacement.dynamic();
6774 queued(&replacement_server).await.accept(&replacement);
6775 let mut source = replacement_dynamic
6776 .requested_track()
6777 .await
6778 .unwrap()
6779 .resolving_start()
6780 .accept(None);
6781 source.start_at(0).unwrap();
6782 let mut group = source.create_group(0u64.into()).unwrap();
6783 group.start_at(2).unwrap();
6784 group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6785 let mut continuation = reading.clone();
6786 continuation.start_at(2);
6787 let frame = tokio::time::timeout(Duration::from_secs(1), continuation.read_frame())
6788 .await
6789 .expect("replacement frame")
6790 .unwrap()
6791 .unwrap();
6792 assert_eq!(&frame.payload[..], b"c");
6793 drop(continuation);
6794 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"b");
6795 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"c");
6796 drop(reading);
6797 drop(subscription);
6798 source.unused().await.unwrap();
6799 drop(group);
6800 drop(source);
6801
6802 let track = resolved.track("log").unwrap();
6803 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6804 let mut next = replacement_dynamic
6805 .requested_track()
6806 .await
6807 .unwrap()
6808 .resolving_start()
6809 .accept(None);
6810 next.start_at(0).unwrap();
6811 let mut subscription = subscribing.await.unwrap().unwrap();
6812 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6813 for expected in [b"a", b"b", b"c"] {
6814 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], expected);
6815 }
6816 }
6817
6818 #[tokio::test]
6821 async fn unannounce_keeps_a_returning_reader_awaiting_its_info() {
6822 let producer = origin(1).produce();
6823 let consumer = producer.consume();
6824
6825 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6826 let mut dynamic = broadcast.dynamic();
6827 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6828 let track = resolved.track("video").unwrap();
6829 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6830 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6831 .await
6832 .expect("the front asked the source")
6833 .expect("request");
6834 let source = request.accept(None);
6835 let mut group = source.append_group().unwrap();
6836 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
6837 group.finish().unwrap();
6838 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6839 subscription.recv_group().await.unwrap().expect("the cached group");
6840 drop(subscription);
6841
6842 tokio::time::timeout(Duration::from_secs(1), source.unused())
6845 .await
6846 .expect("parked")
6847 .expect("source open");
6848 drop(source);
6849 let track = resolved.track("video").unwrap();
6850 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6851 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6852 .await
6853 .expect("the front asked the source again")
6854 .expect("request");
6855
6856 broadcast.unannounce();
6857 settle(|| resolved.is_closed()).await;
6858
6859 let source = request.accept(None);
6861 let mut group = source.create_group(1u64.into()).unwrap();
6862 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
6863 group.finish().unwrap();
6864 source.finish().unwrap();
6865
6866 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
6867 let mut payloads = Vec::new();
6868 while let Some(mut group) = subscription.recv_group().await.expect("ends cleanly") {
6869 payloads.push(group.read_frame().await.unwrap().unwrap().payload);
6870 }
6871 assert_eq!(payloads.last().map(|p| &p[..]), Some(&b"late"[..]));
6872 }
6873
6874 #[tokio::test]
6877 async fn reannounce_before_the_front_acts_keeps_it() {
6878 let producer = origin(1).produce();
6879 let consumer = producer.consume();
6880
6881 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6882 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6883
6884 broadcast.unannounce();
6885 broadcast.announce(Route::default()).unwrap();
6886 for _ in 0..20 {
6887 tokio::task::yield_now().await;
6888 }
6889 assert!(!resolved.is_closed(), "the front ended across a reannouncement");
6890 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6891 assert!(again.is_clone(&resolved));
6892 }
6893
6894 #[tokio::test]
6895 async fn local_broadcast_resolves_once_announced() {
6896 let producer = origin(1).produce();
6897 let consumer = producer.consume();
6898
6899 let broadcast = producer.create_broadcast("room/alice").unwrap();
6901 let err = consumer
6902 .request_broadcast("room/alice")
6903 .now_or_never()
6904 .expect("unroutable is synchronous")
6905 .err()
6906 .unwrap();
6907 assert!(matches!(err, Error::Unroutable));
6908
6909 broadcast.announce(Route::default()).unwrap();
6910 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6911 assert_eq!(resolved.info().path.as_str(), "room/alice");
6912 drop(broadcast);
6913
6914 let err = consumer
6916 .request_broadcast("room/bob")
6917 .now_or_never()
6918 .expect("unroutable is synchronous")
6919 .err()
6920 .unwrap();
6921 assert!(matches!(err, Error::Unroutable));
6922 }
6923
6924 #[test]
6925 fn create_broadcast_accepts_a_max_depth_path() {
6926 let producer = origin(1).produce();
6927 let path = vec!["a"; Path::MAX_PARTS].join("/");
6928 let _broadcast = producer.create_broadcast(path.as_str()).expect("max depth is allowed");
6929 let deeper = vec!["a"; Path::MAX_PARTS + 1].join("/");
6930 assert!(matches!(
6931 producer.create_broadcast(deeper.as_str()),
6932 Err(Error::BoundsExceeded(_))
6933 ));
6934 }
6935
6936 #[tokio::test]
6937 async fn duplicate_routes_aggregate_until_the_last_leaves() {
6938 let producer = origin(1).produce();
6939 let first = producer.dynamic("live", Route::default().with_cost(3)).unwrap();
6940 let second = producer.dynamic("live", Route::default().with_cost(1)).unwrap();
6941
6942 let mut announced = producer.consume().announced();
6943 let update = announced.next().now_or_never().expect("next").expect("no next");
6944 assert_eq!(update.prefix.as_str(), "live");
6945 assert_eq!(update.kind, AnnounceKind::Announced);
6946 assert_eq!(update.route.cost, Cost::new(1));
6947 announced.assert_next_wait();
6948
6949 drop(second);
6950 let update = announced.next().now_or_never().expect("next").expect("no next");
6951 assert_eq!(update.prefix.as_str(), "live");
6952 assert_eq!(update.kind, AnnounceKind::Updated);
6953 assert_eq!(update.route.cost, Cost::new(3));
6954
6955 drop(first);
6956 announced.assert_next_ended("live");
6957 announced.assert_next_wait();
6958 }
6959
6960 #[test]
6961 fn dynamic_may_cover_a_scope_but_disjoint_prefixes_are_refused() {
6962 let producer = origin(1).produce();
6963 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
6964 let _broad = scoped
6965 .dynamic("", Route::default())
6966 .expect("an overlapping prefix is accepted");
6967
6968 let _ok = scoped
6969 .dynamic("room/alice", Route::default())
6970 .expect("a contained prefix is accepted");
6971 assert!(matches!(
6972 scoped.dynamic("other", Route::default()),
6973 Err(Error::Unauthorized)
6974 ));
6975 }
6976
6977 #[tokio::test]
6978 async fn dynamic_route_keeps_its_producer_scope() {
6979 let producer = origin(1).produce();
6980 let scope = Patterns::from("*/chat".parse::<Pattern>().unwrap());
6981 let scoped = producer.scope("", &scope).unwrap();
6982 let dynamic = scoped.dynamic("", Route::default()).unwrap();
6983
6984 let mut matching = producer
6985 .consume()
6986 .scope("", &scopes(&["room/chat"]))
6987 .unwrap()
6988 .announced();
6989 matching.assert_next_active("");
6990 let mut outside = producer
6991 .consume()
6992 .scope("", &scopes(&["room/video"]))
6993 .unwrap()
6994 .announced();
6995 outside.assert_next_wait();
6996
6997 let refused = producer
6998 .consume()
6999 .request_broadcast("room/video")
7000 .now_or_never()
7001 .expect("an out-of-scope request must be refused synchronously");
7002 assert!(matches!(refused, Err(Error::Unroutable)));
7003 assert!(dynamic.requested_broadcast().now_or_never().is_none());
7004
7005 let _pending = producer.consume().request_broadcast("room/chat");
7006 let request = queued(&dynamic).await;
7007 assert_eq!(request.path().as_str(), "room/chat");
7008 }
7009
7010 #[tokio::test]
7011 async fn scoped_cursor_selects_among_the_routes_it_can_see() {
7012 let producer = origin(1).produce();
7013 let scoped = |pattern: &str| {
7014 producer
7015 .scope("", &Patterns::from(pattern.parse::<Pattern>().unwrap()))
7016 .unwrap()
7017 };
7018 let _chat = scoped("*/chat").dynamic("", Route::default().with_cost(1)).unwrap();
7019 let _video = scoped("*/video").dynamic("", Route::default().with_cost(5)).unwrap();
7020
7021 let mut video = producer
7022 .consume()
7023 .scope("", &scopes(&["room/video"]))
7024 .unwrap()
7025 .announced();
7026 assert_eq!(video.assert_next_active("").cost, Cost::new(5));
7027 }
7028
7029 #[tokio::test]
7030 async fn route_changes_preserve_each_cursors_visible_winner() {
7031 let producer = origin(1).produce();
7032 let consumer = producer.consume();
7033 let mut all = consumer.clone().announced();
7034 let mut excluded = consumer.clone().excluding(origin(7)).announced();
7035 let mut video = consumer
7036 .clone()
7037 .scope("", &scopes(&["room/live/video"]))
7038 .unwrap()
7039 .announced();
7040 let mut rooted = consumer
7041 .scope("room/live", &Patterns::from(Pattern::all()))
7042 .unwrap()
7043 .announced();
7044
7045 let chat = producer
7046 .scope("", &scopes(&["room/live/chat"]))
7047 .unwrap()
7048 .dynamic("room/live", Route::default().with_hops(hops(&[7])).with_cost(1))
7049 .unwrap();
7050 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(1));
7051 assert_eq!(rooted.assert_next_active("").cost, Cost::new(1));
7052 excluded.assert_next_wait();
7053 video.assert_next_wait();
7054
7055 let video_route = producer
7056 .scope("", &scopes(&["room/live/video"]))
7057 .unwrap()
7058 .dynamic("room/live", Route::default().with_hops(hops(&[8])).with_cost(5))
7059 .unwrap();
7060 all.assert_next_wait();
7061 rooted.assert_next_wait();
7062 assert_eq!(excluded.assert_next_active("room/live").cost, Cost::new(5));
7063 assert_eq!(video.assert_next_active("room/live").cost, Cost::new(5));
7064
7065 chat.update(Route::default().with_hops(hops(&[7])).with_cost(9))
7066 .unwrap();
7067 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(5));
7068 assert_eq!(rooted.assert_next_active("").cost, Cost::new(5));
7069 excluded.assert_next_wait();
7070 video.assert_next_wait();
7071
7072 drop(video_route);
7073 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(9));
7074 assert_eq!(rooted.assert_next_active("").cost, Cost::new(9));
7075 excluded.assert_next_ended("room/live");
7076 video.assert_next_ended("room/live");
7077 }
7078
7079 #[tokio::test]
7080 async fn root_cursor_keeps_the_more_specific_covering_route() {
7081 let producer = origin(1).produce();
7082 let broad = producer.dynamic("room", Route::default().with_cost(1)).unwrap();
7083 let narrow = producer.dynamic("room/live", Route::default().with_cost(9)).unwrap();
7084 let mut announced = producer
7085 .consume()
7086 .scope("room/live/video", &Patterns::from(Pattern::all()))
7087 .unwrap()
7088 .announced();
7089 assert_eq!(announced.assert_next_active("").cost, Cost::new(9));
7090
7091 broad.update(Route::default().with_cost(0)).unwrap();
7092 announced.assert_next_wait();
7093 narrow.update(Route::default().with_cost(8)).unwrap();
7094 assert_eq!(announced.assert_next_active("").cost, Cost::new(8));
7095 drop(narrow);
7096 assert_eq!(announced.assert_next_active("").cost, Cost::new(0));
7097 }
7098
7099 #[tokio::test]
7100 async fn dynamic_accepts_a_max_depth_prefix() {
7101 let producer = origin(1).produce();
7102 let path = (0..Path::MAX_PARTS)
7103 .map(|i| format!("s{i}"))
7104 .collect::<Vec<_>>()
7105 .join("/");
7106 let mut announced = producer.consume().announced();
7107
7108 let dynamic = producer.dynamic(&path, Route::default()).expect("max depth is allowed");
7109 announced.assert_next_active(&path);
7110
7111 let _pending = producer.consume().request_broadcast(&path);
7112 let request = queued(&dynamic).await;
7113 assert_eq!(request.path().as_str(), path);
7114 }
7115
7116 #[tokio::test]
7117 async fn dynamic_exclusion_skips_routes_through_the_subscriber() {
7118 let producer = origin(1).produce();
7119 let _server = producer
7120 .dynamic("live", Route::default().with_hops(hops(&[7])))
7121 .unwrap();
7122
7123 let mut excluded = producer.consume().excluding(origin(7)).announced();
7124 excluded.assert_next_wait();
7125
7126 let mut clean = producer.consume().excluding(origin(8)).announced();
7127 clean.assert_next_active("live");
7128 }
7129
7130 #[tokio::test]
7132 async fn announce_consumer_is_a_stream() {
7133 use futures::StreamExt;
7134 let producer = origin(1).produce();
7135 let server = producer.dynamic("live", Route::default()).unwrap();
7136 let mut announced = producer.consume().announced();
7137 let update = StreamExt::next(&mut announced)
7138 .now_or_never()
7139 .expect("next")
7140 .expect("no next");
7141 assert_eq!(update.prefix.as_str(), "live");
7142 assert_eq!(update.kind, AnnounceKind::Announced);
7143 assert!(StreamExt::next(&mut announced).now_or_never().is_none());
7144 drop(server);
7145 let update = StreamExt::next(&mut announced)
7146 .now_or_never()
7147 .expect("next")
7148 .expect("no next");
7149 assert_eq!(update.kind, AnnounceKind::Retracted);
7150 }
7151
7152 #[tokio::test]
7153 async fn dynamic_retracts() {
7154 let producer = origin(1).produce();
7155 let server = producer.dynamic("live", Route::default()).unwrap();
7156 let mut announced = producer.consume().announced();
7157 announced.assert_next_active("live");
7158
7159 drop(server);
7160 announced.assert_next_ended("live");
7161 }
7162
7163 #[test]
7164 fn charged_wildcard_cost_accumulates_across_hops() {
7165 let first = Cost::new(4).charged(1);
7166 let second = first.charged(2);
7167 assert_eq!(second, Cost { warm: 7, cold: 7 });
7168 }
7169
7170 #[tokio::test]
7171 async fn local_broadcast_is_invisible_until_announced() {
7172 let producer = origin(1).produce();
7173 let mut local = producer.consume().announced();
7174 let mut peer = producer.consume().excluding(Hop::UNKNOWN).announced();
7175 let broadcast = producer.create_broadcast("room/alice").unwrap();
7176 local.assert_next_wait();
7177 peer.assert_next_wait();
7178
7179 broadcast.announce(Route::default()).unwrap();
7180 local.assert_next_active("room/alice");
7181 peer.assert_next_active("room/alice");
7182
7183 drop(broadcast);
7184 local.assert_next_ended("room/alice");
7185 peer.assert_next_ended("room/alice");
7186 }
7187
7188 #[tokio::test]
7189 async fn served_route_materializes_on_demand() {
7190 let producer = origin(1).produce();
7191 let consumer = producer.consume();
7192
7193 let server = producer.dynamic("room", Route::default()).unwrap();
7194
7195 let pending = consumer.request_broadcast("room/alice");
7196 let request = queued(&server).await;
7197 assert_eq!(request.path().as_str(), "room/alice");
7198
7199 let source = broadcast::Info::new().produce();
7200 request.accept(&source);
7201
7202 let resolved = pending.await.expect("resolves");
7203 assert_eq!(resolved.info().path.as_str(), "room/alice");
7205
7206 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
7208 assert!(again.is_clone(&resolved));
7209 }
7210
7211 #[tokio::test]
7212 async fn served_requests_coalesce() {
7213 let producer = origin(1).produce();
7214 let consumer = producer.consume();
7215 let server = producer.dynamic("room", Route::default()).unwrap();
7216
7217 let first = consumer.request_broadcast("room/alice");
7218 let second = consumer.request_broadcast("room/alice");
7219
7220 let request = queued(&server).await;
7221 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7223
7224 let source = broadcast::Info::new().produce();
7225 request.accept(&source);
7226
7227 let first = first.await.expect("resolves");
7228 let second = second.await.expect("resolves");
7229 assert!(first.is_clone(&second));
7230 }
7231
7232 #[tokio::test]
7233 async fn retract_rejects_pending_requests() {
7234 let producer = origin(1).produce();
7235 let consumer = producer.consume();
7236 let server = producer.dynamic("room", Route::default()).unwrap();
7237
7238 let pending = consumer.request_broadcast("room/alice");
7239 drop(server);
7240
7241 let err = pending.await.err().unwrap();
7242 assert!(matches!(err, Error::Unroutable));
7243
7244 let err = consumer
7246 .request_broadcast("room/alice")
7247 .now_or_never()
7248 .expect("unroutable")
7249 .err()
7250 .unwrap();
7251 assert!(matches!(err, Error::Unroutable));
7252 }
7253
7254 #[tokio::test]
7255 async fn routed_broadcast_survives_serving_route_retraction() {
7256 let producer = origin(1).produce();
7257 let consumer = producer.consume();
7258
7259 let standby_server = producer.dynamic("room", Route::default()).unwrap();
7262 let second_server = producer.dynamic("room", Route::default()).unwrap();
7263 let incumbent_server = producer.dynamic("room", Route::default()).unwrap();
7264
7265 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7266 assert!((&mut resolving).now_or_never().is_none());
7267
7268 drop(incumbent_server);
7274 assert!((&mut resolving).now_or_never().is_none());
7275 drop(second_server);
7276 assert!((&mut resolving).now_or_never().is_none());
7277
7278 let request = queued(&standby_server).await;
7279 let source = broadcast::Info::new().produce();
7280 request.accept(&source);
7281
7282 let resolved = resolving.await.expect("resolves via the standby");
7283 assert_eq!(resolved.info().path.as_str(), "room/alice");
7284 }
7285
7286 #[tokio::test]
7287 async fn split_horizon_skips_routes_through_the_requester() {
7288 let producer = origin(1).produce();
7289 let _server = producer
7290 .dynamic("room", Route::default().with_hops(hops(&[7])))
7291 .unwrap();
7292
7293 let excluded = producer.consume().excluding(origin(7));
7295 let err = excluded
7296 .request_broadcast("room/alice")
7297 .now_or_never()
7298 .expect("unroutable")
7299 .err()
7300 .unwrap();
7301 assert!(matches!(err, Error::Unroutable));
7302
7303 let clean = producer.consume().excluding(origin(8));
7305 let pending = clean.request_broadcast("room/alice");
7306 assert!(pending.now_or_never().is_none());
7307 }
7308
7309 #[tokio::test(start_paused = true)]
7313 async fn viewer_sessions_share_the_plain_front() {
7314 let producer = origin(1).produce();
7315 let server = producer
7317 .dynamic("room", Route::default().with_hops(hops(&[10])))
7318 .unwrap();
7319 let upstream = broadcast::Info::new().produce();
7320 let source = upstream.create_track("video", None).unwrap();
7321 let mut group = source.append_group().unwrap();
7322 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
7323
7324 let counts = || {
7325 let table = producer.shared.lock();
7326 let watches = table
7327 .routes
7328 .root
7329 .find(Path::new("room/alice").parts())
7330 .map_or(0, |node| node.watches.len());
7331 (table.fronts.len(), watches)
7332 };
7333
7334 let mut first = None;
7335 let mut front = None;
7336 for viewer in 0..1000 {
7337 let session = producer.consume().excluding(Hop::random());
7338 let pending = session.request_broadcast("room/alice");
7339 if viewer == 0 {
7340 queued(&server).await.accept(&upstream);
7341 }
7342 let resolved = pending.await.expect("resolves");
7343 let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap();
7344 next_group(&mut subscription).await.unwrap().expect("the live group");
7345 drop(subscription);
7346 first.get_or_insert_with(counts);
7347 front = Some(resolved);
7348 }
7349 assert_eq!(first, Some((1, 1)), "one front and one watch for the first viewer");
7350 assert_eq!(counts(), (1, 1), "viewers minted fronts of their own");
7351 drop(front);
7352 }
7353
7354 #[tokio::test]
7358 async fn a_hop_in_a_chain_gets_its_own_front() {
7359 let producer = origin(1).produce();
7360 let _incumbent = producer
7361 .dynamic("room", Route::default().with_hops(hops(&[10])))
7362 .unwrap();
7363 let fronts = || producer.shared.lock().fronts.len();
7364
7365 let peer = producer.consume().excluding(origin(7));
7366 let _local = producer.consume().request_broadcast("room/alice");
7367 let _peer = peer.request_broadcast("room/alice");
7368 assert_eq!(fronts(), 1, "a hop no chain names excludes nothing");
7369
7370 let _echo = producer
7372 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7373 .unwrap();
7374 let _filtered = peer.request_broadcast("room/alice");
7375 assert_eq!(fronts(), 2, "the peer still shares the plain front");
7376 let _viewer = producer.consume().excluding(origin(8)).request_broadcast("room/alice");
7377 assert_eq!(fronts(), 2, "a viewer left the plain front");
7378 }
7379
7380 #[tokio::test]
7384 async fn a_hairpin_through_the_plain_front_is_withdrawn() {
7385 let producer = origin(1).produce();
7386 let peer = producer.consume().excluding(origin(7));
7387 let mut view = peer.announced();
7388 let incumbent = producer
7389 .dynamic("room", Route::default().with_hops(hops(&[10])))
7390 .unwrap();
7391 view.assert_next_active("room");
7392
7393 let pending = peer.request_broadcast("room/alice");
7394 let source = broadcast::Info::new().produce();
7395 queued(&incumbent).await.accept(&source);
7396 let resolved = pending.await.expect("resolves through the incumbent");
7397
7398 let echo = producer
7400 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7401 .unwrap();
7402 drop(incumbent);
7403 drop(source);
7404
7405 let replacement = broadcast::Info::new().produce();
7406 queued(&echo).await.accept(&replacement);
7407 resolved.assert_not_closed();
7408 view.assert_next_ended("room");
7409 }
7410
7411 #[tokio::test]
7412 async fn routes_report_where_they_entered() {
7413 let producer = origin(1).produce();
7414 let peer = producer.clone().peer();
7415 let mut announced = producer.consume().announced();
7416
7417 let _ingest = producer
7418 .dynamic("client", Route::default().with_hops(hops(&[5])).with_via(origin(5)))
7419 .unwrap();
7420 let _gateway = producer.publish("gateway", Route::default()).unwrap();
7421 let _forwarded = peer
7422 .dynamic(
7423 "forwarded",
7424 Route::default().with_hops(hops(&[5, 7])).with_via(origin(7)),
7425 )
7426 .unwrap();
7427
7428 assert_eq!(announced.assert_next_active("client").source(), Source::Local);
7429 assert_eq!(
7430 announced.assert_next_active("forwarded").source(),
7431 Source::Peer(origin(7))
7432 );
7433 assert_eq!(announced.assert_next_active("gateway").source(), Source::Local);
7434
7435 let scoped = peer.scope("room", &Patterns::from(Pattern::all())).unwrap();
7437 let _nested = scoped.dynamic("x", Route::default().with_via(origin(8))).unwrap();
7438 assert_eq!(announced.assert_next_active("room/x").source(), Source::Peer(origin(8)));
7439 }
7440
7441 #[tokio::test(start_paused = true)]
7445 async fn try_next_delivers_a_held_update() {
7446 async fn step(by: Duration) {
7448 tokio::time::advance(by).await;
7449 for _ in 0..10 {
7450 tokio::task::yield_now().await;
7451 }
7452 }
7453
7454 let producer = origin(1).produce();
7455 let peer = producer.clone().peer();
7456 let mut announced = producer.consume().announced();
7457 step(Duration::from_millis(1)).await;
7458
7459 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7460 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7461 assert!(announced.try_next().unwrap().kind.is_active());
7462
7463 let _far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7464 drop(near);
7465 assert!(announced.try_next().is_none(), "the update is held");
7466 step(Duration::from_millis(1)).await;
7467 assert!(announced.try_next().is_none(), "the update is still held");
7468 step(DEFAULT_UPDATE_HOLD).await;
7469 let update = announced.try_next().expect("the hold passed");
7470 assert_eq!(update.kind, AnnounceKind::Updated);
7471 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7472 }
7473
7474 #[test]
7480 fn a_withdrawal_wave_costs_k_log_k_steps() {
7481 for k in [64usize, 256, 1024] {
7482 let (producer, mut driver) = Producer::new(Config::new(origin(1)));
7483 driver.poll(Instant::now(), &kio::Waiter::noop()).unwrap();
7485 let peer = producer.clone().peer();
7486 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7487 let mut near = Vec::new();
7488 let mut far = Vec::new();
7489 for i in 0..k {
7490 near.push(peer.dynamic(format!("room/{i}"), route(&[9, 2], 2)).unwrap());
7491 far.push(peer.dynamic(format!("room/{i}"), route(&[9, 3, 4], 4)).unwrap());
7492 }
7493 let mut announced = producer.consume().announced();
7494 for _ in 0..k {
7495 assert_eq!(announced.try_next().unwrap().kind, AnnounceKind::Announced);
7496 }
7497 let before = announced.state.read().steps;
7498
7499 for route in near {
7500 drop(route);
7501 assert!(announced.try_next().is_none(), "the update is held");
7502 }
7503 for (i, route) in far.into_iter().enumerate().rev() {
7504 drop(route);
7505 let update = announced.try_next().expect("the retraction is delivered");
7506 assert_eq!(update.kind, AnnounceKind::Retracted);
7507 assert_eq!(update.prefix, Path::new(&format!("room/{i}")));
7508 }
7509
7510 let steps = announced.state.read().steps - before;
7511 let bound = 4 * k * (k.ilog2() as usize + 1);
7512 assert!(steps <= bound, "{k} broadcasts took {steps} steps, over {bound}");
7513 }
7514 }
7515
7516 #[tokio::test(start_paused = true)]
7519 async fn a_changed_route_waits_out_the_hold() {
7520 let producer = origin(1).produce();
7521 let peer = producer.clone().peer();
7522 let mut announced = producer.consume().announced();
7523 tokio::time::sleep(Duration::from_millis(1)).await;
7525
7526 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7527 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7528 announced.assert_next_active("room");
7529
7530 let far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7532 let farther = peer.dynamic("room", route(&[9, 5, 6, 7], 7)).unwrap();
7533 let start = tokio::time::Instant::now();
7534 drop(near);
7535 drop(far);
7536 let update = announced.next().await.unwrap();
7537 assert_eq!(update.kind, AnnounceKind::Updated);
7538 assert_eq!(update.route.hops, hops(&[9, 5, 6, 7]));
7539 assert_eq!(start.elapsed(), DEFAULT_UPDATE_HOLD);
7540 announced.assert_next_wait();
7541
7542 drop(farther);
7544 announced.assert_next_ended("room");
7545 }
7546
7547 #[tokio::test(start_paused = true)]
7550 async fn a_retraction_quarantines_a_named_publisher() {
7551 let producer = origin(1).produce();
7552 let peer = producer.clone().peer();
7553 let mut announced = producer.consume().announced();
7554 tokio::time::sleep(Duration::from_millis(1)).await;
7555
7556 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7557
7558 let near = peer.dynamic("named", route(&[9, 2], 2)).unwrap();
7559 announced.assert_next_active("named");
7560 drop(near);
7561 announced.assert_next_ended("named");
7562 let start = tokio::time::Instant::now();
7563 let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap();
7564 let update = announced.next().await.unwrap();
7565 assert_eq!(update.kind, AnnounceKind::Announced);
7566 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7567 assert!(start.elapsed() >= RETRACT_QUARANTINE);
7568
7569 let near = peer.dynamic("anonymous", route(&[0, 2], 2)).unwrap();
7570 announced.assert_next_active("anonymous");
7571 drop(near);
7572 announced.assert_next_ended("anonymous");
7573 let _other = peer.dynamic("anonymous", route(&[0, 3, 4], 4)).unwrap();
7574 announced.assert_next_active("anonymous");
7575
7576 let queued = announced.anchoring.len();
7579 for chain in [[9, 2], [0, 2], [9, 2], [0, 2]] {
7580 let churn = peer.dynamic("churn", route(&chain, 2)).unwrap();
7581 announced.assert_next_active("churn");
7582 drop(churn);
7583 announced.assert_next_ended("churn");
7584 }
7585 assert_eq!(announced.anchoring.len(), queued + 1);
7586
7587 let gone = peer.dynamic("gone", route(&[9, 2], 2)).unwrap();
7590 announced.assert_next_active("gone");
7591 drop(gone);
7592 announced.assert_next_ended("gone");
7593 assert!(!announced.retracted.is_empty());
7594 for _ in 0..2 {
7595 announced.assert_next_wait();
7596 tokio::time::sleep(RETRACT_QUARANTINE).await;
7597 }
7598 announced.assert_next_wait();
7599 assert!(announced.retracted.is_empty(), "the ended quarantine is still held");
7600 }
7601
7602 #[tokio::test]
7606 async fn withdrawn_peer_hides_routes_through_it() {
7607 let producer = origin(1).produce();
7608 let peer = producer.clone().peer();
7609 let mut announced = producer.consume().announced();
7610
7611 let direct = || Route::default().with_hops(hops(&[9, 2])).with_via(origin(2));
7613 let first = peer.dynamic("room", direct()).unwrap();
7614 let relayed = peer
7615 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])).with_via(origin(3)))
7616 .unwrap();
7617 announced.assert_next_active("room");
7618 announced.assert_next_wait();
7619
7620 first.withdrawn();
7622 announced.assert_next_ended("room");
7623 announced.assert_next_wait();
7624
7625 let second = peer.dynamic("room", direct()).unwrap();
7627 announced.assert_next_active("room");
7628 assert!(producer.shared.lock().withdrawn.is_empty());
7629
7630 second.withdrawn();
7632 drop(relayed);
7633 announced.assert_next_ended("room");
7634 assert!(producer.shared.lock().withdrawn.is_empty());
7635 }
7636
7637 #[tokio::test]
7639 async fn old_session_withdrawal_keeps_newer_route() {
7640 for restart in [false, true] {
7641 let producer = origin(1).produce();
7642 let peer = producer.clone().peer();
7643 let mut announced = producer.consume().announced();
7644 let route = Route::default().with_hops(hops(&[9, 2]));
7645 let first = peer.dynamic("room", route.clone()).unwrap();
7646 let second = peer.dynamic("room", route.clone()).unwrap();
7647 announced.assert_next_active("room");
7648 announced.assert_next_wait();
7649 let remaining = if restart {
7650 first.update(route).unwrap();
7652 second.withdrawn();
7653 first
7654 } else {
7655 first.withdrawn();
7656 second
7657 };
7658 announced.assert_next_wait();
7659 assert!(producer.shared.lock().withdrawn.is_empty());
7660 remaining.withdrawn();
7661 announced.assert_next_ended("room");
7662 assert!(producer.shared.lock().withdrawn.is_empty());
7663 }
7664 }
7665
7666 #[tokio::test(start_paused = true)]
7670 async fn a_reconnect_under_a_new_hop_wins_over_the_old_session() {
7671 for (old, new) in [
7672 (&[10][..], &[11][..]),
7673 (&[11], &[10]),
7674 (&[10, 0], &[11, 0]),
7675 (&[11, 0], &[10, 0]),
7676 ] {
7677 let producer = origin(1).produce();
7678 let mut announced = producer.consume().announced();
7679 tokio::time::sleep(Duration::from_millis(1)).await;
7680
7681 let _old = producer.dynamic("room", Route::default().with_hops(hops(old))).unwrap();
7682 assert_eq!(announced.assert_next_active("room").hops, hops(old));
7683
7684 let _new = producer.dynamic("room", Route::default().with_hops(hops(new))).unwrap();
7685 let update = tokio::time::timeout(DEFAULT_UPDATE_HOLD * 2, announced.next())
7686 .await
7687 .unwrap_or_else(|_| panic!("{new:?} never took over from {old:?}"))
7688 .unwrap();
7689 assert_eq!(update.kind, AnnounceKind::Updated);
7690 assert_eq!(update.route.hops, hops(new));
7691 }
7692 }
7693
7694 #[tokio::test]
7695 async fn withdrawn_route_no_longer_covers_requests() {
7696 let producer = origin(1).produce();
7697 let peer = producer.clone().peer();
7698 let direct = peer.dynamic("room", Route::default().with_hops(hops(&[9, 2]))).unwrap();
7699 let _relayed = peer
7700 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])))
7701 .unwrap();
7702 let path = Path::new("room/video");
7703 let id = producer
7704 .shared
7705 .read()
7706 .routes
7707 .covering(&path)
7708 .find(|entry| entry.hops.iter().last() == Some(&origin(3)))
7709 .unwrap()
7710 .id;
7711 assert!(producer.shared.read().routes.covers(&path, id));
7712 direct.withdrawn();
7713 assert!(!producer.shared.read().routes.covers(&path, id));
7714 }
7715
7716 #[tokio::test]
7719 async fn source_change_is_an_update() {
7720 let producer = origin(1).produce();
7721 let peer = producer.clone().peer();
7722 let mut announced = producer.consume().announced();
7723
7724 let route = Route::default().with_hops(hops(&[7])).with_via(origin(7));
7725 let _forwarded = peer.dynamic("room", route.clone()).unwrap();
7726 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7727
7728 let local = producer.dynamic("room", route).unwrap();
7730 let update = announced.next().now_or_never().expect("next blocked").expect("no next");
7731 assert_eq!(update.kind, AnnounceKind::Updated);
7732 assert_eq!(update.route.source(), Source::Local);
7733
7734 drop(local);
7735 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7736 }
7737
7738 #[tokio::test]
7739 async fn local_view_hides_peer_routes() {
7740 let producer = origin(1).produce();
7741 let peer = producer.clone().peer();
7742 let mut local = producer.consume().local().announced();
7743
7744 let _forwarded = peer
7745 .dynamic("remote", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7746 .unwrap();
7747 local.assert_next_wait();
7748
7749 let _shadow = peer
7753 .dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7754 .unwrap();
7755 let ingest = producer
7756 .dynamic(
7757 "both",
7758 Route::default().with_hops(hops(&[5])).with_via(origin(5)).with_cost(9),
7759 )
7760 .unwrap();
7761 assert_eq!(local.assert_next_active("both").source(), Source::Local);
7762 drop(ingest);
7763 local.assert_next_ended("both");
7764
7765 let err = producer
7768 .consume()
7769 .local()
7770 .request_broadcast("remote/alice")
7771 .now_or_never()
7772 .expect("unroutable")
7773 .err()
7774 .unwrap();
7775 assert!(matches!(err, Error::Unroutable));
7776 assert!(
7777 producer
7778 .consume()
7779 .request_broadcast("remote/alice")
7780 .now_or_never()
7781 .is_none()
7782 );
7783 }
7784
7785 #[tokio::test]
7789 async fn handler_rejection_is_final() {
7790 let producer = origin(1).produce();
7791 let consumer = producer.consume();
7792 let server = producer.dynamic("room", Route::default()).unwrap();
7793
7794 let pending = consumer.request_broadcast("room/alice");
7795 let request = queued(&server).await;
7796 request.reject(Error::Unroutable);
7797 let err = tokio::time::timeout(Duration::from_secs(5), pending)
7798 .await
7799 .expect("the front must give up, not spin")
7800 .err()
7801 .unwrap();
7802 assert!(matches!(err, Error::Unroutable));
7803
7804 let pending = consumer.request_broadcast("room/bob");
7806 let request = queued(&server).await;
7807 assert_eq!(request.path().as_str(), "room/bob");
7808 let served = broadcast::Info::new().produce();
7809 request.accept(&served);
7810 pending.await.expect("resolves");
7811 }
7812
7813 #[tokio::test]
7816 async fn routed_broadcast_waits_out_a_rejection() {
7817 let producer = origin(1).produce();
7818 let consumer = producer.consume();
7819 let server = producer.dynamic("room", Route::default()).unwrap();
7820
7821 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7822 assert!((&mut resolving).now_or_never().is_none());
7823 let request = queued(&server).await;
7824 request.reject(Error::Unroutable);
7825
7826 for _ in 0..20 {
7828 tokio::task::yield_now().await;
7829 }
7830 assert!((&mut resolving).now_or_never().is_none());
7831 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7832
7833 server.update(Route::default().with_cost(2)).unwrap();
7835 assert!((&mut resolving).now_or_never().is_none());
7836 let request = queued(&server).await;
7837 let served = broadcast::Info::new().produce();
7838 request.accept(&served);
7839 resolving.await.expect("resolves");
7840 }
7841
7842 #[tokio::test]
7845 async fn routed_broadcast_reports_teardown_as_closed() {
7846 let (producer, driver) = Producer::new(Config::new(origin(1)));
7847 let consumer = producer.consume();
7848 let _server = producer.dynamic("room", Route::default()).unwrap();
7849
7850 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7852 assert!((&mut resolving).now_or_never().is_none());
7853
7854 drop(driver);
7855
7856 let err = tokio::time::timeout(Duration::from_secs(5), resolving)
7857 .await
7858 .expect("teardown resolves the wait")
7859 .err()
7860 .unwrap();
7861 assert!(matches!(err, Error::Closed), "unexpected end: {err}");
7862 }
7863
7864 #[tokio::test]
7867 async fn routed_broadcast_wakes_for_a_local_broadcast() {
7868 let producer = origin(1).produce();
7869 let consumer = producer.consume();
7870 let server = producer.dynamic("room", Route::default()).unwrap();
7871
7872 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7873 assert!((&mut resolving).now_or_never().is_none());
7874 queued(&server).await.reject(Error::Unroutable);
7875 for _ in 0..20 {
7876 tokio::task::yield_now().await;
7877 }
7878 assert!((&mut resolving).now_or_never().is_none());
7879
7880 let _local = producer.publish("room/alice", Route::default()).unwrap();
7882 let resolved = resolving.await.expect("resolves locally");
7883 assert_eq!(resolved.info().path.as_str(), "room/alice");
7884 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7885 }
7886
7887 #[tokio::test]
7890 async fn late_track_on_a_served_front_replays() {
7891 let producer = origin(1).produce();
7892 let consumer = producer.consume();
7893 let server = producer.dynamic("room", Route::default()).unwrap();
7894
7895 let source = broadcast::Info::new().produce();
7896 for name in ["a", "b"] {
7897 let track = source.create_track(name, None).unwrap();
7898 let mut group = track.append_group().unwrap();
7899 group.write_frame(crate::Timestamp::ZERO, name.as_bytes()).unwrap();
7900 group.finish().unwrap();
7901 std::mem::forget(track);
7903 }
7904
7905 let pending = consumer.request_broadcast("room/alice");
7906 queued(&server).await.accept(&source);
7907 let resolved = pending.await.expect("resolves");
7908
7909 let budget = track::Subscription::default().with_max_age(Duration::from_secs(3600));
7910 for name in ["a", "b"] {
7911 let mut subscription = resolved
7912 .track(name)
7913 .unwrap()
7914 .subscribe(budget.clone())
7915 .await
7916 .expect("subscribe");
7917 let mut group = tokio::time::timeout(Duration::from_secs(5), subscription.recv_group())
7918 .await
7919 .expect("the late track must replay, not park")
7920 .expect("recv group")
7921 .expect("track ended early");
7922 let frame = group.read_frame().await.expect("read frame").expect("frame");
7923 assert_eq!(&frame.payload[..], name.as_bytes());
7924 }
7925 }
7926
7927 #[tokio::test]
7928 async fn most_specific_prefix_shadows() {
7929 let producer = origin(1).produce();
7930 let consumer = producer.consume();
7931
7932 let broad_server = producer.dynamic("", Route::default()).unwrap();
7933 let _narrow = producer.announce(".dash", Route::default()).unwrap();
7936
7937 let err = consumer
7938 .request_broadcast(".dash/pid")
7939 .now_or_never()
7940 .expect("unroutable")
7941 .err()
7942 .unwrap();
7943 assert!(matches!(err, Error::Unroutable));
7944
7945 let _pending = consumer.request_broadcast("room/alice");
7947 let request = queued(&broad_server).await;
7948 assert_eq!(request.path().as_str(), "room/alice");
7949 }
7950
7951 #[tokio::test]
7952 async fn root_dynamic_serves_any_path() {
7953 let producer = origin(1).produce();
7954 let consumer = producer.consume();
7955 let mut announced = consumer.announced();
7956 let dynamic = producer.dynamic("", Route::default()).unwrap();
7957 announced.assert_next_active("");
7959
7960 let pending = consumer.request_broadcast("anything/at/all");
7961 let request = queued(&dynamic).await;
7962 assert_eq!(request.path().as_str(), "anything/at/all");
7963
7964 let source = broadcast::Info::new().produce();
7965 request.accept(&source);
7966 let resolved = pending.await.expect("resolves");
7967 assert_eq!(resolved.info().path.as_str(), "anything/at/all");
7968
7969 drop(dynamic);
7971 announced.assert_next_ended("");
7972 let err = consumer
7973 .request_broadcast("something/else")
7974 .now_or_never()
7975 .expect("unroutable")
7976 .err()
7977 .unwrap();
7978 assert!(matches!(err, Error::Unroutable));
7979 }
7980
7981 #[tokio::test]
7987 async fn out_of_scope_request_never_reaches_the_dynamic_handler() {
7988 let producer = origin(1).produce();
7989 let dynamic = producer.dynamic("", Route::default()).unwrap();
7990 let scoped = producer.consume().scope("", &scopes(&["tenant-a"])).unwrap();
7991
7992 for path in ["tenant-b/live", "tenant-a-other/live"] {
7995 let refused = scoped
7996 .request_broadcast(path)
7997 .now_or_never()
7998 .expect("an out-of-scope request must be refused synchronously, not queued");
7999 assert!(matches!(refused, Err(Error::Unauthorized)));
8000 assert!(
8001 dynamic.requested_broadcast().now_or_never().is_none(),
8002 "the dynamic handler was asked to create a broadcast the requester may not read"
8003 );
8004 }
8005 }
8006
8007 #[tokio::test]
8008 async fn routed_waits_for_coverage() {
8009 let producer = origin(1).produce();
8010 let consumer = producer.consume();
8011
8012 let mut fut = consumer.routed("room/alice").boxed();
8013 assert!((&mut fut).now_or_never().is_none());
8014
8015 let _a = producer.announce("room", Route::default().with_cost(3)).unwrap();
8017 let route = fut.now_or_never().expect("covered").expect("routed");
8018 assert_eq!(route.cost, Cost::new(3));
8019
8020 consumer
8022 .routed("room/alice/cam")
8023 .now_or_never()
8024 .expect("covered")
8025 .expect("routed");
8026 }
8027
8028 #[tokio::test]
8029 async fn routed_ignores_deeper_routes() {
8030 let producer = origin(1).produce();
8031 let consumer = producer.consume();
8032
8033 let _deep = producer.announce("room/alice/cam", Route::default()).unwrap();
8035 let mut fut = consumer.routed("room/alice").boxed();
8036 assert!((&mut fut).now_or_never().is_none());
8037
8038 let _exact = producer.announce("room/alice", Route::default()).unwrap();
8039 fut.now_or_never().expect("covered").expect("routed");
8040 }
8041
8042 #[tokio::test]
8043 async fn routed_accepts_a_max_depth_path() {
8044 let producer = origin(1).produce();
8045 let consumer = producer.consume();
8046 let path = (0..Path::MAX_PARTS)
8047 .map(|i| format!("s{i}"))
8048 .collect::<Vec<_>>()
8049 .join("/");
8050 assert_eq!(Path::new(&path).parts().count(), Path::MAX_PARTS);
8051
8052 assert!(consumer.allowed().matches(&path));
8053
8054 let mut fut = consumer.routed(&path).boxed();
8055 assert!((&mut fut).now_or_never().is_none());
8056
8057 let _a = producer.announce("", Route::default()).unwrap();
8059 fut.now_or_never().expect("covered").expect("routed");
8060 }
8061
8062 #[tokio::test]
8063 async fn teardown_ends_everything() {
8064 let (producer, driver) = Producer::new(Config::new(origin(1)));
8065 let consumer = producer.consume();
8066 let _announcement = producer.announce("room", Route::default()).unwrap();
8067 let mut announced = consumer.announced();
8068 announced.assert_next_active("room");
8069
8070 let _server = producer.dynamic("served", Route::default()).unwrap();
8071 let pending = consumer.request_broadcast("served/path");
8072
8073 drop(driver);
8074
8075 announced.assert_next_active("served");
8077 assert!(announced.next().now_or_never().expect("ended").is_none());
8078
8079 assert!(pending.now_or_never().expect("rejected").is_err());
8081 assert!(matches!(producer.announce("x", Route::default()), Err(Error::Closed)));
8082 assert!(matches!(producer.create_broadcast("x"), Err(Error::Closed)));
8083 let err = consumer
8084 .request_broadcast("y")
8085 .now_or_never()
8086 .expect("closed")
8087 .err()
8088 .unwrap();
8089 assert!(matches!(err, Error::Closed));
8090
8091 let mut late = consumer.announced();
8093 assert!(late.next().now_or_never().expect("ended").is_none());
8094 }
8095
8096 struct ResumeRig {
8099 producer: Producer,
8100 resolved: broadcast::Consumer,
8101 subscription: track::Subscriber,
8102 incumbent_track: track::Producer,
8105 }
8106
8107 impl ResumeRig {
8108 async fn new(first: &[u64]) -> (Self, Dynamic, broadcast::Producer) {
8111 let producer = origin(1).produce();
8112 let consumer = producer.consume();
8113
8114 let server = producer
8115 .dynamic("room", Route::default().with_hops(hops(first)))
8116 .unwrap();
8117
8118 let pending = consumer.request_broadcast("room/alice");
8119 let request = queued(&server).await;
8120 let source = broadcast::Info::new().produce();
8121 let track = source.create_track("video", None).unwrap();
8122 let mut group = track.append_group().unwrap();
8123 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8124 group.finish().unwrap();
8125 request.accept(&source);
8126
8127 let resolved = pending.await.expect("resolves");
8128 let mut subscription = resolved
8129 .track("video")
8130 .unwrap()
8131 .subscribe(None)
8132 .await
8133 .expect("subscribe");
8134 let mut group = subscription
8135 .recv_group()
8136 .await
8137 .expect("recv group")
8138 .expect("track ended early");
8139 let frame = group.read_frame().await.expect("read frame").expect("frame");
8140 assert_eq!(&frame.payload[..], b"before");
8141
8142 (
8143 Self {
8144 producer,
8145 resolved,
8146 subscription,
8147 incumbent_track: track,
8148 },
8149 server,
8150 source,
8151 )
8152 }
8153
8154 fn standby(&self, first: &[u64]) -> Dynamic {
8157 self.producer
8158 .dynamic("room", Route::default().with_hops(hops(first)))
8159 .unwrap()
8160 }
8161 }
8162
8163 async fn assert_resumes(rig: &mut ResumeRig, server: &Dynamic) {
8168 let request = queued(server).await;
8169 let replacement = broadcast::Info::new().produce();
8170 let track = replacement.create_track("video", None).unwrap();
8171 let mut group = track.append_group().unwrap();
8174 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8175 group.finish().unwrap();
8176 request.accept(&replacement);
8177
8178 let mut group = track.append_group().unwrap();
8179 group.write_frame(crate::Timestamp::ZERO, b"resumed".as_ref()).unwrap();
8180 group.finish().unwrap();
8181
8182 let mut group = rig
8183 .subscription
8184 .recv_group()
8185 .await
8186 .expect("subscription survives the failover")
8187 .expect("track ended early");
8188 let frame = group.read_frame().await.expect("read frame").expect("frame");
8189 assert_eq!(&frame.payload[..], b"resumed");
8190 }
8191
8192 #[tokio::test]
8195 async fn driver_resolves_with_live_consumers() {
8196 let (producer, driver) = Producer::new(Config::new(origin(1)));
8197 let consumer = producer.consume();
8198 let run = crate::time::run(driver);
8199 drop(producer);
8200 tokio::time::timeout(Duration::from_secs(5), run)
8201 .await
8202 .expect("driver must finish once the producers are gone");
8203 drop(consumer);
8204 }
8205
8206 #[tokio::test]
8207 async fn remote_source_resumes_through_same_first_hop() {
8208 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8209 let standby_server = rig.standby(&[10, 20]);
8210
8211 drop(incumbent);
8213 drop(source);
8214
8215 assert_resumes(&mut rig, &standby_server).await;
8217 }
8218
8219 #[tokio::test]
8223 async fn incompatible_successor_is_refused() {
8224 for replacement in [
8225 track::Info::default().with_timescale(crate::Timescale::MICRO),
8226 track::Info::default().with_priority(7),
8227 track::Info::default().with_max_age(Duration::from_secs(7)),
8228 ] {
8229 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8230 let standby_server = rig.standby(&[10, 20]);
8231 drop(incumbent);
8232 drop(source);
8233
8234 let request = queued(&standby_server).await;
8237 let successor = broadcast::Info::new().produce();
8238 let track = successor.create_track("video", replacement).unwrap();
8239 let mut group = track.append_group().unwrap();
8240 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8241 group.finish().unwrap();
8242 request.accept(&successor);
8243
8244 assert!(
8245 matches!(rig.subscription.recv_group().await, Err(Error::Unsupported)),
8246 "the subscription must abort rather than resume onto incompatible metadata"
8247 );
8248
8249 let reopened = rig.resolved.track("video").unwrap();
8251 assert!(matches!(reopened.query().await, Err(Error::Unsupported)));
8252 assert!(matches!(reopened.subscribe(None).await, Err(Error::Unsupported)));
8253 }
8254 }
8255
8256 #[tokio::test]
8257 async fn different_first_hop_ends_the_subscription() {
8258 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8259 let rival_server = rig.standby(&[11]);
8261
8262 drop(incumbent);
8265 drop(source);
8266 rig.incumbent_track.abort(Error::Dropped).unwrap();
8267
8268 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
8270 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8271
8272 let consumer = rig.producer.consume();
8274 let pending = consumer.request_broadcast("room/alice");
8275 let request = queued(&rival_server).await;
8276 let replacement = broadcast::Info::new().produce();
8277 request.accept(&replacement);
8278 pending.await.expect("re-request resolves through the rival");
8279 }
8280
8281 #[tokio::test]
8286 async fn a_first_hop_update_drains_the_old_publisher() {
8287 for first in [&[][..], &[0][..], &[10][..]] {
8288 let (mut rig, server, _source) = ResumeRig::new(first).await;
8289 server.update(Route::default().with_hops(hops(&[11]))).unwrap();
8290
8291 let mut group = rig.incumbent_track.append_group().unwrap();
8293 group.write_frame(crate::Timestamp::ZERO, b"draining".as_ref()).unwrap();
8294 group.finish().unwrap();
8295 let mut group = next_group(&mut rig.subscription)
8296 .await
8297 .expect("the in-flight subscription survives the update")
8298 .expect("track ended early");
8299 let frame = group.read_frame().await.expect("read frame").expect("frame");
8300 assert_eq!(&frame.payload[..], b"draining", "first hop {first:?}");
8301
8302 let consumer = rig.producer.consume();
8305 let pending = consumer.request_broadcast("room/alice");
8306 let request = queued(&server).await;
8307 let replacement = broadcast::Info::new().produce();
8308 let track = replacement
8309 .create_track("video", track::Info::default().with_priority(7))
8310 .unwrap();
8311 let mut group = track.append_group().unwrap();
8312 group.write_frame(crate::Timestamp::ZERO, b"new".as_ref()).unwrap();
8313 group.finish().unwrap();
8314 request.accept(&replacement);
8315
8316 let resolved = pending.await.expect("resolves through the new publisher");
8317 assert!(
8318 !resolved.is_clone(&rig.resolved),
8319 "first hop {first:?} joined the old front"
8320 );
8321 let mut fresh = resolved
8322 .track("video")
8323 .unwrap()
8324 .subscribe(None)
8325 .await
8326 .expect("the old publisher's track info does not apply");
8327 let mut group = next_group(&mut fresh)
8328 .await
8329 .expect("recv group")
8330 .expect("track ended early");
8331 let frame = group.read_frame().await.expect("read frame").expect("frame");
8332 assert_eq!(&frame.payload[..], b"new");
8333
8334 rig.incumbent_track.finish().unwrap();
8337 let end = next_group(&mut rig.subscription).await;
8338 assert!(
8339 !matches!(end, Ok(Some(_))),
8340 "first hop {first:?} spliced the new publisher into a live subscription"
8341 );
8342 }
8343 }
8344
8345 #[tokio::test]
8348 async fn a_first_hop_update_resumes_through_the_same_publisher() {
8349 let (mut rig, incumbent, _source) = ResumeRig::new(&[10]).await;
8350 let standby_server = rig.standby(&[10, 20]);
8351
8352 incumbent.update(Route::default().with_hops(hops(&[11]))).unwrap();
8353
8354 assert_resumes(&mut rig, &standby_server).await;
8355 assert!(
8356 incumbent.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8357 "the front never asks the new publisher"
8358 );
8359 }
8360
8361 #[tokio::test]
8362 async fn anonymous_routes_never_resume() {
8363 let (mut rig, incumbent, source) = ResumeRig::new(&[]).await;
8366 let _twin_server = rig.standby(&[]);
8367
8368 drop(incumbent);
8369 drop(source);
8370 rig.incumbent_track.abort(Error::Dropped).unwrap();
8371
8372 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
8373 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8374 }
8375
8376 #[tokio::test]
8383 async fn anonymous_handoff_serves_the_newcomer_immediately() {
8384 let producer = origin(1).produce();
8385
8386 let server_a = producer
8388 .dynamic("room", Route::default().with_hops(hops(&[10])))
8389 .unwrap();
8390
8391 let consumer = producer.consume().excluding(origin(30));
8394 let pending = consumer.request_broadcast("room/alice");
8395 let request = queued(&server_a).await;
8396 let source_a = broadcast::Info::new().produce();
8397 let track_a = source_a.create_track("video", None).unwrap();
8398 let mut group = track_a.append_group().unwrap();
8399 group.write_frame(crate::Timestamp::ZERO, b"from-a".as_ref()).unwrap();
8400 group.finish().unwrap();
8401 request.accept(&source_a);
8402
8403 let resolved_a = pending.await.expect("resolves");
8404 let mut sub_a = resolved_a
8405 .track("video")
8406 .unwrap()
8407 .subscribe(None)
8408 .await
8409 .expect("subscribe");
8410 let mut group = sub_a
8411 .recv_group()
8412 .await
8413 .expect("recv group")
8414 .expect("track ended early");
8415 assert_eq!(
8416 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8417 b"from-a"
8418 );
8419
8420 drop(track_a);
8423 drop(source_a);
8424 drop(server_a);
8425
8426 let err = sub_a.recv_group().await.err().expect("front closed");
8428 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8429
8430 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8434 settle(|| {
8435 matches!(
8436 consumer.request_broadcast("room/alice").now_or_never(),
8437 Some(Err(Error::Unroutable))
8438 )
8439 })
8440 .await;
8441
8442 let server_b = producer
8444 .dynamic("room", Route::default().with_hops(hops(&[20])))
8445 .unwrap();
8446 let pending = consumer.request_broadcast("room/alice");
8447 let request = queued(&server_b).await;
8448 let source_b = broadcast::Info::new().produce();
8449 let track_b = source_b.create_track("video", None).unwrap();
8450 let mut group = track_b.append_group().unwrap();
8451 group.write_frame(crate::Timestamp::ZERO, b"from-b".as_ref()).unwrap();
8452 group.finish().unwrap();
8453 request.accept(&source_b);
8454
8455 let resolved_b = pending.await.expect("B's front is served immediately");
8456 assert!(
8457 !resolved_b.is_clone(&resolved_a),
8458 "B must not splice into A's closed front"
8459 );
8460
8461 let mut sub_b = resolved_b
8462 .track("video")
8463 .unwrap()
8464 .subscribe(None)
8465 .await
8466 .expect("subscribe");
8467 let mut group = sub_b
8468 .recv_group()
8469 .await
8470 .expect("recv group")
8471 .expect("track ended early");
8472 assert_eq!(
8473 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8474 b"from-b"
8475 );
8476 }
8477
8478 #[tokio::test]
8479 async fn reprice_is_invisible_to_the_subscription() {
8480 let (rig, incumbent, source) = ResumeRig::new(&[10]).await;
8481
8482 incumbent
8485 .update(Route::default().with_hops(hops(&[10])).with_cost(9))
8486 .unwrap();
8487
8488 let track = source.create_track("audio", None).unwrap();
8489 let mut group = track.append_group().unwrap();
8490 group.write_frame(crate::Timestamp::ZERO, b"steady".as_ref()).unwrap();
8491 group.finish().unwrap();
8492
8493 let mut audio = rig
8494 .resolved
8495 .track("audio")
8496 .unwrap()
8497 .subscribe(None)
8498 .await
8499 .expect("subscribe survives the reprice");
8500 let mut group = audio
8501 .recv_group()
8502 .await
8503 .expect("recv group")
8504 .expect("track ended early");
8505 let frame = group.read_frame().await.expect("read frame").expect("frame");
8506 assert_eq!(&frame.payload[..], b"steady");
8507 }
8508
8509 #[tokio::test]
8510 async fn drain_reprice_migrates_before_the_session_dies() {
8511 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8512 let standby_server = rig.standby(&[10, 20]);
8513
8514 incumbent
8518 .update(Route::default().with_hops(hops(&[10])).with_cost(Cost::DRAIN))
8519 .unwrap();
8520
8521 assert_resumes(&mut rig, &standby_server).await;
8522
8523 drop(incumbent);
8525 drop(source);
8526 }
8527
8528 #[tokio::test]
8529 async fn local_sources_splice_newest_first() {
8530 let producer = origin(1).produce();
8531 let consumer = producer.consume();
8532
8533 let first = producer.publish("room/alice", Route::default()).unwrap();
8534 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8535
8536 let second = producer.publish("room/alice", Route::default()).unwrap();
8538 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
8539 assert!(again.is_clone(&resolved));
8540
8541 first.close();
8543 settle(|| consumer.get_broadcast("room/alice").is_some()).await;
8544 second.close();
8545 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8546
8547 let _third = producer.publish("room/alice", Route::default()).unwrap();
8549 assert!(consumer.get_broadcast("room/alice").is_some());
8550 }
8551
8552 #[tokio::test]
8560 async fn a_finished_broadcast_concludes_in_flight_subscriptions() {
8561 let producer = origin(1).produce();
8562 let consumer = producer.consume();
8563
8564 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8565 let track = broadcast.create_track("video", None).unwrap();
8566
8567 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8568 let mut subscription = resolved
8569 .track("video")
8570 .unwrap()
8571 .subscribe(None)
8572 .await
8573 .expect("subscribe");
8574 let mut group = track.append_group().unwrap();
8576 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8577 group.finish().unwrap();
8578 track.finish().unwrap();
8579 drop(track);
8580 broadcast.close();
8581
8582 let mut group = next_group(&mut subscription)
8583 .await
8584 .expect("a cleanly finished track was served as an error")
8585 .expect("the track ended before its last group");
8586 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8587 drop(group);
8588
8589 let end = next_group(&mut subscription)
8590 .await
8591 .expect("a cleanly finished track ended as an error");
8592 assert!(end.is_none(), "a group followed the final one");
8593 }
8594
8595 #[tokio::test]
8602 async fn a_retracted_route_concludes_in_flight_subscriptions() {
8603 let producer = origin(1).produce();
8604 let consumer = producer.consume();
8605 let server = producer
8606 .dynamic("room", Route::default().with_hops(hops(&[10])))
8607 .unwrap();
8608
8609 let pending = consumer.request_broadcast("room/alice");
8610 let request = queued(&server).await;
8611 let source = broadcast::Info::new().produce();
8612 let track = source.create_track("video", None).unwrap();
8613 request.accept(&source);
8614
8615 let resolved = pending.await.expect("resolves");
8616 let mut subscription = resolved
8617 .track("video")
8618 .unwrap()
8619 .subscribe(None)
8620 .await
8621 .expect("subscribe");
8622
8623 source.close();
8626 drop(server);
8627 settle(|| resolved.is_closed()).await;
8628 let mut group = track.append_group().unwrap();
8629 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8630 group.finish().unwrap();
8631 track.finish().unwrap();
8632 drop(track);
8633
8634 let mut group = next_group(&mut subscription)
8635 .await
8636 .expect("a retracted route's track was served as an error")
8637 .expect("the track ended before its last group");
8638 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8639 drop(group);
8640
8641 let end = next_group(&mut subscription)
8642 .await
8643 .expect("a cleanly finished track ended as an error");
8644 assert!(end.is_none(), "a group followed the final one");
8645 }
8646
8647 #[tokio::test]
8650 async fn a_closed_source_is_not_requested_again_from_its_standing_route() {
8651 let producer = origin(1).produce();
8652 let server = producer
8653 .dynamic("room", Route::default().with_hops(hops(&[10])))
8654 .unwrap();
8655 let pending = producer.consume().request_broadcast("room/alice");
8656 let source = broadcast::Info::new().produce();
8657 queued(&server).await.accept(&source);
8658 let resolved = pending.await.unwrap();
8659
8660 source.close();
8661 settle(|| resolved.is_closed()).await;
8662 assert!(
8663 server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8664 "the closed source was requested again"
8665 );
8666 }
8667
8668 #[tokio::test]
8673 async fn origin_front_drops_the_source_when_unused() {
8674 let producer = origin(1).produce();
8675 let consumer = producer.consume();
8676
8677 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8678 let track = broadcast.create_track("video", None).unwrap();
8679 let mut group = track.append_group().unwrap();
8680 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8681 group.finish().unwrap();
8682
8683 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8684 let mut subscription = resolved
8685 .track("video")
8686 .unwrap()
8687 .subscribe(None)
8688 .await
8689 .expect("subscribe");
8690 let mut group = subscription.recv_group().await.unwrap().unwrap();
8691 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8692 drop(group);
8693 drop(subscription);
8694
8695 tokio::time::timeout(Duration::from_secs(1), track.unused())
8696 .await
8697 .expect("source unused should resolve far below TRACK_IDLE_LINGER")
8698 .expect("source closed");
8699
8700 let mut again = resolved
8703 .track("video")
8704 .unwrap()
8705 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8706 .await
8707 .expect("resubscribe");
8708 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8709 .await
8710 .expect("cached group is still on the front")
8711 .expect("recv group")
8712 .expect("track ended early");
8713 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8714
8715 tokio::time::timeout(Duration::from_secs(1), track.used())
8716 .await
8717 .expect("returning reader re-splices the source")
8718 .expect("source closed");
8719
8720 let mut group = track.append_group().unwrap();
8721 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8722 group.finish().unwrap();
8723 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8724 .await
8725 .expect("groups past the cached edge come from the re-splice")
8726 .expect("recv group")
8727 .expect("track ended early");
8728 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8729 }
8730
8731 #[tokio::test]
8741 async fn resumed_reader_skips_warm_groups_behind_the_new_edge() {
8742 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
8743 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
8744 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
8745 let write = |source: &track::Producer, sequence: u64, millis: u64| {
8746 let mut group = source.create_group(sequence.into()).unwrap();
8747 group.write_frame(ms(millis), b"x".as_ref()).unwrap();
8748 group.finish().unwrap();
8749 };
8750 let drain = |subscription: &mut track::Subscriber| {
8751 let mut sequences = Vec::new();
8752 while let Poll::Ready(group) = subscription.poll_recv_group(&kio::Waiter::noop()) {
8753 sequences.push(group.unwrap().expect("track ended").sequence);
8754 }
8755 sequences
8756 };
8757
8758 let track = resolved.track("video").unwrap();
8759 let first = budget.clone();
8760 let subscribing = tokio::spawn(async move { track.subscribe(first).await });
8761 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8762 .await
8763 .expect("the front asked the source")
8764 .expect("request");
8765 let source = request.resolving_start().accept(None);
8766 for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] {
8767 write(&source, sequence, millis);
8768 }
8769 let mut subscription = subscribing.await.unwrap().expect("subscribe");
8770 next_group(&mut subscription).await.unwrap().expect("a cached group");
8771 drain(&mut subscription);
8772 drop(subscription);
8773
8774 tokio::time::timeout(Duration::from_secs(1), source.unused())
8775 .await
8776 .expect("parked")
8777 .expect("source open");
8778 drop(source);
8779
8780 let track = resolved.track("video").unwrap();
8781 let second = budget.clone();
8782 let subscribing = tokio::spawn(async move { track.subscribe(second).await });
8783 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8784 .await
8785 .expect("the front asked the source again")
8786 .expect("request");
8787 let mut source = request.resolving_start().accept(None);
8788 for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] {
8789 write(&source, sequence, millis);
8790 }
8791 source.start_at(3).unwrap();
8794 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
8795 settle(|| subscription.latest() == Some(22)).await;
8796
8797 assert_eq!(drain(&mut subscription), [3, 20, 21, 22]);
8800 }
8801
8802 #[tokio::test]
8807 async fn chained_front_drops_the_source_when_unused() {
8808 let leaf = origin(1).produce();
8809 let leaf_consumer = leaf.consume();
8810
8811 let broadcast = leaf.publish("room/alice", Route::default()).unwrap();
8812 let track = broadcast.create_track("video", None).unwrap();
8813 let mut group = track.append_group().unwrap();
8814 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8815 group.finish().unwrap();
8816
8817 let leaf_front = leaf_consumer.request_broadcast("room/alice").await.expect("resolves");
8820
8821 let mid = origin(2).produce();
8822 let mid_server = mid.dynamic("room", Route::default().with_hops(hops(&[10]))).unwrap();
8823 let mid_pending = mid.consume().request_broadcast("room/alice");
8824 queued(&mid_server).await.accept(&leaf_front);
8825 let mid_resolved = mid_pending.await.expect("mid resolves");
8826
8827 let edge = origin(3).produce();
8828 let edge_server = edge.dynamic("room", Route::default().with_hops(hops(&[20]))).unwrap();
8829 let edge_pending = edge.consume().request_broadcast("room/alice");
8830 queued(&edge_server).await.accept(&mid_resolved);
8831 let edge_resolved = edge_pending.await.expect("edge resolves");
8832
8833 let mut subscription = edge_resolved
8834 .track("video")
8835 .unwrap()
8836 .subscribe(None)
8837 .await
8838 .expect("subscribe");
8839 let mut group = subscription.recv_group().await.unwrap().unwrap();
8840 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8841 drop(group);
8842 drop(subscription);
8843
8844 tokio::time::timeout(Duration::from_secs(5), track.unused())
8845 .await
8846 .expect("chained unused should resolve far below TRACK_IDLE_LINGER")
8847 .expect("source closed");
8848
8849 let cached = edge_resolved.track("video").unwrap().cached_groups();
8850 assert_eq!(
8851 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8852 vec![0],
8853 "every front keeps the delivered groups after releasing its source"
8854 );
8855
8856 let mut subscription = edge_resolved
8858 .track("video")
8859 .unwrap()
8860 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8861 .await
8862 .expect("resubscribe");
8863 tokio::time::timeout(Duration::from_secs(5), track.used())
8864 .await
8865 .expect("resubscribe should reach the leaf")
8866 .expect("source open");
8867 let mut group = track.append_group().unwrap();
8868 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8869 group.finish().unwrap();
8870 let mut group = subscription.recv_group().await.unwrap().unwrap();
8871 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8872 drop(group);
8873 let mut group = subscription.recv_group().await.unwrap().unwrap();
8874 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8875 drop(group);
8876 drop(subscription);
8877
8878 tokio::time::timeout(Duration::from_secs(5), track.unused())
8879 .await
8880 .expect("second chained unused should resolve far below TRACK_IDLE_LINGER")
8881 .expect("source closed");
8882
8883 let cached = edge_resolved.track("video").unwrap().cached_groups();
8884 assert_eq!(
8885 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8886 vec![0, 1],
8887 "repeated demand keeps every complete group while releasing its source"
8888 );
8889
8890 let fetch = edge_resolved.track("video").unwrap().fetch_group(2, None);
8891 let mut fetch = std::pin::pin!(fetch);
8892 assert!(futures::poll!(fetch.as_mut()).is_pending(), "fetch should re-splice");
8893 tokio::time::timeout(Duration::from_secs(5), track.used())
8894 .await
8895 .expect("fetch should reach the leaf")
8896 .expect("source open");
8897 let mut group = track.append_group().unwrap();
8898 group.write_frame(crate::Timestamp::ZERO, b"fetched".as_ref()).unwrap();
8899 group.finish().unwrap();
8900 let mut group = tokio::time::timeout(Duration::from_secs(5), fetch)
8901 .await
8902 .expect("re-spliced source should answer the fetch")
8903 .expect("fetch succeeds");
8904 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"fetched");
8905 }
8906
8907 #[tokio::test]
8911 async fn incompatible_local_source_keeps_the_incumbent() {
8912 let producer = origin(1).produce();
8913 let consumer = producer.consume();
8914
8915 let first = producer.publish("room/alice", Route::default()).unwrap();
8916 let track = first.create_track("video", None).unwrap();
8917 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8918 let mut subscription = resolved
8919 .track("video")
8920 .unwrap()
8921 .subscribe(None)
8922 .await
8923 .expect("subscribe");
8924 let mut group = track.append_group().unwrap();
8925 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8926 group.finish().unwrap();
8927 let mut group = subscription.recv_group().await.unwrap().unwrap();
8928 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"before");
8929
8930 let second = producer.publish("room/alice", Route::default()).unwrap();
8932 let _incompatible = second
8933 .create_track("video", track::Info::default().with_timescale(crate::Timescale::MICRO))
8934 .unwrap();
8935 for _ in 0..10 {
8936 tokio::task::yield_now().await;
8937 }
8938
8939 let mut group = track.append_group().unwrap();
8941 group.write_frame(crate::Timestamp::ZERO, b"still".as_ref()).unwrap();
8942 group.finish().unwrap();
8943 let mut group = subscription.recv_group().await.unwrap().unwrap();
8944 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"still");
8945
8946 drop(track);
8948 first.close();
8949 assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported)));
8950 }
8951
8952 #[tokio::test]
8953 async fn multiple_scopes_present_one_broad_prefix() {
8954 let producer = origin(1).produce();
8955 let _a = producer.announce("", Route::default()).unwrap();
8956
8957 let consumer = producer.consume().scope("", &scopes(&["alpha", "beta"])).unwrap();
8958 let mut announced = consumer.announced();
8959 announced.assert_next_active("");
8960 announced.assert_next_wait();
8961 }
8962
8963 #[test]
8964 fn scope_accepts_every_pattern_union() {
8965 let producer = origin(1).produce();
8966
8967 let root = producer.scope("", &Patterns::from(Pattern::all())).unwrap();
8969 assert_eq!(root.allowed(), Patterns::from(Pattern::all()));
8970
8971 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8973 assert_eq!(scoped.allowed(), scopes(&["room"]));
8974
8975 let multi = producer.scope("", &scopes(&["room", "room/chat", "anon"])).unwrap();
8977 assert_eq!(multi.allowed(), scopes(&["room", "anon"]));
8978
8979 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
8981 assert_eq!(consumer.allowed(), scopes(&["room"]));
8982
8983 for text in ["room", "", "*room", "room/*", "*", "**/room", "room/**/chat", "*.hang"] {
8984 let union = Patterns::from(text.parse::<Pattern>().unwrap());
8985 assert_eq!(producer.scope("", &union).expect(text).allowed(), union, "{text}");
8986 assert_eq!(
8987 producer.consume().scope("", &union).expect(text).allowed(),
8988 union,
8989 "{text}"
8990 );
8991 }
8992
8993 let mixed: Patterns = ["room/**".parse().unwrap(), "other".parse().unwrap()]
8994 .into_iter()
8995 .collect();
8996 assert_eq!(producer.scope("", &mixed).unwrap().allowed(), mixed);
8997 }
8998
8999 #[test]
9000 fn route_table_prunes_to_empty() {
9001 let producer = origin(1).produce();
9002 let consumer = producer.consume();
9003
9004 let cursor = consumer
9007 .scope("", &scopes(&["room/a", "other/deep/head"]))
9008 .unwrap()
9009 .announced();
9010 let route = producer.announce("room/a/b/c", Route::default()).unwrap();
9011 {
9012 let table = producer.shared.lock();
9013 assert!(table.routes.root.find(Path::new("room/a/b/c").parts()).is_some());
9014 assert!(table.routes.root.find(Path::new("other/deep/head").parts()).is_some());
9015 assert_eq!(table.routes.root.cursors_below, 2);
9016 }
9017
9018 drop(route);
9019 drop(cursor);
9020 let table = producer.shared.lock();
9021 assert!(table.routes.root.is_empty());
9022 assert_eq!(table.routes.root.cursors_below, 0);
9023 }
9024
9025 #[test]
9029 fn a_published_broadcast_keeps_the_driver_running() {
9030 let (producer, mut driver) = Producer::new(Config::new(origin(1)));
9031 let waiter = kio::Waiter::noop();
9032 let broadcast = producer.create_broadcast("room/a").unwrap();
9033 drop(producer);
9034 assert!(
9035 driver.poll(Instant::now(), &waiter).is_ok(),
9036 "the broadcast is lifecycle work"
9037 );
9038 drop(broadcast);
9039 assert!(matches!(driver.poll(Instant::now(), &waiter), Err(Error::Closed)));
9040 }
9041
9042 #[tokio::test]
9045 async fn a_dynamic_keeps_the_driver_running() {
9046 let (producer, driver) = Producer::new(Config::new(origin(1)));
9047 let run = tokio::spawn(crate::time::run(driver));
9048 let consumer = producer.consume();
9049 let server = producer.dynamic("room", Route::default()).unwrap();
9050 drop(producer);
9051
9052 let pending = consumer.request_broadcast("room/alice");
9053 let request = queued(&server).await;
9054 let source = broadcast::Info::new().produce();
9055 request.accept(&source);
9056 let resolved = pending.await.expect("the dynamic still serves");
9057
9058 drop(resolved);
9059 drop(source);
9060 drop(server);
9061 tokio::time::timeout(Duration::from_secs(5), run)
9062 .await
9063 .expect("driver must finish once the dynamic is gone")
9064 .unwrap();
9065 drop(consumer);
9066 }
9067
9068 #[test]
9069 fn watch_wakes_only_for_covering_changes() {
9070 let producer = origin(1).produce();
9071 let waiter = kio::Waiter::noop();
9072 let watch = producer.shared.lock().watch(&producer.shared, &Path::new("room/a"));
9073 let seen = watch.seen();
9074
9075 let _other = producer.announce("other", Route::default()).unwrap();
9077 let _below = producer.announce("room/a/b", Route::default()).unwrap();
9078 assert!(watch.poll_changed(&waiter, seen).is_pending());
9079
9080 let above = producer.announce("room", Route::default()).unwrap();
9082 assert!(watch.poll_changed(&waiter, seen).is_ready());
9083 let seen = watch.seen();
9084 drop(above);
9085 assert!(watch.poll_changed(&waiter, seen).is_ready());
9086 let seen = watch.seen();
9087
9088 let _beside = producer.create_broadcast("room/b").unwrap();
9090 assert!(watch.poll_changed(&waiter, seen).is_pending());
9091 let _here = producer.create_broadcast("room/a").unwrap();
9092 assert!(watch.poll_changed(&waiter, seen).is_ready());
9093
9094 drop(watch);
9096 let table = producer.shared.lock();
9097 let node = table
9098 .routes
9099 .root
9100 .find(Path::new("room/a").parts())
9101 .expect("route below keeps the node");
9102 assert!(node.watches.is_empty());
9103 assert_eq!(table.routes.root.watches_below, 0);
9104 }
9105
9106 #[test]
9107 fn a_discarded_front_task_unregisters_its_watch() {
9108 let (producer, _driver) = Producer::new(Config {
9109 hop: origin(1),
9110 ..Default::default()
9111 });
9112 let consumer = producer.consume();
9113 let _served = producer.dynamic("room", Route::default()).unwrap();
9114 drop(producer);
9119 let _pending = consumer.request_broadcast("room/a");
9120 }
9121
9122 #[test]
9123 fn create_broadcast_refuses_a_path_no_pattern_can_spell() {
9124 let producer = origin(1).produce();
9125
9126 assert!(matches!(
9130 producer.create_broadcast("room/*"),
9131 Err(Error::InvalidPath(_))
9132 ));
9133 assert!(matches!(
9134 producer.announce("room/**", Route::default()),
9135 Err(Error::InvalidPath(_))
9136 ));
9137 }
9138
9139 #[test]
9140 fn scope_empty_union_grants_nothing() {
9141 let producer = origin(1).produce();
9142
9143 assert!(matches!(producer.scope("", &Patterns::new()), Err(Error::Unauthorized)));
9145 assert!(matches!(
9146 producer.consume().scope("", &Patterns::new()),
9147 Err(Error::Unauthorized)
9148 ));
9149 }
9150
9151 #[test]
9152 fn scope_nests_and_rebases_roots() {
9153 let producer = origin(1).produce();
9154
9155 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
9157 let nested = scoped.scope("", &scopes(&["room/chat"])).unwrap();
9158 assert_eq!(nested.allowed(), scopes(&["room/chat"]));
9159
9160 assert!(matches!(
9162 scoped.scope("", &scopes(&["other"])),
9163 Err(Error::Unauthorized)
9164 ));
9165
9166 let rooted = nested.scope("room/chat", &Patterns::from(Pattern::all())).unwrap();
9168 assert_eq!(rooted.allowed(), scopes(&[""]));
9169
9170 let broadcast = nested.create_broadcast("room/chat/live").unwrap();
9172 assert!(producer.consume().get_broadcast("room/chat/live").is_some());
9173 broadcast.close();
9174 }
9175
9176 #[test]
9177 fn scope_intersects_and_rebases_arbitrary_grants() {
9178 let producer = origin(1).produce();
9179 let rooms = producer
9180 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
9181 .unwrap();
9182 let chats = rooms
9183 .scope("", &Patterns::from("*/chat".parse::<Pattern>().unwrap()))
9184 .unwrap();
9185 assert_eq!(chats.allowed(), Patterns::from("room/chat".parse::<Pattern>().unwrap()));
9186
9187 let exact = producer
9188 .scope("", &Patterns::from("room/alice".parse::<Pattern>().unwrap()))
9189 .unwrap();
9190 let rooted = exact.scope("room", &Patterns::from(Pattern::all())).unwrap();
9191 assert_eq!(rooted.allowed(), Patterns::from("alice".parse::<Pattern>().unwrap()));
9192 assert!(matches!(
9193 exact.scope("room/bob", &Patterns::from(Pattern::all())),
9194 Err(Error::Unauthorized)
9195 ));
9196
9197 let broadcast = exact.create_broadcast("room/alice").unwrap();
9198 assert!(matches!(
9199 exact.create_broadcast("room/alice/cam"),
9200 Err(Error::Unauthorized)
9201 ));
9202 assert!(producer.consume().get_broadcast("room/alice").is_some());
9203 drop(broadcast);
9204 }
9205
9206 #[tokio::test]
9207 async fn wildcard_scope_filters_announcements_and_reports_captures() {
9208 let producer = origin(1).produce();
9209 let consumer = producer
9210 .consume()
9211 .scope("", &Patterns::from("room/*/chat".parse::<Pattern>().unwrap()))
9212 .unwrap();
9213 let mut announced = consumer.announced();
9214
9215 let alice = producer.create_broadcast("room/alice/chat").unwrap();
9216 alice.announce(Route::default()).unwrap();
9217 let update = announced.try_next().expect("alice's chat");
9218 assert_eq!(update.prefix.as_str(), "room/alice/chat");
9219 assert_eq!(update.captures, Some(vec!["alice".parse::<Pattern>().unwrap()]));
9220
9221 let audio = producer.create_broadcast("room/alice/audio").unwrap();
9222 audio.announce(Route::default()).unwrap();
9223 announced.assert_next_wait();
9224
9225 let broad = producer.announce("room", Route::default()).unwrap();
9226 let update = announced.try_next().expect("overlapping broad route");
9227 assert_eq!(update.prefix.as_str(), "room");
9228 assert_eq!(update.captures, None, "an overlap does not pin the wildcard");
9229
9230 drop(broad);
9231 drop(audio);
9232 drop(alice);
9233 }
9234
9235 #[tokio::test]
9236 async fn local_broadcast_wins_announcement_ties() {
9237 let producer = origin(1).produce();
9238 let remote = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
9239 let local = producer.create_broadcast("room/alice").unwrap();
9240 local.announce(Route::default()).unwrap();
9241
9242 let mut announced = producer.consume().announced();
9243 let update = announced.try_next().expect("one winning route");
9244 assert_eq!(update.prefix.as_str(), "room/alice");
9245 assert_eq!(update.route.cost, Cost::default());
9246 announced.assert_next_wait();
9247
9248 drop(local);
9249 drop(remote);
9250 }
9251
9252 #[test]
9257 fn cost_charge_saturates() {
9258 assert_eq!(Cost { warm: 4, cold: 6 }.charged(5), Cost { warm: 9, cold: 11 });
9259 assert_eq!(Cost::new(u64::MAX).charged(10), Cost::new(MAX_COST));
9260
9261 assert_eq!(Cost::UNKNOWN.charged(3).cold, MAX_COST);
9264 }
9265
9266 fn expiring_origin(expiry: Duration) -> Producer {
9268 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
9269 Config {
9270 pool,
9271 ..Config::default()
9272 }
9273 .produce()
9274 }
9275
9276 #[tokio::test(start_paused = true)]
9280 async fn stalled_publisher_open_group_is_reclaimed() {
9281 let expiry = Duration::from_secs(1);
9282 let origin = expiring_origin(expiry);
9283 let broadcast = origin.create_broadcast("test").unwrap();
9284 let track = broadcast.create_track("video", None).unwrap();
9285
9286 let mut stalled = track.append_group().unwrap();
9287 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
9288 let _successor = track.append_group().unwrap();
9292
9293 let mut reading = stalled.consume();
9294 assert!(reading.read_frame().await.unwrap().is_some());
9295
9296 crate::model::clock::advance(expiry * 2);
9298
9299 let reclaimed = tokio::time::timeout(Duration::from_secs(60), reading.read_frame()).await;
9302 assert!(
9303 matches!(reclaimed, Ok(Err(Error::Old))),
9304 "the sweep must reclaim an idle open group and surface the gap, got {reclaimed:?}"
9305 );
9306 }
9307
9308 #[tokio::test(start_paused = true)]
9311 async fn sweep_respects_a_disabled_expiry() {
9312 let origin = Config {
9313 pool: cache::Pool::unbounded(),
9314 ..Config::default()
9315 }
9316 .produce();
9317 let broadcast = origin.create_broadcast("test").unwrap();
9318 let track = broadcast.create_track("video", None).unwrap();
9319
9320 let mut stalled = track.append_group().unwrap();
9321 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
9322 let _successor = track.append_group().unwrap();
9323
9324 let mut reading = stalled.consume();
9325 assert!(reading.read_frame().await.unwrap().is_some());
9326
9327 crate::model::clock::advance(Duration::from_secs(3600));
9328 tokio::time::advance(Duration::from_secs(3600)).await;
9329
9330 assert!(
9331 reading.read_frame().now_or_never().is_none(),
9332 "a pool without an expiry window never reclaims"
9333 );
9334 }
9335
9336 #[test]
9339 fn drain_cost_is_encodable() {
9340 use crate::coding::Encode;
9341
9342 let mut buf = Vec::new();
9343 Cost::DRAIN
9344 .encode(&mut buf, crate::lite::Version::Lite06)
9345 .expect("a draining route is still forwarded, so its cost must encode");
9346 }
9347}