1use crate::{broadcast, cache, group, stats, track};
2use kio::Pollable;
3use std::{
4 cmp::Reverse,
5 collections::{BTreeMap, 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 pool: cache::Pool,
109
110 pub cache_duration: Duration,
118
119 pub default_max_age: Duration,
129
130 pub update_hold: Duration,
147}
148
149pub const DEFAULT_UPDATE_HOLD: Duration = Duration::from_millis(300);
152
153const RETRACT_QUARANTINE: Duration = Duration::from_secs(2);
157
158impl Default for Config {
159 fn default() -> Self {
161 let pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY));
162 Self {
163 hop: Hop::random(),
164 pool,
165 cache_duration: Duration::MAX,
166 default_max_age: track::DEFAULT_MAX_AGE,
167 update_hold: DEFAULT_UPDATE_HOLD,
168 }
169 }
170}
171
172impl Config {
173 pub fn new(hop: Hop) -> Self {
175 Self { hop, ..Self::default() }
176 }
177}
178
179impl From<Hop> for Config {
180 fn from(hop: Hop) -> Self {
182 Self::new(hop)
183 }
184}
185
186impl TryFrom<u64> for Hop {
187 type Error = InvalidHop;
188
189 fn try_from(id: u64) -> Result<Self, Self::Error> {
190 Self::new(id)
191 }
192}
193
194impl fmt::Display for Hop {
195 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
196 self.id.fmt(f)
197 }
198}
199
200impl<V: Copy> Encode<V> for Hop
201where
202 u64: Encode<V>,
203{
204 fn encode<W: bytes::BufMut>(&self, w: &mut W, version: V) -> Result<(), EncodeError> {
205 self.id.encode(w, version)
206 }
207}
208
209impl<V: Copy> Decode<V> for Hop
210where
211 u64: Decode<V>,
212{
213 fn decode<R: bytes::Buf>(r: &mut R, version: V) -> Result<Self, DecodeError> {
214 Self::from_wire(u64::decode(r, version)?)
215 }
216}
217
218pub(crate) const MAX_HOPS: usize = 32;
224
225#[derive(Debug, Clone, Default, PartialEq, Eq)]
233pub struct Hops(Vec<Hop>);
234
235#[derive(Debug, Clone, Copy, PartialEq, Eq)]
237#[non_exhaustive]
238pub enum InvalidHop {
239 Range,
242
243 TooMany,
246
247 Duplicate,
251}
252
253impl fmt::Display for InvalidHop {
254 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
255 match self {
256 Self::Range => write!(f, "local hop id must be non-zero and below 2^62"),
257 Self::TooMany => write!(f, "too many hops (max {MAX_HOPS})"),
258 Self::Duplicate => write!(f, "hop already in the chain"),
259 }
260 }
261}
262
263impl std::error::Error for InvalidHop {}
264
265impl From<InvalidHop> for DecodeError {
266 fn from(err: InvalidHop) -> Self {
267 match err {
268 InvalidHop::TooMany => DecodeError::BoundsExceeded,
269 InvalidHop::Range | InvalidHop::Duplicate => DecodeError::InvalidValue,
270 }
271 }
272}
273
274impl Hops {
275 pub fn new() -> Self {
277 Self(Vec::new())
278 }
279
280 pub fn push(&mut self, hop: Hop) -> Result<(), InvalidHop> {
286 if self.0.len() >= MAX_HOPS {
287 return Err(InvalidHop::TooMany);
288 }
289 if hop != Hop::UNKNOWN && self.0.contains(&hop) {
290 return Err(InvalidHop::Duplicate);
291 }
292 self.0.push(hop);
293 Ok(())
294 }
295
296 pub(crate) fn stamp(&mut self, stamp: Hop) -> Result<(), InvalidHop> {
305 match self.0.first() {
306 None => {
307 self.push(stamp)?;
308 self.push(Hop::UNKNOWN)
309 }
310 Some(first) if *first == Hop::UNKNOWN => {
311 if self.0.len() >= MAX_HOPS {
312 return Err(InvalidHop::TooMany);
313 }
314 if self.0.contains(&stamp) {
315 return Err(InvalidHop::Duplicate);
316 }
317 self.0.insert(0, stamp);
318 Ok(())
319 }
320 Some(_) => Ok(()),
321 }
322 }
323
324 pub fn contains(&self, hop: &Hop) -> bool {
326 self.0.contains(hop)
327 }
328
329 pub fn len(&self) -> usize {
331 self.0.len()
332 }
333
334 pub fn is_empty(&self) -> bool {
336 self.0.is_empty()
337 }
338
339 pub fn iter(&self) -> std::slice::Iter<'_, Hop> {
341 self.0.iter()
342 }
343
344 pub fn as_slice(&self) -> &[Hop] {
346 &self.0
347 }
348}
349
350impl TryFrom<Vec<Hop>> for Hops {
351 type Error = InvalidHop;
352
353 fn try_from(v: Vec<Hop>) -> Result<Self, Self::Error> {
354 if v.len() > MAX_HOPS {
355 return Err(InvalidHop::TooMany);
356 }
357 for (i, hop) in v.iter().enumerate() {
359 if *hop != Hop::UNKNOWN && v[i + 1..].contains(hop) {
360 return Err(InvalidHop::Duplicate);
361 }
362 }
363 Ok(Self(v))
364 }
365}
366
367impl<'a> IntoIterator for &'a Hops {
368 type Item = &'a Hop;
369 type IntoIter = std::slice::Iter<'a, Hop>;
370
371 fn into_iter(self) -> Self::IntoIter {
372 self.iter()
373 }
374}
375
376impl<V: Copy> Encode<V> for Hops
377where
378 u64: Encode<V>,
379 Hop: Encode<V>,
380{
381 fn encode<W: bytes::BufMut>(&self, w: &mut W, version: V) -> Result<(), EncodeError> {
382 (self.0.len() as u64).encode(w, version)?;
383 for origin in &self.0 {
384 origin.encode(w, version)?;
385 }
386 Ok(())
387 }
388}
389
390impl<V: Copy> Decode<V> for Hops
391where
392 u64: Decode<V>,
393 Hop: Decode<V>,
394{
395 fn decode<R: bytes::Buf>(r: &mut R, version: V) -> Result<Self, DecodeError> {
396 let count = u64::decode(r, version)? as usize;
397 if count > MAX_HOPS {
398 return Err(DecodeError::BoundsExceeded);
399 }
400 let mut list = Self(Vec::with_capacity(count));
403 for _ in 0..count {
404 list.push(Hop::decode(r, version)?)?;
405 }
406 Ok(list)
407 }
408}
409
410const MAX_COST: u64 = (1 << 62) - 1;
418
419#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
428pub struct Cost {
429 pub warm: u64,
438
439 pub cold: u64,
446}
447
448impl Cost {
449 pub const fn new(cost: u64) -> Self {
452 Self { warm: cost, cold: cost }
453 }
454
455 pub const MAX: Self = Self::new(MAX_COST);
462
463 pub const DRAIN: Self = Self::MAX;
466
467 pub(crate) const UNKNOWN: Self = Self {
471 warm: 0,
472 cold: MAX_COST,
473 };
474
475 pub(crate) fn charged(self, link_cost: u64) -> Self {
478 Self {
479 warm: self.warm.saturating_add(link_cost).min(MAX_COST),
480 cold: self.cold.saturating_add(link_cost).min(MAX_COST),
481 }
482 }
483
484 pub(crate) fn clamped(self) -> Self {
487 Self {
488 warm: self.warm.min(MAX_COST),
489 cold: self.cold.min(MAX_COST),
490 }
491 }
492}
493
494impl From<u64> for Cost {
495 fn from(cost: u64) -> Self {
496 Self::new(cost)
497 }
498}
499
500#[derive(Clone, Debug, PartialEq, Eq)]
511#[non_exhaustive]
512pub struct Route {
513 pub hops: Hops,
519
520 pub cost: Cost,
525
526 pub(crate) via: Hop,
532
533 pub(crate) source: Source,
535}
536
537impl Default for Route {
538 fn default() -> Self {
539 Self {
540 hops: Hops::new(),
541 cost: Cost::default(),
542 via: Hop::UNKNOWN,
543 source: Source::Local,
544 }
545 }
546}
547
548#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
554pub enum Source {
555 #[default]
557 Local,
558 Peer(Hop),
561}
562
563impl Route {
564 pub fn with_hops(mut self, hops: Hops) -> Self {
566 self.hops = hops;
567 self
568 }
569
570 pub fn with_cost(mut self, cost: impl Into<Cost>) -> Self {
575 self.cost = cost.into();
576 self
577 }
578
579 pub(crate) fn with_via(mut self, via: Hop) -> Self {
584 self.via = via;
585 self
586 }
587
588 pub fn is_anonymous(&self) -> bool {
596 self.hops.iter().any(|hop| *hop == Hop::UNKNOWN)
597 }
598
599 pub fn source(&self) -> Source {
604 self.source
605 }
606}
607
608static NEXT_CONSUMER_ID: AtomicU64 = AtomicU64::new(0);
609
610#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
611struct ConsumerId(u64);
612
613impl ConsumerId {
614 fn new() -> Self {
615 Self(NEXT_CONSUMER_ID.fetch_add(1, Ordering::Relaxed))
616 }
617}
618
619fn fnv_key(name: &str, origins: impl IntoIterator<Item = Hop>) -> u64 {
628 const SEED: u64 = 0x420C0DECB00B; const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
630
631 let mut hash = SEED;
632 for &byte in name.as_bytes() {
633 hash = (hash ^ u64::from(byte)).wrapping_mul(FNV_PRIME);
634 }
635 for origin in origins {
636 for &byte in &origin.id().to_le_bytes() {
637 hash = (hash ^ u64::from(byte)).wrapping_mul(FNV_PRIME);
638 }
639 }
640
641 hash
642}
643
644fn route_order(prefix: &Path, entry: &RouteEntry) -> (bool, Cost, bool, usize, u64, Reverse<u64>) {
652 (
653 entry.is_anonymous(),
654 entry.cost,
655 !entry.local,
656 entry.hops.len(),
657 fnv_key(prefix.as_str(), entry.hops.iter().copied()),
658 Reverse(entry.id),
659 )
660}
661
662type RouteMeta = (Hops, Cost, Source);
664
665fn earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant> {
667 match (a, b) {
668 (Some(a), Some(b)) => Some(a.min(b)),
669 (a, b) => a.or(b),
670 }
671}
672
673#[derive(Clone, Copy)]
677struct Quarantine {
678 held: Held,
679 publisher: Option<Hop>,
680 rank: (Cost, usize),
681}
682
683impl Quarantine {
684 fn holds(&self, meta: &RouteMeta) -> bool {
688 self.publisher.is_some_and(|hop| meta.0.iter().next() == Some(&hop)) && (meta.1, meta.0.len()) > self.rank
689 }
690}
691
692#[derive(Clone, Copy, PartialEq)]
699enum Held {
700 Anchoring(Instant),
701 Until(Instant),
702}
703
704type AnnounceMeta = (RouteMeta, Option<Vec<Pattern>>);
711
712enum PendingUpdate {
713 Announce(AnnounceMeta),
714 Unannounce(AnnounceMeta),
715 UnannounceAnnounce { old: AnnounceMeta, new: AnnounceMeta },
716}
717
718#[derive(Default)]
723struct OriginConsumerState {
724 pending: BTreeMap<PathOwned, PendingUpdate>,
725 delivered: BTreeSet<PathOwned>,
730 ended: bool,
733}
734
735impl OriginConsumerState {
736 fn apply_announce(&mut self, prefix: PathOwned, meta: RouteMeta, captures: Option<Vec<Pattern>>) {
737 let meta = (meta, captures);
738 let new = match self.pending.remove(&prefix) {
739 None | Some(PendingUpdate::Announce(_)) => PendingUpdate::Announce(meta),
741 Some(PendingUpdate::Unannounce(old) | PendingUpdate::UnannounceAnnounce { old, .. }) => {
743 PendingUpdate::UnannounceAnnounce { old, new: meta }
744 }
745 };
746 self.pending.insert(prefix, new);
747 }
748
749 fn apply_unannounce(&mut self, prefix: PathOwned, last: RouteMeta, captures: Option<Vec<Pattern>>) {
750 let last = (last, captures);
751 match self.pending.remove(&prefix) {
752 Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => {}
755 None | Some(PendingUpdate::Announce(_) | PendingUpdate::Unannounce(_)) => {
758 self.pending.insert(prefix, PendingUpdate::Unannounce(last));
759 }
760 Some(PendingUpdate::UnannounceAnnounce { old, .. }) => {
763 self.pending.insert(prefix, PendingUpdate::Unannounce(old));
764 }
765 }
766 }
767
768 fn is_update(&self, prefix: &PathOwned, update: &PendingUpdate) -> bool {
771 matches!(update, PendingUpdate::Announce(_)) && self.delivered.contains(prefix)
772 }
773
774 fn scan(
778 &self,
779 held: &mut HashMap<PathOwned, Held>,
780 retracted: &HashMap<PathOwned, Quarantine>,
781 now: Option<Instant>,
782 hold: Duration,
783 ) -> Result<PathOwned, Option<Instant>> {
784 let mut wake = None;
785 for (prefix, update) in &self.pending {
786 let Some(now) = now.filter(|_| !hold.is_zero() && !self.ended) else {
787 return Ok(prefix.clone());
788 };
789 if !self.is_update(prefix, update) {
790 let quarantine = match update {
791 PendingUpdate::Announce((meta, _)) => retracted.get(prefix).filter(|q| q.holds(meta)),
792 _ => None,
793 };
794 let until = match quarantine.map(|q| q.held) {
795 Some(Held::Anchoring(seen)) => seen + Duration::from_nanos(1),
796 Some(Held::Until(until)) if now < until => until,
797 _ => return Ok(prefix.clone()),
798 };
799 wake = Some(wake.map_or(until, |w: Instant| w.min(until)));
800 continue;
801 }
802 let until = match held.get(prefix).copied() {
805 None => {
806 held.insert(prefix.clone(), Held::Anchoring(now));
807 now + Duration::from_nanos(1)
808 }
809 Some(Held::Anchoring(seen)) if now <= seen => seen + Duration::from_nanos(1),
810 Some(Held::Anchoring(_)) => {
811 held.insert(prefix.clone(), Held::Until(now + hold));
812 now + hold
813 }
814 Some(Held::Until(until)) if until <= now => return Ok(prefix.clone()),
815 Some(Held::Until(until)) => until,
816 };
817 wake = Some(wake.map_or(until, |w: Instant| w.min(until)));
818 }
819 Err(wake)
820 }
821
822 fn take_prefix(&mut self, prefix: PathOwned) -> Option<AnnounceUpdate> {
824 let ((meta, captures), kind) = match self.pending.remove(&prefix).unwrap() {
825 PendingUpdate::Announce(meta) => {
826 let kind = match self.delivered.insert(prefix.clone()) {
828 true => AnnounceKind::Announced,
829 false => AnnounceKind::Updated,
830 };
831 (meta, kind)
832 }
833 PendingUpdate::Unannounce(meta) => {
834 self.delivered.remove(&prefix);
835 (meta, AnnounceKind::Retracted)
836 }
837 PendingUpdate::UnannounceAnnounce { old, new } => {
838 self.delivered.remove(&prefix);
841 self.pending.insert(prefix.clone(), PendingUpdate::Announce(new));
842 (old, AnnounceKind::Retracted)
843 }
844 };
845 Some(AnnounceUpdate {
846 prefix,
847 captures,
848 route: Route {
849 hops: meta.0,
850 cost: meta.1,
851 via: Hop::UNKNOWN,
852 source: meta.2,
853 },
854 kind,
855 })
856 }
857}
858
859struct RouteEntry {
861 id: u64,
862 prefix: PathOwned,
863 scope: Patterns,
866 hops: Hops,
867 cost: Cost,
868 via: Hop,
872 local: bool,
875 peer: bool,
878 server: Option<kio::Shared<ServeState>>,
882 source: Option<broadcast::Consumer>,
886 advertised: bool,
890 stale: bool,
894 claim: Pattern,
900}
901
902impl RouteEntry {
903 fn live(&self) -> bool {
905 self.advertised && !self.stale
906 }
907
908 fn is_anonymous(&self) -> bool {
909 self.hops.iter().any(|hop| *hop == Hop::UNKNOWN)
910 }
911
912 fn entered(&self) -> Source {
914 match self.peer {
915 true => Source::Peer(self.via),
916 false => Source::Local,
917 }
918 }
919
920 fn serves(&self, path: &Path) -> bool {
924 self.server.is_some() || (self.source.is_some() && self.prefix == *path)
925 }
926
927 fn qualifies(&self, pin: Pin) -> bool {
929 match pin {
930 Pin::Any => true,
931 Pin::Local => self.local,
932 Pin::Publisher(first) => self.hops.iter().next() == Some(&first),
933 Pin::Route(id) => self.id == id && self.hops.iter().next().is_none_or(|first| *first == Hop::UNKNOWN),
935 }
936 }
937
938 fn visible_to(&self, exclude: Option<Hop>) -> bool {
943 match exclude {
944 Some(peer) if peer != Hop::UNKNOWN => self.via != peer && !self.hops.contains(&peer),
945 _ => true,
946 }
947 }
948
949 fn overlaps(&self, allowed: &Patterns) -> bool {
951 self.scope.iter().any(|scope| {
952 scope
953 .intersect(&self.claim)
954 .is_ok_and(|scoped| scoped.iter().any(|restriction| allowed.overlaps(restriction)))
955 })
956 }
957}
958
959fn prefix_claim(prefix: &Path) -> Result<Pattern, InvalidPattern> {
961 if prefix.parts().count() == Path::MAX_PARTS {
962 Pattern::literal(prefix.as_str())
963 } else {
964 Pattern::subtree(prefix.as_str())
965 }
966}
967
968#[derive(Default)]
973struct ServeState {
974 requests: Requests<PathOwned, kio::Producer<PendingBroadcast>>,
977
978 served: WeakCache<PathOwned, broadcast::WeakConsumer>,
984
985 closed: bool,
988}
989
990type FrontKey = (PathOwned, Horizon);
997
998#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
1000struct Horizon {
1001 exclude: Option<Hop>,
1005 local: bool,
1007}
1008
1009impl Horizon {
1010 fn admits(&self, entry: &RouteEntry) -> bool {
1012 !(self.local && entry.peer) && entry.visible_to(self.exclude)
1013 }
1014
1015 fn effective(self, routes: &RouteTable, path: &Path) -> Self {
1025 match self.exclude {
1026 Some(peer) if routes.covering(path).any(|entry| !entry.visible_to(Some(peer))) => self,
1027 _ => Self { exclude: None, ..self },
1028 }
1029 }
1030}
1031
1032#[derive(Clone)]
1035struct RemoteFront {
1036 request: kio::Producer<PendingBroadcast>,
1040 broadcast: broadcast::WeakConsumer,
1043 pin: kio::Lock<Pin>,
1046}
1047
1048type CursorRoute = (u64, RouteMeta, bool, Option<Vec<Pattern>>);
1050
1051impl WeakEntry for RemoteFront {
1052 fn is_closed(&self) -> bool {
1053 self.broadcast.is_closed()
1054 }
1055
1056 fn same_channel(&self, other: &Self) -> bool {
1057 self.broadcast.same_channel(&other.broadcast)
1058 }
1059}
1060
1061struct TableCursor {
1064 root: PathOwned,
1066 allowed: Patterns,
1068 heads: Vec<PathOwned>,
1072 horizon: Horizon,
1074 hidden: Hidden,
1076 state: kio::Producer<OriginConsumerState>,
1079 under: PathOwned,
1082 holes: Vec<PathOwned>,
1085 named: Option<(Mount, Patterns)>,
1088 current: HashMap<PathOwned, CursorRoute>,
1093}
1094
1095impl TableCursor {
1096 fn presented(&self, prefix: &Path, claim: &Pattern) -> Option<PathOwned> {
1102 if !self.allowed.overlaps(claim) {
1103 return None;
1104 }
1105
1106 if let Some(relative) = prefix.strip_prefix(&self.root) {
1107 return Some(relative.to_owned());
1108 }
1109 self.root.has_prefix(prefix).then(PathOwned::default)
1110 }
1111
1112 fn captures(&self, prefix: &Path) -> Option<Vec<Pattern>> {
1115 let (literal, allowed) = match &self.named {
1116 Some((mount, allowed)) => (Pattern::literal(mount.name(prefix)?.as_str()).ok()?, allowed),
1117 None => (Pattern::literal(prefix.as_str()).ok()?, &self.allowed),
1118 };
1119 allowed
1120 .iter()
1121 .filter_map(|allowed| {
1122 allowed
1123 .captures(&literal)
1124 .map(|captures| (allowed.specificity(), captures))
1125 })
1126 .max_by_key(|(specificity, _)| *specificity)
1127 .map(|(_, captures)| captures)
1128 }
1129
1130 fn visible(&self, entry: &RouteEntry) -> bool {
1134 entry.live()
1135 && self.horizon.admits(entry)
1136 && entry.overlaps(&self.allowed)
1137 && self.discovers(&entry.prefix)
1138 && !self.holes.iter().any(|hole| entry.prefix.has_prefix(hole))
1139 && self.named.as_ref().is_none_or(|(mount, _)| mount.names(&entry.prefix))
1140 }
1141
1142 fn update(&mut self, presented: &PathOwned, best: Option<&RouteEntry>) {
1144 match best {
1145 Some(entry) => {
1146 let meta = (entry.hops.clone(), entry.cost, entry.entered());
1147 let served = entry.server.is_some();
1148 let captures = self.captures(&entry.prefix);
1149 let previous = self
1150 .current
1151 .insert(presented.clone(), (entry.id, meta.clone(), served, captures.clone()));
1152 match previous {
1153 Some((_, prev, prev_served, prev_captures))
1160 if prev == meta && prev_served == served && prev_captures == captures => {}
1161 Some((_, prev, _, prev_captures)) if prev_captures != captures => {
1164 if let Ok(mut state) = self.state.write() {
1165 state.apply_unannounce(self.under.join(presented), prev, prev_captures);
1166 state.apply_announce(self.under.join(presented), meta, captures);
1167 }
1168 }
1169 _ => {
1170 if let Ok(mut state) = self.state.write() {
1171 state.apply_announce(self.under.join(presented), meta, captures);
1172 }
1173 }
1174 }
1175 }
1176 None => {
1177 if let Some((_, last, _, captures)) = self.current.remove(presented)
1178 && let Ok(mut state) = self.state.write()
1179 {
1180 state.apply_unannounce(self.under.join(presented), last, captures);
1181 }
1182 }
1183 }
1184 }
1185
1186 fn discovers(&self, prefix: &Path) -> bool {
1188 self.hidden.discovers(&self.heads, prefix)
1189 }
1190}
1191
1192#[derive(Clone)]
1195struct OriginScope {
1196 allowed: Patterns,
1199 mounts: Arc<[Mount]>,
1201}
1202
1203#[derive(Clone, Debug)]
1206struct Mount {
1207 at: PathOwned,
1208 target: PathOwned,
1209}
1210
1211impl Mount {
1212 fn resolve(&self, path: &Path) -> Option<PathOwned> {
1215 let resolved = self.target.join(path.strip_prefix(&self.at)?);
1216 (resolved.parts().count() <= Path::MAX_PARTS).then_some(resolved)
1217 }
1218
1219 fn name(&self, path: &Path) -> Option<PathOwned> {
1222 Some(self.at.join(path.strip_prefix(&self.target)?))
1223 }
1224
1225 fn names(&self, path: &Path) -> bool {
1229 path.strip_prefix(&self.target)
1230 .is_none_or(|rest| self.at.parts().count() + rest.parts().count() <= Path::MAX_PARTS)
1231 }
1232
1233 fn translate(&self, patterns: &Patterns) -> Patterns {
1240 let target = self.target.as_str();
1241 patterns
1242 .rebase(self.at.as_str())
1243 .iter()
1244 .filter_map(|member| {
1245 member
1246 .rooted(target)
1247 .or_else(|_| {
1248 let segments = member
1249 .segments()
1250 .iter()
1251 .filter(|segment| **segment != Segment::Globstar);
1252 Pattern::new(segments.cloned())?.rooted(target)
1253 })
1254 .ok()
1255 })
1256 .collect()
1257 }
1258
1259 fn translate_head(&self, head: &Path) -> Option<PathOwned> {
1262 match head.strip_prefix(&self.at) {
1263 Some(rest) => Some(self.target.join(rest)),
1264 None => self.at.has_prefix(head).then(|| self.target.clone()),
1265 }
1266 }
1267}
1268
1269impl OriginScope {
1270 fn empty() -> Self {
1272 Self {
1273 allowed: Patterns::new(),
1274 mounts: Arc::from([]),
1275 }
1276 }
1277
1278 fn narrow(&self, patterns: &Patterns) -> Option<Self> {
1280 let allowed = self.allowed.intersect(patterns).ok()?;
1281 if allowed.is_empty() {
1282 None
1283 } else {
1284 Some(Self {
1285 allowed,
1286 mounts: self.mounts.clone(),
1287 })
1288 }
1289 }
1290
1291 fn mount(&self, path: &Path) -> Option<&Mount> {
1293 self.mounts.iter().find(|mount| path.has_prefix(&mount.at))
1294 }
1295
1296 fn resolve<'a>(&self, path: &'a Path<'a>) -> Option<Path<'a>> {
1299 match self.mount(path) {
1300 Some(mount) => mount.resolve(path),
1301 None => Some(path.borrow()),
1302 }
1303 }
1304
1305 fn publishes(&self, prefix: &Path) -> bool {
1308 self.mount(prefix).is_none()
1309 }
1310
1311 fn permits(&self, path: &Path) -> bool {
1313 self.allowed.matches(path.as_str())
1314 }
1315
1316 fn relative(&self, root: &Path) -> Patterns {
1318 self.allowed.rebase(root.as_str())
1319 }
1320}
1321
1322impl Default for OriginScope {
1323 fn default() -> Self {
1324 Self {
1325 allowed: Patterns::from(Pattern::all()),
1326 mounts: Arc::from([]),
1327 }
1328 }
1329}
1330
1331pub(crate) fn interest_prefixes(allowed: &Patterns) -> Vec<PathOwned> {
1334 let mut heads: Vec<PathOwned> = allowed
1335 .iter()
1336 .map(|pattern| Path::new(pattern.head()).to_owned())
1337 .collect();
1338 heads.sort();
1339 heads.dedup();
1340 let covered = heads.clone();
1341 heads.retain(|head| !covered.iter().any(|other| other != head && head.has_prefix(other)));
1342 heads
1343}
1344
1345#[derive(Clone, Debug, Default, PartialEq, Eq)]
1352pub(crate) struct Hidden {
1353 include: bool,
1355 from: Option<PathOwned>,
1357 beyond: Option<Vec<PathOwned>>,
1360}
1361
1362impl Hidden {
1363 fn discovers(&self, heads: &[PathOwned], prefix: &Path) -> bool {
1365 (self.include || !hides(self.from.as_ref().map(std::slice::from_ref).unwrap_or(heads), prefix))
1366 && self.beyond.as_ref().is_none_or(|outer| hides(outer, prefix))
1367 }
1368
1369 fn translate(&self, mount: &Mount) -> Self {
1373 let beyond = self.beyond.as_ref().and_then(|outer| {
1374 (!hides(outer, &mount.at)).then(|| outer.iter().filter_map(|head| mount.translate_head(head)).collect())
1375 });
1376 Self {
1377 include: self.include,
1378 from: self.from.as_ref().and_then(|head| mount.translate_head(head)),
1379 beyond,
1380 }
1381 }
1382}
1383
1384fn hides(heads: &[PathOwned], prefix: &Path) -> bool {
1387 heads
1388 .iter()
1389 .any(|head| prefix.strip_prefix(head).is_some_and(|below| below.is_hidden()))
1390}
1391
1392#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1394pub enum AnnounceKind {
1395 Announced,
1397 Updated,
1399 Retracted,
1401}
1402
1403impl AnnounceKind {
1404 pub fn is_active(self) -> bool {
1406 !matches!(self, Self::Retracted)
1407 }
1408}
1409
1410#[derive(Clone, Debug)]
1418pub struct AnnounceUpdate {
1419 pub prefix: PathOwned,
1421 pub captures: Option<Vec<Pattern>>,
1424 pub route: Route,
1427 pub kind: AnnounceKind,
1429}
1430
1431#[derive(Clone)]
1433pub struct Producer {
1434 hop: Hop,
1437
1438 scope: OriginScope,
1440
1441 root: PathOwned,
1443
1444 shared: kio::Shared<OriginState>,
1447
1448 pool: cache::Pool,
1451
1452 cache_duration: Duration,
1455
1456 default_max_age: Duration,
1459
1460 stats: stats::Session,
1464
1465 peer: bool,
1468
1469 tasks: Tasks,
1473
1474 timers: Clock,
1476}
1477
1478impl Producer {
1479 pub fn new(config: Config) -> (Self, Driver) {
1486 let (tasks, set) = TaskSet::new();
1487 let scope = OriginScope::default();
1488 let shared = kio::Shared::new(OriginState::new(config.update_hold));
1489 let timers = Clock::default();
1490 let pool = config.pool.clone();
1491 let producer = Self {
1492 hop: config.hop,
1493 scope: scope.clone(),
1494 root: PathOwned::default(),
1495 shared: shared.clone(),
1496 pool: config.pool,
1497 cache_duration: config.cache_duration,
1498 default_max_age: config.default_max_age,
1499 stats: stats::Session::default(),
1500 peer: false,
1501 tasks,
1502 timers: timers.clone(),
1503 };
1504 let driver = Driver {
1505 state: DriverState {
1506 set,
1507 shared,
1508 done: false,
1509 },
1510 timers,
1511 pool,
1512 };
1513 (producer, driver)
1514 }
1515
1516 pub fn with_stats(mut self, session: stats::Session) -> Self {
1520 self.stats = session;
1521 self
1522 }
1523
1524 pub fn peer(mut self) -> Self {
1532 self.peer = true;
1533 self
1534 }
1535
1536 pub fn config(&self) -> Config {
1538 Config {
1539 hop: self.hop,
1540 pool: self.pool.clone(),
1541 cache_duration: self.cache_duration,
1542 default_max_age: self.default_max_age,
1543 update_hold: self.shared.lock().update_hold,
1544 }
1545 }
1546
1547 pub fn hop(&self) -> Hop {
1549 self.hop
1550 }
1551
1552 pub(crate) fn default_max_age(&self) -> Duration {
1555 self.default_max_age
1556 }
1557
1558 pub(crate) fn empty(hop: Hop) -> Self {
1563 let (tasks, _) = TaskSet::new();
1566 Self {
1567 hop,
1568 scope: OriginScope::empty(),
1569 root: PathOwned::default(),
1570 shared: kio::Shared::new(OriginState::new(DEFAULT_UPDATE_HOLD)),
1571 pool: cache::Pool::default(),
1572 cache_duration: Duration::MAX,
1573 default_max_age: track::DEFAULT_MAX_AGE,
1574 stats: stats::Session::default(),
1575 peer: false,
1576 tasks,
1577 timers: Clock::default(),
1578 }
1579 }
1580
1581 pub fn create_broadcast(&self, path: impl AsPath) -> Result<broadcast::Producer, Error> {
1612 let path = path.as_path();
1613
1614 let full = self.root.join(&path).to_owned();
1615 if !self.scope.permits(&full) || !self.scope.publishes(&full) {
1616 return Err(Error::Unauthorized);
1617 }
1618 if full.parts().count() > Path::MAX_PARTS {
1622 return Err(BoundsExceeded.into());
1623 }
1624 let claim = prefix_claim(&full)?;
1627
1628 let ingress = self.stats.ingress(&full);
1630
1631 let announcing = Announcing {
1636 hop: self.hop,
1637 shared: self.shared.clone(),
1638 requested: full.clone(),
1639 prefixes: vec![(full.clone(), claim)],
1640 scope: self.scope.allowed.clone(),
1641 local: true,
1642 peer: self.peer,
1643 stats: self.stats.clone(),
1644 };
1645 let info = broadcast::Info {
1646 pool: self.pool.clone(),
1647 cache_duration: self.cache_duration,
1648 path: full,
1649 };
1650 let source = info.produce().with_stats(ingress.clone());
1651 let entry = announcing.announce(
1652 Route::default(),
1653 Serving {
1654 server: None,
1655 source: Some(source.consume()),
1656 advertised: false,
1657 },
1658 )?;
1659 Ok(source.with_announcer(Announcer {
1660 entry,
1661 ingress,
1662 _keepalive: self.tasks.keepalive(),
1663 }))
1664 }
1665
1666 pub fn publish(&self, path: impl AsPath, route: Route) -> Result<broadcast::Producer, Error> {
1668 let broadcast = self.create_broadcast(path)?;
1669 broadcast.announce(route)?;
1670 Ok(broadcast)
1671 }
1672
1673 pub(crate) fn create_source(&self, path: impl AsPath) -> broadcast::Producer {
1679 let path = path.as_path();
1680 let full = self.root.join(&path).to_owned();
1681 let ingress = self.stats.ingress(&full);
1682 broadcast::Info {
1683 pool: self.pool.clone(),
1684 cache_duration: self.cache_duration,
1685 path: full,
1686 }
1687 .produce()
1688 .with_stats(ingress)
1689 }
1690
1691 #[cfg(test)]
1700 pub(crate) fn announce(&self, prefix: impl AsPath, route: Route) -> Result<AnnounceProducer, Error> {
1701 Announcing::new(self, prefix)?.announce(
1702 route,
1703 Serving {
1704 server: None,
1705 source: None,
1706 advertised: true,
1707 },
1708 )
1709 }
1710
1711 pub fn dynamic(&self, prefix: impl AsPath, route: Route) -> Result<Dynamic, Error> {
1733 let announcing = Announcing::new(self, prefix)?;
1734 let serve = kio::Shared::<ServeState>::default();
1735 serve.lock().requests.add_handler();
1736 let announcement = announcing.announce(
1737 route,
1738 Serving {
1739 server: Some(serve.clone()),
1740 source: None,
1741 advertised: true,
1742 },
1743 )?;
1744 Ok(Dynamic {
1745 announcement,
1746 state: serve,
1747 _keepalive: self.tasks.keepalive(),
1748 })
1749 }
1750
1751 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Producer, Error> {
1758 let root = self.root.join(root).to_owned();
1759 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
1760 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
1761 Ok(Producer {
1762 hop: self.hop,
1763 scope,
1764 root,
1765 shared: self.shared.clone(),
1766 pool: self.pool.clone(),
1767 cache_duration: self.cache_duration,
1768 default_max_age: self.default_max_age,
1769 stats: self.stats.clone(),
1770 peer: self.peer,
1771 tasks: self.tasks.clone(),
1772 timers: self.timers.clone(),
1773 })
1774 }
1775
1776 pub fn mount(&self, at: impl AsPath, target: impl AsPath) -> Result<Producer, Error> {
1795 let at = self.root.join(at).to_owned();
1796 let target = self.root.join(target).to_owned();
1797 if [&at, &target]
1798 .into_iter()
1799 .any(|path| path.parts().count() > Path::MAX_PARTS)
1800 {
1801 return Err(BoundsExceeded.into());
1802 }
1803 Pattern::literal(at.as_str())?;
1805 if !self
1806 .scope
1807 .allowed
1808 .covers(&Patterns::from(Pattern::subtree(target.as_str())?))
1809 {
1810 return Err(Error::Unauthorized);
1811 }
1812 let overlaps = |a: &Path, b: &Path| a.has_prefix(b) || b.has_prefix(a);
1816 if overlaps(&at, &target)
1817 || self
1818 .scope
1819 .mounts
1820 .iter()
1821 .any(|mount| overlaps(&at, &mount.at) || overlaps(&target, &mount.at) || overlaps(&at, &mount.target))
1822 {
1823 return Err(Error::Duplicate);
1824 }
1825 let mounts = self
1826 .scope
1827 .mounts
1828 .iter()
1829 .cloned()
1830 .chain([Mount { at, target }])
1831 .collect();
1832 Ok(Producer {
1833 scope: OriginScope {
1834 allowed: self.scope.allowed.clone(),
1835 mounts,
1836 },
1837 ..self.clone()
1838 })
1839 }
1840
1841 pub fn consume(&self) -> Consumer {
1846 Consumer::from_producer(self, stats::Session::default())
1849 }
1850
1851 pub fn root(&self) -> &Path<'_> {
1853 &self.root
1854 }
1855
1856 pub fn allowed(&self) -> Patterns {
1858 self.scope.relative(&self.root)
1859 }
1860
1861 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
1863 self.root.join(path)
1864 }
1865}
1866
1867struct Announcing {
1871 hop: Hop,
1872 shared: kio::Shared<OriginState>,
1873 requested: PathOwned,
1875 prefixes: Vec<(PathOwned, Pattern)>,
1879 scope: Patterns,
1881 local: bool,
1882 peer: bool,
1884 stats: stats::Session,
1885}
1886
1887impl Announcing {
1888 fn new(producer: &Producer, prefix: impl AsPath) -> Result<Self, Error> {
1890 let requested = producer.root.join(prefix.as_path()).to_owned();
1891 if requested.parts().count() > Path::MAX_PARTS {
1892 return Err(BoundsExceeded.into());
1893 }
1894 let claim = prefix_claim(&requested)?;
1895 if !producer.scope.allowed.overlaps(&claim) || !producer.scope.publishes(&requested) {
1896 return Err(Error::Unauthorized);
1897 }
1898 Ok(Self {
1899 hop: producer.hop,
1900 shared: producer.shared.clone(),
1901 requested: requested.clone(),
1902 prefixes: vec![(requested, claim)],
1903 scope: producer.scope.allowed.clone(),
1904 local: false,
1905 peer: producer.peer,
1906 stats: producer.stats.clone(),
1907 })
1908 }
1909
1910 fn announce(&self, route: Route, serving: Serving) -> Result<AnnounceProducer, Error> {
1911 debug_assert!(
1912 !route.hops.contains(&self.hop),
1913 "announce called with a looping hop chain",
1914 );
1915
1916 let via = route.via;
1917
1918 let mut shared = self.shared.lock();
1919 if shared.closed {
1920 return Err(Error::Closed);
1921 }
1922
1923 let mut entries = Vec::with_capacity(self.prefixes.len());
1924 for (prefix, claim) in &self.prefixes {
1925 shared.reannounced(prefix, &route.hops);
1926 let id = shared.next_route;
1927 shared.next_route += 1;
1928 let stale = shared.withdrawn_through(prefix, &route.hops);
1929 shared.routes.insert(RouteEntry {
1930 id,
1931 prefix: prefix.clone(),
1932 scope: self.scope.clone(),
1933 hops: route.hops.clone(),
1934 cost: route.cost,
1935 via,
1936 local: self.local,
1937 peer: self.peer,
1938 server: serving.server.clone(),
1939 source: serving.source.clone(),
1940 advertised: serving.advertised,
1941 stale,
1942 claim: claim.clone(),
1943 });
1944 shared.sync_route(prefix, claim);
1945 entries.push((prefix.clone(), id));
1946 }
1947 drop(shared);
1948
1949 let guard = serving
1951 .advertised
1952 .then(|| self.stats.ingress(&self.requested).announce());
1953
1954 Ok(AnnounceProducer {
1955 shared: self.shared.clone(),
1956 entries,
1957 guard,
1958 })
1959 }
1960}
1961
1962struct Serving {
1964 server: Option<kio::Shared<ServeState>>,
1965 source: Option<broadcast::Consumer>,
1966 advertised: bool,
1967}
1968
1969pub(crate) struct Announcer {
1976 entry: AnnounceProducer,
1977 ingress: stats::Scope,
1979 _keepalive: Keepalive,
1983}
1984
1985impl Announcer {
1986 pub(crate) fn announce(&mut self, route: Route) -> Result<(), Error> {
1988 self.entry.update(route)?;
1989 if self.entry.guard.is_none() {
1990 self.entry.guard = Some(self.ingress.announce());
1991 }
1992 Ok(())
1993 }
1994
1995 pub(crate) fn withdraw(&mut self) {
1997 self.entry.withdraw();
1998 self.entry.guard = None;
1999 }
2000}
2001
2002#[must_use = "dropping an announcement retracts the route"]
2009pub(crate) struct AnnounceProducer {
2010 shared: kio::Shared<OriginState>,
2011 entries: Vec<(PathOwned, u64)>,
2014 guard: Option<stats::Announce>,
2016}
2017
2018impl AnnounceProducer {
2019 pub fn update(&self, route: Route) -> Result<(), Error> {
2027 let mut shared = self.shared.lock();
2028 if shared.closed {
2029 return Err(Error::Closed);
2030 }
2031 for (prefix, id) in &self.entries {
2032 shared.reannounced(prefix, &route.hops);
2033 let stale = shared.withdrawn_through(prefix, &route.hops);
2034 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2036 return Err(Error::Closed);
2037 };
2038 if entry.hops.iter().next() != route.hops.iter().next()
2041 && let Some(server) = &entry.server
2042 {
2043 drop(std::mem::take(&mut server.lock().served));
2044 }
2045 entry.hops = route.hops.clone();
2046 entry.stale = stale;
2047 entry.cost = route.cost;
2048 entry.via = route.via;
2049 entry.advertised = true;
2050 let claim = entry.claim.clone();
2051 shared.sync_route(prefix, &claim);
2052 shared.prune_withdrawn(prefix);
2053 }
2054 Ok(())
2055 }
2056
2057 fn withdraw(&self) {
2062 let mut shared = self.shared.lock();
2063 for (prefix, id) in &self.entries {
2064 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2065 continue;
2066 };
2067 if !entry.advertised {
2068 continue;
2069 }
2070 entry.advertised = false;
2071 let claim = entry.claim.clone();
2072 shared.sync_route(prefix, &claim);
2073 }
2074 }
2075
2076 fn retract(&self, withdrawn: bool) {
2079 let mut shared = self.shared.lock();
2080 for (prefix, id) in &self.entries {
2081 let Some(entry) = shared.routes.remove(prefix, *id) else {
2082 continue;
2083 };
2084 if let Some(server) = &entry.server {
2087 let mut server = server.lock();
2088 server.closed = true;
2089 for producer in server.requests.drain_all() {
2090 if let Ok(mut request) = producer.write() {
2091 request.resolved.get_or_insert(Err(Error::Unroutable));
2092 }
2093 }
2094 }
2095 if withdrawn
2100 && let Some(&peer) = entry.hops.iter().last()
2101 && peer != Hop::UNKNOWN
2102 && !shared
2103 .routes
2104 .at(prefix)
2105 .any(|other| other.live() && other.hops.iter().last() == Some(&peer))
2106 {
2107 shared.withdrawn.entry(prefix.clone()).or_default().insert(peer);
2108 shared.restale(prefix);
2109 }
2110 shared.sync_route(&entry.prefix, &entry.claim);
2111 shared.prune_withdrawn(&entry.prefix);
2112 }
2113 }
2114}
2115
2116impl Drop for AnnounceProducer {
2117 fn drop(&mut self) {
2118 self.retract(false);
2119 }
2120}
2121
2122#[must_use = "poll the driver or the origin makes no progress"]
2137pub struct Driver {
2138 state: DriverState,
2139 timers: Clock,
2141 pool: cache::Pool,
2144}
2145
2146struct DriverState {
2148 set: TaskSet,
2150 shared: kio::Shared<OriginState>,
2153 done: bool,
2155}
2156
2157impl Driver {
2158 pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2165 self.timers.advance(now);
2166 self.timers.register_driver(waiter);
2167 let result = self.state.poll(waiter);
2168 let gc = self.pool.gc(now);
2169 if result.is_ready() {
2170 return Err(Error::Closed);
2171 }
2172 Ok(self.timers.timeout().into_iter().chain(gc).min())
2173 }
2174}
2175
2176impl crate::time::Driver for Driver {
2177 fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2178 self.poll(now, waiter)
2179 }
2180}
2181
2182impl DriverState {
2183 fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
2184 if !self.done {
2188 ready!(self.set.poll(waiter));
2189 self.done = true;
2190 }
2191 Poll::Ready(())
2192 }
2193
2194 fn teardown(&mut self) {
2198 drop(std::mem::replace(&mut self.set, TaskSet::owned()));
2201
2202 let (servers, cursors, fronts) = {
2207 let mut shared = self.shared.lock();
2208 shared.closed = true;
2209 shared.routes.poke_all();
2211 let servers: Vec<_> = shared
2212 .routes
2213 .entries()
2214 .filter_map(|entry| entry.server.clone())
2215 .collect();
2216 let cursors: Vec<_> = shared.cursors.values().map(|cursor| cursor.state.clone()).collect();
2217 let fronts: Vec<_> = shared.fronts.values().map(|front| front.request.clone()).collect();
2218 (servers, cursors, fronts)
2219 };
2220
2221 for producer in fronts {
2224 if let Ok(mut request) = producer.write() {
2225 request.resolved.get_or_insert(Err(Error::Dropped));
2226 }
2227 }
2228 for server in servers {
2232 let mut server = server.lock();
2233 server.closed = true;
2234 for producer in server.requests.drain_all() {
2235 if let Ok(mut request) = producer.write() {
2236 request.resolved.get_or_insert(Err(Error::Dropped));
2237 }
2238 }
2239 }
2240
2241 for state in cursors {
2244 if let Ok(mut state) = state.write() {
2245 state.ended = true;
2246 }
2247 }
2248 }
2249}
2250
2251impl Drop for DriverState {
2252 fn drop(&mut self) {
2253 self.teardown();
2254 }
2255}
2256
2257const TRACK_IDLE_LINGER: Duration = Duration::from_secs(30);
2271
2272struct WarmCopy {
2277 track: track::Producer,
2278 _dynamic: track::Dynamic,
2279 edge: Option<WarmGroup>,
2282}
2283
2284impl Drop for WarmCopy {
2285 fn drop(&mut self) {
2286 let _ = self.track.finish();
2287 }
2288}
2289
2290struct WarmGroup(group::Producer);
2297
2298impl Drop for WarmGroup {
2299 fn drop(&mut self) {
2300 if !self.0.is_finished() {
2301 let _ = self.0.clone().abort(Error::Cancel);
2302 }
2303 }
2304}
2305
2306struct HeldGroup(group::Producer);
2311
2312impl Drop for HeldGroup {
2313 fn drop(&mut self) {
2314 if !self.0.is_finished() {
2315 let _ = self.0.abort_if_last(Error::Cancel);
2316 }
2317 }
2318}
2319
2320fn warm_copy(source: &track::Consumer, head: Option<&WarmGroup>, held: Option<&group::Producer>) -> Option<WarmCopy> {
2328 let info = source.cached_info()?;
2329 let mut track = track::Producer::new(Arc::new(source.broadcast().clone()), source.name(), info);
2330 let head = head.map(|head| &head.0);
2331 let mut groups = source.cached_groups();
2332 if let Some(held) = held
2333 && !groups.iter().any(|(group, _)| group.sequence == held.sequence)
2334 {
2335 groups.push((held.clone(), true));
2336 }
2337 let latest = groups
2339 .iter()
2340 .filter(|(_, visible)| *visible)
2341 .map(|(group, _)| group.sequence)
2342 .max();
2343 let mut edge = None;
2344 let mut first = None;
2345
2346 if let Some(head) = head
2350 && !groups.iter().any(|(group, _)| group.sequence == head.sequence)
2351 {
2352 let is_latest = latest.is_none_or(|latest| head.sequence >= latest);
2353 if head.is_finished() {
2354 let _ = track.adopt_group(head.clone(), true);
2355 first = Some(head.sequence);
2356 if is_latest {
2357 edge = Some(WarmGroup(head.clone()));
2358 }
2359 } else if is_latest {
2360 edge = warm_rebuild(&track, head, None);
2361 first = edge.as_ref().map(|_| head.sequence);
2362 }
2363 }
2364
2365 for (group, visible) in groups {
2366 let finished = group.is_finished();
2367 let whole = group.live_first_frame() == Some(0);
2368 let is_latest = visible && Some(group.sequence) == latest;
2369 let warm = if finished && whole {
2370 let _ = track.adopt_group(group.clone(), visible);
2371 Some(WarmGroup(group))
2372 } else if (finished || is_latest) && (whole || head.is_some_and(|head| head.sequence == group.sequence)) {
2373 warm_rebuild(&track, &group, head)
2379 } else {
2380 None
2382 };
2383 if visible && let Some(warm) = &warm {
2384 first = Some(first.map_or(warm.0.sequence, |first: u64| first.min(warm.0.sequence)));
2385 }
2386 if is_latest {
2387 edge = warm;
2388 }
2389 }
2390 track.start_at(first).ok()?;
2394 let dynamic = track.dynamic();
2395 Some(WarmCopy {
2396 track,
2397 _dynamic: dynamic,
2398 edge,
2399 })
2400}
2401
2402fn warm_rebuild(track: &track::Producer, live: &group::Producer, head: Option<&group::Producer>) -> Option<WarmGroup> {
2405 let mut start = live.live_first_frame()? as u64;
2407 let mut tail = live.consume();
2408 tail.start_at(start);
2409 let mut frames = Vec::new();
2410 if start > 0
2411 && let Some(head) = head.filter(|head| head.sequence == live.sequence)
2412 && let Some(head_start) = head.live_first_frame()
2413 {
2414 let mut head = head.consume();
2415 head.start_at(head_start as u64);
2416 while head.index() < start {
2417 match head.poll_read_frame(&kio::Waiter::noop()) {
2418 Poll::Ready(Ok(Some(frame))) => frames.push(frame),
2419 _ => break,
2420 }
2421 }
2422 match head.index() == start {
2423 true => start = head_start as u64,
2424 false => frames.clear(),
2426 }
2427 }
2428 while let Poll::Ready(Ok(Some(frame))) = tail.poll_read_frame(&kio::Waiter::noop()) {
2429 frames.push(frame);
2430 }
2431 if frames.is_empty() {
2432 return None;
2433 }
2434
2435 let rebuilt = WarmGroup(
2437 track
2438 .create_group(group::Info {
2439 sequence: live.sequence,
2440 })
2441 .ok()?,
2442 );
2443 let mut writer = rebuilt.0.clone();
2444 if start > 0 {
2445 writer.start_at(start).ok()?;
2446 }
2447 for frame in frames {
2448 writer.write_frame(frame.timestamp, frame.payload).ok()?;
2449 }
2450 if live.is_finished() {
2451 writer.finish().ok()?;
2452 }
2453 Some(rebuilt)
2454}
2455
2456struct FrontTask {
2458 shared: kio::Shared<OriginState>,
2460 broadcast: broadcast::Producer,
2462 path: PathOwned,
2464 horizon: Horizon,
2466 watch: Watch,
2468 request: kio::Producer<PendingBroadcast>,
2470 pin: kio::Lock<Pin>,
2472 timers: Clock,
2473}
2474
2475struct TrackIo {
2478 resume: super::resume::Producer,
2479 staged: Option<(u64, track::Consumer)>,
2481 query: Option<(u64, track::Consumer, track::Querying)>,
2483 copy: Option<(u64, track::Consumer)>,
2485 edge: Option<track::Position>,
2490 warm: Option<WarmCopy>,
2493 head: Option<WarmGroup>,
2496 held: Option<HeldGroup>,
2501 dead: Option<track::Consumer>,
2504 used: bool,
2506}
2507
2508impl TrackIo {
2509 fn end(&mut self) {
2512 self.staged = None;
2513 self.query = None;
2514 self.copy = None;
2515 self.warm = None;
2516 self.head = None;
2517 self.held = None;
2518 self.dead = None;
2519 }
2520
2521 fn bury(&mut self) {
2524 let Some((_, copy)) = self.copy.take() else { return };
2525 let after = self.held.as_ref().map(|held| held.0.sequence);
2526 if let Poll::Ready(group) = copy.poll_latest_group(after, &kio::Waiter::noop()) {
2527 self.held = Some(HeldGroup(group));
2528 }
2529 self.dead = Some(copy);
2530 }
2531
2532 fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> {
2537 let Some(held) = self.held.take() else {
2538 return Ok(());
2539 };
2540 if held.0.is_finished() {
2541 return Ok(());
2542 }
2543 let Some(warm) = warm_copy(copy, self.head.as_ref(), Some(&held.0)) else {
2544 return Ok(());
2545 };
2546 self.head = None;
2547 self.resume.park(&warm.track)?;
2548 self.warm = Some(warm);
2549 Ok(())
2550 }
2551}
2552
2553async fn run_front(task: FrontTask) {
2557 let FrontTask {
2558 shared,
2559 broadcast,
2560 path,
2561 horizon,
2562 watch,
2563 request,
2564 pin,
2565 timers,
2566 } = task;
2567
2568 enum Step {
2570 Assigned(Arc<str>, super::resume::Producer),
2571 Resolved(u64, Result<broadcast::Consumer, Error>),
2572 SourceClosed(u64),
2573 Info(Arc<str>, u64, Result<track::Info, Error>),
2574 Ended(Arc<str>, u64, Result<(), Error>),
2575 Held(Arc<str>, group::Producer),
2576 Demand(Arc<str>),
2577 Deadline,
2578 Table,
2579 }
2580
2581 let mut front = Front::new(TRACK_IDLE_LINGER);
2582 let mut sources: HashMap<u64, broadcast::Consumer> = HashMap::new();
2583 let mut next_source = 0u64;
2584 let mut upstream: Option<(u64, kio::Consumer<PendingBroadcast>)> = None;
2586 let mut tracks: HashMap<Arc<str>, TrackIo> = HashMap::new();
2587 let mut deadline = crate::runtime::Deadline::new(&timers);
2588 let mut seen = 0;
2590 let mut events: VecDeque<Event> = VecDeque::new();
2591
2592 let select = |front: &mut Front, sources: &HashMap<u64, broadcast::Consumer>, seen: &mut u64| -> Event {
2595 let table = shared.read();
2596 if table.closed {
2597 return Event::Closed;
2598 }
2599 *seen = watch.seen();
2601 front.retain_routes(|route| table.routes.covers(&path.as_path(), route));
2602 let best = table
2603 .best_route(&path.as_path(), horizon, front.pin(), front.refused_routes())
2604 .map(|entry| Candidate {
2605 route: entry.id,
2606 first: entry.hops.iter().next().copied(),
2607 local: entry.local,
2608 });
2609 let serving_closing = front
2610 .serving()
2611 .and_then(|id| sources.get(&id))
2612 .is_some_and(|source| source.is_closing());
2613 Event::Selected { best, serving_closing }
2614 };
2615
2616 events.push_back(select(&mut front, &sources, &mut seen));
2617
2618 loop {
2619 while let Some(event) = events.pop_front() {
2620 for action in front.step(event) {
2621 match action {
2622 Action::Reselect => events.push_back(select(&mut front, &sources, &mut seen)),
2623 Action::Request { route } => {
2624 let found = {
2626 let table = shared.read();
2627 table
2628 .routes
2629 .covering(&path.as_path())
2630 .find(|entry| entry.id == route && entry.live())
2631 .map(|entry| {
2632 (
2633 Candidate {
2634 route,
2635 first: entry.hops.iter().next().copied(),
2636 local: entry.local,
2637 },
2638 entry.source.clone(),
2639 entry.server.clone(),
2640 )
2641 })
2642 };
2643 let Some((candidate, source, server)) = found else {
2644 events.push_back(Event::Resolved {
2645 route,
2646 result: Err(Refusal {
2647 err: Error::Unroutable,
2648 standing: false,
2649 }),
2650 });
2651 continue;
2652 };
2653 front.identify(candidate);
2654 *pin.lock() = front.pin();
2655 if let Some(source) = source {
2656 let id = next_source;
2657 next_source += 1;
2658 sources.insert(id, source);
2659 events.push_back(Event::Resolved { route, result: Ok(id) });
2660 continue;
2661 }
2662 let Some(server) = server else {
2663 events.push_back(Event::Resolved {
2664 route,
2665 result: Err(Refusal {
2666 err: Error::Unroutable,
2667 standing: true,
2668 }),
2669 });
2670 continue;
2671 };
2672 let mut serve = server.lock();
2673 if serve.closed {
2674 drop(serve);
2677 events.push_back(Event::Resolved {
2678 route,
2679 result: Err(Refusal {
2680 err: Error::Unroutable,
2681 standing: true,
2682 }),
2683 });
2684 continue;
2685 }
2686 if let Some(weak) = serve.served.get(&path) {
2689 drop(serve);
2690 let id = next_source;
2691 next_source += 1;
2692 sources.insert(id, weak.consume());
2693 events.push_back(Event::Resolved { route, result: Ok(id) });
2694 continue;
2695 }
2696 let pending = match serve.requests.join(&path) {
2697 Some(producer) => producer.consume(),
2698 None => {
2699 let producer = kio::Producer::<PendingBroadcast>::default();
2700 let consumer = producer.consume();
2701 match serve.requests.insert(path.clone(), producer) {
2702 Ok(()) => consumer,
2703 Err(_) => {
2706 drop(serve);
2707 events.push_back(Event::Resolved {
2708 route,
2709 result: Err(Refusal {
2710 err: Error::Unroutable,
2711 standing: true,
2712 }),
2713 });
2714 continue;
2715 }
2716 }
2717 }
2718 };
2719 upstream = Some((route, pending));
2720 }
2721 Action::Detach { source } => {
2722 sources.remove(&source);
2723 for io in tracks.values_mut() {
2726 if io.copy.as_ref().is_some_and(|(s, _)| *s == source) {
2727 io.bury();
2728 }
2729 if io.query.as_ref().is_some_and(|(s, ..)| *s == source) {
2730 io.query = None;
2731 }
2732 if io.staged.as_ref().is_some_and(|(s, _)| *s == source) {
2733 io.staged = None;
2734 }
2735 }
2736 }
2737 Action::Resolve => {
2738 if let Ok(mut pending) = request.write() {
2739 pending.resolved.get_or_insert(Ok(broadcast.consume()));
2740 }
2741 }
2742 Action::Query { track: name, source } => {
2743 let Some(io) = tracks.get_mut(&name) else { continue };
2744 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
2745 match sources.get(&source).map(|s| s.track(&name)) {
2746 Some(Ok(copy)) => {
2747 let query = copy.query().into_inner();
2750 io.query = Some((source, copy, query));
2751 }
2752 Some(Err(err)) => events.push_back(Event::TrackInfo {
2753 track: name,
2754 source,
2755 closing,
2756 result: Err(err),
2757 }),
2758 None => {}
2759 }
2760 }
2761 Action::Splice { track: name, source } => {
2762 let Some(io) = tracks.get_mut(&name) else { continue };
2763 let Some((staged, copy)) = io.staged.take() else {
2764 continue;
2765 };
2766 if staged != source {
2767 continue;
2768 }
2769 let outgoing = io.copy.take().map(|(_, copy)| copy).or_else(|| io.dead.take());
2770 let retired = match outgoing {
2771 Some(outgoing) => io.retire(&outgoing),
2772 None => Ok(()),
2773 };
2774 if let Err(err) = retired.and_then(|()| io.resume.takeover(©)) {
2775 let _ = io.resume.abort(err);
2779 tracks.remove(&name);
2780 continue;
2781 }
2782 if let Some(head) = io.warm.take().and_then(|mut warm| warm.edge.take()) {
2783 io.head = Some(head);
2784 }
2785 io.edge = io.resume.resume_position();
2788 io.held = None;
2789 io.copy = Some((source, copy));
2790 }
2791 Action::Park { track: name } => {
2792 let Some(io) = tracks.get_mut(&name) else { continue };
2793 let Some((_, copy)) = io.copy.take() else { continue };
2794 let warm = warm_copy(©, io.head.as_ref(), io.held.as_ref().map(|held| &held.0));
2798 drop(copy);
2799 io.head = None;
2800 io.held = None;
2801 io.dead = None;
2802 let parked = match &warm {
2803 Some(warm) => io.resume.park(&warm.track),
2804 None => io.resume.release(),
2805 };
2806 if parked.is_err() {
2807 tracks.remove(&name);
2808 continue;
2809 }
2810 io.warm = warm;
2811 }
2812 Action::Forget { track: name } => {
2813 if let Some(io) = tracks.get_mut(&name)
2818 && !broadcast.forget_spliced(&name, &io.resume)
2819 {
2820 if !io.used {
2821 io.used = true;
2822 events.push_back(Event::Used { track: name });
2823 }
2824 continue;
2825 }
2826 tracks.remove(&name);
2827 events.push_back(Event::Forgotten { track: name });
2828 }
2829 Action::Finish { track: name } => {
2830 if let Some(io) = tracks.get_mut(&name) {
2831 io.end();
2832 let _ = io.resume.finish();
2833 }
2834 }
2835 Action::Abort { track: name, err } => {
2836 if let Some(io) = tracks.get_mut(&name) {
2837 tracing::debug!(name = %name, %err, "aborting track");
2838 io.end();
2839 let _ = io.resume.abort(err);
2840 }
2841 }
2842 Action::Arm { at } => deadline.set(at),
2843 Action::End { err } => {
2844 if let Ok(mut pending) = request.write() {
2845 pending.resolved.get_or_insert(Err(err.clone()));
2846 }
2847 broadcast.close();
2854 broadcast.release_spliced(err.clone());
2855 for (_, mut io) in tracks.drain() {
2856 let waiting = io.staged.take().map(|(_, copy)| copy);
2860 let waiting = waiting.or_else(|| io.query.take().map(|(_, copy, _)| copy));
2861 if let Some(copy) = waiting
2862 && io.resume.is_used()
2863 {
2864 if io.resume.takeover(©).is_err() {
2865 continue;
2866 }
2867 io.warm = None;
2868 }
2869 if !io.resume.is_used() || !io.resume.is_spliced() || io.warm.is_some() {
2871 let _ = io.resume.abort(err.clone());
2872 }
2873 }
2874 return;
2875 }
2876 }
2877 }
2878 }
2879
2880 let step = kio::wait(|waiter| {
2881 if let Poll::Ready((name, resume)) = broadcast.poll_spliced_assigned(waiter) {
2882 return Poll::Ready(Step::Assigned(name, resume));
2883 }
2884 if let Some((route, pending)) = &upstream
2885 && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved {
2886 Some(result) => Poll::Ready(result.clone()),
2887 None => Poll::Pending,
2888 }) {
2889 return Poll::Ready(Step::Resolved(
2890 *route,
2891 match result {
2892 Ok(resolved) => resolved,
2893 Err(_closed) => Err(Error::Unroutable),
2896 },
2897 ));
2898 }
2899 if let Some(id) = front.serving()
2900 && let Some(source) = sources.get(&id)
2901 && source.poll_closed(waiter).is_ready()
2902 {
2903 return Poll::Ready(Step::SourceClosed(id));
2904 }
2905 for (name, io) in &tracks {
2906 if let Some((source, _, query)) = &io.query
2907 && let Poll::Ready(result) = query.poll(waiter)
2908 {
2909 return Poll::Ready(Step::Info(name.clone(), *source, result));
2910 }
2911 if let Some((source, copy)) = &io.copy
2912 && let Poll::Ready(result) = copy.poll_complete(waiter)
2913 {
2914 return Poll::Ready(Step::Ended(name.clone(), *source, result));
2915 }
2916 if let Some((_, copy)) = &io.copy
2917 && let Poll::Ready(group) =
2918 copy.poll_latest_group(io.held.as_ref().map(|held| held.0.sequence), waiter)
2919 {
2920 return Poll::Ready(Step::Held(name.clone(), group));
2921 }
2922 let edge = match io.used {
2924 true => io.resume.poll_unused(waiter),
2925 false => io.resume.poll_used(waiter),
2926 };
2927 if edge.is_ready() {
2928 return Poll::Ready(Step::Demand(name.clone()));
2929 }
2930 }
2931 if deadline.poll(waiter).is_ready() {
2932 return Poll::Ready(Step::Deadline);
2933 }
2934 watch.poll_changed(waiter, seen).map(|()| Step::Table)
2935 })
2936 .await;
2937
2938 let event = match step {
2939 Step::Assigned(name, resume) => {
2940 tracks.insert(
2941 name.clone(),
2942 TrackIo {
2943 resume,
2944 staged: None,
2945 query: None,
2946 copy: None,
2947 edge: None,
2948 warm: None,
2949 head: None,
2950 held: None,
2951 dead: None,
2952 used: false,
2953 },
2954 );
2955 Event::TrackAssigned {
2956 track: name,
2957 now: timers.now(),
2958 }
2959 }
2960 Step::Resolved(route, result) => {
2961 upstream = None;
2962 match result {
2963 Ok(source) => {
2964 let id = next_source;
2965 next_source += 1;
2966 sources.insert(id, source);
2967 Event::Resolved { route, result: Ok(id) }
2968 }
2969 Err(err) => {
2970 let standing =
2974 !matches!(err, Error::Unroutable) || shared.read().routes.covers(&path.as_path(), route);
2975 Event::Resolved {
2976 route,
2977 result: Err(Refusal { err, standing }),
2978 }
2979 }
2980 }
2981 }
2982 Step::SourceClosed(source) => Event::SourceClosed { source },
2983 Step::Info(name, source, result) => {
2984 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
2985 let Some(io) = tracks.get_mut(&name) else { continue };
2986 let Some((_, copy, _)) = io.query.take() else { continue };
2987 let result = match result {
2990 Ok(info) => match copy.poll_complete(&kio::Waiter::noop()) {
2991 Poll::Ready(Err(err)) => Err(err),
2992 _ => Ok(info),
2993 },
2994 Err(err) => Err(err),
2995 };
2996 if result.is_ok() && io.used {
2999 io.staged = Some((source, copy));
3000 }
3001 Event::TrackInfo {
3002 track: name,
3003 source,
3004 closing,
3005 result,
3006 }
3007 }
3008 Step::Ended(name, source, result) => {
3009 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3010 let Some(io) = tracks.get_mut(&name) else { continue };
3011 io.bury();
3012 let delivered = io.resume.resume_position() != io.edge;
3013 Event::TrackEnded {
3014 track: name,
3015 source,
3016 closing,
3017 result,
3018 delivered,
3019 }
3020 }
3021 Step::Held(name, group) => {
3022 if let Some(io) = tracks.get_mut(&name) {
3023 io.held = Some(HeldGroup(group));
3024 }
3025 continue;
3026 }
3027 Step::Demand(name) => {
3028 let Some(io) = tracks.get_mut(&name) else { continue };
3029 io.used = io.resume.is_used();
3030 if !io.used {
3031 io.query = None;
3034 io.staged = None;
3035 }
3036 match io.used {
3037 true => Event::Used { track: name },
3038 false => Event::Unused {
3039 track: name,
3040 now: timers.now(),
3041 },
3042 }
3043 }
3044 Step::Deadline => {
3045 deadline.set(None);
3048 Event::Deadline { now: timers.now() }
3049 }
3050 Step::Table => select(&mut front, &sources, &mut seen),
3051 };
3052 events.push_back(event);
3053 }
3054}
3055
3056#[derive(Default)]
3061struct RouteTable {
3062 root: RouteNode,
3063}
3064
3065#[derive(Default)]
3068struct RouteNode {
3069 entries: Vec<RouteEntry>,
3071 cursors: Vec<ConsumerId>,
3073 cursors_below: usize,
3076 watches: Vec<(u64, kio::Producer<Watched>)>,
3079 watches_below: usize,
3082 children: HashMap<String, RouteNode>,
3083}
3084
3085#[derive(Default)]
3088struct Watched {
3089 generation: u64,
3090}
3091
3092struct Watch {
3098 shared: kio::Shared<OriginState>,
3099 path: PathOwned,
3100 id: u64,
3101 signal: kio::Consumer<Watched>,
3102}
3103
3104impl Watch {
3105 fn seen(&self) -> u64 {
3109 self.signal.read().generation
3110 }
3111
3112 fn poll_changed(&self, waiter: &kio::Waiter, seen: u64) -> Poll<()> {
3114 self.signal
3115 .poll(waiter, |watched| match watched.generation != seen {
3116 true => Poll::Ready(()),
3117 false => Poll::Pending,
3118 })
3119 .map(|_| ())
3120 }
3121}
3122
3123impl Drop for Watch {
3124 fn drop(&mut self) {
3125 self.shared.lock().routes.remove_watch(&self.path, self.id);
3126 }
3127}
3128
3129#[derive(Clone, Copy)]
3131struct Below {
3132 cursors: usize,
3133 watches: usize,
3134}
3135
3136impl Below {
3137 const NONE: Self = Self { cursors: 0, watches: 0 };
3138 const CURSOR: Self = Self { cursors: 1, watches: 0 };
3139 const WATCH: Self = Self { cursors: 0, watches: 1 };
3140}
3141
3142impl RouteNode {
3143 fn is_empty(&self) -> bool {
3145 self.entries.is_empty() && self.cursors.is_empty() && self.watches.is_empty() && self.children.is_empty()
3146 }
3147
3148 fn find<'a>(&self, mut parts: impl Iterator<Item = &'a str>) -> Option<&Self> {
3150 match parts.next() {
3151 None => Some(self),
3152 Some(part) => self.children.get(part)?.find(parts),
3153 }
3154 }
3155
3156 fn reach<'a>(&mut self, mut parts: impl Iterator<Item = &'a str>, below: Below) -> &mut Self {
3159 self.cursors_below += below.cursors;
3160 self.watches_below += below.watches;
3161 match parts.next() {
3162 None => self,
3163 Some(part) => self.children.entry(part.to_string()).or_default().reach(parts, below),
3164 }
3165 }
3166
3167 fn edit<'a, R>(
3171 &mut self,
3172 mut parts: impl Iterator<Item = &'a str>,
3173 below: Below,
3174 f: impl FnOnce(&mut Self) -> R,
3175 ) -> Option<R> {
3176 let result = match parts.next() {
3177 None => f(self),
3178 Some(part) => {
3179 let child = self.children.get_mut(part)?;
3180 let result = child.edit(parts, below, f)?;
3181 if child.is_empty() {
3182 self.children.remove(part);
3183 }
3184 result
3185 }
3186 };
3187 self.cursors_below -= below.cursors;
3188 self.watches_below -= below.watches;
3189 Some(result)
3190 }
3191
3192 fn poke(&self) {
3194 for (_, watch) in &self.watches {
3195 if let Ok(mut watched) = watch.write() {
3196 watched.generation += 1;
3197 }
3198 }
3199 }
3200
3201 fn poke_below(&self) {
3204 if self.watches_below == 0 {
3205 return;
3206 }
3207 self.poke();
3208 for child in self.children.values() {
3209 child.poke_below();
3210 }
3211 }
3212
3213 fn walk<'a>(&'a self, visit: &mut impl FnMut(&'a Self)) {
3215 visit(self);
3216 for child in self.children.values() {
3217 child.walk(visit);
3218 }
3219 }
3220
3221 fn collect_cursors(&self, out: &mut Vec<ConsumerId>) {
3223 if self.cursors_below == 0 {
3224 return;
3225 }
3226 out.extend(&self.cursors);
3227 for child in self.children.values() {
3228 child.collect_cursors(out);
3229 }
3230 }
3231}
3232
3233impl RouteTable {
3234 fn split(&self, path: &Path) -> (Vec<&RouteNode>, Option<&RouteNode>) {
3237 let mut above = Vec::new();
3238 let mut node = &self.root;
3239 for part in path.parts() {
3240 above.push(node);
3241 match node.children.get(part) {
3242 Some(child) => node = child,
3243 None => return (above, None),
3244 }
3245 }
3246 (above, Some(node))
3247 }
3248
3249 fn covering(&self, path: &Path) -> impl Iterator<Item = &RouteEntry> {
3251 let (above, at) = self.split(path);
3252 above.into_iter().chain(at).flat_map(|node| node.entries.iter())
3253 }
3254
3255 fn covers(&self, path: &Path, id: u64) -> bool {
3257 self.covering(path).any(|entry| entry.id == id && entry.live())
3258 }
3259
3260 fn at(&self, prefix: &Path) -> impl Iterator<Item = &RouteEntry> {
3262 self.root
3263 .find(prefix.parts())
3264 .into_iter()
3265 .flat_map(|node| node.entries.iter())
3266 }
3267
3268 fn at_mut(&mut self, prefix: &Path) -> impl Iterator<Item = &mut RouteEntry> {
3270 let mut node = Some(&mut self.root);
3271 for part in prefix.parts() {
3272 node = node.and_then(|node| node.children.get_mut(part));
3273 }
3274 node.into_iter().flat_map(|node| node.entries.iter_mut())
3275 }
3276
3277 fn entries(&self) -> impl Iterator<Item = &RouteEntry> {
3279 let mut nodes = Vec::new();
3280 self.root.walk(&mut |node| nodes.push(node));
3281 nodes.into_iter().flat_map(|node| node.entries.iter())
3282 }
3283
3284 fn insert(&mut self, entry: RouteEntry) {
3286 let node = self.root.reach(entry.prefix.parts(), Below::NONE);
3287 node.entries.push(entry);
3288 }
3289
3290 fn entry_mut(&mut self, prefix: &Path, id: u64) -> Option<&mut RouteEntry> {
3292 let mut node = &mut self.root;
3293 for part in prefix.parts() {
3294 node = node.children.get_mut(part)?;
3295 }
3296 node.entries.iter_mut().find(|entry| entry.id == id)
3297 }
3298
3299 fn remove(&mut self, prefix: &Path, id: u64) -> Option<RouteEntry> {
3301 self.root
3302 .edit(prefix.parts(), Below::NONE, |node| {
3303 let index = node.entries.iter().position(|entry| entry.id == id)?;
3304 Some(node.entries.swap_remove(index))
3305 })
3306 .flatten()
3307 }
3308
3309 fn add_cursor(&mut self, head: &Path, id: ConsumerId) {
3311 self.root.reach(head.parts(), Below::CURSOR).cursors.push(id);
3312 }
3313
3314 fn remove_cursor(&mut self, head: &Path, id: ConsumerId) {
3317 self.root.edit(head.parts(), Below::CURSOR, |node| {
3318 node.cursors.retain(|cursor| *cursor != id)
3319 });
3320 }
3321
3322 fn add_watch(&mut self, path: &Path, id: u64) -> kio::Consumer<Watched> {
3324 let producer = kio::Producer::<Watched>::default();
3325 let consumer = producer.consume();
3326 self.root.reach(path.parts(), Below::WATCH).watches.push((id, producer));
3327 consumer
3328 }
3329
3330 fn remove_watch(&mut self, path: &Path, id: u64) {
3333 self.root.edit(path.parts(), Below::WATCH, |node| {
3334 node.watches.retain(|(watch, _)| *watch != id)
3335 });
3336 }
3337
3338 fn poke_below(&self, prefix: &Path) {
3340 if let (_, Some(node)) = self.split(prefix) {
3341 node.poke_below();
3342 }
3343 }
3344
3345 fn poke_all(&self) {
3347 self.root.walk(&mut |node| node.poke());
3348 }
3349
3350 fn cursors_touching(&self, prefix: &Path) -> Vec<ConsumerId> {
3354 let (above, at) = self.split(prefix);
3355 let mut cursors: Vec<ConsumerId> = above.iter().flat_map(|node| node.cursors.iter().copied()).collect();
3356 if let Some(node) = at {
3357 node.collect_cursors(&mut cursors);
3358 }
3359 cursors.sort_unstable();
3361 cursors.dedup();
3362 cursors
3363 }
3364}
3365
3366struct OriginState {
3373 update_hold: Duration,
3376
3377 routes: RouteTable,
3380 next_route: u64,
3381 next_watch: u64,
3382
3383 cursors: HashMap<ConsumerId, TableCursor>,
3387
3388 fronts: WeakCache<FrontKey, RemoteFront>,
3394
3395 withdrawn: HashMap<PathOwned, HashSet<Hop>>,
3398
3399 closed: bool,
3402}
3403
3404impl OriginState {
3405 fn new(update_hold: Duration) -> Self {
3406 Self {
3407 update_hold,
3408 routes: RouteTable::default(),
3409 next_route: 0,
3410 next_watch: 0,
3411 cursors: HashMap::new(),
3412 fronts: WeakCache::default(),
3413 withdrawn: HashMap::new(),
3414 closed: false,
3415 }
3416 }
3417
3418 fn withdrawn_through(&self, prefix: &Path, hops: &Hops) -> bool {
3420 if self.withdrawn.is_empty() {
3421 return false;
3422 }
3423 self.withdrawn
3424 .get(prefix)
3425 .is_some_and(|peers| hops.iter().any(|hop| peers.contains(hop)))
3426 }
3427
3428 fn restale(&mut self, prefix: &Path) {
3430 let peers = self.withdrawn.get(prefix);
3431 let mut changed = None;
3432 for entry in self.routes.at_mut(prefix) {
3433 let stale = peers.is_some_and(|peers| entry.hops.iter().any(|hop| peers.contains(hop)));
3434 if entry.stale != stale {
3435 entry.stale = stale;
3436 changed = Some(entry.claim.clone());
3437 }
3438 }
3439 if let Some(claim) = changed {
3440 self.sync_route(prefix, &claim);
3441 }
3442 }
3443
3444 fn reannounced(&mut self, prefix: &PathOwned, hops: &Hops) {
3448 if let Some(sender) = hops.iter().last()
3449 && self.withdrawn.get_mut(prefix).is_some_and(|peers| peers.remove(sender))
3450 {
3451 self.restale(prefix);
3452 self.prune_withdrawn(prefix);
3453 }
3454 }
3455
3456 fn prune_withdrawn(&mut self, prefix: &PathOwned) {
3458 if self.withdrawn.is_empty() {
3459 return;
3460 }
3461 let Some(peers) = self.withdrawn.get_mut(prefix) else {
3462 return;
3463 };
3464 if peers.len() <= 1 {
3465 peers.retain(|peer| self.routes.at(prefix).any(|entry| entry.hops.contains(peer)));
3467 } else {
3468 let mut unreferenced = peers.clone();
3469 for entry in self.routes.at(prefix) {
3470 if unreferenced.is_empty() {
3471 break;
3472 }
3473 for hop in entry.hops.iter() {
3474 unreferenced.remove(hop);
3475 }
3476 }
3477 peers.retain(|peer| !unreferenced.contains(peer));
3478 }
3479 if peers.is_empty() {
3480 self.withdrawn.remove(prefix);
3481 }
3482 }
3483
3484 fn sync_route(&mut self, prefix: &Path, claim: &Pattern) {
3489 let routes = &self.routes;
3491 let mut candidates = None;
3492 for id in routes.cursors_touching(prefix) {
3493 let Some(cursor) = self.cursors.get_mut(&id) else {
3494 continue;
3495 };
3496 if let Some(presented) = cursor.presented(prefix, claim) {
3497 if presented.is_empty() {
3498 Self::sync_cursor(routes, cursor, &presented);
3500 } else {
3501 let candidates = candidates.get_or_insert_with(|| {
3504 let mut entries: Vec<_> = routes
3505 .at(prefix)
3506 .map(|entry| (route_order(prefix, entry), entry))
3507 .collect();
3508 entries.sort_unstable_by_key(|(order, _)| *order);
3509 entries
3510 });
3511 let best = candidates
3512 .iter()
3513 .map(|(_, entry)| *entry)
3514 .find(|entry| cursor.visible(entry));
3515 cursor.update(&presented, best);
3516 }
3517 }
3518 }
3519 routes.poke_below(prefix);
3521 }
3522
3523 fn watch(&mut self, shared: &kio::Shared<OriginState>, path: &Path) -> Watch {
3525 let id = self.next_watch;
3526 self.next_watch += 1;
3527 let signal = self.routes.add_watch(path, id);
3528 Watch {
3529 shared: shared.clone(),
3530 path: path.to_owned(),
3531 id,
3532 signal,
3533 }
3534 }
3535
3536 fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned) {
3539 let candidates: Vec<&RouteEntry> = match presented.is_empty() {
3545 true => routes
3546 .covering(&cursor.root)
3547 .filter(|entry| cursor.visible(entry))
3548 .collect(),
3549 false => {
3550 let absolute = cursor.root.join(presented);
3551 routes.at(&absolute).filter(|entry| cursor.visible(entry)).collect()
3552 }
3553 };
3554 let most = candidates.iter().map(|entry| entry.prefix.len()).max();
3555 let best = most.and_then(|most| {
3556 candidates
3557 .into_iter()
3558 .filter(|entry| entry.prefix.len() == most)
3559 .min_by_key(|entry| route_order(&entry.prefix, entry))
3560 });
3561
3562 cursor.update(presented, best);
3563 }
3564
3565 fn register_cursor(&mut self, id: ConsumerId, mut cursor: TableCursor) {
3567 let mut presented: BTreeSet<PathOwned> = BTreeSet::new();
3570 for head in &cursor.heads {
3571 let (above, at) = self.routes.split(head);
3572 let mut nodes = above;
3573 if let Some(node) = at {
3574 node.walk(&mut |node| nodes.push(node));
3575 }
3576 for entry in nodes.into_iter().flat_map(|node| node.entries.iter()) {
3577 if let Some(p) = cursor.presented(&entry.prefix, &entry.claim) {
3578 presented.insert(p);
3579 }
3580 }
3581 }
3582 for p in &presented {
3583 Self::sync_cursor(&self.routes, &mut cursor, p);
3584 }
3585 for head in &cursor.heads {
3586 self.routes.add_cursor(head, id);
3587 }
3588 self.cursors.insert(id, cursor);
3589 }
3590
3591 fn best_route(&self, path: &Path, horizon: Horizon, pin: Pin, refused: &HashSet<u64>) -> Option<&RouteEntry> {
3606 let (above, at) = self.routes.split(path);
3610 let mut best = None;
3611 for node in above.into_iter().chain(at) {
3612 let mut candidates = node
3613 .entries
3614 .iter()
3615 .filter(|entry| entry.live())
3616 .filter(|entry| entry.scope.matches(path.as_str()))
3617 .filter(|entry| horizon.admits(entry))
3618 .filter(|entry| entry.qualifies(pin))
3619 .filter(|entry| !refused.contains(&entry.id))
3620 .peekable();
3621 if candidates.peek().is_some() {
3622 best = candidates
3623 .filter(|entry| entry.serves(path))
3624 .min_by_key(|entry| route_order(&entry.prefix, entry));
3625 }
3626 }
3627 best
3628 }
3629}
3630
3631#[derive(Default)]
3638struct PendingBroadcast {
3639 resolved: Option<Result<broadcast::Consumer, Error>>,
3640}
3641
3642#[must_use = "dropping an origin::Dynamic retracts the route"]
3657pub struct Dynamic {
3658 announcement: AnnounceProducer,
3660 state: kio::Shared<ServeState>,
3661 _keepalive: Keepalive,
3665}
3666
3667impl Dynamic {
3668 pub(crate) fn withdrawn(self) {
3671 self.announcement.retract(true);
3672 }
3673
3674 pub fn update(&self, route: Route) -> Result<(), Error> {
3684 self.announcement.update(route)
3685 }
3686
3687 pub fn poll_requested_broadcast(&self, waiter: &kio::Waiter) -> Poll<Result<Request, Error>> {
3692 let mut state = ready!(self.state.poll(waiter, |state| {
3693 if state.closed || state.requests.has_queued() {
3694 Poll::Ready(())
3695 } else {
3696 Poll::Pending
3697 }
3698 }));
3699
3700 if state.closed {
3702 return Poll::Ready(Err(Error::Closed));
3703 }
3704
3705 let path = state.requests.pop().expect("predicate guaranteed a request");
3706 let producer = state.requests.get(&path).expect("popped key must be pending").clone();
3712 Poll::Ready(Ok(Request {
3713 path,
3714 producer,
3715 home: self.state.clone(),
3716 }))
3717 }
3718
3719 pub async fn requested_broadcast(&self) -> Result<Request, Error> {
3725 kio::wait(|waiter| self.poll_requested_broadcast(waiter)).await
3726 }
3727}
3728
3729impl ServeState {
3730 fn resolve(
3741 shared: &kio::Shared<Self>,
3742 path: &PathOwned,
3743 producer: &kio::Producer<PendingBroadcast>,
3744 result: Result<broadcast::Consumer, Error>,
3745 ) {
3746 let mut state = shared.lock();
3747 if state.closed {
3748 return;
3749 }
3750 let resolved = match result {
3751 Ok(broadcast) => {
3752 let existing = state.served.insert(path.clone(), broadcast.weak());
3756 Ok(existing.map(|weak| weak.consume()).unwrap_or(broadcast))
3757 }
3758 Err(err) => Err(err),
3759 };
3760 state.requests.remove_if(path, |p| p.same_channel(producer));
3761 if let Ok(mut pending) = producer.write() {
3762 pending.resolved.get_or_insert(resolved);
3763 drop(state);
3764 }
3765 }
3766
3767 fn forget(shared: &kio::Shared<Self>, path: &PathOwned, producer: &kio::Producer<PendingBroadcast>) {
3769 shared.lock().requests.remove_if(path, |p| p.same_channel(producer));
3770 }
3771}
3772
3773pub struct Request {
3780 path: PathOwned,
3782
3783 producer: kio::Producer<PendingBroadcast>,
3786
3787 home: kio::Shared<ServeState>,
3790}
3791
3792impl Request {
3793 pub fn path(&self) -> &Path<'_> {
3795 &self.path
3796 }
3797
3798 pub fn accept(self, broadcast: impl Consume<broadcast::Consumer>) {
3804 let broadcast = broadcast.consume();
3805 ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast));
3806 }
3808
3809 pub fn reject(self, err: Error) {
3811 ServeState::resolve(&self.home, &self.path, &self.producer, Err(err));
3812 }
3813}
3814
3815impl Drop for Request {
3816 fn drop(&mut self) {
3817 ServeState::forget(&self.home, &self.path, &self.producer);
3825 }
3826}
3827
3828pub struct Requesting {
3835 inner: RequestState,
3836 path: PathOwned,
3840 stats: stats::Scope,
3843}
3844
3845enum RequestState {
3846 Failed(Error),
3849 Pending(kio::Consumer<PendingBroadcast>),
3851}
3852
3853impl Requesting {
3854 fn failed(error: Error) -> Self {
3855 Self::new(RequestState::Failed(error))
3856 }
3857
3858 fn queued(consumer: kio::Consumer<PendingBroadcast>) -> Self {
3859 Self::new(RequestState::Pending(consumer))
3860 }
3861
3862 pub fn is_queued(&self) -> bool {
3871 matches!(self.inner, RequestState::Pending(_))
3872 }
3873
3874 fn new(inner: RequestState) -> Self {
3875 Self {
3876 inner,
3877 path: PathOwned::default(),
3878 stats: stats::Scope::default(),
3879 }
3880 }
3881
3882 fn with_path(mut self, path: PathOwned) -> Self {
3883 self.path = path;
3884 self
3885 }
3886
3887 fn with_stats(mut self, scope: stats::Scope) -> Self {
3889 self.stats = scope;
3890 self
3891 }
3892
3893 fn hand_out(&self, broadcast: broadcast::Consumer) -> broadcast::Consumer {
3895 broadcast.with_path(self.path.clone()).with_stats(self.stats.clone())
3896 }
3897
3898 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<broadcast::Consumer, Error>> {
3900 match &self.inner {
3901 RequestState::Failed(error) => Poll::Ready(Err(error.clone())),
3902 RequestState::Pending(consumer) => Poll::Ready(
3903 match ready!(consumer.poll(waiter, |state| match &state.resolved {
3904 Some(result) => Poll::Ready(result.clone()),
3905 None => Poll::Pending,
3906 })) {
3907 Ok(result) => result.map(|broadcast| self.hand_out(broadcast)),
3908 Err(_closed) => Err(Error::Unroutable),
3910 },
3911 ),
3912 }
3913 }
3914}
3915
3916impl kio::Pollable for Requesting {
3917 type Output = Result<broadcast::Consumer, Error>;
3918
3919 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3920 self.poll_ok(waiter)
3921 }
3922}
3923
3924pub trait Consume<T> {
3932 fn consume(&self) -> T;
3934}
3935
3936impl<T, U: Consume<T>> Consume<T> for &U {
3937 fn consume(&self) -> T {
3938 (**self).consume()
3939 }
3940}
3941
3942impl Consume<Consumer> for Producer {
3943 fn consume(&self) -> Consumer {
3944 Consumer::from_producer(self, stats::Session::default())
3948 }
3949}
3950
3951impl Consume<Consumer> for Consumer {
3952 fn consume(&self) -> Consumer {
3953 self.clone()
3954 }
3955}
3956
3957impl Consume<broadcast::Consumer> for broadcast::Producer {
3958 fn consume(&self) -> broadcast::Consumer {
3959 self.consume()
3961 }
3962}
3963
3964impl Consume<broadcast::Consumer> for broadcast::Consumer {
3965 fn consume(&self) -> broadcast::Consumer {
3966 self.clone()
3967 }
3968}
3969
3970impl Consume<track::Consumer> for track::Producer {
3971 fn consume(&self) -> track::Consumer {
3972 self.consume()
3973 }
3974}
3975
3976impl Consume<track::Consumer> for track::Consumer {
3977 fn consume(&self) -> track::Consumer {
3978 self.clone()
3979 }
3980}
3981
3982#[derive(Clone)]
3988pub struct Consumer {
3989 hop: Hop,
3991 scope: OriginScope,
3992
3993 root: PathOwned,
3995
3996 shared: kio::Shared<OriginState>,
3999
4000 stats: stats::Session,
4004
4005 horizon: Horizon,
4010
4011 hidden: Hidden,
4013
4014 pool: cache::Pool,
4017 cache_duration: Duration,
4018
4019 tasks: TasksWeak,
4023
4024 timers: Clock,
4027}
4028
4029impl Consumer {
4030 fn from_producer(producer: &Producer, stats: stats::Session) -> Self {
4031 Self {
4032 hop: producer.hop,
4033 scope: producer.scope.clone(),
4034 root: producer.root.clone(),
4035 shared: producer.shared.clone(),
4036 stats,
4037 horizon: Horizon::default(),
4038 hidden: Hidden::default(),
4039 pool: producer.pool.clone(),
4040 cache_duration: producer.cache_duration,
4041 tasks: producer.tasks.downgrade(),
4042 timers: producer.timers.clone(),
4043 }
4044 }
4045
4046 pub fn hop(&self) -> Hop {
4048 self.hop
4049 }
4050
4051 pub(crate) fn excluding(mut self, peer: Hop) -> Self {
4059 self.horizon.exclude = Some(peer);
4060 self
4061 }
4062
4063 pub fn local(mut self) -> Self {
4070 self.horizon.local = true;
4071 self
4072 }
4073
4074 pub fn with_hidden(mut self, hidden: bool) -> Self {
4079 self.hidden.include = hidden;
4080 self
4081 }
4082
4083 pub(crate) fn beyond(mut self, outer: &Consumer) -> Self {
4086 self.hidden.beyond = Some(
4087 outer
4088 .hidden
4089 .from
4090 .clone()
4091 .map(|head| vec![head])
4092 .unwrap_or_else(|| interest_prefixes(&outer.scope.allowed)),
4093 );
4094 self
4095 }
4096
4097 pub(crate) fn discovery(mut self, hidden: bool) -> Self {
4099 self.hidden.include = hidden;
4100 self.hidden.from = Some(self.root.clone());
4101 self
4102 }
4103
4104 pub(crate) fn includes_hidden(&self) -> bool {
4106 self.hidden.include
4107 }
4108
4109 pub fn with_stats(mut self, session: stats::Session) -> Self {
4113 self.stats = session;
4114 self
4115 }
4116
4117 fn untagged(&self) -> Self {
4121 Self {
4122 stats: stats::Session::default(),
4123 ..self.clone()
4124 }
4125 }
4126
4127 pub(crate) fn empty(&self) -> Self {
4132 Self {
4133 scope: OriginScope::empty(),
4134 ..self.clone()
4135 }
4136 }
4137
4138 pub fn announced(&self) -> AnnounceConsumer {
4147 let state = kio::Producer::<OriginConsumerState>::default();
4148 let cursor = |root: PathOwned,
4149 allowed: Patterns,
4150 hidden: Hidden,
4151 mount: Option<&Mount>,
4152 under: PathOwned,
4153 holes: Vec<PathOwned>| TableCursor {
4154 root,
4155 heads: interest_prefixes(&allowed),
4156 allowed,
4157 horizon: self.horizon,
4158 hidden,
4159 state: state.clone(),
4160 under,
4161 holes,
4162 named: mount.map(|mount| (mount.clone(), self.scope.allowed.clone())),
4163 current: HashMap::new(),
4164 };
4165
4166 if let Some(mount) = self.scope.mount(&self.root) {
4169 let Some(root) = mount.resolve(&self.root) else {
4171 return AnnounceConsumer::new(
4172 self.root.clone(),
4173 Vec::new(),
4174 state,
4175 self.stats.clone(),
4176 &self.shared,
4177 self.timers.clone(),
4178 );
4179 };
4180 let cursors = vec![cursor(
4181 root,
4182 mount.translate(&self.scope.allowed),
4183 self.hidden.translate(mount),
4184 Some(mount),
4185 PathOwned::default(),
4186 Vec::new(),
4187 )];
4188 return AnnounceConsumer::new(
4189 self.root.clone(),
4190 cursors,
4191 state,
4192 self.stats.clone(),
4193 &self.shared,
4194 self.timers.clone(),
4195 );
4196 }
4197
4198 let heads = interest_prefixes(&self.scope.allowed);
4201 let mut holes = Vec::new();
4202 let mut cursors = Vec::new();
4203 for mount in self.scope.mounts.iter() {
4204 let Some(under) = mount.at.strip_prefix(&self.root) else {
4205 continue;
4206 };
4207 holes.push(mount.at.clone());
4208 let allowed = mount.translate(&self.scope.allowed);
4209 if allowed.is_empty()
4212 || !(self.hidden.include
4213 || !hides(
4214 self.hidden.from.as_ref().map(std::slice::from_ref).unwrap_or(&heads),
4215 &mount.at,
4216 )) {
4217 continue;
4218 }
4219 cursors.push(cursor(
4220 mount.target.clone(),
4221 allowed,
4222 self.hidden.translate(mount),
4223 Some(mount),
4224 under.to_owned(),
4225 Vec::new(),
4226 ));
4227 }
4228 cursors.insert(
4229 0,
4230 cursor(
4231 self.root.clone(),
4232 self.scope.allowed.clone(),
4233 self.hidden.clone(),
4234 None,
4235 PathOwned::default(),
4236 holes,
4237 ),
4238 );
4239 AnnounceConsumer::new(
4240 self.root.clone(),
4241 cursors,
4242 state,
4243 self.stats.clone(),
4244 &self.shared,
4245 self.timers.clone(),
4246 )
4247 }
4248
4249 pub fn consume(&self) -> Self {
4251 self.clone()
4252 }
4253
4254 #[cfg(test)]
4257 pub(crate) fn get_broadcast(&self, path: impl AsPath) -> Option<broadcast::Consumer> {
4258 let full = self.root.join(path).to_owned();
4259 if !self.scope.permits(&full) {
4260 return None;
4261 }
4262 let full = self.scope.resolve(&full)?;
4263 let table = self.shared.lock();
4264 table
4265 .routes
4266 .at(&full)
4267 .filter(|entry| entry.local)
4268 .min_by_key(|entry| route_order(&entry.prefix, entry))
4269 .and_then(|entry| entry.source.clone())
4270 }
4271
4272 pub async fn routed(&self, path: impl AsPath) -> Option<Route> {
4282 let path = path.as_path();
4283
4284 let consumer = match Pattern::subtree(path.as_str()) {
4289 Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?,
4290 Err(InvalidPattern::TooManySegments) => self.clone(),
4291 Err(_) => return None,
4292 };
4293
4294 if !consumer.allowed().matches(path.as_str()) {
4298 return None;
4299 }
4300
4301 let mut announced = consumer.untagged().with_hidden(true).announced();
4305 loop {
4306 let update = announced.next().await?;
4307 if update.kind.is_active() && path.has_prefix(&update.prefix) {
4308 return Some(update.route);
4309 }
4310 }
4311 }
4312
4313 pub async fn routed_broadcast(&self, path: impl AsPath) -> Result<broadcast::Consumer, Error> {
4327 let path = path.as_path();
4328
4329 if !self.allowed().matches(path.as_str()) {
4332 return Err(Error::Unauthorized);
4333 }
4334 loop {
4335 let (watch, seen) = {
4344 let mut table = self.shared.lock();
4345 if table.closed {
4346 return Err(Error::Closed);
4347 }
4348 let named = self.root.join(&path);
4349 let resolved = self.scope.resolve(&named).ok_or(BoundsExceeded)?;
4350 let watch = table.watch(&self.shared, &resolved);
4351 let seen = watch.seen();
4352 (watch, seen)
4353 };
4354 match self.request_broadcast(&path).await {
4355 Ok(broadcast) => return Ok(broadcast),
4356 Err(Error::Unroutable) => {
4357 kio::wait(|waiter| watch.poll_changed(waiter, seen)).await;
4358 }
4359 Err(Error::Dropped) if self.shared.lock().closed => return Err(Error::Closed),
4362 Err(err) => return Err(err),
4363 }
4364 }
4365 }
4366
4367 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Consumer, Error> {
4374 let root = self.root.join(root).to_owned();
4375 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
4376 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
4377 Ok(Consumer {
4378 scope,
4379 root,
4380 ..self.clone()
4381 })
4382 }
4383
4384 pub fn request_broadcast(&self, path: impl AsPath) -> kio::Pending<Requesting> {
4407 let path = path.as_path();
4408
4409 let named = self.root.join(&path).to_owned();
4413 let scope = self.stats.egress(&named);
4414 let requested = path.to_owned();
4418
4419 if !self.scope.permits(&named) {
4421 return kio::Pending::new(Requesting::failed(Error::Unauthorized));
4422 }
4423
4424 let Some(absolute) = self.scope.resolve(&named).map(|path| path.to_owned()) else {
4427 return kio::Pending::new(Requesting::failed(BoundsExceeded.into()));
4428 };
4429
4430 let mut state = self.shared.lock();
4431
4432 if state.closed {
4434 return kio::Pending::new(Requesting::failed(Error::Closed));
4435 }
4436
4437 let horizon = self.horizon.effective(&state.routes, &absolute.as_path());
4438
4439 if state
4443 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4444 .is_none()
4445 {
4446 return kio::Pending::new(Requesting::failed(Error::Unroutable));
4447 }
4448
4449 let key = (absolute.clone(), horizon);
4457 if let Some(front) = state.fronts.get(&key) {
4458 let pin = *front.pin.lock();
4459 let current = state
4460 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4461 .is_some_and(|entry| entry.qualifies(pin));
4462 if current {
4463 let pending = Requesting::queued(front.request.consume())
4464 .with_path(requested)
4465 .with_stats(scope);
4466 return kio::Pending::new(pending);
4467 }
4468 state.fronts.remove(&key);
4469 }
4470
4471 let broadcast = broadcast::Producer::new_spliced(broadcast::Info {
4476 pool: self.pool.clone(),
4477 cache_duration: self.cache_duration,
4478 path: absolute.clone(),
4479 });
4480 let request = kio::Producer::<PendingBroadcast>::default();
4481 let consumer = request.consume();
4482 let watch = state.watch(&self.shared, &absolute);
4483 let pin = kio::Lock::new(Pin::Any);
4484 state.fronts.insert(
4485 key,
4486 RemoteFront {
4487 request: request.clone(),
4488 broadcast: broadcast.consume().weak(),
4489 pin: pin.clone(),
4490 },
4491 );
4492 drop(state);
4495 self.tasks.push(run_front(FrontTask {
4496 shared: self.shared.clone(),
4497 broadcast,
4498 path: absolute,
4499 horizon,
4500 watch,
4501 request,
4502 pin,
4503 timers: self.timers.clone(),
4504 }));
4505 kio::Pending::new(Requesting::queued(consumer).with_path(requested).with_stats(scope))
4506 }
4507
4508 pub fn root(&self) -> &Path<'_> {
4510 &self.root
4511 }
4512
4513 pub fn allowed(&self) -> Patterns {
4515 self.scope.relative(&self.root)
4516 }
4517
4518 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
4520 self.root.join(path)
4521 }
4522}
4523
4524pub struct AnnounceConsumer {
4529 ids: Vec<ConsumerId>,
4532 shared: kio::Shared<OriginState>,
4533 root: PathOwned,
4534
4535 state: kio::Producer<OriginConsumerState>,
4538
4539 stats: stats::Session,
4542
4543 guards: HashMap<PathOwned, stats::Announce>,
4547
4548 park: kio::Park,
4551
4552 update_hold: Duration,
4555 held: HashMap<PathOwned, Held>,
4556 retracted: HashMap<PathOwned, Quarantine>,
4559 anchoring: VecDeque<(Instant, PathOwned)>,
4560 expiring: VecDeque<(Instant, PathOwned)>,
4561 timers: Clock,
4562 hold: crate::runtime::Deadline<Clock>,
4563}
4564
4565impl AnnounceConsumer {
4566 fn new(
4567 root: PathOwned,
4568 cursors: Vec<TableCursor>,
4569 state: kio::Producer<OriginConsumerState>,
4570 stats: stats::Session,
4571 shared: &kio::Shared<OriginState>,
4572 timers: Clock,
4573 ) -> Self {
4574 let mut ids = Vec::with_capacity(cursors.len());
4575 let update_hold;
4576 {
4577 let mut table = shared.lock();
4578 update_hold = table.update_hold;
4579 if table.closed {
4580 if let Ok(mut state) = state.write() {
4582 state.ended = true;
4583 }
4584 } else {
4585 for cursor in cursors {
4586 let id = ConsumerId::new();
4587 table.register_cursor(id, cursor);
4588 ids.push(id);
4589 }
4590 }
4591 }
4592
4593 Self {
4594 ids,
4595 shared: shared.clone(),
4596 root,
4597 state,
4598 stats,
4599 guards: HashMap::new(),
4600 park: kio::Park::default(),
4601 update_hold,
4602 held: HashMap::new(),
4603 retracted: HashMap::new(),
4604 anchoring: VecDeque::new(),
4605 expiring: VecDeque::new(),
4606 hold: crate::runtime::Deadline::new(&timers),
4607 timers,
4608 }
4609 }
4610
4611 fn quarantine(&mut self, update: &AnnounceUpdate, now: Option<Instant>) {
4614 if update.kind.is_active() {
4615 return;
4616 }
4617 let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else {
4618 return;
4619 };
4620 let publisher = update
4622 .route
4623 .hops
4624 .iter()
4625 .next()
4626 .copied()
4627 .filter(|hop| *hop != Hop::UNKNOWN);
4628 let quarantine = Quarantine {
4630 held: Held::Anchoring(now),
4631 publisher,
4632 rank: (update.route.cost, update.route.hops.len()),
4633 };
4634 let replaced = self.retracted.insert(update.prefix.clone(), quarantine);
4635 if !replaced.is_some_and(|q| q.held == Held::Anchoring(now)) {
4638 self.anchoring.push_back((now, update.prefix.clone()));
4639 }
4640 }
4641
4642 fn settle(&mut self, now: Option<Instant>) -> Option<Instant> {
4645 let now = now?;
4646 while let Some((seen, prefix)) = self.anchoring.front() {
4649 if *seen >= now {
4650 break;
4651 }
4652 if let Some(quarantine) = self.retracted.get_mut(prefix)
4653 && quarantine.held == Held::Anchoring(*seen)
4654 {
4655 quarantine.held = Held::Until(now + RETRACT_QUARANTINE);
4656 self.expiring.push_back((now + RETRACT_QUARANTINE, prefix.clone()));
4657 }
4658 self.anchoring.pop_front();
4659 }
4660 while let Some((until, prefix)) = self.expiring.front() {
4661 if *until > now {
4662 break;
4663 }
4664 if self
4665 .retracted
4666 .get(prefix)
4667 .is_some_and(|q| q.held == Held::Until(*until))
4668 {
4669 self.retracted.remove(prefix);
4670 }
4671 self.expiring.pop_front();
4672 }
4673 let anchoring = self.anchoring.front().map(|(seen, _)| *seen + Duration::from_nanos(1));
4674 earliest(anchoring, self.expiring.front().map(|(until, _)| *until))
4675 }
4676
4677 fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate {
4679 let absolute = self.root.join(&update.prefix).to_owned();
4680 if update.kind.is_active() {
4681 let scope = self.stats.egress(&absolute);
4682 self.guards
4683 .entry(update.prefix.clone())
4684 .or_insert_with(|| scope.announce());
4685 } else {
4686 self.guards.remove(&update.prefix);
4687 }
4688 update
4689 }
4690
4691 pub async fn next(&mut self) -> Option<AnnounceUpdate> {
4699 kio::wait(|waiter| self.poll_next(waiter)).await
4700 }
4701
4702 pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<AnnounceUpdate>> {
4708 loop {
4709 let now = self.timers.try_now();
4710 let settling = self.settle(now);
4711 let hold = self.update_hold;
4712 let held = &mut self.held;
4713 let retracted = &self.retracted;
4714 let mut ready = None;
4715 let mut wake = None;
4716 let update = match self.state.poll(waiter, |state| {
4717 if state.pending.is_empty() {
4718 return match state.ended {
4719 true => Poll::Ready(()),
4720 false => Poll::Pending,
4721 };
4722 }
4723 match state.scan(held, retracted, now, hold) {
4724 Ok(prefix) => {
4725 ready = Some(prefix);
4726 Poll::Ready(())
4727 }
4728 Err(at) => {
4729 wake = at;
4730 Poll::Pending
4731 }
4732 }
4733 }) {
4734 Poll::Ready(Ok(mut state)) => match ready {
4735 Some(prefix) => state.take_prefix(prefix),
4736 None => {
4737 state.close();
4740 None
4741 }
4742 },
4743 Poll::Ready(Err(_)) => None,
4745 Poll::Pending => {
4746 self.hold.set(earliest(wake, settling));
4747 ready!(self.hold.poll(waiter));
4748 continue;
4749 }
4750 };
4751 let Some(update) = update else {
4752 return Poll::Ready(None);
4753 };
4754 self.held.remove(&update.prefix);
4755 self.quarantine(&update, now);
4756 return Poll::Ready(Some(self.hand_out(update)));
4757 }
4758 }
4759
4760 pub fn try_next(&mut self) -> Option<AnnounceUpdate> {
4765 let now = self.timers.try_now();
4766 let settling = self.settle(now);
4767 let mut state = self.state.write().ok()?;
4768 let prefix = match state.scan(&mut self.held, &self.retracted, now, self.update_hold) {
4769 Ok(prefix) => prefix,
4770 Err(wake) => {
4771 let wake = earliest(wake, settling);
4772 drop(state);
4775 self.hold.set(wake);
4776 return None;
4777 }
4778 };
4779 self.held.remove(&prefix);
4780 let update = state.take_prefix(prefix)?;
4781 drop(state);
4782 self.quarantine(&update, now);
4783 Some(self.hand_out(update))
4784 }
4785
4786 pub fn is_closed(&self) -> bool {
4788 let state = self.state.read();
4789 state.is_closed() || state.ended
4790 }
4791
4792 pub fn root(&self) -> &Path<'_> {
4794 &self.root
4795 }
4796
4797 pub fn absolute(&self, prefix: impl AsPath) -> Path<'_> {
4799 self.root.join(prefix)
4800 }
4801}
4802
4803impl futures::Stream for AnnounceConsumer {
4804 type Item = AnnounceUpdate;
4805
4806 fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Option<Self::Item>> {
4807 let this = self.get_mut();
4808 let waiter = this.park.hold(cx).clone();
4809 this.poll_next(&waiter)
4810 }
4811}
4812
4813impl Drop for AnnounceConsumer {
4814 fn drop(&mut self) {
4815 let mut shared = self.shared.lock();
4816 for id in &self.ids {
4817 if let Some(cursor) = shared.cursors.remove(id) {
4818 for head in &cursor.heads {
4819 shared.routes.remove_cursor(head, *id);
4820 }
4821 }
4822 }
4823 }
4824}
4825
4826#[cfg(test)]
4827use futures::FutureExt;
4828
4829#[cfg(test)]
4830#[allow(missing_docs)] impl AnnounceConsumer {
4832 pub fn assert_next_active(&mut self, expected: impl AsPath) -> Route {
4834 let expected = expected.as_path();
4835 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4836 assert_eq!(update.prefix, expected, "wrong prefix");
4837 assert!(update.kind.is_active(), "should be an active route");
4838 update.route
4839 }
4840
4841 pub fn assert_try_next_active(&mut self, expected: impl AsPath) -> Route {
4843 let expected = expected.as_path();
4844 let update = self.try_next().expect("no next");
4845 assert_eq!(update.prefix, expected, "wrong prefix");
4846 assert!(update.kind.is_active(), "should be an active route");
4847 update.route
4848 }
4849
4850 pub fn assert_next_ended(&mut self, expected: impl AsPath) {
4852 let expected = expected.as_path();
4853 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4854 assert_eq!(update.prefix, expected, "wrong prefix");
4855 assert_eq!(update.kind, AnnounceKind::Retracted, "should be a retraction");
4856 }
4857
4858 pub fn assert_next_wait(&mut self) {
4859 if let Some(res) = self.next().now_or_never() {
4860 panic!("next should block: got {:?}", res.map(|u| u.prefix));
4861 }
4862 }
4863}
4864
4865#[cfg(test)]
4869pub(crate) trait ProduceTest {
4870 fn produce(self) -> Producer;
4871}
4872
4873#[cfg(test)]
4874impl ProduceTest for Config {
4875 fn produce(self) -> Producer {
4876 let (producer, driver) = Producer::new(self);
4877 if tokio::runtime::Handle::try_current().is_ok() {
4878 tokio::spawn(crate::time::run(driver));
4879 } else {
4880 std::mem::forget(driver);
4883 }
4884 producer
4885 }
4886}
4887
4888#[cfg(test)]
4889impl ProduceTest for Hop {
4890 fn produce(self) -> Producer {
4891 Config::new(self).produce()
4892 }
4893}
4894
4895#[cfg(test)]
4896mod tests {
4897 use super::*;
4898 use futures::FutureExt;
4899
4900 fn origin(id: u64) -> Hop {
4901 Hop::new(id).unwrap()
4902 }
4903
4904 fn hops(ids: &[u64]) -> Hops {
4905 let mut list = Hops::new();
4906 for &id in ids {
4907 list.push(if id == 0 { Hop::UNKNOWN } else { origin(id) }).unwrap();
4908 }
4909 list
4910 }
4911
4912 fn scopes(prefixes: &[&str]) -> Patterns {
4914 prefixes
4915 .iter()
4916 .map(|prefix| Pattern::subtree(prefix).unwrap())
4917 .collect()
4918 }
4919
4920 #[test]
4921 fn default_config_mints_a_real_hop() {
4922 let config = Config::default();
4923 assert_ne!(config.hop, Hop::UNKNOWN);
4924 let (producer, _driver) = Producer::new(config.clone());
4925 assert_eq!(producer.hop(), config.hop);
4926 assert_eq!(producer.consume().hop(), config.hop);
4927 }
4928
4929 #[test]
4930 fn random_hops_fit_legacy_lite_clients() {
4931 for _ in 0..32 {
4932 assert!(Hop::random().id() < 1u64 << 53);
4933 }
4934 }
4935
4936 async fn settle(mut check: impl FnMut() -> bool) {
4939 for _ in 0..100 {
4940 if check() {
4941 return;
4942 }
4943 tokio::task::yield_now().await;
4944 }
4945 panic!("condition never settled");
4946 }
4947
4948 async fn queued(server: &Dynamic) -> Request {
4950 let mut request = None;
4951 settle(|| match server.poll_requested_broadcast(&kio::Waiter::noop()) {
4952 Poll::Ready(Ok(popped)) => {
4953 request = Some(popped);
4954 true
4955 }
4956 _ => false,
4957 })
4958 .await;
4959 request.unwrap()
4960 }
4961
4962 async fn next_group(subscription: &mut crate::track::Subscriber) -> Result<Option<crate::group::Consumer>, Error> {
4964 let mut next = None;
4965 settle(|| match subscription.poll_recv_group(&kio::Waiter::noop()) {
4966 Poll::Ready(result) => {
4967 next = Some(result);
4968 true
4969 }
4970 Poll::Pending => false,
4971 })
4972 .await;
4973 next.unwrap()
4974 }
4975
4976 #[tokio::test]
4977 async fn announce_and_retract() {
4978 let producer = origin(1).produce();
4979 let consumer = producer.consume();
4980 let mut announced = consumer.announced();
4981 announced.assert_next_wait();
4982
4983 let announcement = producer.announce("room/alice", Route::default()).unwrap();
4984 let route = announced.assert_next_active("room/alice");
4985 assert!(route.hops.is_empty());
4986 assert_eq!(route.cost, Cost::default());
4987 announced.assert_next_wait();
4988
4989 drop(announcement);
4990 announced.assert_next_ended("room/alice");
4991 announced.assert_next_wait();
4992 }
4993
4994 #[tokio::test]
4997 async fn hidden_routes_need_an_opt_in() {
4998 let producer = origin(1).produce();
4999 let consumer = producer.consume();
5000 let _visible = producer.announce("room/alice", Route::default()).unwrap();
5001 let _stats = producer.announce(".stats/node", Route::default()).unwrap();
5002 let _nested = producer.announce("room/.internal", Route::default()).unwrap();
5003 let _suffix = producer.announce("room/catalog.pro", Route::default()).unwrap();
5005
5006 let mut announced = consumer.announced();
5007 announced.assert_next_active("room/alice");
5008 announced.assert_next_active("room/catalog.pro");
5009 announced.assert_next_wait();
5010
5011 let mut announced = consumer.clone().with_hidden(true).announced();
5012 announced.assert_next_active(".stats/node");
5013 announced.assert_next_active("room/.internal");
5014 announced.assert_next_active("room/alice");
5015 announced.assert_next_active("room/catalog.pro");
5016 announced.assert_next_wait();
5017
5018 let mut announced = consumer
5020 .scope(".stats", &Patterns::from(Pattern::all()))
5021 .unwrap()
5022 .announced();
5023 announced.assert_next_active("node");
5024 announced.assert_next_wait();
5025 let mut announced = consumer.scope("", &scopes(&["room/.internal"])).unwrap().announced();
5026 announced.assert_next_active("room/.internal");
5027 announced.assert_next_wait();
5028
5029 let mut announced = consumer.clone().with_hidden(true).beyond(&consumer).announced();
5032 announced.assert_next_active(".stats/node");
5033 announced.assert_next_active("room/.internal");
5034 announced.assert_next_wait();
5035 let room = consumer.scope("", &scopes(&["room"])).unwrap().beyond(&consumer);
5036 let mut announced = room.announced();
5037 announced.assert_next_wait();
5038 let stats = consumer.scope("", &scopes(&[".stats"])).unwrap().beyond(&consumer);
5039 let mut announced = stats.announced();
5040 announced.assert_next_active(".stats/node");
5041 announced.assert_next_wait();
5042 }
5043
5044 #[tokio::test]
5046 async fn hidden_broadcast_resolves_by_path() {
5047 let producer = origin(1).produce();
5048 let consumer = producer.consume();
5049 let broadcast = producer.create_broadcast(".stats/node").unwrap();
5050 broadcast.announce(Route::default()).unwrap();
5051
5052 consumer.announced().assert_next_wait();
5053 let resolved = consumer.request_broadcast(".stats/node").await.expect("resolves");
5054 assert_eq!(resolved.info().path.as_str(), ".stats/node");
5055 }
5056
5057 fn mounted(producer: &Producer, pid: &str, patterns: &[&str]) -> Consumer {
5059 let patterns: Patterns = patterns.iter().map(|pattern| pattern.parse().unwrap()).collect();
5060 producer
5061 .mount(format!("{pid}/.svc"), format!(".svc/{pid}"))
5062 .unwrap()
5063 .scope(pid, &patterns)
5064 .unwrap()
5065 .consume()
5066 }
5067
5068 #[tokio::test]
5071 async fn mount_resolves_on_the_target_front() {
5072 let producer = origin(1).produce();
5073 let server = producer.dynamic(".svc", Route::default()).unwrap();
5074 let project = mounted(&producer, "p1", &["**"]);
5075
5076 let through = project.request_broadcast(".svc/foo");
5077 let direct = producer.consume().request_broadcast(".svc/p1/foo");
5078
5079 let request = queued(&server).await;
5080 assert_eq!(request.path().as_str(), ".svc/p1/foo");
5081 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5082 let source = broadcast::Info::new().produce();
5083 request.accept(&source);
5084
5085 let through = through.await.expect("resolves");
5086 let direct = direct.await.expect("resolves");
5087 assert!(through.is_clone(&direct));
5088 assert_eq!(through.info().path.as_str(), ".svc/foo");
5090 }
5091
5092 #[tokio::test]
5095 async fn mount_presents_target_routes_under_the_mount() {
5096 let producer = origin(1).produce();
5097 let _claim = producer.announce(".svc", Route::default()).unwrap();
5098 let foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5099 let _other = producer.publish(".svc/p2/bar", Route::default()).unwrap();
5100 let _cam = producer.publish("p1/cam", Route::default()).unwrap();
5101 let _shadowed = producer.publish("p1/.svc/forged", Route::default()).unwrap();
5102 let project = mounted(&producer, "p1", &["**"]);
5103
5104 let mut announced = project.announced();
5106 announced.assert_next_active("cam");
5107 announced.assert_next_wait();
5108
5109 let mut announced = project.clone().with_hidden(true).announced();
5110 announced.assert_next_active(".svc");
5111 announced.assert_next_active(".svc/foo");
5112 announced.assert_next_active("cam");
5113 announced.assert_next_wait();
5114
5115 let mut inside = project
5117 .scope(".svc", &Patterns::from(Pattern::all()))
5118 .unwrap()
5119 .announced();
5120 inside.assert_next_active("");
5121 inside.assert_next_active("foo");
5122 inside.assert_next_wait();
5123
5124 drop(foo);
5125 announced.assert_next_ended(".svc/foo");
5126 inside.assert_next_ended("foo");
5127 announced.assert_next_wait();
5128
5129 let err = project.request_broadcast(".svc/forged").await.err().unwrap();
5131 assert!(matches!(err, Error::NotFound | Error::Unroutable), "{err:?}");
5132 }
5133
5134 #[tokio::test]
5137 async fn mount_authorizes_the_named_path() {
5138 let producer = origin(1).produce();
5139 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5140 let _bar = producer.publish(".svc/p1/bar", Route::default()).unwrap();
5141
5142 let granted = mounted(&producer, "p1", &["foo", ".svc/foo"]);
5143 granted.request_broadcast(".svc/foo").await.expect("granted");
5144 let refused = granted
5145 .request_broadcast(".svc/bar")
5146 .now_or_never()
5147 .expect("refused at once");
5148 assert!(matches!(refused, Err(Error::Unauthorized)));
5149 let mut announced = granted.clone().with_hidden(true).announced();
5150 announced.assert_next_active(".svc/foo");
5151 announced.assert_next_wait();
5152
5153 let narrower = mounted(&producer, "p1", &["foo"]);
5154 let refused = narrower
5155 .request_broadcast(".svc/foo")
5156 .now_or_never()
5157 .expect("refused at once");
5158 assert!(matches!(refused, Err(Error::Unauthorized)));
5159 narrower.clone().with_hidden(true).announced().assert_next_wait();
5160
5161 let other = mounted(&producer, "p2", &["**"]);
5163 other.clone().with_hidden(true).announced().assert_next_wait();
5164 let err = other.request_broadcast(".svc/foo").await.err().unwrap();
5165 assert!(matches!(err, Error::Unroutable));
5166 }
5167
5168 #[tokio::test]
5171 async fn mount_captures_the_named_path() {
5172 let producer = origin(1).produce();
5173 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5174 let project = mounted(&producer, "p1", &["**"]).with_hidden(true);
5175
5176 let update = project.announced().try_next().expect("foo");
5177 assert_eq!(update.prefix.as_str(), ".svc/foo");
5178 assert_eq!(update.captures, Some(vec![".svc/foo".parse::<Pattern>().unwrap()]));
5179
5180 let inside = project.scope(".svc", &Patterns::from(Pattern::all())).unwrap();
5181 let update = inside.announced().try_next().expect("foo");
5182 assert_eq!(update.prefix.as_str(), "foo");
5183 assert_eq!(update.captures, Some(vec!["foo".parse::<Pattern>().unwrap()]));
5184 }
5185
5186 #[tokio::test]
5189 async fn mount_keeps_a_max_depth_target() {
5190 let producer = origin(1).produce();
5191 let deep = vec!["d"; Path::MAX_PARTS].join("/");
5192 let _leaf = producer.publish(deep.as_str(), Route::default()).unwrap();
5193 let project = producer
5194 .mount("p1/.svc", deep.as_str())
5195 .unwrap()
5196 .scope("p1", &Patterns::from(Pattern::all()))
5197 .unwrap()
5198 .consume()
5199 .with_hidden(true);
5200
5201 let mut announced = project.announced();
5202 announced.assert_next_active(".svc");
5203 announced.assert_next_wait();
5204
5205 let err = project.request_broadcast(".svc/x").await.err().unwrap();
5207 assert!(matches!(err, Error::BoundsExceeded(_)), "{err:?}");
5208 let inside = project.scope(".svc/x", &Patterns::from(Pattern::all())).unwrap();
5209 inside.announced().assert_next_wait();
5210 }
5211
5212 #[tokio::test]
5215 async fn mount_bounds_the_named_path() {
5216 let producer = origin(1).produce();
5217 let _near = producer.publish("t/x", Route::default()).unwrap();
5218 let deep = format!("t/{}", vec!["d"; Path::MAX_PARTS - 1].join("/"));
5219 let _deep = producer.publish(deep.as_str(), Route::default()).unwrap();
5220 let project = producer
5221 .mount("p1/a/b", "t")
5222 .unwrap()
5223 .scope("p1", &Patterns::from(Pattern::all()))
5224 .unwrap()
5225 .consume();
5226
5227 let mut announced = project.announced();
5228 announced.assert_next_active("a/b/x");
5229 announced.assert_next_wait();
5230
5231 let over = vec!["d"; Path::MAX_PARTS + 1].join("/");
5232 assert!(matches!(
5233 producer.mount(over.as_str(), "t"),
5234 Err(Error::BoundsExceeded(_))
5235 ));
5236 }
5237
5238 #[tokio::test]
5240 async fn mount_is_read_only() {
5241 let producer = origin(1).produce();
5242 let project = producer
5243 .mount("p1/.svc", ".svc/p1")
5244 .unwrap()
5245 .scope("p1", &Patterns::from(Pattern::all()))
5246 .unwrap();
5247 assert!(matches!(project.create_broadcast(".svc/foo"), Err(Error::Unauthorized)));
5248 assert!(matches!(
5249 project.dynamic(".svc", Route::default()),
5250 Err(Error::Unauthorized)
5251 ));
5252 assert!(matches!(
5253 project.dynamic(".svc/foo", Route::default()),
5254 Err(Error::Unauthorized)
5255 ));
5256 project.publish("cam", Route::default()).unwrap();
5257 }
5258
5259 #[test]
5261 fn mount_never_widens_a_scope() {
5262 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5263 let project = producer.scope("", &scopes(&["p1"])).unwrap();
5264 assert!(matches!(project.mount("p1/.svc", ".svc/p1"), Err(Error::Unauthorized)));
5265
5266 let mounted = producer.mount("p1/.svc", ".svc/p1").unwrap();
5267 assert!(matches!(mounted.mount("p1/.svc/x", ".other"), Err(Error::Duplicate)));
5268 assert!(matches!(mounted.mount("p1", ".other"), Err(Error::Duplicate)));
5269 assert!(matches!(mounted.mount("p2", "p1/.svc/x"), Err(Error::Duplicate)));
5271 assert!(matches!(mounted.mount("p2", "p1"), Err(Error::Duplicate)));
5272 assert!(matches!(mounted.mount(".svc/p1/x", ".other"), Err(Error::Duplicate)));
5274 assert!(matches!(mounted.mount(".svc", ".other"), Err(Error::Duplicate)));
5275 mounted.mount("p1/.other", ".other/p1").unwrap();
5276 mounted.mount("p2/.svc", ".svc/p1").unwrap();
5278 for (at, target) in [("a", "a/b"), ("a/b", "a"), ("a", "a")] {
5280 assert!(
5281 matches!(producer.mount(at, target), Err(Error::Duplicate)),
5282 "{at} -> {target}"
5283 );
5284 }
5285 }
5286
5287 #[test]
5289 fn mount_refuses_a_wildcard_mount_point() {
5290 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5291 for at in ["*", "p1/*", "p1/**", "p1/a*"] {
5292 assert!(
5293 matches!(producer.mount(at, ".svc/p1"), Err(Error::InvalidPath(_))),
5294 "{at}"
5295 );
5296 }
5297 }
5298
5299 #[tokio::test]
5302 async fn mount_egress_counts_under_the_named_path() {
5303 let registry = stats::Registry::new(stats::Config::new());
5304 let producer = origin(1).produce();
5305 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5306 let project = mounted(&producer, "p1", &["**"])
5307 .with_stats(registry.tier(stats::Tier::default()).session("p1"))
5308 .with_hidden(true);
5309
5310 let mut announced = project.announced();
5311 announced.assert_next_active(".svc/foo");
5312 project.request_broadcast(".svc/foo").await.expect("resolves");
5313
5314 let mut report = stats::Report::default();
5315 registry.report(&mut report);
5316 let paths: Vec<_> = report
5317 .traffic
5318 .iter()
5319 .map(|entry| entry.path.as_str().to_string())
5320 .collect();
5321 assert_eq!(paths, ["p1/.svc/foo"]);
5322 }
5323
5324 #[tokio::test]
5326 async fn hidden_route_announced_later_stays_hidden() {
5327 let producer = origin(1).produce();
5328 let consumer = producer.consume();
5329 let mut announced = consumer.announced();
5330 let mut opted = consumer.clone().with_hidden(true).announced();
5331
5332 let hidden = producer.announce(".stats/node", Route::default()).unwrap();
5333 announced.assert_next_wait();
5334 opted.assert_next_active(".stats/node");
5335
5336 drop(hidden);
5337 announced.assert_next_wait();
5338 opted.assert_next_ended(".stats/node");
5339 }
5340
5341 #[tokio::test]
5342 async fn broadcast_announces_its_own_path() {
5343 let producer = origin(1).produce();
5344 let consumer = producer.consume();
5345 let mut announced = consumer.announced();
5346 let mut peer = consumer.clone().excluding(Hop::UNKNOWN).announced();
5347
5348 let broadcast = producer.create_broadcast("room/alice").unwrap();
5350 announced.assert_next_wait();
5351 peer.assert_next_wait();
5352
5353 broadcast.announce(Route::default().with_cost(3)).unwrap();
5354 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(3));
5355 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(3));
5356
5357 broadcast.announce(Route::default().with_cost(1)).unwrap();
5359 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(1));
5360 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(1));
5361
5362 broadcast.unannounce();
5364 announced.assert_next_ended("room/alice");
5365 peer.assert_next_ended("room/alice");
5366 broadcast.unannounce();
5367 announced.assert_next_wait();
5368 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5369 assert!(matches!(err, Error::Unroutable));
5370
5371 broadcast.announce(Route::default()).unwrap();
5373 announced.assert_next_active("room/alice");
5374 peer.assert_next_active("room/alice");
5375 broadcast.close();
5376 announced.assert_next_ended("room/alice");
5377 peer.assert_next_ended("room/alice");
5378 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5379 announced.assert_next_wait();
5380 }
5381
5382 #[tokio::test]
5383 async fn broadcast_announcement_retracts_with_the_last_producer() {
5384 let producer = origin(1).produce();
5385 let consumer = producer.consume();
5386 let mut announced = consumer.announced();
5387
5388 let broadcast = producer.create_broadcast("room/alice").unwrap();
5389 let clone = broadcast.clone();
5390 broadcast.announce(Route::default()).unwrap();
5391 announced.assert_next_active("room/alice");
5392
5393 drop(broadcast);
5395 announced.assert_next_wait();
5396 drop(clone);
5397 announced.assert_next_ended("room/alice");
5398 }
5399
5400 #[tokio::test]
5401 async fn publish_creates_and_announces_together() {
5402 let producer = origin(1).produce();
5403 let mut announced = producer.consume().announced();
5404 let _broadcast = producer.publish("room/alice", Route::default()).unwrap();
5405 announced.assert_next_active("room/alice");
5406 }
5407
5408 #[tokio::test]
5409 async fn standalone_broadcast_cannot_announce() {
5410 let broadcast = broadcast::Info::new().produce();
5411 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5412 broadcast.unannounce();
5414 }
5415
5416 #[tokio::test]
5417 async fn announce_replays_to_late_cursor() {
5418 let producer = origin(1).produce();
5419 let _a = producer.announce("room/alice", Route::default()).unwrap();
5420 let _b = producer.announce("room/bob", Route::default()).unwrap();
5421
5422 let mut announced = producer.consume().announced();
5423 announced.assert_next_active("room/alice");
5425 announced.assert_next_active("room/bob");
5426 announced.assert_next_wait();
5427 }
5428
5429 #[tokio::test]
5430 async fn announce_keeps_its_prefix_under_a_producer_scope() {
5431 let producer = origin(1).produce();
5432 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
5433
5434 let _a = scoped.announce("", Route::default()).unwrap();
5436 let mut announced = producer.consume().announced();
5437 announced.assert_next_active("");
5438
5439 assert!(matches!(
5441 scoped.announce("other", Route::default()),
5442 Err(Error::Unauthorized)
5443 ));
5444 }
5445
5446 #[tokio::test]
5447 async fn cursor_keeps_an_overlapping_prefix_above_its_scope() {
5448 let producer = origin(1).produce();
5449 let _a = producer.announce("", Route::default()).unwrap();
5450
5451 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
5452 let mut announced = consumer.announced();
5453 announced.assert_next_active("");
5454 }
5455
5456 #[tokio::test]
5457 async fn cursor_root_strips_prefix() {
5458 let producer = origin(1).produce();
5459 let _a = producer.announce("room/alice", Route::default()).unwrap();
5460
5461 let consumer = producer
5462 .consume()
5463 .scope("room", &Patterns::from(Pattern::all()))
5464 .unwrap();
5465 let mut announced = consumer.announced();
5466 announced.assert_next_active("alice");
5467 }
5468
5469 #[tokio::test]
5470 async fn best_route_wins_and_fails_over() {
5471 let producer = origin(1).produce();
5472 let mut announced = producer.consume().announced();
5473
5474 let expensive = producer
5475 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5476 .unwrap();
5477 let route = announced.assert_next_active("room");
5478 assert_eq!(route.cost, Cost::new(5));
5479
5480 let cheap = producer
5482 .announce("room", Route::default().with_hops(hops(&[20])).with_cost(1))
5483 .unwrap();
5484 let route = announced.assert_next_active("room");
5485 assert_eq!(route.cost, Cost::new(1));
5486
5487 drop(cheap);
5489 let route = announced.assert_next_active("room");
5490 assert_eq!(route.cost, Cost::new(5));
5491
5492 drop(expensive);
5494 announced.assert_next_ended("room");
5495 }
5496
5497 #[tokio::test]
5498 async fn identical_reannounce_is_invisible() {
5499 let producer = origin(1).produce();
5500 let mut announced = producer.consume().announced();
5501
5502 let old = producer
5503 .announce("room", Route::default().with_hops(hops(&[10])))
5504 .unwrap();
5505 let first = announced.assert_next_active("room");
5506 assert_eq!(first.hops.as_slice(), hops(&[10]).as_slice());
5507
5508 let _new = producer
5512 .announce("room", Route::default().with_hops(hops(&[10])))
5513 .unwrap();
5514 announced.assert_next_wait();
5515
5516 drop(old);
5518 announced.assert_next_wait();
5519 }
5520
5521 #[tokio::test]
5522 async fn exclude_hides_routes_through_the_peer() {
5523 let producer = origin(1).produce();
5524 let _a = producer
5525 .announce("room", Route::default().with_hops(hops(&[7])))
5526 .unwrap();
5527
5528 let mut hidden = producer.consume().excluding(origin(7)).announced();
5529 hidden.assert_next_wait();
5530
5531 let mut visible = producer.consume().excluding(origin(8)).announced();
5532 visible.assert_next_active("room");
5533 }
5534
5535 #[tokio::test]
5536 async fn exclude_matches_via_when_the_chain_is_anonymous() {
5537 let producer = origin(1).produce();
5538 let assigned = origin(777);
5539 let _echoed = producer
5540 .announce("echoed", Route::default().with_hops(hops(&[0])).with_via(assigned))
5541 .unwrap();
5542 let _local = producer
5543 .announce("local", Route::default().with_hops(hops(&[10])))
5544 .unwrap();
5545
5546 let mut hidden = producer.consume().excluding(assigned).announced();
5547 hidden.assert_next_active("local");
5548 hidden.assert_next_wait();
5549 }
5550
5551 #[tokio::test]
5552 async fn anonymous_route_loses_to_identified_at_any_cost() {
5553 let producer = origin(1).produce();
5554 let mut announced = producer.consume().announced();
5555
5556 let _anonymous = producer
5557 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5558 .unwrap();
5559 let route = announced.assert_next_active("room");
5560 assert!(route.is_anonymous());
5561 assert_eq!(route.cost, Cost::new(1));
5562
5563 let _identified = producer
5564 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5565 .unwrap();
5566 let route = announced.assert_next_active("room");
5567 assert!(!route.is_anonymous());
5568 assert_eq!(route.cost, Cost::new(5));
5569 }
5570
5571 #[tokio::test]
5574 async fn a_stamped_route_ranks_below_an_identified_one() {
5575 let producer = origin(1).produce();
5576 let mut announced = producer.consume().announced();
5577
5578 let mut stamped = Hops::new();
5579 stamped.stamp(origin(5)).unwrap();
5580 assert_eq!(stamped.as_slice(), &[origin(5), Hop::UNKNOWN]);
5581
5582 let _legacy = producer
5583 .announce("room", Route::default().with_hops(stamped).with_cost(0))
5584 .unwrap();
5585 let route = announced.assert_next_active("room");
5586 assert!(route.is_anonymous());
5587
5588 let _identified = producer
5589 .announce("room", Route::default().with_hops(hops(&[10, 11])).with_cost(5))
5590 .unwrap();
5591 let route = announced.assert_next_active("room");
5592 assert_eq!(route.hops.as_slice(), hops(&[10, 11]).as_slice());
5593 }
5594
5595 #[test]
5596 fn stamping_keeps_the_leading_zero_and_names_nothing_else() {
5597 let mut chain = hops(&[0, 7]);
5598 chain.stamp(origin(5)).unwrap();
5599 assert_eq!(chain.as_slice(), hops(&[5, 0, 7]).as_slice());
5600
5601 let mut named = hops(&[7, 0]);
5603 named.stamp(origin(5)).unwrap();
5604 assert_eq!(named.as_slice(), hops(&[7, 0]).as_slice());
5605
5606 let mut full = Hops::try_from(vec![Hop::UNKNOWN; MAX_HOPS]).unwrap();
5608 assert_eq!(full.stamp(origin(5)), Err(InvalidHop::TooMany));
5609 }
5610
5611 #[tokio::test]
5612 async fn anonymous_routes_order_by_cost() {
5613 let producer = origin(1).produce();
5614 let mut announced = producer.consume().announced();
5615
5616 let expensive = producer
5617 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(5))
5618 .unwrap();
5619 let route = announced.assert_next_active("room");
5620 assert_eq!(route.cost, Cost::new(5));
5621
5622 let _cheap = producer
5623 .announce("room", Route::default().with_hops(hops(&[0, 7])).with_cost(1))
5624 .unwrap();
5625 let route = announced.assert_next_active("room");
5626 assert!(route.is_anonymous());
5627 assert_eq!(route.cost, Cost::new(1));
5628
5629 drop(expensive);
5630 announced.assert_next_wait();
5631 }
5632
5633 #[tokio::test]
5634 async fn anonymous_chain_from_identified_peer_still_ranks_last() {
5635 let producer = origin(1).produce();
5636 let mut announced = producer.consume().announced();
5637
5638 let _anonymous = producer
5639 .announce(
5640 "room",
5641 Route::default()
5642 .with_hops(hops(&[0, 7]))
5643 .with_cost(1)
5644 .with_via(origin(7)),
5645 )
5646 .unwrap();
5647 announced.assert_next_active("room");
5648
5649 let _identified = producer
5650 .announce("room", Route::default().with_hops(hops(&[10, 20])).with_cost(5))
5651 .unwrap();
5652 let route = announced.assert_next_active("room");
5653 assert!(!route.is_anonymous());
5654 assert_eq!(route.cost, Cost::new(5));
5655 }
5656
5657 #[tokio::test]
5658 async fn request_prefers_identified_over_cheaper_anonymous() {
5659 let producer = origin(1).produce();
5660 let consumer = producer.consume();
5661
5662 let anonymous = producer
5663 .dynamic("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5664 .unwrap();
5665 let identified = producer
5666 .dynamic("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5667 .unwrap();
5668
5669 let _pending = consumer.request_broadcast("room/alice");
5670 let request = queued(&identified).await;
5671 assert_eq!(request.path().as_str(), "room/alice");
5672 assert!(
5673 anonymous.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
5674 "the cheaper anonymous route must not serve"
5675 );
5676 }
5677
5678 #[tokio::test]
5679 async fn update_reprices_in_place() {
5680 let producer = origin(1).produce();
5681 let mut announced = producer.consume().announced();
5682
5683 let announcement = producer.announce("room", Route::default()).unwrap();
5684 announced.assert_next_active("room");
5685
5686 announcement.update(Route::default().with_cost(9)).unwrap();
5687 let route = announced.assert_next_active("room");
5688 assert_eq!(route.cost, Cost::new(9));
5689 }
5690
5691 #[tokio::test]
5692 async fn retract_after_undelivered_reprice_still_delivered() {
5693 let producer = origin(1).produce();
5694 let mut announced = producer.consume().announced();
5695
5696 let announcement = producer.announce("room", Route::default()).unwrap();
5697 announced.assert_next_active("room");
5698
5699 announcement.update(Route::default().with_cost(9)).unwrap();
5703 drop(announcement);
5704 announced.assert_next_ended("room");
5705 announced.assert_next_wait();
5706 }
5707
5708 #[tokio::test]
5709 async fn scoped_cursor_advertises_most_specific_covering_route() {
5710 let producer = origin(1).produce();
5711 let _broad = producer.announce("room", Route::default().with_cost(1)).unwrap();
5714 let _narrow = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
5715
5716 let consumer = producer
5717 .consume()
5718 .scope("room/alice", &Patterns::from(Pattern::all()))
5719 .unwrap();
5720 let mut announced = consumer.announced();
5721 let route = announced.assert_next_active("");
5722 assert_eq!(route.cost, Cost::new(9));
5723 announced.assert_next_wait();
5724 }
5725
5726 #[tokio::test]
5727 async fn capture_change_retracts_before_reannouncing_a_presented_prefix() {
5728 let producer = origin(1).produce();
5729 let _broad = producer.announce("room", Route::default()).unwrap();
5730 let exact = producer.announce("room/alice", Route::default()).unwrap();
5731 let consumer = producer
5732 .consume()
5733 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
5734 .unwrap()
5735 .scope("room/alice", &Patterns::from(Pattern::all()))
5736 .unwrap();
5737 let mut announced = consumer.announced();
5738
5739 let first = announced.next().now_or_never().expect("next").expect("announce");
5740 assert_eq!(first.prefix.as_str(), "");
5741 assert_eq!(first.kind, AnnounceKind::Announced);
5742 assert_eq!(first.captures, Some(Vec::new()));
5743
5744 drop(exact);
5745 let retracted = announced.next().now_or_never().expect("next").expect("retract");
5746 assert_eq!(retracted.prefix.as_str(), "");
5747 assert_eq!(retracted.kind, AnnounceKind::Retracted);
5748 assert_eq!(retracted.captures, Some(Vec::new()));
5749 let replacement = announced.next().now_or_never().expect("next").expect("announce");
5750 assert_eq!(replacement.prefix.as_str(), "");
5751 assert_eq!(replacement.kind, AnnounceKind::Announced);
5752 assert_eq!(replacement.captures, None);
5753 }
5754
5755 #[tokio::test]
5756 async fn routed_broadcast_resolves_once_announced() {
5757 let producer = origin(1).produce();
5758 let consumer = producer.consume();
5759
5760 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
5762 assert!((&mut resolving).now_or_never().is_none());
5763
5764 let broadcast = producer.create_broadcast("room/alice").unwrap();
5766 for _ in 0..20 {
5767 tokio::task::yield_now().await;
5768 }
5769 assert!((&mut resolving).now_or_never().is_none());
5770
5771 broadcast.announce(Route::default()).unwrap();
5772 let resolved = resolving.await.expect("resolves once announced");
5773 assert_eq!(resolved.info().path.as_str(), "room/alice");
5774 drop(broadcast);
5775 }
5776
5777 #[tokio::test]
5780 async fn cheaper_remote_route_beats_a_local_broadcast() {
5781 let producer = origin(1).produce();
5782 let consumer = producer.consume();
5783 let mut announced = consumer.announced();
5784
5785 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5786 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(5));
5787
5788 let server = producer
5789 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5790 .unwrap();
5791 let route = announced.assert_next_active("room/alice");
5792 assert_eq!(route.cost, Cost::new(1));
5793 assert_eq!(route.hops, hops(&[10]));
5794
5795 let pending = consumer.request_broadcast("room/alice");
5797 let request = queued(&server).await;
5798 let upstream = broadcast::Info::new().produce();
5799 request.accept(&upstream);
5800 pending.await.expect("resolves through the cheaper route");
5801 }
5802
5803 #[tokio::test]
5807 async fn cheaper_route_after_a_front_wins_new_requests() {
5808 let producer = origin(1).produce();
5809 let consumer = producer.consume();
5810
5811 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5812 let first = consumer
5813 .request_broadcast("room/alice")
5814 .await
5815 .expect("resolves locally");
5816
5817 let server = producer
5818 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5819 .unwrap();
5820
5821 let pending = consumer.request_broadcast("room/alice");
5822 let request = queued(&server).await;
5823 let upstream = broadcast::Info::new().produce();
5824 request.accept(&upstream);
5825 let second = pending.await.expect("resolves through the cheaper route");
5826
5827 assert!(!first.is_closed(), "the old front must keep serving its readers");
5828 assert!(!first.is_clone(&second), "the newcomer must not join the old front");
5829 }
5830
5831 #[tokio::test]
5835 async fn local_broadcast_wins_a_tie_with_a_hopless_route() {
5836 let producer = origin(1).produce();
5837 let consumer = producer.consume();
5838
5839 let _local = producer.publish("room/alice", Route::default()).unwrap();
5840 let server = producer.dynamic("room/alice", Route::default()).unwrap();
5841
5842 let resolved = tokio::time::timeout(Duration::from_secs(1), consumer.request_broadcast("room/alice"))
5844 .await
5845 .expect("the newer hopless route won the tie")
5846 .expect("resolves");
5847 assert_eq!(resolved.info().path.as_str(), "room/alice");
5848 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5849 }
5850
5851 #[tokio::test]
5854 async fn announce_stats_follow_the_advertisement() {
5855 let registry = stats::Registry::new(stats::Config::new());
5856 let producer = origin(1)
5857 .produce()
5858 .with_stats(registry.tier(stats::Tier::default()).session("root"));
5859 let announces = || {
5860 registry
5861 .snapshot()
5862 .traffic()
5863 .into_iter()
5864 .find(|(_, role, _)| *role == stats::Role::Subscriber)
5865 .map(|(_, _, traffic)| (traffic.announces_started, traffic.announces_ended))
5866 .unwrap_or_default()
5867 };
5868
5869 let broadcast = producer.create_broadcast("room/alice").unwrap();
5870 assert_eq!(announces(), (0, 0), "a hidden broadcast is not announced");
5871 broadcast.announce(Route::default()).unwrap();
5872 broadcast
5873 .announce(Route {
5874 cost: Cost::new(3),
5875 ..Route::default()
5876 })
5877 .unwrap();
5878 assert_eq!(announces(), (1, 0), "a re-price is not another announce");
5879 broadcast.unannounce();
5880 assert_eq!(announces(), (1, 1));
5881 broadcast.announce(Route::default()).unwrap();
5882 drop(broadcast);
5883 assert_eq!(announces(), (2, 2));
5884 }
5885
5886 #[tokio::test]
5888 async fn local_broadcast_wins_a_cost_tie() {
5889 let producer = origin(1).produce();
5890 let consumer = producer.consume();
5891 let mut announced = consumer.announced();
5892
5893 let server = producer
5894 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(2))
5895 .unwrap();
5896 announced.assert_next_active("room/alice");
5897 let _local = producer.publish("room/alice", Route::default().with_cost(2)).unwrap();
5898 assert!(announced.assert_next_active("room/alice").hops.is_empty());
5899
5900 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5901 assert_eq!(resolved.info().path.as_str(), "room/alice");
5902 for _ in 0..20 {
5903 tokio::task::yield_now().await;
5904 }
5905 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5906 }
5907
5908 #[tokio::test]
5911 async fn unannounce_ends_the_front() {
5912 let producer = origin(1).produce();
5913 let consumer = producer.consume();
5914
5915 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5916 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5917
5918 broadcast.unannounce();
5919 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5920 assert!(matches!(err, Error::Unroutable), "joined a retracted front: {err}");
5921 settle(|| resolved.is_closed()).await;
5922 assert!(!broadcast.consume().is_closed(), "the broadcast itself lives on");
5923
5924 broadcast.announce(Route::default()).unwrap();
5926 let again = consumer.request_broadcast("room/alice").await.expect("resolves again");
5927 assert!(!again.is_clone(&resolved));
5928 }
5929
5930 #[tokio::test]
5934 async fn unannounce_keeps_a_track_awaiting_its_info() {
5935 let producer = origin(1).produce();
5936 let consumer = producer.consume();
5937
5938 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5939 let mut dynamic = broadcast.dynamic();
5940 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5941 let track = resolved.track("video").unwrap();
5942 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
5943 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
5944 .await
5945 .expect("the front asked the source")
5946 .expect("request");
5947
5948 broadcast.unannounce();
5949 settle(|| resolved.is_closed()).await;
5950
5951 let source = request.accept(None);
5952 let mut group = source.append_group().unwrap();
5953 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
5954 group.finish().unwrap();
5955 source.finish().unwrap();
5956
5957 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
5958 let mut group = subscription.recv_group().await.unwrap().expect("the source's group");
5959 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late");
5960 assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly");
5961 }
5962
5963 async fn served_front() -> (Dynamic, broadcast::Producer, broadcast::Dynamic, broadcast::Consumer) {
5966 let producer = origin(1).produce();
5967 let consumer = producer.consume();
5968 let server = producer
5969 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
5970 .unwrap();
5971 let pending = consumer.request_broadcast("room/alice");
5972 let upstream = broadcast::Info::new().produce();
5973 let dynamic = upstream.dynamic();
5974 queued(&server).await.accept(&upstream);
5975 let resolved = pending.await.expect("resolves");
5976 (server, upstream, dynamic, resolved)
5977 }
5978
5979 #[tokio::test]
5984 async fn returning_reader_skips_a_warm_cache_the_copy_resolved_past() {
5985 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
5986 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
5987 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
5988
5989 let track = resolved.track("audio").unwrap();
5990 let b = budget.clone();
5991 let subscribing = tokio::spawn(async move { track.subscribe(b).await });
5992 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
5993 .await
5994 .expect("the front asked the source")
5995 .expect("request");
5996 let source = request.resolving_start().accept(None);
5997 for seq in 0..4u64 {
5998 let mut group = source.create_group(seq.into()).unwrap();
5999 group.write_frame(ms(seq * 20), b"old".as_ref()).unwrap();
6000 group.finish().unwrap();
6001 }
6002 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6003 subscription.recv_group().await.unwrap().expect("the live group");
6004 drop(subscription);
6005 tokio::time::timeout(Duration::from_secs(1), source.unused())
6006 .await
6007 .expect("parked")
6008 .expect("source open");
6009 drop(source);
6010
6011 let track = resolved.track("audio").unwrap();
6012 let subscribing = tokio::spawn(async move { track.subscribe(budget).await });
6013 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6014 .await
6015 .expect("the front asked the source again")
6016 .expect("request");
6017 let mut source = request.resolving_start().accept(None);
6018 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6019
6020 assert!(
6022 tokio::time::timeout(Duration::from_millis(50), subscription.recv_group())
6023 .await
6024 .is_err(),
6025 "the warm cache was served before the copy resolved its start"
6026 );
6027
6028 source.start_at(20).unwrap();
6030 let mut group = source.create_group(20u64.into()).unwrap();
6031 group.write_frame(ms(2000), b"new".as_ref()).unwrap();
6032 group.finish().unwrap();
6033 let group = subscription.recv_group().await.unwrap().expect("the live group");
6034 assert_eq!(group.sequence, 20, "a stale warm group was served");
6035 }
6036
6037 #[tokio::test]
6042 async fn returning_reader_replays_a_current_warm_cache() {
6043 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6044
6045 let track = resolved.track("catalog").unwrap();
6046 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6047 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6048 .await
6049 .expect("the front asked the source")
6050 .expect("request");
6051 let source = request.resolving_start().accept(None);
6052 let mut group = source.create_group(0u64.into()).unwrap();
6053 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6054 group.finish().unwrap();
6055 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6056 subscription.recv_group().await.unwrap().expect("the catalog");
6057 drop(subscription);
6058 tokio::time::timeout(Duration::from_secs(1), source.unused())
6059 .await
6060 .expect("parked")
6061 .expect("source open");
6062 drop(source);
6063
6064 let track = resolved.track("catalog").unwrap();
6065 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6066 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6067 .await
6068 .expect("the front asked the source again")
6069 .expect("request");
6070 let mut source = request.resolving_start().accept(None);
6071 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6072
6073 let reading = tokio::spawn(async move {
6076 let mut group = subscription.recv_group().await.unwrap().expect("the catalog");
6077 assert_eq!(group.sequence, 0);
6078 group.read_frame().await.unwrap().expect("the snapshot").payload
6079 });
6080 tokio::task::yield_now().await;
6081 assert_eq!(
6082 source.subscription().and_then(|sub| sub.start),
6083 Some(track::Position { group: 0, frame: 1 }),
6084 "the re-splice asked past the cached catalog"
6085 );
6086 source.start_at(0).unwrap();
6087 let mut tail = source.create_group(0u64.into()).unwrap();
6088 tail.start_at(1).unwrap();
6089 tail.finish().unwrap();
6090 let payload = tokio::time::timeout(Duration::from_secs(1), reading)
6091 .await
6092 .expect("the returning reader never got the catalog")
6093 .unwrap();
6094 assert_eq!(&payload[..], b"snapshot");
6095 }
6096
6097 #[tokio::test(start_paused = true)]
6101 async fn finished_track_is_forgotten_after_the_linger() {
6102 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6103
6104 let track = resolved.track("catalog").unwrap();
6105 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6106 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6107 .await
6108 .expect("the front asked the source")
6109 .expect("request");
6110 let source = request.resolving_start().accept(None);
6111 let mut group = source.create_group(0u64.into()).unwrap();
6112 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6113 group.finish().unwrap();
6114 source.finish().unwrap();
6115 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6116 assert_eq!(
6117 next_group(&mut subscription)
6118 .await
6119 .unwrap()
6120 .expect("the catalog")
6121 .sequence,
6122 0
6123 );
6124 assert!(next_group(&mut subscription).await.unwrap().is_none());
6125 drop(subscription);
6126 drop(source);
6127
6128 let mut subscription = resolved.track("catalog").unwrap().subscribe(None).await.unwrap();
6130 assert_eq!(
6131 next_group(&mut subscription)
6132 .await
6133 .unwrap()
6134 .expect("the catalog")
6135 .sequence,
6136 0
6137 );
6138 assert!(next_group(&mut subscription).await.unwrap().is_none());
6139 drop(subscription);
6140 assert!(
6141 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6142 .await
6143 .is_err(),
6144 "a finished track within the linger asked the source again"
6145 );
6146
6147 tokio::time::sleep(TRACK_IDLE_LINGER).await;
6149
6150 let track = resolved.track("catalog").unwrap();
6151 let _subscribing = tokio::spawn(async move { track.subscribe(None).await });
6152 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6153 .await
6154 .expect("the finished track outlived the linger")
6155 .expect("request");
6156 }
6157
6158 #[tokio::test]
6163 async fn returning_reader_continues_an_open_warm_group() {
6164 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6165
6166 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6167 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6168 .await
6169 .expect("frame")
6170 .unwrap()
6171 .expect("group ended");
6172 frame.payload.to_vec()
6173 }
6174
6175 let mut expect: Vec<&[u8]> = Vec::new();
6176 let mut floor: Option<track::Position> = None;
6177 for (round, payload) in [b"a".as_ref(), b"b", b"c"].into_iter().enumerate() {
6178 let track = resolved.track("log").unwrap();
6179 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6180 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6181 .await
6182 .expect("the front asked the source")
6183 .expect("request");
6184 let mut source = request.resolving_start().accept(None);
6185 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6186
6187 source.start_at(0).unwrap();
6189 let mut group = source.create_group(0u64.into()).unwrap();
6190 if let Some(floor) = floor {
6191 group.start_at(floor.frame).unwrap();
6192 }
6193 group.write_frame(crate::Timestamp::ZERO, payload).unwrap();
6194 expect.push(payload);
6195 source
6196 .insert_datagram(10, crate::Timestamp::ZERO, b"datagram".as_ref())
6197 .unwrap();
6198
6199 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6200 .await
6201 .expect("group 0")
6202 .unwrap()
6203 .expect("track ended");
6204 assert_eq!(reading.sequence, 0);
6205 for frame in &expect {
6206 assert_eq!(read(&mut reading).await, *frame, "round {round}");
6207 }
6208 assert_eq!(
6209 source.subscription().and_then(|sub| sub.start),
6210 floor,
6211 "round {round} asked for the wrong continuation"
6212 );
6213
6214 drop(reading);
6215 drop(subscription);
6216 tokio::time::timeout(Duration::from_secs(1), source.unused())
6217 .await
6218 .expect("parked")
6219 .expect("source open");
6220 drop(group);
6221 drop(source);
6222 floor = Some(track::Position {
6223 group: 0,
6224 frame: expect.len() as u64,
6225 });
6226 }
6227 }
6228
6229 #[tokio::test]
6236 async fn a_takeover_keeps_the_open_group_head_for_later_readers() {
6237 tokio::time::pause();
6238 for dies_first in [true, false] {
6239 takeover_keeps_the_open_group_head(dies_first).await;
6240 }
6241 }
6242
6243 async fn takeover_keeps_the_open_group_head(dies_first: bool) {
6244 let producer = origin(1).produce();
6245 let first_server = producer
6246 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6247 .unwrap();
6248 let pending = producer.consume().request_broadcast("room/alice");
6249 let upstream = broadcast::Info::new().produce();
6250 let mut dynamic = upstream.dynamic();
6251 queued(&first_server).await.accept(&upstream);
6252 let resolved = pending.await.unwrap();
6253
6254 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6255 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6256 .await
6257 .expect("frame")
6258 .unwrap()
6259 .expect("group ended");
6260 frame.payload.to_vec()
6261 }
6262
6263 let track = resolved.track("log").unwrap();
6265 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6266 let source = dynamic.requested_track().await.unwrap().accept(None);
6267 let mut group = source.create_group(0u64.into()).unwrap();
6268 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6269 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6270 let mut subscription = subscribing.await.unwrap().unwrap();
6271 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6272 assert_eq!(read(&mut reading).await, b"a");
6273 assert_eq!(read(&mut reading).await, b"b");
6274
6275 let mut first = Some((group, source, upstream, dynamic));
6278 let mut joining = None;
6279 if dies_first {
6280 drop(first.take());
6281 assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none());
6284 let track = resolved.track("log").unwrap();
6285 joining = Some(tokio::spawn(async move {
6286 let mut joined = track.subscribe(None).await.unwrap();
6287 let group = joined.recv_group().await.unwrap().expect("track ended");
6288 (joined, group)
6289 }));
6290 }
6291 let cost = if dies_first { 5 } else { 0 };
6292 let replacement_server = producer
6293 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(cost))
6294 .unwrap();
6295 let replacement = broadcast::Info::new().produce();
6296 let mut replacement_dynamic = replacement.dynamic();
6297 queued(&replacement_server).await.accept(&replacement);
6298 let mut next = tokio::time::timeout(Duration::from_secs(1), replacement_dynamic.requested_track())
6299 .await
6300 .expect("the front asked the replacement")
6301 .unwrap()
6302 .resolving_start()
6303 .accept(None);
6304 next.start_at(0).unwrap();
6305 let mut next_group = next.create_group(0u64.into()).unwrap();
6306 next_group.start_at(2).unwrap();
6307 next_group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6308 assert_eq!(read(&mut reading).await, b"c", "dies_first={dies_first}");
6309
6310 if let Some((group, ..)) = &mut first {
6312 let mut independent = group.consume();
6313 group.write_frame(crate::Timestamp::ZERO, b"x".as_ref()).unwrap();
6314 for expect in [b"a", b"b", b"x"] {
6315 assert_eq!(read(&mut independent).await, expect);
6316 }
6317 group.finish().unwrap();
6318 }
6319 drop(first);
6320
6321 let track = resolved.track("log").unwrap();
6322 let mut later = tokio::time::timeout(Duration::from_secs(1), track.subscribe(None))
6323 .await
6324 .expect("subscribe")
6325 .unwrap();
6326 let mut late = tokio::time::timeout(Duration::from_secs(1), later.recv_group())
6327 .await
6328 .unwrap_or_else(|_| panic!("dies_first={dies_first}: the later reader never got the group"))
6329 .unwrap()
6330 .expect("track ended");
6331 assert_eq!(late.sequence, 0);
6332 for expect in [b"a", b"b", b"c"] {
6333 assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}");
6334 }
6335 let mut joined = None;
6336 if let Some(joining) = joining {
6337 let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining)
6338 .await
6339 .expect("the reader joining mid-outage never got the group")
6340 .unwrap();
6341 assert_eq!(group.sequence, 0);
6342 for expect in [b"a", b"b", b"c"] {
6343 assert_eq!(read(&mut group).await, expect);
6344 }
6345 joined = Some((subscription, group));
6346 }
6347
6348 assert!(
6350 tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6351 .await
6352 .is_err(),
6353 "dies_first={dies_first}: the group was handed out twice"
6354 );
6355
6356 next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap();
6358 assert_eq!(read(&mut reading).await, b"d");
6359 assert_eq!(read(&mut late).await, b"d");
6360 if let Some((_, group)) = &mut joined {
6361 assert_eq!(read(group).await, b"d");
6362 }
6363 next_group.finish().unwrap();
6364 drop((
6365 joined,
6366 reading,
6367 late,
6368 subscription,
6369 later,
6370 next,
6371 replacement,
6372 replacement_server,
6373 first_server,
6374 ));
6375 }
6376
6377 #[tokio::test]
6382 async fn returning_reader_gets_a_live_open_warm_group() {
6383 use futures::FutureExt;
6384
6385 tokio::time::pause();
6386 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6387
6388 let track = resolved.track("video").unwrap();
6390 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6391 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6392 .await
6393 .expect("the front asked the source")
6394 .expect("request");
6395 let mut source = request.resolving_start().accept(None);
6396 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6397 source.start_at(0).unwrap();
6398 let mut group = source.create_group(0u64.into()).unwrap();
6399 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6400 let mut reading = subscription.recv_group().await.unwrap().expect("group 0");
6401 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6402 drop(reading);
6403 drop(subscription);
6404 tokio::time::timeout(Duration::from_secs(1), source.unused())
6405 .await
6406 .expect("parked")
6407 .expect("source open");
6408 drop(group);
6409 drop(source);
6410
6411 let track = resolved.track("video").unwrap();
6413 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6414 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6415 .await
6416 .expect("the front asked the source")
6417 .expect("request");
6418 let mut source = request.resolving_start().accept(None);
6419 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6420 source.start_at(0).unwrap();
6421 let mut group = source.create_group(0u64.into()).unwrap();
6422 group.start_at(1).unwrap();
6423 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6424
6425 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6426 .await
6427 .expect("group 0")
6428 .unwrap()
6429 .expect("track ended");
6430 assert_eq!(reading.sequence, 0);
6431 let early = reading.finished().now_or_never();
6432 assert!(
6433 early.is_none(),
6434 "the live group already resolved its end before a frame was read: {early:?}"
6435 );
6436
6437 group.finish().unwrap();
6439 assert_eq!(reading.finished().await.unwrap(), 0);
6440 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6441 assert_eq!(reading.finished().await.unwrap(), 1);
6442 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"b");
6443 assert_eq!(reading.finished().await.unwrap(), 2);
6444 }
6445
6446 #[tokio::test]
6447 async fn warm_head_survives_another_takeover_before_park() {
6448 tokio::time::pause();
6449 let producer = origin(1).produce();
6450 let _server = producer
6451 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6452 .unwrap();
6453 let pending = producer.consume().request_broadcast("room/alice");
6454 let upstream = broadcast::Info::new().produce();
6455 let mut dynamic = upstream.dynamic();
6456 queued(&_server).await.accept(&upstream);
6457 let resolved = pending.await.unwrap();
6458
6459 let track = resolved.track("log").unwrap();
6460 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6461 let source = dynamic.requested_track().await.unwrap().accept(None);
6462 let mut group = source.create_group(0u64.into()).unwrap();
6463 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6464 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6465 let mut subscription = subscribing.await.unwrap().unwrap();
6466 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6467 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6468 drop(reading);
6469 drop(subscription);
6470 source.unused().await.unwrap();
6471 drop(group);
6472 drop(source);
6473
6474 let track = resolved.track("log").unwrap();
6475 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6476 let mut resumed = dynamic.requested_track().await.unwrap().resolving_start().accept(None);
6477 resumed.start_at(0).unwrap();
6478 let mut subscription = subscribing.await.unwrap().unwrap();
6479 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6480 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6481
6482 let replacement_server = producer
6483 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
6484 .unwrap();
6485 let replacement = broadcast::Info::new().produce();
6486 let mut replacement_dynamic = replacement.dynamic();
6487 queued(&replacement_server).await.accept(&replacement);
6488 let mut source = replacement_dynamic
6489 .requested_track()
6490 .await
6491 .unwrap()
6492 .resolving_start()
6493 .accept(None);
6494 source.start_at(0).unwrap();
6495 let mut group = source.create_group(0u64.into()).unwrap();
6496 group.start_at(2).unwrap();
6497 group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6498 let mut continuation = reading.clone();
6499 continuation.start_at(2);
6500 let frame = tokio::time::timeout(Duration::from_secs(1), continuation.read_frame())
6501 .await
6502 .expect("replacement frame")
6503 .unwrap()
6504 .unwrap();
6505 assert_eq!(&frame.payload[..], b"c");
6506 drop(continuation);
6507 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"b");
6508 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"c");
6509 drop(reading);
6510 drop(subscription);
6511 source.unused().await.unwrap();
6512 drop(group);
6513 drop(source);
6514
6515 let track = resolved.track("log").unwrap();
6516 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6517 let mut next = replacement_dynamic
6518 .requested_track()
6519 .await
6520 .unwrap()
6521 .resolving_start()
6522 .accept(None);
6523 next.start_at(0).unwrap();
6524 let mut subscription = subscribing.await.unwrap().unwrap();
6525 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6526 for expected in [b"a", b"b", b"c"] {
6527 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], expected);
6528 }
6529 }
6530
6531 #[tokio::test]
6534 async fn unannounce_keeps_a_returning_reader_awaiting_its_info() {
6535 let producer = origin(1).produce();
6536 let consumer = producer.consume();
6537
6538 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6539 let mut dynamic = broadcast.dynamic();
6540 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6541 let track = resolved.track("video").unwrap();
6542 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6543 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6544 .await
6545 .expect("the front asked the source")
6546 .expect("request");
6547 let source = request.accept(None);
6548 let mut group = source.append_group().unwrap();
6549 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
6550 group.finish().unwrap();
6551 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6552 subscription.recv_group().await.unwrap().expect("the cached group");
6553 drop(subscription);
6554
6555 tokio::time::timeout(Duration::from_secs(1), source.unused())
6558 .await
6559 .expect("parked")
6560 .expect("source open");
6561 drop(source);
6562 let track = resolved.track("video").unwrap();
6563 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6564 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6565 .await
6566 .expect("the front asked the source again")
6567 .expect("request");
6568
6569 broadcast.unannounce();
6570 settle(|| resolved.is_closed()).await;
6571
6572 let source = request.accept(None);
6574 let mut group = source.create_group(1u64.into()).unwrap();
6575 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
6576 group.finish().unwrap();
6577 source.finish().unwrap();
6578
6579 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
6580 let mut payloads = Vec::new();
6581 while let Some(mut group) = subscription.recv_group().await.expect("ends cleanly") {
6582 payloads.push(group.read_frame().await.unwrap().unwrap().payload);
6583 }
6584 assert_eq!(payloads.last().map(|p| &p[..]), Some(&b"late"[..]));
6585 }
6586
6587 #[tokio::test]
6590 async fn reannounce_before_the_front_acts_keeps_it() {
6591 let producer = origin(1).produce();
6592 let consumer = producer.consume();
6593
6594 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6595 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6596
6597 broadcast.unannounce();
6598 broadcast.announce(Route::default()).unwrap();
6599 for _ in 0..20 {
6600 tokio::task::yield_now().await;
6601 }
6602 assert!(!resolved.is_closed(), "the front ended across a reannouncement");
6603 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6604 assert!(again.is_clone(&resolved));
6605 }
6606
6607 #[tokio::test]
6608 async fn local_broadcast_resolves_once_announced() {
6609 let producer = origin(1).produce();
6610 let consumer = producer.consume();
6611
6612 let broadcast = producer.create_broadcast("room/alice").unwrap();
6614 let err = consumer
6615 .request_broadcast("room/alice")
6616 .now_or_never()
6617 .expect("unroutable is synchronous")
6618 .err()
6619 .unwrap();
6620 assert!(matches!(err, Error::Unroutable));
6621
6622 broadcast.announce(Route::default()).unwrap();
6623 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6624 assert_eq!(resolved.info().path.as_str(), "room/alice");
6625 drop(broadcast);
6626
6627 let err = consumer
6629 .request_broadcast("room/bob")
6630 .now_or_never()
6631 .expect("unroutable is synchronous")
6632 .err()
6633 .unwrap();
6634 assert!(matches!(err, Error::Unroutable));
6635 }
6636
6637 #[test]
6638 fn create_broadcast_accepts_a_max_depth_path() {
6639 let producer = origin(1).produce();
6640 let path = vec!["a"; Path::MAX_PARTS].join("/");
6641 let _broadcast = producer.create_broadcast(path.as_str()).expect("max depth is allowed");
6642 let deeper = vec!["a"; Path::MAX_PARTS + 1].join("/");
6643 assert!(matches!(
6644 producer.create_broadcast(deeper.as_str()),
6645 Err(Error::BoundsExceeded(_))
6646 ));
6647 }
6648
6649 #[tokio::test]
6650 async fn duplicate_routes_aggregate_until_the_last_leaves() {
6651 let producer = origin(1).produce();
6652 let first = producer.dynamic("live", Route::default().with_cost(3)).unwrap();
6653 let second = producer.dynamic("live", Route::default().with_cost(1)).unwrap();
6654
6655 let mut announced = producer.consume().announced();
6656 let update = announced.next().now_or_never().expect("next").expect("no next");
6657 assert_eq!(update.prefix.as_str(), "live");
6658 assert_eq!(update.kind, AnnounceKind::Announced);
6659 assert_eq!(update.route.cost, Cost::new(1));
6660 announced.assert_next_wait();
6661
6662 drop(second);
6663 let update = announced.next().now_or_never().expect("next").expect("no next");
6664 assert_eq!(update.prefix.as_str(), "live");
6665 assert_eq!(update.kind, AnnounceKind::Updated);
6666 assert_eq!(update.route.cost, Cost::new(3));
6667
6668 drop(first);
6669 announced.assert_next_ended("live");
6670 announced.assert_next_wait();
6671 }
6672
6673 #[test]
6674 fn dynamic_may_cover_a_scope_but_disjoint_prefixes_are_refused() {
6675 let producer = origin(1).produce();
6676 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
6677 let _broad = scoped
6678 .dynamic("", Route::default())
6679 .expect("an overlapping prefix is accepted");
6680
6681 let _ok = scoped
6682 .dynamic("room/alice", Route::default())
6683 .expect("a contained prefix is accepted");
6684 assert!(matches!(
6685 scoped.dynamic("other", Route::default()),
6686 Err(Error::Unauthorized)
6687 ));
6688 }
6689
6690 #[tokio::test]
6691 async fn dynamic_route_keeps_its_producer_scope() {
6692 let producer = origin(1).produce();
6693 let scope = Patterns::from("*/chat".parse::<Pattern>().unwrap());
6694 let scoped = producer.scope("", &scope).unwrap();
6695 let dynamic = scoped.dynamic("", Route::default()).unwrap();
6696
6697 let mut matching = producer
6698 .consume()
6699 .scope("", &scopes(&["room/chat"]))
6700 .unwrap()
6701 .announced();
6702 matching.assert_next_active("");
6703 let mut outside = producer
6704 .consume()
6705 .scope("", &scopes(&["room/video"]))
6706 .unwrap()
6707 .announced();
6708 outside.assert_next_wait();
6709
6710 let refused = producer
6711 .consume()
6712 .request_broadcast("room/video")
6713 .now_or_never()
6714 .expect("an out-of-scope request must be refused synchronously");
6715 assert!(matches!(refused, Err(Error::Unroutable)));
6716 assert!(dynamic.requested_broadcast().now_or_never().is_none());
6717
6718 let _pending = producer.consume().request_broadcast("room/chat");
6719 let request = queued(&dynamic).await;
6720 assert_eq!(request.path().as_str(), "room/chat");
6721 }
6722
6723 #[tokio::test]
6724 async fn scoped_cursor_selects_among_the_routes_it_can_see() {
6725 let producer = origin(1).produce();
6726 let scoped = |pattern: &str| {
6727 producer
6728 .scope("", &Patterns::from(pattern.parse::<Pattern>().unwrap()))
6729 .unwrap()
6730 };
6731 let _chat = scoped("*/chat").dynamic("", Route::default().with_cost(1)).unwrap();
6732 let _video = scoped("*/video").dynamic("", Route::default().with_cost(5)).unwrap();
6733
6734 let mut video = producer
6735 .consume()
6736 .scope("", &scopes(&["room/video"]))
6737 .unwrap()
6738 .announced();
6739 assert_eq!(video.assert_next_active("").cost, Cost::new(5));
6740 }
6741
6742 #[tokio::test]
6743 async fn route_changes_preserve_each_cursors_visible_winner() {
6744 let producer = origin(1).produce();
6745 let consumer = producer.consume();
6746 let mut all = consumer.clone().announced();
6747 let mut excluded = consumer.clone().excluding(origin(7)).announced();
6748 let mut video = consumer
6749 .clone()
6750 .scope("", &scopes(&["room/live/video"]))
6751 .unwrap()
6752 .announced();
6753 let mut rooted = consumer
6754 .scope("room/live", &Patterns::from(Pattern::all()))
6755 .unwrap()
6756 .announced();
6757
6758 let chat = producer
6759 .scope("", &scopes(&["room/live/chat"]))
6760 .unwrap()
6761 .dynamic("room/live", Route::default().with_hops(hops(&[7])).with_cost(1))
6762 .unwrap();
6763 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(1));
6764 assert_eq!(rooted.assert_next_active("").cost, Cost::new(1));
6765 excluded.assert_next_wait();
6766 video.assert_next_wait();
6767
6768 let video_route = producer
6769 .scope("", &scopes(&["room/live/video"]))
6770 .unwrap()
6771 .dynamic("room/live", Route::default().with_hops(hops(&[8])).with_cost(5))
6772 .unwrap();
6773 all.assert_next_wait();
6774 rooted.assert_next_wait();
6775 assert_eq!(excluded.assert_next_active("room/live").cost, Cost::new(5));
6776 assert_eq!(video.assert_next_active("room/live").cost, Cost::new(5));
6777
6778 chat.update(Route::default().with_hops(hops(&[7])).with_cost(9))
6779 .unwrap();
6780 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(5));
6781 assert_eq!(rooted.assert_next_active("").cost, Cost::new(5));
6782 excluded.assert_next_wait();
6783 video.assert_next_wait();
6784
6785 drop(video_route);
6786 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(9));
6787 assert_eq!(rooted.assert_next_active("").cost, Cost::new(9));
6788 excluded.assert_next_ended("room/live");
6789 video.assert_next_ended("room/live");
6790 }
6791
6792 #[tokio::test]
6793 async fn root_cursor_keeps_the_more_specific_covering_route() {
6794 let producer = origin(1).produce();
6795 let broad = producer.dynamic("room", Route::default().with_cost(1)).unwrap();
6796 let narrow = producer.dynamic("room/live", Route::default().with_cost(9)).unwrap();
6797 let mut announced = producer
6798 .consume()
6799 .scope("room/live/video", &Patterns::from(Pattern::all()))
6800 .unwrap()
6801 .announced();
6802 assert_eq!(announced.assert_next_active("").cost, Cost::new(9));
6803
6804 broad.update(Route::default().with_cost(0)).unwrap();
6805 announced.assert_next_wait();
6806 narrow.update(Route::default().with_cost(8)).unwrap();
6807 assert_eq!(announced.assert_next_active("").cost, Cost::new(8));
6808 drop(narrow);
6809 assert_eq!(announced.assert_next_active("").cost, Cost::new(0));
6810 }
6811
6812 #[tokio::test]
6813 async fn dynamic_accepts_a_max_depth_prefix() {
6814 let producer = origin(1).produce();
6815 let path = (0..Path::MAX_PARTS)
6816 .map(|i| format!("s{i}"))
6817 .collect::<Vec<_>>()
6818 .join("/");
6819 let mut announced = producer.consume().announced();
6820
6821 let dynamic = producer.dynamic(&path, Route::default()).expect("max depth is allowed");
6822 announced.assert_next_active(&path);
6823
6824 let _pending = producer.consume().request_broadcast(&path);
6825 let request = queued(&dynamic).await;
6826 assert_eq!(request.path().as_str(), path);
6827 }
6828
6829 #[tokio::test]
6830 async fn dynamic_exclusion_skips_routes_through_the_subscriber() {
6831 let producer = origin(1).produce();
6832 let _server = producer
6833 .dynamic("live", Route::default().with_hops(hops(&[7])))
6834 .unwrap();
6835
6836 let mut excluded = producer.consume().excluding(origin(7)).announced();
6837 excluded.assert_next_wait();
6838
6839 let mut clean = producer.consume().excluding(origin(8)).announced();
6840 clean.assert_next_active("live");
6841 }
6842
6843 #[tokio::test]
6845 async fn announce_consumer_is_a_stream() {
6846 use futures::StreamExt;
6847 let producer = origin(1).produce();
6848 let server = producer.dynamic("live", Route::default()).unwrap();
6849 let mut announced = producer.consume().announced();
6850 let update = StreamExt::next(&mut announced)
6851 .now_or_never()
6852 .expect("next")
6853 .expect("no next");
6854 assert_eq!(update.prefix.as_str(), "live");
6855 assert_eq!(update.kind, AnnounceKind::Announced);
6856 assert!(StreamExt::next(&mut announced).now_or_never().is_none());
6857 drop(server);
6858 let update = StreamExt::next(&mut announced)
6859 .now_or_never()
6860 .expect("next")
6861 .expect("no next");
6862 assert_eq!(update.kind, AnnounceKind::Retracted);
6863 }
6864
6865 #[tokio::test]
6866 async fn dynamic_retracts() {
6867 let producer = origin(1).produce();
6868 let server = producer.dynamic("live", Route::default()).unwrap();
6869 let mut announced = producer.consume().announced();
6870 announced.assert_next_active("live");
6871
6872 drop(server);
6873 announced.assert_next_ended("live");
6874 }
6875
6876 #[test]
6877 fn charged_wildcard_cost_accumulates_across_hops() {
6878 let first = Cost::new(4).charged(1);
6879 let second = first.charged(2);
6880 assert_eq!(second, Cost { warm: 7, cold: 7 });
6881 }
6882
6883 #[tokio::test]
6884 async fn local_broadcast_is_invisible_until_announced() {
6885 let producer = origin(1).produce();
6886 let mut local = producer.consume().announced();
6887 let mut peer = producer.consume().excluding(Hop::UNKNOWN).announced();
6888 let broadcast = producer.create_broadcast("room/alice").unwrap();
6889 local.assert_next_wait();
6890 peer.assert_next_wait();
6891
6892 broadcast.announce(Route::default()).unwrap();
6893 local.assert_next_active("room/alice");
6894 peer.assert_next_active("room/alice");
6895
6896 drop(broadcast);
6897 local.assert_next_ended("room/alice");
6898 peer.assert_next_ended("room/alice");
6899 }
6900
6901 #[tokio::test]
6902 async fn served_route_materializes_on_demand() {
6903 let producer = origin(1).produce();
6904 let consumer = producer.consume();
6905
6906 let server = producer.dynamic("room", Route::default()).unwrap();
6907
6908 let pending = consumer.request_broadcast("room/alice");
6909 let request = queued(&server).await;
6910 assert_eq!(request.path().as_str(), "room/alice");
6911
6912 let source = broadcast::Info::new().produce();
6913 request.accept(&source);
6914
6915 let resolved = pending.await.expect("resolves");
6916 assert_eq!(resolved.info().path.as_str(), "room/alice");
6918
6919 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6921 assert!(again.is_clone(&resolved));
6922 }
6923
6924 #[tokio::test]
6925 async fn served_requests_coalesce() {
6926 let producer = origin(1).produce();
6927 let consumer = producer.consume();
6928 let server = producer.dynamic("room", Route::default()).unwrap();
6929
6930 let first = consumer.request_broadcast("room/alice");
6931 let second = consumer.request_broadcast("room/alice");
6932
6933 let request = queued(&server).await;
6934 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
6936
6937 let source = broadcast::Info::new().produce();
6938 request.accept(&source);
6939
6940 let first = first.await.expect("resolves");
6941 let second = second.await.expect("resolves");
6942 assert!(first.is_clone(&second));
6943 }
6944
6945 #[tokio::test]
6946 async fn retract_rejects_pending_requests() {
6947 let producer = origin(1).produce();
6948 let consumer = producer.consume();
6949 let server = producer.dynamic("room", Route::default()).unwrap();
6950
6951 let pending = consumer.request_broadcast("room/alice");
6952 drop(server);
6953
6954 let err = pending.await.err().unwrap();
6955 assert!(matches!(err, Error::Unroutable));
6956
6957 let err = consumer
6959 .request_broadcast("room/alice")
6960 .now_or_never()
6961 .expect("unroutable")
6962 .err()
6963 .unwrap();
6964 assert!(matches!(err, Error::Unroutable));
6965 }
6966
6967 #[tokio::test]
6968 async fn routed_broadcast_survives_serving_route_retraction() {
6969 let producer = origin(1).produce();
6970 let consumer = producer.consume();
6971
6972 let standby_server = producer.dynamic("room", Route::default()).unwrap();
6975 let second_server = producer.dynamic("room", Route::default()).unwrap();
6976 let incumbent_server = producer.dynamic("room", Route::default()).unwrap();
6977
6978 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
6979 assert!((&mut resolving).now_or_never().is_none());
6980
6981 drop(incumbent_server);
6987 assert!((&mut resolving).now_or_never().is_none());
6988 drop(second_server);
6989 assert!((&mut resolving).now_or_never().is_none());
6990
6991 let request = queued(&standby_server).await;
6992 let source = broadcast::Info::new().produce();
6993 request.accept(&source);
6994
6995 let resolved = resolving.await.expect("resolves via the standby");
6996 assert_eq!(resolved.info().path.as_str(), "room/alice");
6997 }
6998
6999 #[tokio::test]
7000 async fn split_horizon_skips_routes_through_the_requester() {
7001 let producer = origin(1).produce();
7002 let _server = producer
7003 .dynamic("room", Route::default().with_hops(hops(&[7])))
7004 .unwrap();
7005
7006 let excluded = producer.consume().excluding(origin(7));
7008 let err = excluded
7009 .request_broadcast("room/alice")
7010 .now_or_never()
7011 .expect("unroutable")
7012 .err()
7013 .unwrap();
7014 assert!(matches!(err, Error::Unroutable));
7015
7016 let clean = producer.consume().excluding(origin(8));
7018 let pending = clean.request_broadcast("room/alice");
7019 assert!(pending.now_or_never().is_none());
7020 }
7021
7022 #[tokio::test(start_paused = true)]
7026 async fn viewer_sessions_share_the_plain_front() {
7027 let producer = origin(1).produce();
7028 let server = producer
7030 .dynamic("room", Route::default().with_hops(hops(&[10])))
7031 .unwrap();
7032 let upstream = broadcast::Info::new().produce();
7033 let source = upstream.create_track("video", None).unwrap();
7034 let mut group = source.append_group().unwrap();
7035 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
7036
7037 let counts = || {
7038 let table = producer.shared.lock();
7039 let watches = table
7040 .routes
7041 .root
7042 .find(Path::new("room/alice").parts())
7043 .map_or(0, |node| node.watches.len());
7044 (table.fronts.len(), watches)
7045 };
7046
7047 let mut first = None;
7048 let mut front = None;
7049 for viewer in 0..1000 {
7050 let session = producer.consume().excluding(Hop::random());
7051 let pending = session.request_broadcast("room/alice");
7052 if viewer == 0 {
7053 queued(&server).await.accept(&upstream);
7054 }
7055 let resolved = pending.await.expect("resolves");
7056 let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap();
7057 next_group(&mut subscription).await.unwrap().expect("the live group");
7058 drop(subscription);
7059 first.get_or_insert_with(counts);
7060 front = Some(resolved);
7061 }
7062 assert_eq!(first, Some((1, 1)), "one front and one watch for the first viewer");
7063 assert_eq!(counts(), (1, 1), "viewers minted fronts of their own");
7064 drop(front);
7065 }
7066
7067 #[tokio::test]
7071 async fn a_hop_in_a_chain_gets_its_own_front() {
7072 let producer = origin(1).produce();
7073 let _incumbent = producer
7074 .dynamic("room", Route::default().with_hops(hops(&[10])))
7075 .unwrap();
7076 let fronts = || producer.shared.lock().fronts.len();
7077
7078 let peer = producer.consume().excluding(origin(7));
7079 let _local = producer.consume().request_broadcast("room/alice");
7080 let _peer = peer.request_broadcast("room/alice");
7081 assert_eq!(fronts(), 1, "a hop no chain names excludes nothing");
7082
7083 let _echo = producer
7085 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7086 .unwrap();
7087 let _filtered = peer.request_broadcast("room/alice");
7088 assert_eq!(fronts(), 2, "the peer still shares the plain front");
7089 let _viewer = producer.consume().excluding(origin(8)).request_broadcast("room/alice");
7090 assert_eq!(fronts(), 2, "a viewer left the plain front");
7091 }
7092
7093 #[tokio::test]
7097 async fn a_hairpin_through_the_plain_front_is_withdrawn() {
7098 let producer = origin(1).produce();
7099 let peer = producer.consume().excluding(origin(7));
7100 let mut view = peer.announced();
7101 let incumbent = producer
7102 .dynamic("room", Route::default().with_hops(hops(&[10])))
7103 .unwrap();
7104 view.assert_next_active("room");
7105
7106 let pending = peer.request_broadcast("room/alice");
7107 let source = broadcast::Info::new().produce();
7108 queued(&incumbent).await.accept(&source);
7109 let resolved = pending.await.expect("resolves through the incumbent");
7110
7111 let echo = producer
7113 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7114 .unwrap();
7115 drop(incumbent);
7116 drop(source);
7117
7118 let replacement = broadcast::Info::new().produce();
7119 queued(&echo).await.accept(&replacement);
7120 resolved.assert_not_closed();
7121 view.assert_next_ended("room");
7122 }
7123
7124 #[tokio::test]
7125 async fn routes_report_where_they_entered() {
7126 let producer = origin(1).produce();
7127 let peer = producer.clone().peer();
7128 let mut announced = producer.consume().announced();
7129
7130 let _ingest = producer
7131 .dynamic("client", Route::default().with_hops(hops(&[5])).with_via(origin(5)))
7132 .unwrap();
7133 let _gateway = producer.publish("gateway", Route::default()).unwrap();
7134 let _forwarded = peer
7135 .dynamic(
7136 "forwarded",
7137 Route::default().with_hops(hops(&[5, 7])).with_via(origin(7)),
7138 )
7139 .unwrap();
7140
7141 assert_eq!(announced.assert_next_active("client").source(), Source::Local);
7142 assert_eq!(
7143 announced.assert_next_active("forwarded").source(),
7144 Source::Peer(origin(7))
7145 );
7146 assert_eq!(announced.assert_next_active("gateway").source(), Source::Local);
7147
7148 let scoped = peer.scope("room", &Patterns::from(Pattern::all())).unwrap();
7150 let _nested = scoped.dynamic("x", Route::default().with_via(origin(8))).unwrap();
7151 assert_eq!(announced.assert_next_active("room/x").source(), Source::Peer(origin(8)));
7152 }
7153
7154 #[tokio::test(start_paused = true)]
7158 async fn try_next_delivers_a_held_update() {
7159 async fn step(by: Duration) {
7161 tokio::time::advance(by).await;
7162 for _ in 0..10 {
7163 tokio::task::yield_now().await;
7164 }
7165 }
7166
7167 let producer = origin(1).produce();
7168 let peer = producer.clone().peer();
7169 let mut announced = producer.consume().announced();
7170 step(Duration::from_millis(1)).await;
7171
7172 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7173 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7174 assert!(announced.try_next().unwrap().kind.is_active());
7175
7176 let _far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7177 drop(near);
7178 assert!(announced.try_next().is_none(), "the update is held");
7179 step(Duration::from_millis(1)).await;
7180 assert!(announced.try_next().is_none(), "the update is still held");
7181 step(DEFAULT_UPDATE_HOLD).await;
7182 let update = announced.try_next().expect("the hold passed");
7183 assert_eq!(update.kind, AnnounceKind::Updated);
7184 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7185 }
7186
7187 #[tokio::test(start_paused = true)]
7190 async fn a_changed_route_waits_out_the_hold() {
7191 let producer = origin(1).produce();
7192 let peer = producer.clone().peer();
7193 let mut announced = producer.consume().announced();
7194 tokio::time::sleep(Duration::from_millis(1)).await;
7196
7197 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7198 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7199 announced.assert_next_active("room");
7200
7201 let far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7203 let farther = peer.dynamic("room", route(&[9, 5, 6, 7], 7)).unwrap();
7204 let start = tokio::time::Instant::now();
7205 drop(near);
7206 drop(far);
7207 let update = announced.next().await.unwrap();
7208 assert_eq!(update.kind, AnnounceKind::Updated);
7209 assert_eq!(update.route.hops, hops(&[9, 5, 6, 7]));
7210 assert_eq!(start.elapsed(), DEFAULT_UPDATE_HOLD);
7211 announced.assert_next_wait();
7212
7213 drop(farther);
7215 announced.assert_next_ended("room");
7216 }
7217
7218 #[tokio::test(start_paused = true)]
7221 async fn a_retraction_quarantines_a_named_publisher() {
7222 let producer = origin(1).produce();
7223 let peer = producer.clone().peer();
7224 let mut announced = producer.consume().announced();
7225 tokio::time::sleep(Duration::from_millis(1)).await;
7226
7227 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7228
7229 let near = peer.dynamic("named", route(&[9, 2], 2)).unwrap();
7230 announced.assert_next_active("named");
7231 drop(near);
7232 announced.assert_next_ended("named");
7233 let start = tokio::time::Instant::now();
7234 let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap();
7235 let update = announced.next().await.unwrap();
7236 assert_eq!(update.kind, AnnounceKind::Announced);
7237 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7238 assert!(start.elapsed() >= RETRACT_QUARANTINE);
7239
7240 let near = peer.dynamic("anonymous", route(&[0, 2], 2)).unwrap();
7241 announced.assert_next_active("anonymous");
7242 drop(near);
7243 announced.assert_next_ended("anonymous");
7244 let _other = peer.dynamic("anonymous", route(&[0, 3, 4], 4)).unwrap();
7245 announced.assert_next_active("anonymous");
7246
7247 let queued = announced.anchoring.len();
7250 for chain in [[9, 2], [0, 2], [9, 2], [0, 2]] {
7251 let churn = peer.dynamic("churn", route(&chain, 2)).unwrap();
7252 announced.assert_next_active("churn");
7253 drop(churn);
7254 announced.assert_next_ended("churn");
7255 }
7256 assert_eq!(announced.anchoring.len(), queued + 1);
7257
7258 let gone = peer.dynamic("gone", route(&[9, 2], 2)).unwrap();
7261 announced.assert_next_active("gone");
7262 drop(gone);
7263 announced.assert_next_ended("gone");
7264 assert!(!announced.retracted.is_empty());
7265 for _ in 0..2 {
7266 announced.assert_next_wait();
7267 tokio::time::sleep(RETRACT_QUARANTINE).await;
7268 }
7269 announced.assert_next_wait();
7270 assert!(announced.retracted.is_empty(), "the ended quarantine is still held");
7271 }
7272
7273 #[tokio::test]
7277 async fn withdrawn_peer_hides_routes_through_it() {
7278 let producer = origin(1).produce();
7279 let peer = producer.clone().peer();
7280 let mut announced = producer.consume().announced();
7281
7282 let direct = || Route::default().with_hops(hops(&[9, 2])).with_via(origin(2));
7284 let first = peer.dynamic("room", direct()).unwrap();
7285 let relayed = peer
7286 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])).with_via(origin(3)))
7287 .unwrap();
7288 announced.assert_next_active("room");
7289 announced.assert_next_wait();
7290
7291 first.withdrawn();
7293 announced.assert_next_ended("room");
7294 announced.assert_next_wait();
7295
7296 let second = peer.dynamic("room", direct()).unwrap();
7298 announced.assert_next_active("room");
7299 assert!(producer.shared.lock().withdrawn.is_empty());
7300
7301 second.withdrawn();
7303 drop(relayed);
7304 announced.assert_next_ended("room");
7305 assert!(producer.shared.lock().withdrawn.is_empty());
7306 }
7307
7308 #[tokio::test]
7310 async fn old_session_withdrawal_keeps_newer_route() {
7311 for restart in [false, true] {
7312 let producer = origin(1).produce();
7313 let peer = producer.clone().peer();
7314 let mut announced = producer.consume().announced();
7315 let route = Route::default().with_hops(hops(&[9, 2]));
7316 let first = peer.dynamic("room", route.clone()).unwrap();
7317 let second = peer.dynamic("room", route.clone()).unwrap();
7318 announced.assert_next_active("room");
7319 announced.assert_next_wait();
7320 let remaining = if restart {
7321 first.update(route).unwrap();
7323 second.withdrawn();
7324 first
7325 } else {
7326 first.withdrawn();
7327 second
7328 };
7329 announced.assert_next_wait();
7330 assert!(producer.shared.lock().withdrawn.is_empty());
7331 remaining.withdrawn();
7332 announced.assert_next_ended("room");
7333 assert!(producer.shared.lock().withdrawn.is_empty());
7334 }
7335 }
7336
7337 #[tokio::test]
7338 async fn withdrawn_route_no_longer_covers_requests() {
7339 let producer = origin(1).produce();
7340 let peer = producer.clone().peer();
7341 let direct = peer.dynamic("room", Route::default().with_hops(hops(&[9, 2]))).unwrap();
7342 let _relayed = peer
7343 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])))
7344 .unwrap();
7345 let path = Path::new("room/video");
7346 let id = producer
7347 .shared
7348 .read()
7349 .routes
7350 .covering(&path)
7351 .find(|entry| entry.hops.iter().last() == Some(&origin(3)))
7352 .unwrap()
7353 .id;
7354 assert!(producer.shared.read().routes.covers(&path, id));
7355 direct.withdrawn();
7356 assert!(!producer.shared.read().routes.covers(&path, id));
7357 }
7358
7359 #[tokio::test]
7362 async fn source_change_is_an_update() {
7363 let producer = origin(1).produce();
7364 let peer = producer.clone().peer();
7365 let mut announced = producer.consume().announced();
7366
7367 let route = Route::default().with_hops(hops(&[7])).with_via(origin(7));
7368 let _forwarded = peer.dynamic("room", route.clone()).unwrap();
7369 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7370
7371 let local = producer.dynamic("room", route).unwrap();
7373 let update = announced.next().now_or_never().expect("next blocked").expect("no next");
7374 assert_eq!(update.kind, AnnounceKind::Updated);
7375 assert_eq!(update.route.source(), Source::Local);
7376
7377 drop(local);
7378 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7379 }
7380
7381 #[tokio::test]
7382 async fn local_view_hides_peer_routes() {
7383 let producer = origin(1).produce();
7384 let peer = producer.clone().peer();
7385 let mut local = producer.consume().local().announced();
7386
7387 let _forwarded = peer
7388 .dynamic("remote", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7389 .unwrap();
7390 local.assert_next_wait();
7391
7392 let _shadow = peer
7396 .dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7397 .unwrap();
7398 let ingest = producer
7399 .dynamic(
7400 "both",
7401 Route::default().with_hops(hops(&[5])).with_via(origin(5)).with_cost(9),
7402 )
7403 .unwrap();
7404 assert_eq!(local.assert_next_active("both").source(), Source::Local);
7405 drop(ingest);
7406 local.assert_next_ended("both");
7407
7408 let err = producer
7411 .consume()
7412 .local()
7413 .request_broadcast("remote/alice")
7414 .now_or_never()
7415 .expect("unroutable")
7416 .err()
7417 .unwrap();
7418 assert!(matches!(err, Error::Unroutable));
7419 assert!(
7420 producer
7421 .consume()
7422 .request_broadcast("remote/alice")
7423 .now_or_never()
7424 .is_none()
7425 );
7426 }
7427
7428 #[tokio::test]
7432 async fn handler_rejection_is_final() {
7433 let producer = origin(1).produce();
7434 let consumer = producer.consume();
7435 let server = producer.dynamic("room", Route::default()).unwrap();
7436
7437 let pending = consumer.request_broadcast("room/alice");
7438 let request = queued(&server).await;
7439 request.reject(Error::Unroutable);
7440 let err = tokio::time::timeout(Duration::from_secs(5), pending)
7441 .await
7442 .expect("the front must give up, not spin")
7443 .err()
7444 .unwrap();
7445 assert!(matches!(err, Error::Unroutable));
7446
7447 let pending = consumer.request_broadcast("room/bob");
7449 let request = queued(&server).await;
7450 assert_eq!(request.path().as_str(), "room/bob");
7451 let served = broadcast::Info::new().produce();
7452 request.accept(&served);
7453 pending.await.expect("resolves");
7454 }
7455
7456 #[tokio::test]
7459 async fn routed_broadcast_waits_out_a_rejection() {
7460 let producer = origin(1).produce();
7461 let consumer = producer.consume();
7462 let server = producer.dynamic("room", Route::default()).unwrap();
7463
7464 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7465 assert!((&mut resolving).now_or_never().is_none());
7466 let request = queued(&server).await;
7467 request.reject(Error::Unroutable);
7468
7469 for _ in 0..20 {
7471 tokio::task::yield_now().await;
7472 }
7473 assert!((&mut resolving).now_or_never().is_none());
7474 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7475
7476 server.update(Route::default().with_cost(2)).unwrap();
7478 assert!((&mut resolving).now_or_never().is_none());
7479 let request = queued(&server).await;
7480 let served = broadcast::Info::new().produce();
7481 request.accept(&served);
7482 resolving.await.expect("resolves");
7483 }
7484
7485 #[tokio::test]
7488 async fn routed_broadcast_reports_teardown_as_closed() {
7489 let (producer, driver) = Producer::new(Config::new(origin(1)));
7490 let consumer = producer.consume();
7491 let _server = producer.dynamic("room", Route::default()).unwrap();
7492
7493 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7495 assert!((&mut resolving).now_or_never().is_none());
7496
7497 drop(driver);
7498
7499 let err = tokio::time::timeout(Duration::from_secs(5), resolving)
7500 .await
7501 .expect("teardown resolves the wait")
7502 .err()
7503 .unwrap();
7504 assert!(matches!(err, Error::Closed), "unexpected end: {err}");
7505 }
7506
7507 #[tokio::test]
7510 async fn routed_broadcast_wakes_for_a_local_broadcast() {
7511 let producer = origin(1).produce();
7512 let consumer = producer.consume();
7513 let server = producer.dynamic("room", Route::default()).unwrap();
7514
7515 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7516 assert!((&mut resolving).now_or_never().is_none());
7517 queued(&server).await.reject(Error::Unroutable);
7518 for _ in 0..20 {
7519 tokio::task::yield_now().await;
7520 }
7521 assert!((&mut resolving).now_or_never().is_none());
7522
7523 let _local = producer.publish("room/alice", Route::default()).unwrap();
7525 let resolved = resolving.await.expect("resolves locally");
7526 assert_eq!(resolved.info().path.as_str(), "room/alice");
7527 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7528 }
7529
7530 #[tokio::test]
7533 async fn late_track_on_a_served_front_replays() {
7534 let producer = origin(1).produce();
7535 let consumer = producer.consume();
7536 let server = producer.dynamic("room", Route::default()).unwrap();
7537
7538 let source = broadcast::Info::new().produce();
7539 for name in ["a", "b"] {
7540 let track = source.create_track(name, None).unwrap();
7541 let mut group = track.append_group().unwrap();
7542 group.write_frame(crate::Timestamp::ZERO, name.as_bytes()).unwrap();
7543 group.finish().unwrap();
7544 std::mem::forget(track);
7546 }
7547
7548 let pending = consumer.request_broadcast("room/alice");
7549 queued(&server).await.accept(&source);
7550 let resolved = pending.await.expect("resolves");
7551
7552 let budget = track::Subscription::default().with_max_age(Duration::from_secs(3600));
7553 for name in ["a", "b"] {
7554 let mut subscription = resolved
7555 .track(name)
7556 .unwrap()
7557 .subscribe(budget.clone())
7558 .await
7559 .expect("subscribe");
7560 let mut group = tokio::time::timeout(Duration::from_secs(5), subscription.recv_group())
7561 .await
7562 .expect("the late track must replay, not park")
7563 .expect("recv group")
7564 .expect("track ended early");
7565 let frame = group.read_frame().await.expect("read frame").expect("frame");
7566 assert_eq!(&frame.payload[..], name.as_bytes());
7567 }
7568 }
7569
7570 #[tokio::test]
7571 async fn most_specific_prefix_shadows() {
7572 let producer = origin(1).produce();
7573 let consumer = producer.consume();
7574
7575 let broad_server = producer.dynamic("", Route::default()).unwrap();
7576 let _narrow = producer.announce(".dash", Route::default()).unwrap();
7579
7580 let err = consumer
7581 .request_broadcast(".dash/pid")
7582 .now_or_never()
7583 .expect("unroutable")
7584 .err()
7585 .unwrap();
7586 assert!(matches!(err, Error::Unroutable));
7587
7588 let _pending = consumer.request_broadcast("room/alice");
7590 let request = queued(&broad_server).await;
7591 assert_eq!(request.path().as_str(), "room/alice");
7592 }
7593
7594 #[tokio::test]
7595 async fn root_dynamic_serves_any_path() {
7596 let producer = origin(1).produce();
7597 let consumer = producer.consume();
7598 let mut announced = consumer.announced();
7599 let dynamic = producer.dynamic("", Route::default()).unwrap();
7600 announced.assert_next_active("");
7602
7603 let pending = consumer.request_broadcast("anything/at/all");
7604 let request = queued(&dynamic).await;
7605 assert_eq!(request.path().as_str(), "anything/at/all");
7606
7607 let source = broadcast::Info::new().produce();
7608 request.accept(&source);
7609 let resolved = pending.await.expect("resolves");
7610 assert_eq!(resolved.info().path.as_str(), "anything/at/all");
7611
7612 drop(dynamic);
7614 announced.assert_next_ended("");
7615 let err = consumer
7616 .request_broadcast("something/else")
7617 .now_or_never()
7618 .expect("unroutable")
7619 .err()
7620 .unwrap();
7621 assert!(matches!(err, Error::Unroutable));
7622 }
7623
7624 #[tokio::test]
7630 async fn out_of_scope_request_never_reaches_the_dynamic_handler() {
7631 let producer = origin(1).produce();
7632 let dynamic = producer.dynamic("", Route::default()).unwrap();
7633 let scoped = producer.consume().scope("", &scopes(&["tenant-a"])).unwrap();
7634
7635 for path in ["tenant-b/live", "tenant-a-other/live"] {
7638 let refused = scoped
7639 .request_broadcast(path)
7640 .now_or_never()
7641 .expect("an out-of-scope request must be refused synchronously, not queued");
7642 assert!(matches!(refused, Err(Error::Unauthorized)));
7643 assert!(
7644 dynamic.requested_broadcast().now_or_never().is_none(),
7645 "the dynamic handler was asked to create a broadcast the requester may not read"
7646 );
7647 }
7648 }
7649
7650 #[tokio::test]
7651 async fn routed_waits_for_coverage() {
7652 let producer = origin(1).produce();
7653 let consumer = producer.consume();
7654
7655 let mut fut = consumer.routed("room/alice").boxed();
7656 assert!((&mut fut).now_or_never().is_none());
7657
7658 let _a = producer.announce("room", Route::default().with_cost(3)).unwrap();
7660 let route = fut.now_or_never().expect("covered").expect("routed");
7661 assert_eq!(route.cost, Cost::new(3));
7662
7663 consumer
7665 .routed("room/alice/cam")
7666 .now_or_never()
7667 .expect("covered")
7668 .expect("routed");
7669 }
7670
7671 #[tokio::test]
7672 async fn routed_ignores_deeper_routes() {
7673 let producer = origin(1).produce();
7674 let consumer = producer.consume();
7675
7676 let _deep = producer.announce("room/alice/cam", Route::default()).unwrap();
7678 let mut fut = consumer.routed("room/alice").boxed();
7679 assert!((&mut fut).now_or_never().is_none());
7680
7681 let _exact = producer.announce("room/alice", Route::default()).unwrap();
7682 fut.now_or_never().expect("covered").expect("routed");
7683 }
7684
7685 #[tokio::test]
7686 async fn routed_accepts_a_max_depth_path() {
7687 let producer = origin(1).produce();
7688 let consumer = producer.consume();
7689 let path = (0..Path::MAX_PARTS)
7690 .map(|i| format!("s{i}"))
7691 .collect::<Vec<_>>()
7692 .join("/");
7693 assert_eq!(Path::new(&path).parts().count(), Path::MAX_PARTS);
7694
7695 assert!(consumer.allowed().matches(&path));
7696
7697 let mut fut = consumer.routed(&path).boxed();
7698 assert!((&mut fut).now_or_never().is_none());
7699
7700 let _a = producer.announce("", Route::default()).unwrap();
7702 fut.now_or_never().expect("covered").expect("routed");
7703 }
7704
7705 #[tokio::test]
7706 async fn teardown_ends_everything() {
7707 let (producer, driver) = Producer::new(Config::new(origin(1)));
7708 let consumer = producer.consume();
7709 let _announcement = producer.announce("room", Route::default()).unwrap();
7710 let mut announced = consumer.announced();
7711 announced.assert_next_active("room");
7712
7713 let _server = producer.dynamic("served", Route::default()).unwrap();
7714 let pending = consumer.request_broadcast("served/path");
7715
7716 drop(driver);
7717
7718 announced.assert_next_active("served");
7720 assert!(announced.next().now_or_never().expect("ended").is_none());
7721
7722 assert!(pending.now_or_never().expect("rejected").is_err());
7724 assert!(matches!(producer.announce("x", Route::default()), Err(Error::Closed)));
7725 assert!(matches!(producer.create_broadcast("x"), Err(Error::Closed)));
7726 let err = consumer
7727 .request_broadcast("y")
7728 .now_or_never()
7729 .expect("closed")
7730 .err()
7731 .unwrap();
7732 assert!(matches!(err, Error::Closed));
7733
7734 let mut late = consumer.announced();
7736 assert!(late.next().now_or_never().expect("ended").is_none());
7737 }
7738
7739 struct ResumeRig {
7742 producer: Producer,
7743 resolved: broadcast::Consumer,
7744 subscription: track::Subscriber,
7745 incumbent_track: track::Producer,
7748 }
7749
7750 impl ResumeRig {
7751 async fn new(first: &[u64]) -> (Self, Dynamic, broadcast::Producer) {
7754 let producer = origin(1).produce();
7755 let consumer = producer.consume();
7756
7757 let server = producer
7758 .dynamic("room", Route::default().with_hops(hops(first)))
7759 .unwrap();
7760
7761 let pending = consumer.request_broadcast("room/alice");
7762 let request = queued(&server).await;
7763 let source = broadcast::Info::new().produce();
7764 let track = source.create_track("video", None).unwrap();
7765 let mut group = track.append_group().unwrap();
7766 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7767 group.finish().unwrap();
7768 request.accept(&source);
7769
7770 let resolved = pending.await.expect("resolves");
7771 let mut subscription = resolved
7772 .track("video")
7773 .unwrap()
7774 .subscribe(None)
7775 .await
7776 .expect("subscribe");
7777 let mut group = subscription
7778 .recv_group()
7779 .await
7780 .expect("recv group")
7781 .expect("track ended early");
7782 let frame = group.read_frame().await.expect("read frame").expect("frame");
7783 assert_eq!(&frame.payload[..], b"before");
7784
7785 (
7786 Self {
7787 producer,
7788 resolved,
7789 subscription,
7790 incumbent_track: track,
7791 },
7792 server,
7793 source,
7794 )
7795 }
7796
7797 fn standby(&self, first: &[u64]) -> Dynamic {
7800 self.producer
7801 .dynamic("room", Route::default().with_hops(hops(first)))
7802 .unwrap()
7803 }
7804 }
7805
7806 async fn assert_resumes(rig: &mut ResumeRig, server: &Dynamic) {
7811 let request = queued(server).await;
7812 let replacement = broadcast::Info::new().produce();
7813 let track = replacement.create_track("video", None).unwrap();
7814 let mut group = track.append_group().unwrap();
7817 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7818 group.finish().unwrap();
7819 request.accept(&replacement);
7820
7821 let mut group = track.append_group().unwrap();
7822 group.write_frame(crate::Timestamp::ZERO, b"resumed".as_ref()).unwrap();
7823 group.finish().unwrap();
7824
7825 let mut group = rig
7826 .subscription
7827 .recv_group()
7828 .await
7829 .expect("subscription survives the failover")
7830 .expect("track ended early");
7831 let frame = group.read_frame().await.expect("read frame").expect("frame");
7832 assert_eq!(&frame.payload[..], b"resumed");
7833 }
7834
7835 #[tokio::test]
7838 async fn driver_resolves_with_live_consumers() {
7839 let (producer, driver) = Producer::new(Config::new(origin(1)));
7840 let consumer = producer.consume();
7841 let run = crate::time::run(driver);
7842 drop(producer);
7843 tokio::time::timeout(Duration::from_secs(5), run)
7844 .await
7845 .expect("driver must finish once the producers are gone");
7846 drop(consumer);
7847 }
7848
7849 #[tokio::test]
7850 async fn remote_source_resumes_through_same_first_hop() {
7851 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7852 let standby_server = rig.standby(&[10, 20]);
7853
7854 drop(incumbent);
7856 drop(source);
7857
7858 assert_resumes(&mut rig, &standby_server).await;
7860 }
7861
7862 #[tokio::test]
7866 async fn incompatible_successor_is_refused() {
7867 for replacement in [
7868 track::Info::default().with_timescale(crate::Timescale::MICRO),
7869 track::Info::default().with_priority(7),
7870 track::Info::default().with_max_age(Duration::from_secs(7)),
7871 ] {
7872 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7873 let standby_server = rig.standby(&[10, 20]);
7874 drop(incumbent);
7875 drop(source);
7876
7877 let request = queued(&standby_server).await;
7880 let successor = broadcast::Info::new().produce();
7881 let track = successor.create_track("video", replacement).unwrap();
7882 let mut group = track.append_group().unwrap();
7883 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7884 group.finish().unwrap();
7885 request.accept(&successor);
7886
7887 assert!(
7888 matches!(rig.subscription.recv_group().await, Err(Error::Unsupported)),
7889 "the subscription must abort rather than resume onto incompatible metadata"
7890 );
7891
7892 let reopened = rig.resolved.track("video").unwrap();
7894 assert!(matches!(reopened.query().await, Err(Error::Unsupported)));
7895 assert!(matches!(reopened.subscribe(None).await, Err(Error::Unsupported)));
7896 }
7897 }
7898
7899 #[tokio::test]
7900 async fn different_first_hop_ends_the_subscription() {
7901 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7902 let rival_server = rig.standby(&[11]);
7904
7905 drop(incumbent);
7908 drop(source);
7909 rig.incumbent_track.abort(Error::Dropped).unwrap();
7910
7911 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
7913 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
7914
7915 let consumer = rig.producer.consume();
7917 let pending = consumer.request_broadcast("room/alice");
7918 let request = queued(&rival_server).await;
7919 let replacement = broadcast::Info::new().produce();
7920 request.accept(&replacement);
7921 pending.await.expect("re-request resolves through the rival");
7922 }
7923
7924 #[tokio::test]
7929 async fn a_first_hop_update_drains_the_old_publisher() {
7930 for first in [&[][..], &[0][..], &[10][..]] {
7931 let (mut rig, server, _source) = ResumeRig::new(first).await;
7932 server.update(Route::default().with_hops(hops(&[11]))).unwrap();
7933
7934 let mut group = rig.incumbent_track.append_group().unwrap();
7936 group.write_frame(crate::Timestamp::ZERO, b"draining".as_ref()).unwrap();
7937 group.finish().unwrap();
7938 let mut group = next_group(&mut rig.subscription)
7939 .await
7940 .expect("the in-flight subscription survives the update")
7941 .expect("track ended early");
7942 let frame = group.read_frame().await.expect("read frame").expect("frame");
7943 assert_eq!(&frame.payload[..], b"draining", "first hop {first:?}");
7944
7945 let consumer = rig.producer.consume();
7948 let pending = consumer.request_broadcast("room/alice");
7949 let request = queued(&server).await;
7950 let replacement = broadcast::Info::new().produce();
7951 let track = replacement
7952 .create_track("video", track::Info::default().with_priority(7))
7953 .unwrap();
7954 let mut group = track.append_group().unwrap();
7955 group.write_frame(crate::Timestamp::ZERO, b"new".as_ref()).unwrap();
7956 group.finish().unwrap();
7957 request.accept(&replacement);
7958
7959 let resolved = pending.await.expect("resolves through the new publisher");
7960 assert!(
7961 !resolved.is_clone(&rig.resolved),
7962 "first hop {first:?} joined the old front"
7963 );
7964 let mut fresh = resolved
7965 .track("video")
7966 .unwrap()
7967 .subscribe(None)
7968 .await
7969 .expect("the old publisher's track info does not apply");
7970 let mut group = next_group(&mut fresh)
7971 .await
7972 .expect("recv group")
7973 .expect("track ended early");
7974 let frame = group.read_frame().await.expect("read frame").expect("frame");
7975 assert_eq!(&frame.payload[..], b"new");
7976
7977 rig.incumbent_track.finish().unwrap();
7980 let end = next_group(&mut rig.subscription).await;
7981 assert!(
7982 !matches!(end, Ok(Some(_))),
7983 "first hop {first:?} spliced the new publisher into a live subscription"
7984 );
7985 }
7986 }
7987
7988 #[tokio::test]
7991 async fn a_first_hop_update_resumes_through_the_same_publisher() {
7992 let (mut rig, incumbent, _source) = ResumeRig::new(&[10]).await;
7993 let standby_server = rig.standby(&[10, 20]);
7994
7995 incumbent.update(Route::default().with_hops(hops(&[11]))).unwrap();
7996
7997 assert_resumes(&mut rig, &standby_server).await;
7998 assert!(
7999 incumbent.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8000 "the front never asks the new publisher"
8001 );
8002 }
8003
8004 #[tokio::test]
8005 async fn anonymous_routes_never_resume() {
8006 let (mut rig, incumbent, source) = ResumeRig::new(&[]).await;
8009 let _twin_server = rig.standby(&[]);
8010
8011 drop(incumbent);
8012 drop(source);
8013 rig.incumbent_track.abort(Error::Dropped).unwrap();
8014
8015 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
8016 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8017 }
8018
8019 #[tokio::test]
8026 async fn anonymous_handoff_serves_the_newcomer_immediately() {
8027 let producer = origin(1).produce();
8028
8029 let server_a = producer
8031 .dynamic("room", Route::default().with_hops(hops(&[10])))
8032 .unwrap();
8033
8034 let consumer = producer.consume().excluding(origin(30));
8037 let pending = consumer.request_broadcast("room/alice");
8038 let request = queued(&server_a).await;
8039 let source_a = broadcast::Info::new().produce();
8040 let track_a = source_a.create_track("video", None).unwrap();
8041 let mut group = track_a.append_group().unwrap();
8042 group.write_frame(crate::Timestamp::ZERO, b"from-a".as_ref()).unwrap();
8043 group.finish().unwrap();
8044 request.accept(&source_a);
8045
8046 let resolved_a = pending.await.expect("resolves");
8047 let mut sub_a = resolved_a
8048 .track("video")
8049 .unwrap()
8050 .subscribe(None)
8051 .await
8052 .expect("subscribe");
8053 let mut group = sub_a
8054 .recv_group()
8055 .await
8056 .expect("recv group")
8057 .expect("track ended early");
8058 assert_eq!(
8059 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8060 b"from-a"
8061 );
8062
8063 drop(track_a);
8066 drop(source_a);
8067 drop(server_a);
8068
8069 let err = sub_a.recv_group().await.err().expect("front closed");
8071 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8072
8073 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8077 settle(|| {
8078 matches!(
8079 consumer.request_broadcast("room/alice").now_or_never(),
8080 Some(Err(Error::Unroutable))
8081 )
8082 })
8083 .await;
8084
8085 let server_b = producer
8087 .dynamic("room", Route::default().with_hops(hops(&[20])))
8088 .unwrap();
8089 let pending = consumer.request_broadcast("room/alice");
8090 let request = queued(&server_b).await;
8091 let source_b = broadcast::Info::new().produce();
8092 let track_b = source_b.create_track("video", None).unwrap();
8093 let mut group = track_b.append_group().unwrap();
8094 group.write_frame(crate::Timestamp::ZERO, b"from-b".as_ref()).unwrap();
8095 group.finish().unwrap();
8096 request.accept(&source_b);
8097
8098 let resolved_b = pending.await.expect("B's front is served immediately");
8099 assert!(
8100 !resolved_b.is_clone(&resolved_a),
8101 "B must not splice into A's closed front"
8102 );
8103
8104 let mut sub_b = resolved_b
8105 .track("video")
8106 .unwrap()
8107 .subscribe(None)
8108 .await
8109 .expect("subscribe");
8110 let mut group = sub_b
8111 .recv_group()
8112 .await
8113 .expect("recv group")
8114 .expect("track ended early");
8115 assert_eq!(
8116 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8117 b"from-b"
8118 );
8119 }
8120
8121 #[tokio::test]
8122 async fn reprice_is_invisible_to_the_subscription() {
8123 let (rig, incumbent, source) = ResumeRig::new(&[10]).await;
8124
8125 incumbent
8128 .update(Route::default().with_hops(hops(&[10])).with_cost(9))
8129 .unwrap();
8130
8131 let track = source.create_track("audio", None).unwrap();
8132 let mut group = track.append_group().unwrap();
8133 group.write_frame(crate::Timestamp::ZERO, b"steady".as_ref()).unwrap();
8134 group.finish().unwrap();
8135
8136 let mut audio = rig
8137 .resolved
8138 .track("audio")
8139 .unwrap()
8140 .subscribe(None)
8141 .await
8142 .expect("subscribe survives the reprice");
8143 let mut group = audio
8144 .recv_group()
8145 .await
8146 .expect("recv group")
8147 .expect("track ended early");
8148 let frame = group.read_frame().await.expect("read frame").expect("frame");
8149 assert_eq!(&frame.payload[..], b"steady");
8150 }
8151
8152 #[tokio::test]
8153 async fn drain_reprice_migrates_before_the_session_dies() {
8154 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8155 let standby_server = rig.standby(&[10, 20]);
8156
8157 incumbent
8161 .update(Route::default().with_hops(hops(&[10])).with_cost(Cost::DRAIN))
8162 .unwrap();
8163
8164 assert_resumes(&mut rig, &standby_server).await;
8165
8166 drop(incumbent);
8168 drop(source);
8169 }
8170
8171 #[tokio::test]
8172 async fn local_sources_splice_newest_first() {
8173 let producer = origin(1).produce();
8174 let consumer = producer.consume();
8175
8176 let first = producer.publish("room/alice", Route::default()).unwrap();
8177 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8178
8179 let second = producer.publish("room/alice", Route::default()).unwrap();
8181 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
8182 assert!(again.is_clone(&resolved));
8183
8184 first.close();
8186 settle(|| consumer.get_broadcast("room/alice").is_some()).await;
8187 second.close();
8188 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8189
8190 let _third = producer.publish("room/alice", Route::default()).unwrap();
8192 assert!(consumer.get_broadcast("room/alice").is_some());
8193 }
8194
8195 #[tokio::test]
8203 async fn a_finished_broadcast_concludes_in_flight_subscriptions() {
8204 let producer = origin(1).produce();
8205 let consumer = producer.consume();
8206
8207 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8208 let track = broadcast.create_track("video", None).unwrap();
8209
8210 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8211 let mut subscription = resolved
8212 .track("video")
8213 .unwrap()
8214 .subscribe(None)
8215 .await
8216 .expect("subscribe");
8217 let mut group = track.append_group().unwrap();
8219 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8220 group.finish().unwrap();
8221 track.finish().unwrap();
8222 drop(track);
8223 broadcast.close();
8224
8225 let mut group = next_group(&mut subscription)
8226 .await
8227 .expect("a cleanly finished track was served as an error")
8228 .expect("the track ended before its last group");
8229 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8230 drop(group);
8231
8232 let end = next_group(&mut subscription)
8233 .await
8234 .expect("a cleanly finished track ended as an error");
8235 assert!(end.is_none(), "a group followed the final one");
8236 }
8237
8238 #[tokio::test]
8245 async fn a_retracted_route_concludes_in_flight_subscriptions() {
8246 let producer = origin(1).produce();
8247 let consumer = producer.consume();
8248 let server = producer
8249 .dynamic("room", Route::default().with_hops(hops(&[10])))
8250 .unwrap();
8251
8252 let pending = consumer.request_broadcast("room/alice");
8253 let request = queued(&server).await;
8254 let source = broadcast::Info::new().produce();
8255 let track = source.create_track("video", None).unwrap();
8256 request.accept(&source);
8257
8258 let resolved = pending.await.expect("resolves");
8259 let mut subscription = resolved
8260 .track("video")
8261 .unwrap()
8262 .subscribe(None)
8263 .await
8264 .expect("subscribe");
8265
8266 source.close();
8269 drop(server);
8270 settle(|| resolved.is_closed()).await;
8271 let mut group = track.append_group().unwrap();
8272 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8273 group.finish().unwrap();
8274 track.finish().unwrap();
8275 drop(track);
8276
8277 let mut group = next_group(&mut subscription)
8278 .await
8279 .expect("a retracted route's track was served as an error")
8280 .expect("the track ended before its last group");
8281 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8282 drop(group);
8283
8284 let end = next_group(&mut subscription)
8285 .await
8286 .expect("a cleanly finished track ended as an error");
8287 assert!(end.is_none(), "a group followed the final one");
8288 }
8289
8290 #[tokio::test]
8293 async fn a_closed_source_is_not_requested_again_from_its_standing_route() {
8294 let producer = origin(1).produce();
8295 let server = producer
8296 .dynamic("room", Route::default().with_hops(hops(&[10])))
8297 .unwrap();
8298 let pending = producer.consume().request_broadcast("room/alice");
8299 let source = broadcast::Info::new().produce();
8300 queued(&server).await.accept(&source);
8301 let resolved = pending.await.unwrap();
8302
8303 source.close();
8304 settle(|| resolved.is_closed()).await;
8305 assert!(
8306 server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8307 "the closed source was requested again"
8308 );
8309 }
8310
8311 #[tokio::test]
8316 async fn origin_front_drops_the_source_when_unused() {
8317 let producer = origin(1).produce();
8318 let consumer = producer.consume();
8319
8320 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8321 let track = broadcast.create_track("video", None).unwrap();
8322 let mut group = track.append_group().unwrap();
8323 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8324 group.finish().unwrap();
8325
8326 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8327 let mut subscription = resolved
8328 .track("video")
8329 .unwrap()
8330 .subscribe(None)
8331 .await
8332 .expect("subscribe");
8333 let mut group = subscription.recv_group().await.unwrap().unwrap();
8334 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8335 drop(group);
8336 drop(subscription);
8337
8338 tokio::time::timeout(Duration::from_secs(1), track.unused())
8339 .await
8340 .expect("source unused should resolve far below TRACK_IDLE_LINGER")
8341 .expect("source closed");
8342
8343 let mut again = resolved
8346 .track("video")
8347 .unwrap()
8348 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8349 .await
8350 .expect("resubscribe");
8351 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8352 .await
8353 .expect("cached group is still on the front")
8354 .expect("recv group")
8355 .expect("track ended early");
8356 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8357
8358 tokio::time::timeout(Duration::from_secs(1), track.used())
8359 .await
8360 .expect("returning reader re-splices the source")
8361 .expect("source closed");
8362
8363 let mut group = track.append_group().unwrap();
8364 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8365 group.finish().unwrap();
8366 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8367 .await
8368 .expect("groups past the cached edge come from the re-splice")
8369 .expect("recv group")
8370 .expect("track ended early");
8371 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8372 }
8373
8374 #[tokio::test]
8384 async fn resumed_reader_skips_warm_groups_behind_the_new_edge() {
8385 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
8386 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
8387 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
8388 let write = |source: &track::Producer, sequence: u64, millis: u64| {
8389 let mut group = source.create_group(sequence.into()).unwrap();
8390 group.write_frame(ms(millis), b"x".as_ref()).unwrap();
8391 group.finish().unwrap();
8392 };
8393 let drain = |subscription: &mut track::Subscriber| {
8394 let mut sequences = Vec::new();
8395 while let Poll::Ready(group) = subscription.poll_recv_group(&kio::Waiter::noop()) {
8396 sequences.push(group.unwrap().expect("track ended").sequence);
8397 }
8398 sequences
8399 };
8400
8401 let track = resolved.track("video").unwrap();
8402 let first = budget.clone();
8403 let subscribing = tokio::spawn(async move { track.subscribe(first).await });
8404 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8405 .await
8406 .expect("the front asked the source")
8407 .expect("request");
8408 let source = request.resolving_start().accept(None);
8409 for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] {
8410 write(&source, sequence, millis);
8411 }
8412 let mut subscription = subscribing.await.unwrap().expect("subscribe");
8413 next_group(&mut subscription).await.unwrap().expect("a cached group");
8414 drain(&mut subscription);
8415 drop(subscription);
8416
8417 tokio::time::timeout(Duration::from_secs(1), source.unused())
8418 .await
8419 .expect("parked")
8420 .expect("source open");
8421 drop(source);
8422
8423 let track = resolved.track("video").unwrap();
8424 let second = budget.clone();
8425 let subscribing = tokio::spawn(async move { track.subscribe(second).await });
8426 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8427 .await
8428 .expect("the front asked the source again")
8429 .expect("request");
8430 let mut source = request.resolving_start().accept(None);
8431 for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] {
8432 write(&source, sequence, millis);
8433 }
8434 source.start_at(3).unwrap();
8437 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
8438 settle(|| subscription.latest() == Some(22)).await;
8439
8440 assert_eq!(drain(&mut subscription), [3, 20, 21, 22]);
8443 }
8444
8445 #[tokio::test]
8450 async fn chained_front_drops_the_source_when_unused() {
8451 let leaf = origin(1).produce();
8452 let leaf_consumer = leaf.consume();
8453
8454 let broadcast = leaf.publish("room/alice", Route::default()).unwrap();
8455 let track = broadcast.create_track("video", None).unwrap();
8456 let mut group = track.append_group().unwrap();
8457 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8458 group.finish().unwrap();
8459
8460 let leaf_front = leaf_consumer.request_broadcast("room/alice").await.expect("resolves");
8463
8464 let mid = origin(2).produce();
8465 let mid_server = mid.dynamic("room", Route::default().with_hops(hops(&[10]))).unwrap();
8466 let mid_pending = mid.consume().request_broadcast("room/alice");
8467 queued(&mid_server).await.accept(&leaf_front);
8468 let mid_resolved = mid_pending.await.expect("mid resolves");
8469
8470 let edge = origin(3).produce();
8471 let edge_server = edge.dynamic("room", Route::default().with_hops(hops(&[20]))).unwrap();
8472 let edge_pending = edge.consume().request_broadcast("room/alice");
8473 queued(&edge_server).await.accept(&mid_resolved);
8474 let edge_resolved = edge_pending.await.expect("edge resolves");
8475
8476 let mut subscription = edge_resolved
8477 .track("video")
8478 .unwrap()
8479 .subscribe(None)
8480 .await
8481 .expect("subscribe");
8482 let mut group = subscription.recv_group().await.unwrap().unwrap();
8483 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8484 drop(group);
8485 drop(subscription);
8486
8487 tokio::time::timeout(Duration::from_secs(5), track.unused())
8488 .await
8489 .expect("chained unused should resolve far below TRACK_IDLE_LINGER")
8490 .expect("source closed");
8491
8492 let cached = edge_resolved.track("video").unwrap().cached_groups();
8493 assert_eq!(
8494 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8495 vec![0],
8496 "every front keeps the delivered groups after releasing its source"
8497 );
8498
8499 let mut subscription = edge_resolved
8501 .track("video")
8502 .unwrap()
8503 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8504 .await
8505 .expect("resubscribe");
8506 tokio::time::timeout(Duration::from_secs(5), track.used())
8507 .await
8508 .expect("resubscribe should reach the leaf")
8509 .expect("source open");
8510 let mut group = track.append_group().unwrap();
8511 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8512 group.finish().unwrap();
8513 let mut group = subscription.recv_group().await.unwrap().unwrap();
8514 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8515 drop(group);
8516 let mut group = subscription.recv_group().await.unwrap().unwrap();
8517 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8518 drop(group);
8519 drop(subscription);
8520
8521 tokio::time::timeout(Duration::from_secs(5), track.unused())
8522 .await
8523 .expect("second chained unused should resolve far below TRACK_IDLE_LINGER")
8524 .expect("source closed");
8525
8526 let cached = edge_resolved.track("video").unwrap().cached_groups();
8527 assert_eq!(
8528 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8529 vec![0, 1],
8530 "repeated demand keeps every complete group while releasing its source"
8531 );
8532
8533 let fetch = edge_resolved.track("video").unwrap().fetch_group(2, None);
8534 let mut fetch = std::pin::pin!(fetch);
8535 assert!(futures::poll!(fetch.as_mut()).is_pending(), "fetch should re-splice");
8536 tokio::time::timeout(Duration::from_secs(5), track.used())
8537 .await
8538 .expect("fetch should reach the leaf")
8539 .expect("source open");
8540 let mut group = track.append_group().unwrap();
8541 group.write_frame(crate::Timestamp::ZERO, b"fetched".as_ref()).unwrap();
8542 group.finish().unwrap();
8543 let mut group = tokio::time::timeout(Duration::from_secs(5), fetch)
8544 .await
8545 .expect("re-spliced source should answer the fetch")
8546 .expect("fetch succeeds");
8547 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"fetched");
8548 }
8549
8550 #[tokio::test]
8554 async fn incompatible_local_source_keeps_the_incumbent() {
8555 let producer = origin(1).produce();
8556 let consumer = producer.consume();
8557
8558 let first = producer.publish("room/alice", Route::default()).unwrap();
8559 let track = first.create_track("video", None).unwrap();
8560 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8561 let mut subscription = resolved
8562 .track("video")
8563 .unwrap()
8564 .subscribe(None)
8565 .await
8566 .expect("subscribe");
8567 let mut group = track.append_group().unwrap();
8568 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8569 group.finish().unwrap();
8570 let mut group = subscription.recv_group().await.unwrap().unwrap();
8571 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"before");
8572
8573 let second = producer.publish("room/alice", Route::default()).unwrap();
8575 let _incompatible = second
8576 .create_track("video", track::Info::default().with_timescale(crate::Timescale::MICRO))
8577 .unwrap();
8578 for _ in 0..10 {
8579 tokio::task::yield_now().await;
8580 }
8581
8582 let mut group = track.append_group().unwrap();
8584 group.write_frame(crate::Timestamp::ZERO, b"still".as_ref()).unwrap();
8585 group.finish().unwrap();
8586 let mut group = subscription.recv_group().await.unwrap().unwrap();
8587 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"still");
8588
8589 drop(track);
8591 first.close();
8592 assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported)));
8593 }
8594
8595 #[tokio::test]
8596 async fn multiple_scopes_present_one_broad_prefix() {
8597 let producer = origin(1).produce();
8598 let _a = producer.announce("", Route::default()).unwrap();
8599
8600 let consumer = producer.consume().scope("", &scopes(&["alpha", "beta"])).unwrap();
8601 let mut announced = consumer.announced();
8602 announced.assert_next_active("");
8603 announced.assert_next_wait();
8604 }
8605
8606 #[test]
8607 fn scope_accepts_every_pattern_union() {
8608 let producer = origin(1).produce();
8609
8610 let root = producer.scope("", &Patterns::from(Pattern::all())).unwrap();
8612 assert_eq!(root.allowed(), Patterns::from(Pattern::all()));
8613
8614 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8616 assert_eq!(scoped.allowed(), scopes(&["room"]));
8617
8618 let multi = producer.scope("", &scopes(&["room", "room/chat", "anon"])).unwrap();
8620 assert_eq!(multi.allowed(), scopes(&["room", "anon"]));
8621
8622 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
8624 assert_eq!(consumer.allowed(), scopes(&["room"]));
8625
8626 for text in ["room", "", "*room", "room/*", "*", "**/room", "room/**/chat", "*.hang"] {
8627 let union = Patterns::from(text.parse::<Pattern>().unwrap());
8628 assert_eq!(producer.scope("", &union).expect(text).allowed(), union, "{text}");
8629 assert_eq!(
8630 producer.consume().scope("", &union).expect(text).allowed(),
8631 union,
8632 "{text}"
8633 );
8634 }
8635
8636 let mixed: Patterns = ["room/**".parse().unwrap(), "other".parse().unwrap()]
8637 .into_iter()
8638 .collect();
8639 assert_eq!(producer.scope("", &mixed).unwrap().allowed(), mixed);
8640 }
8641
8642 #[test]
8643 fn route_table_prunes_to_empty() {
8644 let producer = origin(1).produce();
8645 let consumer = producer.consume();
8646
8647 let cursor = consumer
8650 .scope("", &scopes(&["room/a", "other/deep/head"]))
8651 .unwrap()
8652 .announced();
8653 let route = producer.announce("room/a/b/c", Route::default()).unwrap();
8654 {
8655 let table = producer.shared.lock();
8656 assert!(table.routes.root.find(Path::new("room/a/b/c").parts()).is_some());
8657 assert!(table.routes.root.find(Path::new("other/deep/head").parts()).is_some());
8658 assert_eq!(table.routes.root.cursors_below, 2);
8659 }
8660
8661 drop(route);
8662 drop(cursor);
8663 let table = producer.shared.lock();
8664 assert!(table.routes.root.is_empty());
8665 assert_eq!(table.routes.root.cursors_below, 0);
8666 }
8667
8668 #[test]
8672 fn a_published_broadcast_keeps_the_driver_running() {
8673 let (producer, mut driver) = Producer::new(Config::new(origin(1)));
8674 let waiter = kio::Waiter::noop();
8675 let broadcast = producer.create_broadcast("room/a").unwrap();
8676 drop(producer);
8677 assert!(
8678 driver.poll(Instant::now(), &waiter).is_ok(),
8679 "the broadcast is lifecycle work"
8680 );
8681 drop(broadcast);
8682 assert!(matches!(driver.poll(Instant::now(), &waiter), Err(Error::Closed)));
8683 }
8684
8685 #[tokio::test]
8688 async fn a_dynamic_keeps_the_driver_running() {
8689 let (producer, driver) = Producer::new(Config::new(origin(1)));
8690 let run = tokio::spawn(crate::time::run(driver));
8691 let consumer = producer.consume();
8692 let server = producer.dynamic("room", Route::default()).unwrap();
8693 drop(producer);
8694
8695 let pending = consumer.request_broadcast("room/alice");
8696 let request = queued(&server).await;
8697 let source = broadcast::Info::new().produce();
8698 request.accept(&source);
8699 let resolved = pending.await.expect("the dynamic still serves");
8700
8701 drop(resolved);
8702 drop(source);
8703 drop(server);
8704 tokio::time::timeout(Duration::from_secs(5), run)
8705 .await
8706 .expect("driver must finish once the dynamic is gone")
8707 .unwrap();
8708 drop(consumer);
8709 }
8710
8711 #[test]
8712 fn watch_wakes_only_for_covering_changes() {
8713 let producer = origin(1).produce();
8714 let waiter = kio::Waiter::noop();
8715 let watch = producer.shared.lock().watch(&producer.shared, &Path::new("room/a"));
8716 let seen = watch.seen();
8717
8718 let _other = producer.announce("other", Route::default()).unwrap();
8720 let _below = producer.announce("room/a/b", Route::default()).unwrap();
8721 assert!(watch.poll_changed(&waiter, seen).is_pending());
8722
8723 let above = producer.announce("room", Route::default()).unwrap();
8725 assert!(watch.poll_changed(&waiter, seen).is_ready());
8726 let seen = watch.seen();
8727 drop(above);
8728 assert!(watch.poll_changed(&waiter, seen).is_ready());
8729 let seen = watch.seen();
8730
8731 let _beside = producer.create_broadcast("room/b").unwrap();
8733 assert!(watch.poll_changed(&waiter, seen).is_pending());
8734 let _here = producer.create_broadcast("room/a").unwrap();
8735 assert!(watch.poll_changed(&waiter, seen).is_ready());
8736
8737 drop(watch);
8739 let table = producer.shared.lock();
8740 let node = table
8741 .routes
8742 .root
8743 .find(Path::new("room/a").parts())
8744 .expect("route below keeps the node");
8745 assert!(node.watches.is_empty());
8746 assert_eq!(table.routes.root.watches_below, 0);
8747 }
8748
8749 #[test]
8750 fn a_discarded_front_task_unregisters_its_watch() {
8751 let (producer, _driver) = Producer::new(Config {
8752 hop: origin(1),
8753 ..Default::default()
8754 });
8755 let consumer = producer.consume();
8756 let _served = producer.dynamic("room", Route::default()).unwrap();
8757 drop(producer);
8762 let _pending = consumer.request_broadcast("room/a");
8763 }
8764
8765 #[test]
8766 fn create_broadcast_refuses_a_path_no_pattern_can_spell() {
8767 let producer = origin(1).produce();
8768
8769 assert!(matches!(
8773 producer.create_broadcast("room/*"),
8774 Err(Error::InvalidPath(_))
8775 ));
8776 assert!(matches!(
8777 producer.announce("room/**", Route::default()),
8778 Err(Error::InvalidPath(_))
8779 ));
8780 }
8781
8782 #[test]
8783 fn scope_empty_union_grants_nothing() {
8784 let producer = origin(1).produce();
8785
8786 assert!(matches!(producer.scope("", &Patterns::new()), Err(Error::Unauthorized)));
8788 assert!(matches!(
8789 producer.consume().scope("", &Patterns::new()),
8790 Err(Error::Unauthorized)
8791 ));
8792 }
8793
8794 #[test]
8795 fn scope_nests_and_rebases_roots() {
8796 let producer = origin(1).produce();
8797
8798 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8800 let nested = scoped.scope("", &scopes(&["room/chat"])).unwrap();
8801 assert_eq!(nested.allowed(), scopes(&["room/chat"]));
8802
8803 assert!(matches!(
8805 scoped.scope("", &scopes(&["other"])),
8806 Err(Error::Unauthorized)
8807 ));
8808
8809 let rooted = nested.scope("room/chat", &Patterns::from(Pattern::all())).unwrap();
8811 assert_eq!(rooted.allowed(), scopes(&[""]));
8812
8813 let broadcast = nested.create_broadcast("room/chat/live").unwrap();
8815 assert!(producer.consume().get_broadcast("room/chat/live").is_some());
8816 broadcast.close();
8817 }
8818
8819 #[test]
8820 fn scope_intersects_and_rebases_arbitrary_grants() {
8821 let producer = origin(1).produce();
8822 let rooms = producer
8823 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
8824 .unwrap();
8825 let chats = rooms
8826 .scope("", &Patterns::from("*/chat".parse::<Pattern>().unwrap()))
8827 .unwrap();
8828 assert_eq!(chats.allowed(), Patterns::from("room/chat".parse::<Pattern>().unwrap()));
8829
8830 let exact = producer
8831 .scope("", &Patterns::from("room/alice".parse::<Pattern>().unwrap()))
8832 .unwrap();
8833 let rooted = exact.scope("room", &Patterns::from(Pattern::all())).unwrap();
8834 assert_eq!(rooted.allowed(), Patterns::from("alice".parse::<Pattern>().unwrap()));
8835 assert!(matches!(
8836 exact.scope("room/bob", &Patterns::from(Pattern::all())),
8837 Err(Error::Unauthorized)
8838 ));
8839
8840 let broadcast = exact.create_broadcast("room/alice").unwrap();
8841 assert!(matches!(
8842 exact.create_broadcast("room/alice/cam"),
8843 Err(Error::Unauthorized)
8844 ));
8845 assert!(producer.consume().get_broadcast("room/alice").is_some());
8846 drop(broadcast);
8847 }
8848
8849 #[tokio::test]
8850 async fn wildcard_scope_filters_announcements_and_reports_captures() {
8851 let producer = origin(1).produce();
8852 let consumer = producer
8853 .consume()
8854 .scope("", &Patterns::from("room/*/chat".parse::<Pattern>().unwrap()))
8855 .unwrap();
8856 let mut announced = consumer.announced();
8857
8858 let alice = producer.create_broadcast("room/alice/chat").unwrap();
8859 alice.announce(Route::default()).unwrap();
8860 let update = announced.try_next().expect("alice's chat");
8861 assert_eq!(update.prefix.as_str(), "room/alice/chat");
8862 assert_eq!(update.captures, Some(vec!["alice".parse::<Pattern>().unwrap()]));
8863
8864 let audio = producer.create_broadcast("room/alice/audio").unwrap();
8865 audio.announce(Route::default()).unwrap();
8866 announced.assert_next_wait();
8867
8868 let broad = producer.announce("room", Route::default()).unwrap();
8869 let update = announced.try_next().expect("overlapping broad route");
8870 assert_eq!(update.prefix.as_str(), "room");
8871 assert_eq!(update.captures, None, "an overlap does not pin the wildcard");
8872
8873 drop(broad);
8874 drop(audio);
8875 drop(alice);
8876 }
8877
8878 #[tokio::test]
8879 async fn local_broadcast_wins_announcement_ties() {
8880 let producer = origin(1).produce();
8881 let remote = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
8882 let local = producer.create_broadcast("room/alice").unwrap();
8883 local.announce(Route::default()).unwrap();
8884
8885 let mut announced = producer.consume().announced();
8886 let update = announced.try_next().expect("one winning route");
8887 assert_eq!(update.prefix.as_str(), "room/alice");
8888 assert_eq!(update.route.cost, Cost::default());
8889 announced.assert_next_wait();
8890
8891 drop(local);
8892 drop(remote);
8893 }
8894
8895 #[test]
8900 fn cost_charge_saturates() {
8901 assert_eq!(Cost { warm: 4, cold: 6 }.charged(5), Cost { warm: 9, cold: 11 });
8902 assert_eq!(Cost::new(u64::MAX).charged(10), Cost::new(MAX_COST));
8903
8904 assert_eq!(Cost::UNKNOWN.charged(3).cold, MAX_COST);
8907 }
8908
8909 fn expiring_origin(expiry: Duration) -> Producer {
8911 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
8912 Config {
8913 pool,
8914 ..Config::default()
8915 }
8916 .produce()
8917 }
8918
8919 #[tokio::test(start_paused = true)]
8923 async fn stalled_publisher_open_group_is_reclaimed() {
8924 let expiry = Duration::from_secs(1);
8925 let origin = expiring_origin(expiry);
8926 let broadcast = origin.create_broadcast("test").unwrap();
8927 let track = broadcast.create_track("video", None).unwrap();
8928
8929 let mut stalled = track.append_group().unwrap();
8930 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
8931 let _successor = track.append_group().unwrap();
8935
8936 let mut reading = stalled.consume();
8937 assert!(reading.read_frame().await.unwrap().is_some());
8938
8939 crate::model::clock::advance(expiry * 2);
8941
8942 let reclaimed = tokio::time::timeout(Duration::from_secs(60), reading.read_frame()).await;
8945 assert!(
8946 matches!(reclaimed, Ok(Err(Error::Old))),
8947 "the sweep must reclaim an idle open group and surface the gap, got {reclaimed:?}"
8948 );
8949 }
8950
8951 #[tokio::test(start_paused = true)]
8954 async fn sweep_respects_a_disabled_expiry() {
8955 let origin = Config {
8956 pool: cache::Pool::unbounded(),
8957 ..Config::default()
8958 }
8959 .produce();
8960 let broadcast = origin.create_broadcast("test").unwrap();
8961 let track = broadcast.create_track("video", None).unwrap();
8962
8963 let mut stalled = track.append_group().unwrap();
8964 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
8965 let _successor = track.append_group().unwrap();
8966
8967 let mut reading = stalled.consume();
8968 assert!(reading.read_frame().await.unwrap().is_some());
8969
8970 crate::model::clock::advance(Duration::from_secs(3600));
8971 tokio::time::advance(Duration::from_secs(3600)).await;
8972
8973 assert!(
8974 reading.read_frame().now_or_never().is_none(),
8975 "a pool without an expiry window never reclaims"
8976 );
8977 }
8978
8979 #[test]
8982 fn drain_cost_is_encodable() {
8983 use crate::coding::Encode;
8984
8985 let mut buf = Vec::new();
8986 Cost::DRAIN
8987 .encode(&mut buf, crate::lite::Version::Lite06)
8988 .expect("a draining route is still forwarded, so its cost must encode");
8989 }
8990}