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 {
2278 track: track::Producer,
2279 _dynamic: track::Dynamic,
2280 edge: Option<WarmGroup>,
2283}
2284
2285impl Drop for WarmCopy {
2286 fn drop(&mut self) {
2287 let _ = self.track.finish();
2288 }
2289}
2290
2291struct WarmGroup(group::Producer);
2298
2299impl Drop for WarmGroup {
2300 fn drop(&mut self) {
2301 if !self.0.is_finished() {
2302 let _ = self.0.clone().abort(Error::Cancel);
2303 }
2304 }
2305}
2306
2307struct HeldGroup(group::Producer);
2312
2313impl Drop for HeldGroup {
2314 fn drop(&mut self) {
2315 if !self.0.is_finished() {
2316 let _ = self.0.abort_if_last(Error::Cancel);
2317 }
2318 }
2319}
2320
2321fn warm_copy(source: &track::Consumer, head: Option<&WarmGroup>, held: Option<&group::Producer>) -> Option<WarmCopy> {
2329 let info = source.cached_info()?;
2330 let mut track = track::Producer::new(Arc::new(source.broadcast().clone()), source.name(), info);
2331 let head = head.map(|head| &head.0);
2332 let mut groups = source.cached_groups();
2333 if let Some(held) = held
2334 && !groups.iter().any(|(group, _)| group.sequence == held.sequence)
2335 {
2336 groups.push((held.clone(), true));
2337 }
2338 let latest = groups
2340 .iter()
2341 .filter(|(_, visible)| *visible)
2342 .map(|(group, _)| group.sequence)
2343 .max();
2344 let mut edge = None;
2345 let mut first = None;
2346
2347 if let Some(head) = head
2351 && !groups.iter().any(|(group, _)| group.sequence == head.sequence)
2352 {
2353 let is_latest = latest.is_none_or(|latest| head.sequence >= latest);
2354 if head.is_finished() {
2355 let _ = track.adopt_group(head.clone(), true);
2356 first = Some(head.sequence);
2357 if is_latest {
2358 edge = Some(WarmGroup(head.clone()));
2359 }
2360 } else if is_latest {
2361 edge = warm_rebuild(&track, head, None);
2362 first = edge.as_ref().map(|_| head.sequence);
2363 }
2364 }
2365
2366 for (group, visible) in groups {
2367 let finished = group.is_finished();
2368 let whole = group.live_first_frame() == Some(0);
2369 let is_latest = visible && Some(group.sequence) == latest;
2370 let warm = if finished && whole {
2371 let _ = track.adopt_group(group.clone(), visible);
2372 Some(WarmGroup(group))
2373 } else if (finished || is_latest) && (whole || head.is_some_and(|head| head.sequence == group.sequence)) {
2374 warm_rebuild(&track, &group, head)
2380 } else {
2381 None
2383 };
2384 if visible && let Some(warm) = &warm {
2385 first = Some(first.map_or(warm.0.sequence, |first: u64| first.min(warm.0.sequence)));
2386 }
2387 if is_latest {
2388 edge = warm;
2389 }
2390 }
2391 track.start_at(first).ok()?;
2395 let dynamic = track.dynamic();
2396 Some(WarmCopy {
2397 track,
2398 _dynamic: dynamic,
2399 edge,
2400 })
2401}
2402
2403fn warm_rebuild(track: &track::Producer, live: &group::Producer, head: Option<&group::Producer>) -> Option<WarmGroup> {
2406 let mut start = live.live_first_frame()? as u64;
2408 let mut tail = live.consume();
2409 tail.start_at(start);
2410 let mut frames = Vec::new();
2411 if start > 0
2412 && let Some(head) = head.filter(|head| head.sequence == live.sequence)
2413 && let Some(head_start) = head.live_first_frame()
2414 {
2415 let mut head = head.consume();
2416 head.start_at(head_start as u64);
2417 while head.index() < start {
2418 match head.poll_read_frame(&kio::Waiter::noop()) {
2419 Poll::Ready(Ok(Some(frame))) => frames.push(frame),
2420 _ => break,
2421 }
2422 }
2423 match head.index() == start {
2424 true => start = head_start as u64,
2425 false => frames.clear(),
2427 }
2428 }
2429 while let Poll::Ready(Ok(Some(frame))) = tail.poll_read_frame(&kio::Waiter::noop()) {
2430 frames.push(frame);
2431 }
2432 if frames.is_empty() {
2433 return None;
2434 }
2435
2436 let rebuilt = WarmGroup(
2438 track
2439 .create_group(group::Info {
2440 sequence: live.sequence,
2441 })
2442 .ok()?,
2443 );
2444 let mut writer = rebuilt.0.clone();
2445 if start > 0 {
2446 writer.start_at(start).ok()?;
2447 }
2448 for frame in frames {
2449 writer.write_frame(frame.timestamp, frame.payload).ok()?;
2450 }
2451 if live.is_finished() {
2452 writer.finish().ok()?;
2453 }
2454 Some(rebuilt)
2455}
2456
2457struct FrontTask {
2459 shared: kio::Shared<OriginState>,
2461 broadcast: broadcast::Producer,
2463 path: PathOwned,
2465 horizon: Horizon,
2467 watch: Watch,
2469 request: kio::Producer<PendingBroadcast>,
2471 pin: kio::Lock<Pin>,
2473 timers: Clock,
2474}
2475
2476struct TrackIo {
2479 resume: super::resume::Producer,
2480 staged: Option<(u64, track::Consumer)>,
2482 query: Option<(u64, track::Consumer, track::Querying)>,
2484 copy: Option<(u64, track::Consumer)>,
2486 edge: Option<track::Position>,
2491 warm: Option<WarmCopy>,
2494 head: Option<WarmGroup>,
2497 prior: Option<(u64, WarmCopy)>,
2502 held: Option<HeldGroup>,
2507 dead: Option<track::Consumer>,
2510 used: bool,
2512}
2513
2514impl TrackIo {
2515 fn end(&mut self) {
2518 self.staged = None;
2519 self.query = None;
2520 self.copy = None;
2521 self.warm = None;
2522 self.head = None;
2523 self.prior = None;
2524 self.held = None;
2525 self.dead = None;
2526 }
2527
2528 fn hold(&mut self, group: group::Producer) {
2531 if self.prior.as_ref().is_some_and(|(edge, _)| group.sequence > *edge) {
2532 self.prior = None;
2533 }
2534 self.held = Some(HeldGroup(group));
2535 }
2536
2537 fn bury(&mut self) {
2540 let Some((_, copy)) = self.copy.take() else { return };
2541 let after = self.held.as_ref().map(|held| held.0.sequence);
2542 if let Poll::Ready(group) = copy.poll_latest_group(after, &kio::Waiter::noop()) {
2543 self.hold(group);
2544 }
2545 self.dead = Some(copy);
2546 }
2547
2548 fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> {
2554 let Some(held) = self.held.take() else {
2555 return Ok(());
2556 };
2557 let Some(warm) = warm_copy(copy, self.head.as_ref(), Some(&held.0)) else {
2558 return Ok(());
2559 };
2560 self.head = None;
2561 self.resume.park(&warm.track)?;
2562 self.warm = Some(warm);
2563 Ok(())
2564 }
2565}
2566
2567async fn run_front(task: FrontTask) {
2571 let FrontTask {
2572 shared,
2573 broadcast,
2574 path,
2575 horizon,
2576 watch,
2577 request,
2578 pin,
2579 timers,
2580 } = task;
2581
2582 enum Step {
2584 Assigned(Arc<str>, super::resume::Producer),
2585 Resolved(u64, Result<broadcast::Consumer, Error>),
2586 SourceClosed(u64),
2587 Info(Arc<str>, u64, Result<track::Info, Error>),
2588 Ended(Arc<str>, u64, Result<(), Error>),
2589 Held(Arc<str>, group::Producer),
2590 Demand(Arc<str>),
2591 Deadline,
2592 Table,
2593 }
2594
2595 let mut front = Front::new(TRACK_IDLE_LINGER);
2596 let mut sources: HashMap<u64, broadcast::Consumer> = HashMap::new();
2597 let mut next_source = 0u64;
2598 let mut upstream: Option<(u64, kio::Consumer<PendingBroadcast>)> = None;
2600 let mut tracks: HashMap<Arc<str>, TrackIo> = HashMap::new();
2601 let mut deadline = crate::runtime::Deadline::new(&timers);
2602 let mut seen = 0;
2604 let mut events: VecDeque<Event> = VecDeque::new();
2605
2606 let select = |front: &mut Front, sources: &HashMap<u64, broadcast::Consumer>, seen: &mut u64| -> Event {
2609 let table = shared.read();
2610 if table.closed {
2611 return Event::Closed;
2612 }
2613 *seen = watch.seen();
2615 front.retain_routes(|route| table.routes.covers(&path.as_path(), route));
2616 let best = table
2617 .best_route(&path.as_path(), horizon, front.pin(), front.refused_routes())
2618 .map(|entry| Candidate {
2619 route: entry.id,
2620 first: entry.hops.iter().next().copied(),
2621 local: entry.local,
2622 });
2623 let serving_closing = front
2624 .serving()
2625 .and_then(|id| sources.get(&id))
2626 .is_some_and(|source| source.is_closing());
2627 Event::Selected { best, serving_closing }
2628 };
2629
2630 events.push_back(select(&mut front, &sources, &mut seen));
2631
2632 loop {
2633 while let Some(event) = events.pop_front() {
2634 for action in front.step(event) {
2635 match action {
2636 Action::Reselect => events.push_back(select(&mut front, &sources, &mut seen)),
2637 Action::Request { route } => {
2638 let found = {
2640 let table = shared.read();
2641 table
2642 .routes
2643 .covering(&path.as_path())
2644 .find(|entry| entry.id == route && entry.live())
2645 .map(|entry| {
2646 (
2647 Candidate {
2648 route,
2649 first: entry.hops.iter().next().copied(),
2650 local: entry.local,
2651 },
2652 entry.source.clone(),
2653 entry.server.clone(),
2654 )
2655 })
2656 };
2657 let Some((candidate, source, server)) = found else {
2658 events.push_back(Event::Resolved {
2659 route,
2660 result: Err(Refusal {
2661 err: Error::Unroutable,
2662 standing: false,
2663 }),
2664 });
2665 continue;
2666 };
2667 front.identify(candidate);
2668 *pin.lock() = front.pin();
2669 if let Some(source) = source {
2670 let id = next_source;
2671 next_source += 1;
2672 sources.insert(id, source);
2673 events.push_back(Event::Resolved { route, result: Ok(id) });
2674 continue;
2675 }
2676 let Some(server) = server else {
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 let mut serve = server.lock();
2687 if serve.closed {
2688 drop(serve);
2691 events.push_back(Event::Resolved {
2692 route,
2693 result: Err(Refusal {
2694 err: Error::Unroutable,
2695 standing: true,
2696 }),
2697 });
2698 continue;
2699 }
2700 if let Some(weak) = serve.served.get(&path) {
2703 drop(serve);
2704 let id = next_source;
2705 next_source += 1;
2706 sources.insert(id, weak.consume());
2707 events.push_back(Event::Resolved { route, result: Ok(id) });
2708 continue;
2709 }
2710 let pending = match serve.requests.join(&path) {
2711 Some(producer) => producer.consume(),
2712 None => {
2713 let producer = kio::Producer::<PendingBroadcast>::default();
2714 let consumer = producer.consume();
2715 match serve.requests.insert(path.clone(), producer) {
2716 Ok(()) => consumer,
2717 Err(_) => {
2720 drop(serve);
2721 events.push_back(Event::Resolved {
2722 route,
2723 result: Err(Refusal {
2724 err: Error::Unroutable,
2725 standing: true,
2726 }),
2727 });
2728 continue;
2729 }
2730 }
2731 }
2732 };
2733 upstream = Some((route, pending));
2734 }
2735 Action::Detach { source } => {
2736 sources.remove(&source);
2737 for io in tracks.values_mut() {
2740 if io.copy.as_ref().is_some_and(|(s, _)| *s == source) {
2741 io.bury();
2742 }
2743 if io.query.as_ref().is_some_and(|(s, ..)| *s == source) {
2744 io.query = None;
2745 }
2746 if io.staged.as_ref().is_some_and(|(s, _)| *s == source) {
2747 io.staged = None;
2748 }
2749 }
2750 }
2751 Action::Resolve => {
2752 if let Ok(mut pending) = request.write() {
2753 pending.resolved.get_or_insert(Ok(broadcast.consume()));
2754 }
2755 }
2756 Action::Query { track: name, source } => {
2757 let Some(io) = tracks.get_mut(&name) else { continue };
2758 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
2759 match sources.get(&source).map(|s| s.track(&name)) {
2760 Some(Ok(copy)) => {
2761 let query = copy.query().into_inner();
2764 io.query = Some((source, copy, query));
2765 }
2766 Some(Err(err)) => events.push_back(Event::TrackInfo {
2767 track: name,
2768 source,
2769 closing,
2770 result: Err(err),
2771 }),
2772 None => {}
2773 }
2774 }
2775 Action::Splice { track: name, source } => {
2776 let Some(io) = tracks.get_mut(&name) else { continue };
2777 let Some((staged, copy)) = io.staged.take() else {
2778 continue;
2779 };
2780 if staged != source {
2781 continue;
2782 }
2783 let outgoing = io.copy.take().map(|(_, copy)| copy).or_else(|| io.dead.take());
2784 let retired = match outgoing {
2785 Some(outgoing) => io.retire(&outgoing),
2786 None => Ok(()),
2787 };
2788 if let Err(err) = retired.and_then(|()| io.resume.takeover(©)) {
2789 let _ = io.resume.abort(err);
2793 tracks.remove(&name);
2794 continue;
2795 }
2796 if let Some(mut warm) = io.warm.take()
2797 && let Some(head) = warm.edge.take()
2798 {
2799 io.prior = Some((head.0.sequence, warm));
2800 io.head = Some(head);
2801 }
2802 io.edge = io.resume.resume_position();
2805 io.held = None;
2806 io.copy = Some((source, copy));
2807 }
2808 Action::Park { track: name } => {
2809 let Some(io) = tracks.get_mut(&name) else { continue };
2810 let Some((_, copy)) = io.copy.take() else { continue };
2811 let warm = warm_copy(©, io.head.as_ref(), io.held.as_ref().map(|held| &held.0));
2815 drop(copy);
2816 io.head = None;
2817 io.prior = None;
2818 io.held = None;
2819 io.dead = None;
2820 let parked = match &warm {
2821 Some(warm) => io.resume.park(&warm.track),
2822 None => io.resume.release(),
2823 };
2824 if parked.is_err() {
2825 tracks.remove(&name);
2826 continue;
2827 }
2828 io.warm = warm;
2829 }
2830 Action::Forget { track: name } => {
2831 if let Some(io) = tracks.get_mut(&name)
2836 && !broadcast.forget_spliced(&name, &io.resume)
2837 {
2838 if !io.used {
2839 io.used = true;
2840 events.push_back(Event::Used { track: name });
2841 }
2842 continue;
2843 }
2844 tracks.remove(&name);
2845 events.push_back(Event::Forgotten { track: name });
2846 }
2847 Action::Finish { track: name } => {
2848 if let Some(io) = tracks.get_mut(&name) {
2849 io.end();
2850 let _ = io.resume.finish();
2851 }
2852 }
2853 Action::Abort { track: name, err } => {
2854 if let Some(io) = tracks.get_mut(&name) {
2855 tracing::debug!(name = %name, %err, "aborting track");
2856 io.end();
2857 let _ = io.resume.abort(err);
2858 }
2859 }
2860 Action::Arm { at } => deadline.set(at),
2861 Action::End { err } => {
2862 if let Ok(mut pending) = request.write() {
2863 pending.resolved.get_or_insert(Err(err.clone()));
2864 }
2865 broadcast.close();
2872 broadcast.release_spliced(err.clone());
2873 for (_, mut io) in tracks.drain() {
2874 let waiting = io.staged.take().map(|(_, copy)| copy);
2878 let waiting = waiting.or_else(|| io.query.take().map(|(_, copy, _)| copy));
2879 if let Some(copy) = waiting
2880 && io.resume.is_used()
2881 {
2882 if io.resume.takeover(©).is_err() {
2883 continue;
2884 }
2885 io.warm = None;
2886 }
2887 if !io.resume.is_used() || !io.resume.is_spliced() || io.warm.is_some() {
2889 let _ = io.resume.abort(err.clone());
2890 }
2891 }
2892 return;
2893 }
2894 }
2895 }
2896 }
2897
2898 let step = kio::wait(|waiter| {
2899 if let Poll::Ready((name, resume)) = broadcast.poll_spliced_assigned(waiter) {
2900 return Poll::Ready(Step::Assigned(name, resume));
2901 }
2902 if let Some((route, pending)) = &upstream
2903 && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved {
2904 Some(result) => Poll::Ready(result.clone()),
2905 None => Poll::Pending,
2906 }) {
2907 return Poll::Ready(Step::Resolved(
2908 *route,
2909 match result {
2910 Ok(resolved) => resolved,
2911 Err(_closed) => Err(Error::Unroutable),
2914 },
2915 ));
2916 }
2917 if let Some(id) = front.serving()
2918 && let Some(source) = sources.get(&id)
2919 && source.poll_closed(waiter).is_ready()
2920 {
2921 return Poll::Ready(Step::SourceClosed(id));
2922 }
2923 for (name, io) in &tracks {
2924 if let Some((source, _, query)) = &io.query
2925 && let Poll::Ready(result) = query.poll(waiter)
2926 {
2927 return Poll::Ready(Step::Info(name.clone(), *source, result));
2928 }
2929 if let Some((source, copy)) = &io.copy
2930 && let Poll::Ready(result) = copy.poll_complete(waiter)
2931 {
2932 return Poll::Ready(Step::Ended(name.clone(), *source, result));
2933 }
2934 if let Some((_, copy)) = &io.copy
2935 && let Poll::Ready(group) =
2936 copy.poll_latest_group(io.held.as_ref().map(|held| held.0.sequence), waiter)
2937 {
2938 return Poll::Ready(Step::Held(name.clone(), group));
2939 }
2940 let edge = match io.used {
2942 true => io.resume.poll_unused(waiter),
2943 false => io.resume.poll_used(waiter),
2944 };
2945 if edge.is_ready() {
2946 return Poll::Ready(Step::Demand(name.clone()));
2947 }
2948 }
2949 if deadline.poll(waiter).is_ready() {
2950 return Poll::Ready(Step::Deadline);
2951 }
2952 watch.poll_changed(waiter, seen).map(|()| Step::Table)
2953 })
2954 .await;
2955
2956 let event = match step {
2957 Step::Assigned(name, resume) => {
2958 tracks.insert(
2959 name.clone(),
2960 TrackIo {
2961 resume,
2962 staged: None,
2963 query: None,
2964 copy: None,
2965 edge: None,
2966 warm: None,
2967 head: None,
2968 prior: None,
2969 held: None,
2970 dead: None,
2971 used: false,
2972 },
2973 );
2974 Event::TrackAssigned {
2975 track: name,
2976 now: timers.now(),
2977 }
2978 }
2979 Step::Resolved(route, result) => {
2980 upstream = None;
2981 match result {
2982 Ok(source) => {
2983 let id = next_source;
2984 next_source += 1;
2985 sources.insert(id, source);
2986 Event::Resolved { route, result: Ok(id) }
2987 }
2988 Err(err) => {
2989 let standing =
2993 !matches!(err, Error::Unroutable) || shared.read().routes.covers(&path.as_path(), route);
2994 Event::Resolved {
2995 route,
2996 result: Err(Refusal { err, standing }),
2997 }
2998 }
2999 }
3000 }
3001 Step::SourceClosed(source) => Event::SourceClosed { source },
3002 Step::Info(name, source, result) => {
3003 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3004 let Some(io) = tracks.get_mut(&name) else { continue };
3005 let Some((_, copy, _)) = io.query.take() else { continue };
3006 let result = match result {
3009 Ok(info) => match copy.poll_complete(&kio::Waiter::noop()) {
3010 Poll::Ready(Err(err)) => Err(err),
3011 _ => Ok(info),
3012 },
3013 Err(err) => Err(err),
3014 };
3015 if result.is_ok() && io.used {
3018 io.staged = Some((source, copy));
3019 }
3020 Event::TrackInfo {
3021 track: name,
3022 source,
3023 closing,
3024 result,
3025 }
3026 }
3027 Step::Ended(name, source, result) => {
3028 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3029 let Some(io) = tracks.get_mut(&name) else { continue };
3030 io.bury();
3031 let delivered = io.resume.resume_position() != io.edge;
3032 Event::TrackEnded {
3033 track: name,
3034 source,
3035 closing,
3036 result,
3037 delivered,
3038 }
3039 }
3040 Step::Held(name, group) => {
3041 if let Some(io) = tracks.get_mut(&name) {
3042 io.hold(group);
3043 }
3044 continue;
3045 }
3046 Step::Demand(name) => {
3047 let Some(io) = tracks.get_mut(&name) else { continue };
3048 io.used = io.resume.is_used();
3049 if !io.used {
3050 io.query = None;
3053 io.staged = None;
3054 }
3055 match io.used {
3056 true => Event::Used { track: name },
3057 false => Event::Unused {
3058 track: name,
3059 now: timers.now(),
3060 },
3061 }
3062 }
3063 Step::Deadline => {
3064 deadline.set(None);
3067 Event::Deadline { now: timers.now() }
3068 }
3069 Step::Table => select(&mut front, &sources, &mut seen),
3070 };
3071 events.push_back(event);
3072 }
3073}
3074
3075#[derive(Default)]
3080struct RouteTable {
3081 root: RouteNode,
3082}
3083
3084#[derive(Default)]
3087struct RouteNode {
3088 entries: Vec<RouteEntry>,
3090 cursors: Vec<ConsumerId>,
3092 cursors_below: usize,
3095 watches: Vec<(u64, kio::Producer<Watched>)>,
3098 watches_below: usize,
3101 children: HashMap<String, RouteNode>,
3102}
3103
3104#[derive(Default)]
3107struct Watched {
3108 generation: u64,
3109}
3110
3111struct Watch {
3117 shared: kio::Shared<OriginState>,
3118 path: PathOwned,
3119 id: u64,
3120 signal: kio::Consumer<Watched>,
3121}
3122
3123impl Watch {
3124 fn seen(&self) -> u64 {
3128 self.signal.read().generation
3129 }
3130
3131 fn poll_changed(&self, waiter: &kio::Waiter, seen: u64) -> Poll<()> {
3133 self.signal
3134 .poll(waiter, |watched| match watched.generation != seen {
3135 true => Poll::Ready(()),
3136 false => Poll::Pending,
3137 })
3138 .map(|_| ())
3139 }
3140}
3141
3142impl Drop for Watch {
3143 fn drop(&mut self) {
3144 self.shared.lock().routes.remove_watch(&self.path, self.id);
3145 }
3146}
3147
3148#[derive(Clone, Copy)]
3150struct Below {
3151 cursors: usize,
3152 watches: usize,
3153}
3154
3155impl Below {
3156 const NONE: Self = Self { cursors: 0, watches: 0 };
3157 const CURSOR: Self = Self { cursors: 1, watches: 0 };
3158 const WATCH: Self = Self { cursors: 0, watches: 1 };
3159}
3160
3161impl RouteNode {
3162 fn is_empty(&self) -> bool {
3164 self.entries.is_empty() && self.cursors.is_empty() && self.watches.is_empty() && self.children.is_empty()
3165 }
3166
3167 fn find<'a>(&self, mut parts: impl Iterator<Item = &'a str>) -> Option<&Self> {
3169 match parts.next() {
3170 None => Some(self),
3171 Some(part) => self.children.get(part)?.find(parts),
3172 }
3173 }
3174
3175 fn reach<'a>(&mut self, mut parts: impl Iterator<Item = &'a str>, below: Below) -> &mut Self {
3178 self.cursors_below += below.cursors;
3179 self.watches_below += below.watches;
3180 match parts.next() {
3181 None => self,
3182 Some(part) => self.children.entry(part.to_string()).or_default().reach(parts, below),
3183 }
3184 }
3185
3186 fn edit<'a, R>(
3190 &mut self,
3191 mut parts: impl Iterator<Item = &'a str>,
3192 below: Below,
3193 f: impl FnOnce(&mut Self) -> R,
3194 ) -> Option<R> {
3195 let result = match parts.next() {
3196 None => f(self),
3197 Some(part) => {
3198 let child = self.children.get_mut(part)?;
3199 let result = child.edit(parts, below, f)?;
3200 if child.is_empty() {
3201 self.children.remove(part);
3202 }
3203 result
3204 }
3205 };
3206 self.cursors_below -= below.cursors;
3207 self.watches_below -= below.watches;
3208 Some(result)
3209 }
3210
3211 fn poke(&self) {
3213 for (_, watch) in &self.watches {
3214 if let Ok(mut watched) = watch.write() {
3215 watched.generation += 1;
3216 }
3217 }
3218 }
3219
3220 fn poke_below(&self) {
3223 if self.watches_below == 0 {
3224 return;
3225 }
3226 self.poke();
3227 for child in self.children.values() {
3228 child.poke_below();
3229 }
3230 }
3231
3232 fn walk<'a>(&'a self, visit: &mut impl FnMut(&'a Self)) {
3234 visit(self);
3235 for child in self.children.values() {
3236 child.walk(visit);
3237 }
3238 }
3239
3240 fn collect_cursors(&self, out: &mut Vec<ConsumerId>) {
3242 if self.cursors_below == 0 {
3243 return;
3244 }
3245 out.extend(&self.cursors);
3246 for child in self.children.values() {
3247 child.collect_cursors(out);
3248 }
3249 }
3250}
3251
3252impl RouteTable {
3253 fn split(&self, path: &Path) -> (Vec<&RouteNode>, Option<&RouteNode>) {
3256 let mut above = Vec::new();
3257 let mut node = &self.root;
3258 for part in path.parts() {
3259 above.push(node);
3260 match node.children.get(part) {
3261 Some(child) => node = child,
3262 None => return (above, None),
3263 }
3264 }
3265 (above, Some(node))
3266 }
3267
3268 fn covering(&self, path: &Path) -> impl Iterator<Item = &RouteEntry> {
3270 let (above, at) = self.split(path);
3271 above.into_iter().chain(at).flat_map(|node| node.entries.iter())
3272 }
3273
3274 fn covers(&self, path: &Path, id: u64) -> bool {
3276 self.covering(path).any(|entry| entry.id == id && entry.live())
3277 }
3278
3279 fn at(&self, prefix: &Path) -> impl Iterator<Item = &RouteEntry> {
3281 self.root
3282 .find(prefix.parts())
3283 .into_iter()
3284 .flat_map(|node| node.entries.iter())
3285 }
3286
3287 fn at_mut(&mut self, prefix: &Path) -> impl Iterator<Item = &mut RouteEntry> {
3289 let mut node = Some(&mut self.root);
3290 for part in prefix.parts() {
3291 node = node.and_then(|node| node.children.get_mut(part));
3292 }
3293 node.into_iter().flat_map(|node| node.entries.iter_mut())
3294 }
3295
3296 fn entries(&self) -> impl Iterator<Item = &RouteEntry> {
3298 let mut nodes = Vec::new();
3299 self.root.walk(&mut |node| nodes.push(node));
3300 nodes.into_iter().flat_map(|node| node.entries.iter())
3301 }
3302
3303 fn insert(&mut self, entry: RouteEntry) {
3305 let node = self.root.reach(entry.prefix.parts(), Below::NONE);
3306 node.entries.push(entry);
3307 }
3308
3309 fn entry_mut(&mut self, prefix: &Path, id: u64) -> Option<&mut RouteEntry> {
3311 let mut node = &mut self.root;
3312 for part in prefix.parts() {
3313 node = node.children.get_mut(part)?;
3314 }
3315 node.entries.iter_mut().find(|entry| entry.id == id)
3316 }
3317
3318 fn remove(&mut self, prefix: &Path, id: u64) -> Option<RouteEntry> {
3320 self.root
3321 .edit(prefix.parts(), Below::NONE, |node| {
3322 let index = node.entries.iter().position(|entry| entry.id == id)?;
3323 Some(node.entries.swap_remove(index))
3324 })
3325 .flatten()
3326 }
3327
3328 fn add_cursor(&mut self, head: &Path, id: ConsumerId) {
3330 self.root.reach(head.parts(), Below::CURSOR).cursors.push(id);
3331 }
3332
3333 fn remove_cursor(&mut self, head: &Path, id: ConsumerId) {
3336 self.root.edit(head.parts(), Below::CURSOR, |node| {
3337 node.cursors.retain(|cursor| *cursor != id)
3338 });
3339 }
3340
3341 fn add_watch(&mut self, path: &Path, id: u64) -> kio::Consumer<Watched> {
3343 let producer = kio::Producer::<Watched>::default();
3344 let consumer = producer.consume();
3345 self.root.reach(path.parts(), Below::WATCH).watches.push((id, producer));
3346 consumer
3347 }
3348
3349 fn remove_watch(&mut self, path: &Path, id: u64) {
3352 self.root.edit(path.parts(), Below::WATCH, |node| {
3353 node.watches.retain(|(watch, _)| *watch != id)
3354 });
3355 }
3356
3357 fn poke_below(&self, prefix: &Path) {
3359 if let (_, Some(node)) = self.split(prefix) {
3360 node.poke_below();
3361 }
3362 }
3363
3364 fn poke_all(&self) {
3366 self.root.walk(&mut |node| node.poke());
3367 }
3368
3369 fn cursors_touching(&self, prefix: &Path) -> Vec<ConsumerId> {
3373 let (above, at) = self.split(prefix);
3374 let mut cursors: Vec<ConsumerId> = above.iter().flat_map(|node| node.cursors.iter().copied()).collect();
3375 if let Some(node) = at {
3376 node.collect_cursors(&mut cursors);
3377 }
3378 cursors.sort_unstable();
3380 cursors.dedup();
3381 cursors
3382 }
3383}
3384
3385struct OriginState {
3392 update_hold: Duration,
3395
3396 routes: RouteTable,
3399 next_route: u64,
3400 next_watch: u64,
3401
3402 cursors: HashMap<ConsumerId, TableCursor>,
3406
3407 fronts: WeakCache<FrontKey, RemoteFront>,
3413
3414 withdrawn: HashMap<PathOwned, HashSet<Hop>>,
3417
3418 closed: bool,
3421}
3422
3423impl OriginState {
3424 fn new(update_hold: Duration) -> Self {
3425 Self {
3426 update_hold,
3427 routes: RouteTable::default(),
3428 next_route: 0,
3429 next_watch: 0,
3430 cursors: HashMap::new(),
3431 fronts: WeakCache::default(),
3432 withdrawn: HashMap::new(),
3433 closed: false,
3434 }
3435 }
3436
3437 fn withdrawn_through(&self, prefix: &Path, hops: &Hops) -> bool {
3439 if self.withdrawn.is_empty() {
3440 return false;
3441 }
3442 self.withdrawn
3443 .get(prefix)
3444 .is_some_and(|peers| hops.iter().any(|hop| peers.contains(hop)))
3445 }
3446
3447 fn restale(&mut self, prefix: &Path) {
3449 let peers = self.withdrawn.get(prefix);
3450 let mut changed = None;
3451 for entry in self.routes.at_mut(prefix) {
3452 let stale = peers.is_some_and(|peers| entry.hops.iter().any(|hop| peers.contains(hop)));
3453 if entry.stale != stale {
3454 entry.stale = stale;
3455 changed = Some(entry.claim.clone());
3456 }
3457 }
3458 if let Some(claim) = changed {
3459 self.sync_route(prefix, &claim);
3460 }
3461 }
3462
3463 fn reannounced(&mut self, prefix: &PathOwned, hops: &Hops) {
3467 if let Some(sender) = hops.iter().last()
3468 && self.withdrawn.get_mut(prefix).is_some_and(|peers| peers.remove(sender))
3469 {
3470 self.restale(prefix);
3471 self.prune_withdrawn(prefix);
3472 }
3473 }
3474
3475 fn prune_withdrawn(&mut self, prefix: &PathOwned) {
3477 if self.withdrawn.is_empty() {
3478 return;
3479 }
3480 let Some(peers) = self.withdrawn.get_mut(prefix) else {
3481 return;
3482 };
3483 if peers.len() <= 1 {
3484 peers.retain(|peer| self.routes.at(prefix).any(|entry| entry.hops.contains(peer)));
3486 } else {
3487 let mut unreferenced = peers.clone();
3488 for entry in self.routes.at(prefix) {
3489 if unreferenced.is_empty() {
3490 break;
3491 }
3492 for hop in entry.hops.iter() {
3493 unreferenced.remove(hop);
3494 }
3495 }
3496 peers.retain(|peer| !unreferenced.contains(peer));
3497 }
3498 if peers.is_empty() {
3499 self.withdrawn.remove(prefix);
3500 }
3501 }
3502
3503 fn sync_route(&mut self, prefix: &Path, claim: &Pattern) {
3508 let routes = &self.routes;
3510 let mut candidates = None;
3511 for id in routes.cursors_touching(prefix) {
3512 let Some(cursor) = self.cursors.get_mut(&id) else {
3513 continue;
3514 };
3515 if let Some(presented) = cursor.presented(prefix, claim) {
3516 if presented.is_empty() {
3517 Self::sync_cursor(routes, cursor, &presented);
3519 } else {
3520 let candidates = candidates.get_or_insert_with(|| {
3523 let mut entries: Vec<_> = routes
3524 .at(prefix)
3525 .map(|entry| (route_order(prefix, entry), entry))
3526 .collect();
3527 entries.sort_unstable_by_key(|(order, _)| *order);
3528 entries
3529 });
3530 let best = candidates
3531 .iter()
3532 .map(|(_, entry)| *entry)
3533 .find(|entry| cursor.visible(entry));
3534 cursor.update(&presented, best);
3535 }
3536 }
3537 }
3538 routes.poke_below(prefix);
3540 }
3541
3542 fn watch(&mut self, shared: &kio::Shared<OriginState>, path: &Path) -> Watch {
3544 let id = self.next_watch;
3545 self.next_watch += 1;
3546 let signal = self.routes.add_watch(path, id);
3547 Watch {
3548 shared: shared.clone(),
3549 path: path.to_owned(),
3550 id,
3551 signal,
3552 }
3553 }
3554
3555 fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned) {
3558 let candidates: Vec<&RouteEntry> = match presented.is_empty() {
3564 true => routes
3565 .covering(&cursor.root)
3566 .filter(|entry| cursor.visible(entry))
3567 .collect(),
3568 false => {
3569 let absolute = cursor.root.join(presented);
3570 routes.at(&absolute).filter(|entry| cursor.visible(entry)).collect()
3571 }
3572 };
3573 let most = candidates.iter().map(|entry| entry.prefix.len()).max();
3574 let best = most.and_then(|most| {
3575 candidates
3576 .into_iter()
3577 .filter(|entry| entry.prefix.len() == most)
3578 .min_by_key(|entry| route_order(&entry.prefix, entry))
3579 });
3580
3581 cursor.update(presented, best);
3582 }
3583
3584 fn register_cursor(&mut self, id: ConsumerId, mut cursor: TableCursor) {
3586 let mut presented: BTreeSet<PathOwned> = BTreeSet::new();
3589 for head in &cursor.heads {
3590 let (above, at) = self.routes.split(head);
3591 let mut nodes = above;
3592 if let Some(node) = at {
3593 node.walk(&mut |node| nodes.push(node));
3594 }
3595 for entry in nodes.into_iter().flat_map(|node| node.entries.iter()) {
3596 if let Some(p) = cursor.presented(&entry.prefix, &entry.claim) {
3597 presented.insert(p);
3598 }
3599 }
3600 }
3601 for p in &presented {
3602 Self::sync_cursor(&self.routes, &mut cursor, p);
3603 }
3604 for head in &cursor.heads {
3605 self.routes.add_cursor(head, id);
3606 }
3607 self.cursors.insert(id, cursor);
3608 }
3609
3610 fn best_route(&self, path: &Path, horizon: Horizon, pin: Pin, refused: &HashSet<u64>) -> Option<&RouteEntry> {
3625 let (above, at) = self.routes.split(path);
3629 let mut best = None;
3630 for node in above.into_iter().chain(at) {
3631 let mut candidates = node
3632 .entries
3633 .iter()
3634 .filter(|entry| entry.live())
3635 .filter(|entry| entry.scope.matches(path.as_str()))
3636 .filter(|entry| horizon.admits(entry))
3637 .filter(|entry| entry.qualifies(pin))
3638 .filter(|entry| !refused.contains(&entry.id))
3639 .peekable();
3640 if candidates.peek().is_some() {
3641 best = candidates
3642 .filter(|entry| entry.serves(path))
3643 .min_by_key(|entry| route_order(&entry.prefix, entry));
3644 }
3645 }
3646 best
3647 }
3648}
3649
3650#[derive(Default)]
3657struct PendingBroadcast {
3658 resolved: Option<Result<broadcast::Consumer, Error>>,
3659}
3660
3661#[must_use = "dropping an origin::Dynamic retracts the route"]
3676pub struct Dynamic {
3677 announcement: AnnounceProducer,
3679 state: kio::Shared<ServeState>,
3680 _keepalive: Keepalive,
3684}
3685
3686impl Dynamic {
3687 pub(crate) fn withdrawn(self) {
3690 self.announcement.retract(true);
3691 }
3692
3693 pub fn update(&self, route: Route) -> Result<(), Error> {
3703 self.announcement.update(route)
3704 }
3705
3706 pub fn poll_requested_broadcast(&self, waiter: &kio::Waiter) -> Poll<Result<Request, Error>> {
3711 let mut state = ready!(self.state.poll(waiter, |state| {
3712 if state.closed || state.requests.has_queued() {
3713 Poll::Ready(())
3714 } else {
3715 Poll::Pending
3716 }
3717 }));
3718
3719 if state.closed {
3721 return Poll::Ready(Err(Error::Closed));
3722 }
3723
3724 let path = state.requests.pop().expect("predicate guaranteed a request");
3725 let producer = state.requests.get(&path).expect("popped key must be pending").clone();
3731 Poll::Ready(Ok(Request {
3732 path,
3733 producer,
3734 home: self.state.clone(),
3735 }))
3736 }
3737
3738 pub async fn requested_broadcast(&self) -> Result<Request, Error> {
3744 kio::wait(|waiter| self.poll_requested_broadcast(waiter)).await
3745 }
3746}
3747
3748impl ServeState {
3749 fn resolve(
3760 shared: &kio::Shared<Self>,
3761 path: &PathOwned,
3762 producer: &kio::Producer<PendingBroadcast>,
3763 result: Result<broadcast::Consumer, Error>,
3764 ) {
3765 let mut state = shared.lock();
3766 if state.closed {
3767 return;
3768 }
3769 let resolved = match result {
3770 Ok(broadcast) => {
3771 let existing = state.served.insert(path.clone(), broadcast.weak());
3775 Ok(existing.map(|weak| weak.consume()).unwrap_or(broadcast))
3776 }
3777 Err(err) => Err(err),
3778 };
3779 state.requests.remove_if(path, |p| p.same_channel(producer));
3780 if let Ok(mut pending) = producer.write() {
3781 pending.resolved.get_or_insert(resolved);
3782 drop(state);
3783 }
3784 }
3785
3786 fn forget(shared: &kio::Shared<Self>, path: &PathOwned, producer: &kio::Producer<PendingBroadcast>) {
3788 shared.lock().requests.remove_if(path, |p| p.same_channel(producer));
3789 }
3790}
3791
3792pub struct Request {
3799 path: PathOwned,
3801
3802 producer: kio::Producer<PendingBroadcast>,
3805
3806 home: kio::Shared<ServeState>,
3809}
3810
3811impl Request {
3812 pub fn path(&self) -> &Path<'_> {
3814 &self.path
3815 }
3816
3817 pub fn accept(self, broadcast: impl Consume<broadcast::Consumer>) {
3823 let broadcast = broadcast.consume();
3824 ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast));
3825 }
3827
3828 pub fn reject(self, err: Error) {
3830 ServeState::resolve(&self.home, &self.path, &self.producer, Err(err));
3831 }
3832}
3833
3834impl Drop for Request {
3835 fn drop(&mut self) {
3836 ServeState::forget(&self.home, &self.path, &self.producer);
3844 }
3845}
3846
3847pub struct Requesting {
3854 inner: RequestState,
3855 path: PathOwned,
3859 stats: stats::Scope,
3862}
3863
3864enum RequestState {
3865 Failed(Error),
3868 Pending(kio::Consumer<PendingBroadcast>),
3870}
3871
3872impl Requesting {
3873 fn failed(error: Error) -> Self {
3874 Self::new(RequestState::Failed(error))
3875 }
3876
3877 fn queued(consumer: kio::Consumer<PendingBroadcast>) -> Self {
3878 Self::new(RequestState::Pending(consumer))
3879 }
3880
3881 pub fn is_queued(&self) -> bool {
3890 matches!(self.inner, RequestState::Pending(_))
3891 }
3892
3893 fn new(inner: RequestState) -> Self {
3894 Self {
3895 inner,
3896 path: PathOwned::default(),
3897 stats: stats::Scope::default(),
3898 }
3899 }
3900
3901 fn with_path(mut self, path: PathOwned) -> Self {
3902 self.path = path;
3903 self
3904 }
3905
3906 fn with_stats(mut self, scope: stats::Scope) -> Self {
3908 self.stats = scope;
3909 self
3910 }
3911
3912 fn hand_out(&self, broadcast: broadcast::Consumer) -> broadcast::Consumer {
3914 broadcast.with_path(self.path.clone()).with_stats(self.stats.clone())
3915 }
3916
3917 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<broadcast::Consumer, Error>> {
3919 match &self.inner {
3920 RequestState::Failed(error) => Poll::Ready(Err(error.clone())),
3921 RequestState::Pending(consumer) => Poll::Ready(
3922 match ready!(consumer.poll(waiter, |state| match &state.resolved {
3923 Some(result) => Poll::Ready(result.clone()),
3924 None => Poll::Pending,
3925 })) {
3926 Ok(result) => result.map(|broadcast| self.hand_out(broadcast)),
3927 Err(_closed) => Err(Error::Unroutable),
3929 },
3930 ),
3931 }
3932 }
3933}
3934
3935impl kio::Pollable for Requesting {
3936 type Output = Result<broadcast::Consumer, Error>;
3937
3938 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3939 self.poll_ok(waiter)
3940 }
3941}
3942
3943pub trait Consume<T> {
3951 fn consume(&self) -> T;
3953}
3954
3955impl<T, U: Consume<T>> Consume<T> for &U {
3956 fn consume(&self) -> T {
3957 (**self).consume()
3958 }
3959}
3960
3961impl Consume<Consumer> for Producer {
3962 fn consume(&self) -> Consumer {
3963 Consumer::from_producer(self, stats::Session::default())
3967 }
3968}
3969
3970impl Consume<Consumer> for Consumer {
3971 fn consume(&self) -> Consumer {
3972 self.clone()
3973 }
3974}
3975
3976impl Consume<broadcast::Consumer> for broadcast::Producer {
3977 fn consume(&self) -> broadcast::Consumer {
3978 self.consume()
3980 }
3981}
3982
3983impl Consume<broadcast::Consumer> for broadcast::Consumer {
3984 fn consume(&self) -> broadcast::Consumer {
3985 self.clone()
3986 }
3987}
3988
3989impl Consume<track::Consumer> for track::Producer {
3990 fn consume(&self) -> track::Consumer {
3991 self.consume()
3992 }
3993}
3994
3995impl Consume<track::Consumer> for track::Consumer {
3996 fn consume(&self) -> track::Consumer {
3997 self.clone()
3998 }
3999}
4000
4001#[derive(Clone)]
4007pub struct Consumer {
4008 hop: Hop,
4010 scope: OriginScope,
4011
4012 root: PathOwned,
4014
4015 shared: kio::Shared<OriginState>,
4018
4019 stats: stats::Session,
4023
4024 horizon: Horizon,
4029
4030 hidden: Hidden,
4032
4033 pool: cache::Pool,
4036 cache_duration: Duration,
4037
4038 tasks: TasksWeak,
4042
4043 timers: Clock,
4046}
4047
4048impl Consumer {
4049 fn from_producer(producer: &Producer, stats: stats::Session) -> Self {
4050 Self {
4051 hop: producer.hop,
4052 scope: producer.scope.clone(),
4053 root: producer.root.clone(),
4054 shared: producer.shared.clone(),
4055 stats,
4056 horizon: Horizon::default(),
4057 hidden: Hidden::default(),
4058 pool: producer.pool.clone(),
4059 cache_duration: producer.cache_duration,
4060 tasks: producer.tasks.downgrade(),
4061 timers: producer.timers.clone(),
4062 }
4063 }
4064
4065 pub fn hop(&self) -> Hop {
4067 self.hop
4068 }
4069
4070 pub(crate) fn excluding(mut self, peer: Hop) -> Self {
4078 self.horizon.exclude = Some(peer);
4079 self
4080 }
4081
4082 pub fn local(mut self) -> Self {
4089 self.horizon.local = true;
4090 self
4091 }
4092
4093 pub fn with_hidden(mut self, hidden: bool) -> Self {
4098 self.hidden.include = hidden;
4099 self
4100 }
4101
4102 pub(crate) fn beyond(mut self, outer: &Consumer) -> Self {
4105 self.hidden.beyond = Some(
4106 outer
4107 .hidden
4108 .from
4109 .clone()
4110 .map(|head| vec![head])
4111 .unwrap_or_else(|| interest_prefixes(&outer.scope.allowed)),
4112 );
4113 self
4114 }
4115
4116 pub(crate) fn discovery(mut self, hidden: bool) -> Self {
4118 self.hidden.include = hidden;
4119 self.hidden.from = Some(self.root.clone());
4120 self
4121 }
4122
4123 pub(crate) fn includes_hidden(&self) -> bool {
4125 self.hidden.include
4126 }
4127
4128 pub fn with_stats(mut self, session: stats::Session) -> Self {
4132 self.stats = session;
4133 self
4134 }
4135
4136 fn untagged(&self) -> Self {
4140 Self {
4141 stats: stats::Session::default(),
4142 ..self.clone()
4143 }
4144 }
4145
4146 pub(crate) fn empty(&self) -> Self {
4151 Self {
4152 scope: OriginScope::empty(),
4153 ..self.clone()
4154 }
4155 }
4156
4157 pub fn announced(&self) -> AnnounceConsumer {
4166 let state = kio::Producer::<OriginConsumerState>::default();
4167 let cursor = |root: PathOwned,
4168 allowed: Patterns,
4169 hidden: Hidden,
4170 mount: Option<&Mount>,
4171 under: PathOwned,
4172 holes: Vec<PathOwned>| TableCursor {
4173 root,
4174 heads: interest_prefixes(&allowed),
4175 allowed,
4176 horizon: self.horizon,
4177 hidden,
4178 state: state.clone(),
4179 under,
4180 holes,
4181 named: mount.map(|mount| (mount.clone(), self.scope.allowed.clone())),
4182 current: HashMap::new(),
4183 };
4184
4185 if let Some(mount) = self.scope.mount(&self.root) {
4188 let Some(root) = mount.resolve(&self.root) else {
4190 return AnnounceConsumer::new(
4191 self.root.clone(),
4192 Vec::new(),
4193 state,
4194 self.stats.clone(),
4195 &self.shared,
4196 self.timers.clone(),
4197 );
4198 };
4199 let cursors = vec![cursor(
4200 root,
4201 mount.translate(&self.scope.allowed),
4202 self.hidden.translate(mount),
4203 Some(mount),
4204 PathOwned::default(),
4205 Vec::new(),
4206 )];
4207 return AnnounceConsumer::new(
4208 self.root.clone(),
4209 cursors,
4210 state,
4211 self.stats.clone(),
4212 &self.shared,
4213 self.timers.clone(),
4214 );
4215 }
4216
4217 let heads = interest_prefixes(&self.scope.allowed);
4220 let mut holes = Vec::new();
4221 let mut cursors = Vec::new();
4222 for mount in self.scope.mounts.iter() {
4223 let Some(under) = mount.at.strip_prefix(&self.root) else {
4224 continue;
4225 };
4226 holes.push(mount.at.clone());
4227 let allowed = mount.translate(&self.scope.allowed);
4228 if allowed.is_empty()
4231 || !(self.hidden.include
4232 || !hides(
4233 self.hidden.from.as_ref().map(std::slice::from_ref).unwrap_or(&heads),
4234 &mount.at,
4235 )) {
4236 continue;
4237 }
4238 cursors.push(cursor(
4239 mount.target.clone(),
4240 allowed,
4241 self.hidden.translate(mount),
4242 Some(mount),
4243 under.to_owned(),
4244 Vec::new(),
4245 ));
4246 }
4247 cursors.insert(
4248 0,
4249 cursor(
4250 self.root.clone(),
4251 self.scope.allowed.clone(),
4252 self.hidden.clone(),
4253 None,
4254 PathOwned::default(),
4255 holes,
4256 ),
4257 );
4258 AnnounceConsumer::new(
4259 self.root.clone(),
4260 cursors,
4261 state,
4262 self.stats.clone(),
4263 &self.shared,
4264 self.timers.clone(),
4265 )
4266 }
4267
4268 pub fn consume(&self) -> Self {
4270 self.clone()
4271 }
4272
4273 #[cfg(test)]
4276 pub(crate) fn get_broadcast(&self, path: impl AsPath) -> Option<broadcast::Consumer> {
4277 let full = self.root.join(path).to_owned();
4278 if !self.scope.permits(&full) {
4279 return None;
4280 }
4281 let full = self.scope.resolve(&full)?;
4282 let table = self.shared.lock();
4283 table
4284 .routes
4285 .at(&full)
4286 .filter(|entry| entry.local)
4287 .min_by_key(|entry| route_order(&entry.prefix, entry))
4288 .and_then(|entry| entry.source.clone())
4289 }
4290
4291 pub async fn routed(&self, path: impl AsPath) -> Option<Route> {
4301 let path = path.as_path();
4302
4303 let consumer = match Pattern::subtree(path.as_str()) {
4308 Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?,
4309 Err(InvalidPattern::TooManySegments) => self.clone(),
4310 Err(_) => return None,
4311 };
4312
4313 if !consumer.allowed().matches(path.as_str()) {
4317 return None;
4318 }
4319
4320 let mut announced = consumer.untagged().with_hidden(true).announced();
4324 loop {
4325 let update = announced.next().await?;
4326 if update.kind.is_active() && path.has_prefix(&update.prefix) {
4327 return Some(update.route);
4328 }
4329 }
4330 }
4331
4332 pub async fn routed_broadcast(&self, path: impl AsPath) -> Result<broadcast::Consumer, Error> {
4346 let path = path.as_path();
4347
4348 if !self.allowed().matches(path.as_str()) {
4351 return Err(Error::Unauthorized);
4352 }
4353 loop {
4354 let (watch, seen) = {
4363 let mut table = self.shared.lock();
4364 if table.closed {
4365 return Err(Error::Closed);
4366 }
4367 let named = self.root.join(&path);
4368 let resolved = self.scope.resolve(&named).ok_or(BoundsExceeded)?;
4369 let watch = table.watch(&self.shared, &resolved);
4370 let seen = watch.seen();
4371 (watch, seen)
4372 };
4373 match self.request_broadcast(&path).await {
4374 Ok(broadcast) => return Ok(broadcast),
4375 Err(Error::Unroutable) => {
4376 kio::wait(|waiter| watch.poll_changed(waiter, seen)).await;
4377 }
4378 Err(Error::Dropped) if self.shared.lock().closed => return Err(Error::Closed),
4381 Err(err) => return Err(err),
4382 }
4383 }
4384 }
4385
4386 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Consumer, Error> {
4393 let root = self.root.join(root).to_owned();
4394 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
4395 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
4396 Ok(Consumer {
4397 scope,
4398 root,
4399 ..self.clone()
4400 })
4401 }
4402
4403 pub fn request_broadcast(&self, path: impl AsPath) -> kio::Pending<Requesting> {
4426 let path = path.as_path();
4427
4428 let named = self.root.join(&path).to_owned();
4432 let scope = self.stats.egress(&named);
4433 let requested = path.to_owned();
4437
4438 if !self.scope.permits(&named) {
4440 return kio::Pending::new(Requesting::failed(Error::Unauthorized));
4441 }
4442
4443 let Some(absolute) = self.scope.resolve(&named).map(|path| path.to_owned()) else {
4446 return kio::Pending::new(Requesting::failed(BoundsExceeded.into()));
4447 };
4448
4449 let mut state = self.shared.lock();
4450
4451 if state.closed {
4453 return kio::Pending::new(Requesting::failed(Error::Closed));
4454 }
4455
4456 let horizon = self.horizon.effective(&state.routes, &absolute.as_path());
4457
4458 if state
4462 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4463 .is_none()
4464 {
4465 return kio::Pending::new(Requesting::failed(Error::Unroutable));
4466 }
4467
4468 let key = (absolute.clone(), horizon);
4476 if let Some(front) = state.fronts.get(&key) {
4477 let pin = *front.pin.lock();
4478 let current = state
4479 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4480 .is_some_and(|entry| entry.qualifies(pin));
4481 if current {
4482 let pending = Requesting::queued(front.request.consume())
4483 .with_path(requested)
4484 .with_stats(scope);
4485 return kio::Pending::new(pending);
4486 }
4487 state.fronts.remove(&key);
4488 }
4489
4490 let broadcast = broadcast::Producer::new_spliced(broadcast::Info {
4495 pool: self.pool.clone(),
4496 cache_duration: self.cache_duration,
4497 path: absolute.clone(),
4498 });
4499 let request = kio::Producer::<PendingBroadcast>::default();
4500 let consumer = request.consume();
4501 let watch = state.watch(&self.shared, &absolute);
4502 let pin = kio::Lock::new(Pin::Any);
4503 state.fronts.insert(
4504 key,
4505 RemoteFront {
4506 request: request.clone(),
4507 broadcast: broadcast.consume().weak(),
4508 pin: pin.clone(),
4509 },
4510 );
4511 drop(state);
4514 self.tasks.push(run_front(FrontTask {
4515 shared: self.shared.clone(),
4516 broadcast,
4517 path: absolute,
4518 horizon,
4519 watch,
4520 request,
4521 pin,
4522 timers: self.timers.clone(),
4523 }));
4524 kio::Pending::new(Requesting::queued(consumer).with_path(requested).with_stats(scope))
4525 }
4526
4527 pub fn root(&self) -> &Path<'_> {
4529 &self.root
4530 }
4531
4532 pub fn allowed(&self) -> Patterns {
4534 self.scope.relative(&self.root)
4535 }
4536
4537 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
4539 self.root.join(path)
4540 }
4541}
4542
4543pub struct AnnounceConsumer {
4548 ids: Vec<ConsumerId>,
4551 shared: kio::Shared<OriginState>,
4552 root: PathOwned,
4553
4554 state: kio::Producer<OriginConsumerState>,
4557
4558 stats: stats::Session,
4561
4562 guards: HashMap<PathOwned, stats::Announce>,
4566
4567 park: kio::Park,
4570
4571 update_hold: Duration,
4574 held: HashMap<PathOwned, Held>,
4575 retracted: HashMap<PathOwned, Quarantine>,
4578 anchoring: VecDeque<(Instant, PathOwned)>,
4579 expiring: VecDeque<(Instant, PathOwned)>,
4580 timers: Clock,
4581 hold: crate::runtime::Deadline<Clock>,
4582}
4583
4584impl AnnounceConsumer {
4585 fn new(
4586 root: PathOwned,
4587 cursors: Vec<TableCursor>,
4588 state: kio::Producer<OriginConsumerState>,
4589 stats: stats::Session,
4590 shared: &kio::Shared<OriginState>,
4591 timers: Clock,
4592 ) -> Self {
4593 let mut ids = Vec::with_capacity(cursors.len());
4594 let update_hold;
4595 {
4596 let mut table = shared.lock();
4597 update_hold = table.update_hold;
4598 if table.closed {
4599 if let Ok(mut state) = state.write() {
4601 state.ended = true;
4602 }
4603 } else {
4604 for cursor in cursors {
4605 let id = ConsumerId::new();
4606 table.register_cursor(id, cursor);
4607 ids.push(id);
4608 }
4609 }
4610 }
4611
4612 Self {
4613 ids,
4614 shared: shared.clone(),
4615 root,
4616 state,
4617 stats,
4618 guards: HashMap::new(),
4619 park: kio::Park::default(),
4620 update_hold,
4621 held: HashMap::new(),
4622 retracted: HashMap::new(),
4623 anchoring: VecDeque::new(),
4624 expiring: VecDeque::new(),
4625 hold: crate::runtime::Deadline::new(&timers),
4626 timers,
4627 }
4628 }
4629
4630 fn quarantine(&mut self, update: &AnnounceUpdate, now: Option<Instant>) {
4633 if update.kind.is_active() {
4634 return;
4635 }
4636 let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else {
4637 return;
4638 };
4639 let publisher = update
4641 .route
4642 .hops
4643 .iter()
4644 .next()
4645 .copied()
4646 .filter(|hop| *hop != Hop::UNKNOWN);
4647 let quarantine = Quarantine {
4649 held: Held::Anchoring(now),
4650 publisher,
4651 rank: (update.route.cost, update.route.hops.len()),
4652 };
4653 let replaced = self.retracted.insert(update.prefix.clone(), quarantine);
4654 if !replaced.is_some_and(|q| q.held == Held::Anchoring(now)) {
4657 self.anchoring.push_back((now, update.prefix.clone()));
4658 }
4659 }
4660
4661 fn settle(&mut self, now: Option<Instant>) -> Option<Instant> {
4664 let now = now?;
4665 while let Some((seen, prefix)) = self.anchoring.front() {
4668 if *seen >= now {
4669 break;
4670 }
4671 if let Some(quarantine) = self.retracted.get_mut(prefix)
4672 && quarantine.held == Held::Anchoring(*seen)
4673 {
4674 quarantine.held = Held::Until(now + RETRACT_QUARANTINE);
4675 self.expiring.push_back((now + RETRACT_QUARANTINE, prefix.clone()));
4676 }
4677 self.anchoring.pop_front();
4678 }
4679 while let Some((until, prefix)) = self.expiring.front() {
4680 if *until > now {
4681 break;
4682 }
4683 if self
4684 .retracted
4685 .get(prefix)
4686 .is_some_and(|q| q.held == Held::Until(*until))
4687 {
4688 self.retracted.remove(prefix);
4689 }
4690 self.expiring.pop_front();
4691 }
4692 let anchoring = self.anchoring.front().map(|(seen, _)| *seen + Duration::from_nanos(1));
4693 earliest(anchoring, self.expiring.front().map(|(until, _)| *until))
4694 }
4695
4696 fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate {
4698 let absolute = self.root.join(&update.prefix).to_owned();
4699 if update.kind.is_active() {
4700 let scope = self.stats.egress(&absolute);
4701 self.guards
4702 .entry(update.prefix.clone())
4703 .or_insert_with(|| scope.announce());
4704 } else {
4705 self.guards.remove(&update.prefix);
4706 }
4707 update
4708 }
4709
4710 pub async fn next(&mut self) -> Option<AnnounceUpdate> {
4718 kio::wait(|waiter| self.poll_next(waiter)).await
4719 }
4720
4721 pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<AnnounceUpdate>> {
4727 loop {
4728 let now = self.timers.try_now();
4729 let settling = self.settle(now);
4730 let hold = self.update_hold;
4731 let held = &mut self.held;
4732 let retracted = &self.retracted;
4733 let mut ready = None;
4734 let mut wake = None;
4735 let update = match self.state.poll(waiter, |state| {
4736 if state.pending.is_empty() {
4737 return match state.ended {
4738 true => Poll::Ready(()),
4739 false => Poll::Pending,
4740 };
4741 }
4742 match state.scan(held, retracted, now, hold) {
4743 Ok(prefix) => {
4744 ready = Some(prefix);
4745 Poll::Ready(())
4746 }
4747 Err(at) => {
4748 wake = at;
4749 Poll::Pending
4750 }
4751 }
4752 }) {
4753 Poll::Ready(Ok(mut state)) => match ready {
4754 Some(prefix) => state.take_prefix(prefix),
4755 None => {
4756 state.close();
4759 None
4760 }
4761 },
4762 Poll::Ready(Err(_)) => None,
4764 Poll::Pending => {
4765 self.hold.set(earliest(wake, settling));
4766 ready!(self.hold.poll(waiter));
4767 continue;
4768 }
4769 };
4770 let Some(update) = update else {
4771 return Poll::Ready(None);
4772 };
4773 self.held.remove(&update.prefix);
4774 self.quarantine(&update, now);
4775 return Poll::Ready(Some(self.hand_out(update)));
4776 }
4777 }
4778
4779 pub fn try_next(&mut self) -> Option<AnnounceUpdate> {
4784 let now = self.timers.try_now();
4785 let settling = self.settle(now);
4786 let mut state = self.state.write().ok()?;
4787 let prefix = match state.scan(&mut self.held, &self.retracted, now, self.update_hold) {
4788 Ok(prefix) => prefix,
4789 Err(wake) => {
4790 let wake = earliest(wake, settling);
4791 drop(state);
4794 self.hold.set(wake);
4795 return None;
4796 }
4797 };
4798 self.held.remove(&prefix);
4799 let update = state.take_prefix(prefix)?;
4800 drop(state);
4801 self.quarantine(&update, now);
4802 Some(self.hand_out(update))
4803 }
4804
4805 pub fn is_closed(&self) -> bool {
4807 let state = self.state.read();
4808 state.is_closed() || state.ended
4809 }
4810
4811 pub fn root(&self) -> &Path<'_> {
4813 &self.root
4814 }
4815
4816 pub fn absolute(&self, prefix: impl AsPath) -> Path<'_> {
4818 self.root.join(prefix)
4819 }
4820}
4821
4822impl futures::Stream for AnnounceConsumer {
4823 type Item = AnnounceUpdate;
4824
4825 fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Option<Self::Item>> {
4826 let this = self.get_mut();
4827 let waiter = this.park.hold(cx).clone();
4828 this.poll_next(&waiter)
4829 }
4830}
4831
4832impl Drop for AnnounceConsumer {
4833 fn drop(&mut self) {
4834 let mut shared = self.shared.lock();
4835 for id in &self.ids {
4836 if let Some(cursor) = shared.cursors.remove(id) {
4837 for head in &cursor.heads {
4838 shared.routes.remove_cursor(head, *id);
4839 }
4840 }
4841 }
4842 }
4843}
4844
4845#[cfg(test)]
4846use futures::FutureExt;
4847
4848#[cfg(test)]
4849#[allow(missing_docs)] impl AnnounceConsumer {
4851 pub fn assert_next_active(&mut self, expected: impl AsPath) -> Route {
4853 let expected = expected.as_path();
4854 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4855 assert_eq!(update.prefix, expected, "wrong prefix");
4856 assert!(update.kind.is_active(), "should be an active route");
4857 update.route
4858 }
4859
4860 pub fn assert_try_next_active(&mut self, expected: impl AsPath) -> Route {
4862 let expected = expected.as_path();
4863 let update = self.try_next().expect("no next");
4864 assert_eq!(update.prefix, expected, "wrong prefix");
4865 assert!(update.kind.is_active(), "should be an active route");
4866 update.route
4867 }
4868
4869 pub fn assert_next_ended(&mut self, expected: impl AsPath) {
4871 let expected = expected.as_path();
4872 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4873 assert_eq!(update.prefix, expected, "wrong prefix");
4874 assert_eq!(update.kind, AnnounceKind::Retracted, "should be a retraction");
4875 }
4876
4877 pub fn assert_next_wait(&mut self) {
4878 if let Some(res) = self.next().now_or_never() {
4879 panic!("next should block: got {:?}", res.map(|u| u.prefix));
4880 }
4881 }
4882}
4883
4884#[cfg(test)]
4888pub(crate) trait ProduceTest {
4889 fn produce(self) -> Producer;
4890}
4891
4892#[cfg(test)]
4893impl ProduceTest for Config {
4894 fn produce(self) -> Producer {
4895 let (producer, driver) = Producer::new(self);
4896 if tokio::runtime::Handle::try_current().is_ok() {
4897 tokio::spawn(crate::time::run(driver));
4898 } else {
4899 std::mem::forget(driver);
4902 }
4903 producer
4904 }
4905}
4906
4907#[cfg(test)]
4908impl ProduceTest for Hop {
4909 fn produce(self) -> Producer {
4910 Config::new(self).produce()
4911 }
4912}
4913
4914#[cfg(test)]
4915mod tests {
4916 use super::*;
4917 use futures::FutureExt;
4918
4919 fn origin(id: u64) -> Hop {
4920 Hop::new(id).unwrap()
4921 }
4922
4923 fn hops(ids: &[u64]) -> Hops {
4924 let mut list = Hops::new();
4925 for &id in ids {
4926 list.push(if id == 0 { Hop::UNKNOWN } else { origin(id) }).unwrap();
4927 }
4928 list
4929 }
4930
4931 fn scopes(prefixes: &[&str]) -> Patterns {
4933 prefixes
4934 .iter()
4935 .map(|prefix| Pattern::subtree(prefix).unwrap())
4936 .collect()
4937 }
4938
4939 #[test]
4940 fn default_config_mints_a_real_hop() {
4941 let config = Config::default();
4942 assert_ne!(config.hop, Hop::UNKNOWN);
4943 let (producer, _driver) = Producer::new(config.clone());
4944 assert_eq!(producer.hop(), config.hop);
4945 assert_eq!(producer.consume().hop(), config.hop);
4946 }
4947
4948 #[test]
4949 fn random_hops_fit_legacy_lite_clients() {
4950 for _ in 0..32 {
4951 assert!(Hop::random().id() < 1u64 << 53);
4952 }
4953 }
4954
4955 async fn settle(mut check: impl FnMut() -> bool) {
4958 for _ in 0..100 {
4959 if check() {
4960 return;
4961 }
4962 tokio::task::yield_now().await;
4963 }
4964 panic!("condition never settled");
4965 }
4966
4967 async fn queued(server: &Dynamic) -> Request {
4969 let mut request = None;
4970 settle(|| match server.poll_requested_broadcast(&kio::Waiter::noop()) {
4971 Poll::Ready(Ok(popped)) => {
4972 request = Some(popped);
4973 true
4974 }
4975 _ => false,
4976 })
4977 .await;
4978 request.unwrap()
4979 }
4980
4981 async fn next_group(subscription: &mut crate::track::Subscriber) -> Result<Option<crate::group::Consumer>, Error> {
4983 let mut next = None;
4984 settle(|| match subscription.poll_recv_group(&kio::Waiter::noop()) {
4985 Poll::Ready(result) => {
4986 next = Some(result);
4987 true
4988 }
4989 Poll::Pending => false,
4990 })
4991 .await;
4992 next.unwrap()
4993 }
4994
4995 #[tokio::test]
4996 async fn announce_and_retract() {
4997 let producer = origin(1).produce();
4998 let consumer = producer.consume();
4999 let mut announced = consumer.announced();
5000 announced.assert_next_wait();
5001
5002 let announcement = producer.announce("room/alice", Route::default()).unwrap();
5003 let route = announced.assert_next_active("room/alice");
5004 assert!(route.hops.is_empty());
5005 assert_eq!(route.cost, Cost::default());
5006 announced.assert_next_wait();
5007
5008 drop(announcement);
5009 announced.assert_next_ended("room/alice");
5010 announced.assert_next_wait();
5011 }
5012
5013 #[tokio::test]
5016 async fn hidden_routes_need_an_opt_in() {
5017 let producer = origin(1).produce();
5018 let consumer = producer.consume();
5019 let _visible = producer.announce("room/alice", Route::default()).unwrap();
5020 let _stats = producer.announce(".stats/node", Route::default()).unwrap();
5021 let _nested = producer.announce("room/.internal", Route::default()).unwrap();
5022 let _suffix = producer.announce("room/catalog.pro", Route::default()).unwrap();
5024
5025 let mut announced = consumer.announced();
5026 announced.assert_next_active("room/alice");
5027 announced.assert_next_active("room/catalog.pro");
5028 announced.assert_next_wait();
5029
5030 let mut announced = consumer.clone().with_hidden(true).announced();
5031 announced.assert_next_active(".stats/node");
5032 announced.assert_next_active("room/.internal");
5033 announced.assert_next_active("room/alice");
5034 announced.assert_next_active("room/catalog.pro");
5035 announced.assert_next_wait();
5036
5037 let mut announced = consumer
5039 .scope(".stats", &Patterns::from(Pattern::all()))
5040 .unwrap()
5041 .announced();
5042 announced.assert_next_active("node");
5043 announced.assert_next_wait();
5044 let mut announced = consumer.scope("", &scopes(&["room/.internal"])).unwrap().announced();
5045 announced.assert_next_active("room/.internal");
5046 announced.assert_next_wait();
5047
5048 let mut announced = consumer.clone().with_hidden(true).beyond(&consumer).announced();
5051 announced.assert_next_active(".stats/node");
5052 announced.assert_next_active("room/.internal");
5053 announced.assert_next_wait();
5054 let room = consumer.scope("", &scopes(&["room"])).unwrap().beyond(&consumer);
5055 let mut announced = room.announced();
5056 announced.assert_next_wait();
5057 let stats = consumer.scope("", &scopes(&[".stats"])).unwrap().beyond(&consumer);
5058 let mut announced = stats.announced();
5059 announced.assert_next_active(".stats/node");
5060 announced.assert_next_wait();
5061 }
5062
5063 #[tokio::test]
5065 async fn hidden_broadcast_resolves_by_path() {
5066 let producer = origin(1).produce();
5067 let consumer = producer.consume();
5068 let broadcast = producer.create_broadcast(".stats/node").unwrap();
5069 broadcast.announce(Route::default()).unwrap();
5070
5071 consumer.announced().assert_next_wait();
5072 let resolved = consumer.request_broadcast(".stats/node").await.expect("resolves");
5073 assert_eq!(resolved.info().path.as_str(), ".stats/node");
5074 }
5075
5076 fn mounted(producer: &Producer, pid: &str, patterns: &[&str]) -> Consumer {
5078 let patterns: Patterns = patterns.iter().map(|pattern| pattern.parse().unwrap()).collect();
5079 producer
5080 .mount(format!("{pid}/.svc"), format!(".svc/{pid}"))
5081 .unwrap()
5082 .scope(pid, &patterns)
5083 .unwrap()
5084 .consume()
5085 }
5086
5087 #[tokio::test]
5090 async fn mount_resolves_on_the_target_front() {
5091 let producer = origin(1).produce();
5092 let server = producer.dynamic(".svc", Route::default()).unwrap();
5093 let project = mounted(&producer, "p1", &["**"]);
5094
5095 let through = project.request_broadcast(".svc/foo");
5096 let direct = producer.consume().request_broadcast(".svc/p1/foo");
5097
5098 let request = queued(&server).await;
5099 assert_eq!(request.path().as_str(), ".svc/p1/foo");
5100 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5101 let source = broadcast::Info::new().produce();
5102 request.accept(&source);
5103
5104 let through = through.await.expect("resolves");
5105 let direct = direct.await.expect("resolves");
5106 assert!(through.is_clone(&direct));
5107 assert_eq!(through.info().path.as_str(), ".svc/foo");
5109 }
5110
5111 #[tokio::test]
5114 async fn mount_presents_target_routes_under_the_mount() {
5115 let producer = origin(1).produce();
5116 let _claim = producer.announce(".svc", Route::default()).unwrap();
5117 let foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5118 let _other = producer.publish(".svc/p2/bar", Route::default()).unwrap();
5119 let _cam = producer.publish("p1/cam", Route::default()).unwrap();
5120 let _shadowed = producer.publish("p1/.svc/forged", Route::default()).unwrap();
5121 let project = mounted(&producer, "p1", &["**"]);
5122
5123 let mut announced = project.announced();
5125 announced.assert_next_active("cam");
5126 announced.assert_next_wait();
5127
5128 let mut announced = project.clone().with_hidden(true).announced();
5129 announced.assert_next_active(".svc");
5130 announced.assert_next_active(".svc/foo");
5131 announced.assert_next_active("cam");
5132 announced.assert_next_wait();
5133
5134 let mut inside = project
5136 .scope(".svc", &Patterns::from(Pattern::all()))
5137 .unwrap()
5138 .announced();
5139 inside.assert_next_active("");
5140 inside.assert_next_active("foo");
5141 inside.assert_next_wait();
5142
5143 drop(foo);
5144 announced.assert_next_ended(".svc/foo");
5145 inside.assert_next_ended("foo");
5146 announced.assert_next_wait();
5147
5148 let err = project.request_broadcast(".svc/forged").await.err().unwrap();
5150 assert!(matches!(err, Error::NotFound | Error::Unroutable), "{err:?}");
5151 }
5152
5153 #[tokio::test]
5156 async fn mount_authorizes_the_named_path() {
5157 let producer = origin(1).produce();
5158 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5159 let _bar = producer.publish(".svc/p1/bar", Route::default()).unwrap();
5160
5161 let granted = mounted(&producer, "p1", &["foo", ".svc/foo"]);
5162 granted.request_broadcast(".svc/foo").await.expect("granted");
5163 let refused = granted
5164 .request_broadcast(".svc/bar")
5165 .now_or_never()
5166 .expect("refused at once");
5167 assert!(matches!(refused, Err(Error::Unauthorized)));
5168 let mut announced = granted.clone().with_hidden(true).announced();
5169 announced.assert_next_active(".svc/foo");
5170 announced.assert_next_wait();
5171
5172 let narrower = mounted(&producer, "p1", &["foo"]);
5173 let refused = narrower
5174 .request_broadcast(".svc/foo")
5175 .now_or_never()
5176 .expect("refused at once");
5177 assert!(matches!(refused, Err(Error::Unauthorized)));
5178 narrower.clone().with_hidden(true).announced().assert_next_wait();
5179
5180 let other = mounted(&producer, "p2", &["**"]);
5182 other.clone().with_hidden(true).announced().assert_next_wait();
5183 let err = other.request_broadcast(".svc/foo").await.err().unwrap();
5184 assert!(matches!(err, Error::Unroutable));
5185 }
5186
5187 #[tokio::test]
5190 async fn mount_captures_the_named_path() {
5191 let producer = origin(1).produce();
5192 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5193 let project = mounted(&producer, "p1", &["**"]).with_hidden(true);
5194
5195 let update = project.announced().try_next().expect("foo");
5196 assert_eq!(update.prefix.as_str(), ".svc/foo");
5197 assert_eq!(update.captures, Some(vec![".svc/foo".parse::<Pattern>().unwrap()]));
5198
5199 let inside = project.scope(".svc", &Patterns::from(Pattern::all())).unwrap();
5200 let update = inside.announced().try_next().expect("foo");
5201 assert_eq!(update.prefix.as_str(), "foo");
5202 assert_eq!(update.captures, Some(vec!["foo".parse::<Pattern>().unwrap()]));
5203 }
5204
5205 #[tokio::test]
5208 async fn mount_keeps_a_max_depth_target() {
5209 let producer = origin(1).produce();
5210 let deep = vec!["d"; Path::MAX_PARTS].join("/");
5211 let _leaf = producer.publish(deep.as_str(), Route::default()).unwrap();
5212 let project = producer
5213 .mount("p1/.svc", deep.as_str())
5214 .unwrap()
5215 .scope("p1", &Patterns::from(Pattern::all()))
5216 .unwrap()
5217 .consume()
5218 .with_hidden(true);
5219
5220 let mut announced = project.announced();
5221 announced.assert_next_active(".svc");
5222 announced.assert_next_wait();
5223
5224 let err = project.request_broadcast(".svc/x").await.err().unwrap();
5226 assert!(matches!(err, Error::BoundsExceeded(_)), "{err:?}");
5227 let inside = project.scope(".svc/x", &Patterns::from(Pattern::all())).unwrap();
5228 inside.announced().assert_next_wait();
5229 }
5230
5231 #[tokio::test]
5234 async fn mount_bounds_the_named_path() {
5235 let producer = origin(1).produce();
5236 let _near = producer.publish("t/x", Route::default()).unwrap();
5237 let deep = format!("t/{}", vec!["d"; Path::MAX_PARTS - 1].join("/"));
5238 let _deep = producer.publish(deep.as_str(), Route::default()).unwrap();
5239 let project = producer
5240 .mount("p1/a/b", "t")
5241 .unwrap()
5242 .scope("p1", &Patterns::from(Pattern::all()))
5243 .unwrap()
5244 .consume();
5245
5246 let mut announced = project.announced();
5247 announced.assert_next_active("a/b/x");
5248 announced.assert_next_wait();
5249
5250 let over = vec!["d"; Path::MAX_PARTS + 1].join("/");
5251 assert!(matches!(
5252 producer.mount(over.as_str(), "t"),
5253 Err(Error::BoundsExceeded(_))
5254 ));
5255 }
5256
5257 #[tokio::test]
5259 async fn mount_is_read_only() {
5260 let producer = origin(1).produce();
5261 let project = producer
5262 .mount("p1/.svc", ".svc/p1")
5263 .unwrap()
5264 .scope("p1", &Patterns::from(Pattern::all()))
5265 .unwrap();
5266 assert!(matches!(project.create_broadcast(".svc/foo"), Err(Error::Unauthorized)));
5267 assert!(matches!(
5268 project.dynamic(".svc", Route::default()),
5269 Err(Error::Unauthorized)
5270 ));
5271 assert!(matches!(
5272 project.dynamic(".svc/foo", Route::default()),
5273 Err(Error::Unauthorized)
5274 ));
5275 project.publish("cam", Route::default()).unwrap();
5276 }
5277
5278 #[test]
5280 fn mount_never_widens_a_scope() {
5281 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5282 let project = producer.scope("", &scopes(&["p1"])).unwrap();
5283 assert!(matches!(project.mount("p1/.svc", ".svc/p1"), Err(Error::Unauthorized)));
5284
5285 let mounted = producer.mount("p1/.svc", ".svc/p1").unwrap();
5286 assert!(matches!(mounted.mount("p1/.svc/x", ".other"), Err(Error::Duplicate)));
5287 assert!(matches!(mounted.mount("p1", ".other"), Err(Error::Duplicate)));
5288 assert!(matches!(mounted.mount("p2", "p1/.svc/x"), Err(Error::Duplicate)));
5290 assert!(matches!(mounted.mount("p2", "p1"), Err(Error::Duplicate)));
5291 assert!(matches!(mounted.mount(".svc/p1/x", ".other"), Err(Error::Duplicate)));
5293 assert!(matches!(mounted.mount(".svc", ".other"), Err(Error::Duplicate)));
5294 mounted.mount("p1/.other", ".other/p1").unwrap();
5295 mounted.mount("p2/.svc", ".svc/p1").unwrap();
5297 for (at, target) in [("a", "a/b"), ("a/b", "a"), ("a", "a")] {
5299 assert!(
5300 matches!(producer.mount(at, target), Err(Error::Duplicate)),
5301 "{at} -> {target}"
5302 );
5303 }
5304 }
5305
5306 #[test]
5308 fn mount_refuses_a_wildcard_mount_point() {
5309 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5310 for at in ["*", "p1/*", "p1/**", "p1/a*"] {
5311 assert!(
5312 matches!(producer.mount(at, ".svc/p1"), Err(Error::InvalidPath(_))),
5313 "{at}"
5314 );
5315 }
5316 }
5317
5318 #[tokio::test]
5321 async fn mount_egress_counts_under_the_named_path() {
5322 let registry = stats::Registry::new(stats::Config::new());
5323 let producer = origin(1).produce();
5324 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5325 let project = mounted(&producer, "p1", &["**"])
5326 .with_stats(registry.tier(stats::Tier::default()).session("p1"))
5327 .with_hidden(true);
5328
5329 let mut announced = project.announced();
5330 announced.assert_next_active(".svc/foo");
5331 project.request_broadcast(".svc/foo").await.expect("resolves");
5332
5333 let mut report = stats::Report::default();
5334 registry.report(&mut report);
5335 let paths: Vec<_> = report
5336 .traffic
5337 .iter()
5338 .map(|entry| entry.path.as_str().to_string())
5339 .collect();
5340 assert_eq!(paths, ["p1/.svc/foo"]);
5341 }
5342
5343 #[tokio::test]
5345 async fn hidden_route_announced_later_stays_hidden() {
5346 let producer = origin(1).produce();
5347 let consumer = producer.consume();
5348 let mut announced = consumer.announced();
5349 let mut opted = consumer.clone().with_hidden(true).announced();
5350
5351 let hidden = producer.announce(".stats/node", Route::default()).unwrap();
5352 announced.assert_next_wait();
5353 opted.assert_next_active(".stats/node");
5354
5355 drop(hidden);
5356 announced.assert_next_wait();
5357 opted.assert_next_ended(".stats/node");
5358 }
5359
5360 #[tokio::test]
5361 async fn broadcast_announces_its_own_path() {
5362 let producer = origin(1).produce();
5363 let consumer = producer.consume();
5364 let mut announced = consumer.announced();
5365 let mut peer = consumer.clone().excluding(Hop::UNKNOWN).announced();
5366
5367 let broadcast = producer.create_broadcast("room/alice").unwrap();
5369 announced.assert_next_wait();
5370 peer.assert_next_wait();
5371
5372 broadcast.announce(Route::default().with_cost(3)).unwrap();
5373 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(3));
5374 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(3));
5375
5376 broadcast.announce(Route::default().with_cost(1)).unwrap();
5378 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(1));
5379 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(1));
5380
5381 broadcast.unannounce();
5383 announced.assert_next_ended("room/alice");
5384 peer.assert_next_ended("room/alice");
5385 broadcast.unannounce();
5386 announced.assert_next_wait();
5387 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5388 assert!(matches!(err, Error::Unroutable));
5389
5390 broadcast.announce(Route::default()).unwrap();
5392 announced.assert_next_active("room/alice");
5393 peer.assert_next_active("room/alice");
5394 broadcast.close();
5395 announced.assert_next_ended("room/alice");
5396 peer.assert_next_ended("room/alice");
5397 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5398 announced.assert_next_wait();
5399 }
5400
5401 #[tokio::test]
5402 async fn broadcast_announcement_retracts_with_the_last_producer() {
5403 let producer = origin(1).produce();
5404 let consumer = producer.consume();
5405 let mut announced = consumer.announced();
5406
5407 let broadcast = producer.create_broadcast("room/alice").unwrap();
5408 let clone = broadcast.clone();
5409 broadcast.announce(Route::default()).unwrap();
5410 announced.assert_next_active("room/alice");
5411
5412 drop(broadcast);
5414 announced.assert_next_wait();
5415 drop(clone);
5416 announced.assert_next_ended("room/alice");
5417 }
5418
5419 #[tokio::test]
5420 async fn publish_creates_and_announces_together() {
5421 let producer = origin(1).produce();
5422 let mut announced = producer.consume().announced();
5423 let _broadcast = producer.publish("room/alice", Route::default()).unwrap();
5424 announced.assert_next_active("room/alice");
5425 }
5426
5427 #[tokio::test]
5428 async fn standalone_broadcast_cannot_announce() {
5429 let broadcast = broadcast::Info::new().produce();
5430 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5431 broadcast.unannounce();
5433 }
5434
5435 #[tokio::test]
5436 async fn announce_replays_to_late_cursor() {
5437 let producer = origin(1).produce();
5438 let _a = producer.announce("room/alice", Route::default()).unwrap();
5439 let _b = producer.announce("room/bob", Route::default()).unwrap();
5440
5441 let mut announced = producer.consume().announced();
5442 announced.assert_next_active("room/alice");
5444 announced.assert_next_active("room/bob");
5445 announced.assert_next_wait();
5446 }
5447
5448 #[tokio::test]
5449 async fn announce_keeps_its_prefix_under_a_producer_scope() {
5450 let producer = origin(1).produce();
5451 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
5452
5453 let _a = scoped.announce("", Route::default()).unwrap();
5455 let mut announced = producer.consume().announced();
5456 announced.assert_next_active("");
5457
5458 assert!(matches!(
5460 scoped.announce("other", Route::default()),
5461 Err(Error::Unauthorized)
5462 ));
5463 }
5464
5465 #[tokio::test]
5466 async fn cursor_keeps_an_overlapping_prefix_above_its_scope() {
5467 let producer = origin(1).produce();
5468 let _a = producer.announce("", Route::default()).unwrap();
5469
5470 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
5471 let mut announced = consumer.announced();
5472 announced.assert_next_active("");
5473 }
5474
5475 #[tokio::test]
5476 async fn cursor_root_strips_prefix() {
5477 let producer = origin(1).produce();
5478 let _a = producer.announce("room/alice", Route::default()).unwrap();
5479
5480 let consumer = producer
5481 .consume()
5482 .scope("room", &Patterns::from(Pattern::all()))
5483 .unwrap();
5484 let mut announced = consumer.announced();
5485 announced.assert_next_active("alice");
5486 }
5487
5488 #[tokio::test]
5489 async fn best_route_wins_and_fails_over() {
5490 let producer = origin(1).produce();
5491 let mut announced = producer.consume().announced();
5492
5493 let expensive = producer
5494 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5495 .unwrap();
5496 let route = announced.assert_next_active("room");
5497 assert_eq!(route.cost, Cost::new(5));
5498
5499 let cheap = producer
5501 .announce("room", Route::default().with_hops(hops(&[20])).with_cost(1))
5502 .unwrap();
5503 let route = announced.assert_next_active("room");
5504 assert_eq!(route.cost, Cost::new(1));
5505
5506 drop(cheap);
5508 let route = announced.assert_next_active("room");
5509 assert_eq!(route.cost, Cost::new(5));
5510
5511 drop(expensive);
5513 announced.assert_next_ended("room");
5514 }
5515
5516 #[tokio::test]
5517 async fn identical_reannounce_is_invisible() {
5518 let producer = origin(1).produce();
5519 let mut announced = producer.consume().announced();
5520
5521 let old = producer
5522 .announce("room", Route::default().with_hops(hops(&[10])))
5523 .unwrap();
5524 let first = announced.assert_next_active("room");
5525 assert_eq!(first.hops.as_slice(), hops(&[10]).as_slice());
5526
5527 let _new = producer
5531 .announce("room", Route::default().with_hops(hops(&[10])))
5532 .unwrap();
5533 announced.assert_next_wait();
5534
5535 drop(old);
5537 announced.assert_next_wait();
5538 }
5539
5540 #[tokio::test]
5541 async fn exclude_hides_routes_through_the_peer() {
5542 let producer = origin(1).produce();
5543 let _a = producer
5544 .announce("room", Route::default().with_hops(hops(&[7])))
5545 .unwrap();
5546
5547 let mut hidden = producer.consume().excluding(origin(7)).announced();
5548 hidden.assert_next_wait();
5549
5550 let mut visible = producer.consume().excluding(origin(8)).announced();
5551 visible.assert_next_active("room");
5552 }
5553
5554 #[tokio::test]
5555 async fn exclude_matches_via_when_the_chain_is_anonymous() {
5556 let producer = origin(1).produce();
5557 let assigned = origin(777);
5558 let _echoed = producer
5559 .announce("echoed", Route::default().with_hops(hops(&[0])).with_via(assigned))
5560 .unwrap();
5561 let _local = producer
5562 .announce("local", Route::default().with_hops(hops(&[10])))
5563 .unwrap();
5564
5565 let mut hidden = producer.consume().excluding(assigned).announced();
5566 hidden.assert_next_active("local");
5567 hidden.assert_next_wait();
5568 }
5569
5570 #[tokio::test]
5571 async fn anonymous_route_loses_to_identified_at_any_cost() {
5572 let producer = origin(1).produce();
5573 let mut announced = producer.consume().announced();
5574
5575 let _anonymous = producer
5576 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5577 .unwrap();
5578 let route = announced.assert_next_active("room");
5579 assert!(route.is_anonymous());
5580 assert_eq!(route.cost, Cost::new(1));
5581
5582 let _identified = producer
5583 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5584 .unwrap();
5585 let route = announced.assert_next_active("room");
5586 assert!(!route.is_anonymous());
5587 assert_eq!(route.cost, Cost::new(5));
5588 }
5589
5590 #[tokio::test]
5593 async fn a_stamped_route_ranks_below_an_identified_one() {
5594 let producer = origin(1).produce();
5595 let mut announced = producer.consume().announced();
5596
5597 let mut stamped = Hops::new();
5598 stamped.stamp(origin(5)).unwrap();
5599 assert_eq!(stamped.as_slice(), &[origin(5), Hop::UNKNOWN]);
5600
5601 let _legacy = producer
5602 .announce("room", Route::default().with_hops(stamped).with_cost(0))
5603 .unwrap();
5604 let route = announced.assert_next_active("room");
5605 assert!(route.is_anonymous());
5606
5607 let _identified = producer
5608 .announce("room", Route::default().with_hops(hops(&[10, 11])).with_cost(5))
5609 .unwrap();
5610 let route = announced.assert_next_active("room");
5611 assert_eq!(route.hops.as_slice(), hops(&[10, 11]).as_slice());
5612 }
5613
5614 #[test]
5615 fn stamping_keeps_the_leading_zero_and_names_nothing_else() {
5616 let mut chain = hops(&[0, 7]);
5617 chain.stamp(origin(5)).unwrap();
5618 assert_eq!(chain.as_slice(), hops(&[5, 0, 7]).as_slice());
5619
5620 let mut named = hops(&[7, 0]);
5622 named.stamp(origin(5)).unwrap();
5623 assert_eq!(named.as_slice(), hops(&[7, 0]).as_slice());
5624
5625 let mut full = Hops::try_from(vec![Hop::UNKNOWN; MAX_HOPS]).unwrap();
5627 assert_eq!(full.stamp(origin(5)), Err(InvalidHop::TooMany));
5628 }
5629
5630 #[tokio::test]
5631 async fn anonymous_routes_order_by_cost() {
5632 let producer = origin(1).produce();
5633 let mut announced = producer.consume().announced();
5634
5635 let expensive = producer
5636 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(5))
5637 .unwrap();
5638 let route = announced.assert_next_active("room");
5639 assert_eq!(route.cost, Cost::new(5));
5640
5641 let _cheap = producer
5642 .announce("room", Route::default().with_hops(hops(&[0, 7])).with_cost(1))
5643 .unwrap();
5644 let route = announced.assert_next_active("room");
5645 assert!(route.is_anonymous());
5646 assert_eq!(route.cost, Cost::new(1));
5647
5648 drop(expensive);
5649 announced.assert_next_wait();
5650 }
5651
5652 #[tokio::test]
5653 async fn anonymous_chain_from_identified_peer_still_ranks_last() {
5654 let producer = origin(1).produce();
5655 let mut announced = producer.consume().announced();
5656
5657 let _anonymous = producer
5658 .announce(
5659 "room",
5660 Route::default()
5661 .with_hops(hops(&[0, 7]))
5662 .with_cost(1)
5663 .with_via(origin(7)),
5664 )
5665 .unwrap();
5666 announced.assert_next_active("room");
5667
5668 let _identified = producer
5669 .announce("room", Route::default().with_hops(hops(&[10, 20])).with_cost(5))
5670 .unwrap();
5671 let route = announced.assert_next_active("room");
5672 assert!(!route.is_anonymous());
5673 assert_eq!(route.cost, Cost::new(5));
5674 }
5675
5676 #[tokio::test]
5677 async fn request_prefers_identified_over_cheaper_anonymous() {
5678 let producer = origin(1).produce();
5679 let consumer = producer.consume();
5680
5681 let anonymous = producer
5682 .dynamic("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5683 .unwrap();
5684 let identified = producer
5685 .dynamic("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5686 .unwrap();
5687
5688 let _pending = consumer.request_broadcast("room/alice");
5689 let request = queued(&identified).await;
5690 assert_eq!(request.path().as_str(), "room/alice");
5691 assert!(
5692 anonymous.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
5693 "the cheaper anonymous route must not serve"
5694 );
5695 }
5696
5697 #[tokio::test]
5698 async fn update_reprices_in_place() {
5699 let producer = origin(1).produce();
5700 let mut announced = producer.consume().announced();
5701
5702 let announcement = producer.announce("room", Route::default()).unwrap();
5703 announced.assert_next_active("room");
5704
5705 announcement.update(Route::default().with_cost(9)).unwrap();
5706 let route = announced.assert_next_active("room");
5707 assert_eq!(route.cost, Cost::new(9));
5708 }
5709
5710 #[tokio::test]
5711 async fn retract_after_undelivered_reprice_still_delivered() {
5712 let producer = origin(1).produce();
5713 let mut announced = producer.consume().announced();
5714
5715 let announcement = producer.announce("room", Route::default()).unwrap();
5716 announced.assert_next_active("room");
5717
5718 announcement.update(Route::default().with_cost(9)).unwrap();
5722 drop(announcement);
5723 announced.assert_next_ended("room");
5724 announced.assert_next_wait();
5725 }
5726
5727 #[tokio::test]
5728 async fn scoped_cursor_advertises_most_specific_covering_route() {
5729 let producer = origin(1).produce();
5730 let _broad = producer.announce("room", Route::default().with_cost(1)).unwrap();
5733 let _narrow = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
5734
5735 let consumer = producer
5736 .consume()
5737 .scope("room/alice", &Patterns::from(Pattern::all()))
5738 .unwrap();
5739 let mut announced = consumer.announced();
5740 let route = announced.assert_next_active("");
5741 assert_eq!(route.cost, Cost::new(9));
5742 announced.assert_next_wait();
5743 }
5744
5745 #[tokio::test]
5746 async fn capture_change_retracts_before_reannouncing_a_presented_prefix() {
5747 let producer = origin(1).produce();
5748 let _broad = producer.announce("room", Route::default()).unwrap();
5749 let exact = producer.announce("room/alice", Route::default()).unwrap();
5750 let consumer = producer
5751 .consume()
5752 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
5753 .unwrap()
5754 .scope("room/alice", &Patterns::from(Pattern::all()))
5755 .unwrap();
5756 let mut announced = consumer.announced();
5757
5758 let first = announced.next().now_or_never().expect("next").expect("announce");
5759 assert_eq!(first.prefix.as_str(), "");
5760 assert_eq!(first.kind, AnnounceKind::Announced);
5761 assert_eq!(first.captures, Some(Vec::new()));
5762
5763 drop(exact);
5764 let retracted = announced.next().now_or_never().expect("next").expect("retract");
5765 assert_eq!(retracted.prefix.as_str(), "");
5766 assert_eq!(retracted.kind, AnnounceKind::Retracted);
5767 assert_eq!(retracted.captures, Some(Vec::new()));
5768 let replacement = announced.next().now_or_never().expect("next").expect("announce");
5769 assert_eq!(replacement.prefix.as_str(), "");
5770 assert_eq!(replacement.kind, AnnounceKind::Announced);
5771 assert_eq!(replacement.captures, None);
5772 }
5773
5774 #[tokio::test]
5775 async fn routed_broadcast_resolves_once_announced() {
5776 let producer = origin(1).produce();
5777 let consumer = producer.consume();
5778
5779 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
5781 assert!((&mut resolving).now_or_never().is_none());
5782
5783 let broadcast = producer.create_broadcast("room/alice").unwrap();
5785 for _ in 0..20 {
5786 tokio::task::yield_now().await;
5787 }
5788 assert!((&mut resolving).now_or_never().is_none());
5789
5790 broadcast.announce(Route::default()).unwrap();
5791 let resolved = resolving.await.expect("resolves once announced");
5792 assert_eq!(resolved.info().path.as_str(), "room/alice");
5793 drop(broadcast);
5794 }
5795
5796 #[tokio::test]
5799 async fn cheaper_remote_route_beats_a_local_broadcast() {
5800 let producer = origin(1).produce();
5801 let consumer = producer.consume();
5802 let mut announced = consumer.announced();
5803
5804 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5805 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(5));
5806
5807 let server = producer
5808 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5809 .unwrap();
5810 let route = announced.assert_next_active("room/alice");
5811 assert_eq!(route.cost, Cost::new(1));
5812 assert_eq!(route.hops, hops(&[10]));
5813
5814 let pending = consumer.request_broadcast("room/alice");
5816 let request = queued(&server).await;
5817 let upstream = broadcast::Info::new().produce();
5818 request.accept(&upstream);
5819 pending.await.expect("resolves through the cheaper route");
5820 }
5821
5822 #[tokio::test]
5826 async fn cheaper_route_after_a_front_wins_new_requests() {
5827 let producer = origin(1).produce();
5828 let consumer = producer.consume();
5829
5830 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5831 let first = consumer
5832 .request_broadcast("room/alice")
5833 .await
5834 .expect("resolves locally");
5835
5836 let server = producer
5837 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5838 .unwrap();
5839
5840 let pending = consumer.request_broadcast("room/alice");
5841 let request = queued(&server).await;
5842 let upstream = broadcast::Info::new().produce();
5843 request.accept(&upstream);
5844 let second = pending.await.expect("resolves through the cheaper route");
5845
5846 assert!(!first.is_closed(), "the old front must keep serving its readers");
5847 assert!(!first.is_clone(&second), "the newcomer must not join the old front");
5848 }
5849
5850 #[tokio::test]
5854 async fn local_broadcast_wins_a_tie_with_a_hopless_route() {
5855 let producer = origin(1).produce();
5856 let consumer = producer.consume();
5857
5858 let _local = producer.publish("room/alice", Route::default()).unwrap();
5859 let server = producer.dynamic("room/alice", Route::default()).unwrap();
5860
5861 let resolved = tokio::time::timeout(Duration::from_secs(1), consumer.request_broadcast("room/alice"))
5863 .await
5864 .expect("the newer hopless route won the tie")
5865 .expect("resolves");
5866 assert_eq!(resolved.info().path.as_str(), "room/alice");
5867 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5868 }
5869
5870 #[tokio::test]
5873 async fn announce_stats_follow_the_advertisement() {
5874 let registry = stats::Registry::new(stats::Config::new());
5875 let producer = origin(1)
5876 .produce()
5877 .with_stats(registry.tier(stats::Tier::default()).session("root"));
5878 let announces = || {
5879 registry
5880 .snapshot()
5881 .traffic()
5882 .into_iter()
5883 .find(|(_, role, _)| *role == stats::Role::Subscriber)
5884 .map(|(_, _, traffic)| (traffic.announces_started, traffic.announces_ended))
5885 .unwrap_or_default()
5886 };
5887
5888 let broadcast = producer.create_broadcast("room/alice").unwrap();
5889 assert_eq!(announces(), (0, 0), "a hidden broadcast is not announced");
5890 broadcast.announce(Route::default()).unwrap();
5891 broadcast
5892 .announce(Route {
5893 cost: Cost::new(3),
5894 ..Route::default()
5895 })
5896 .unwrap();
5897 assert_eq!(announces(), (1, 0), "a re-price is not another announce");
5898 broadcast.unannounce();
5899 assert_eq!(announces(), (1, 1));
5900 broadcast.announce(Route::default()).unwrap();
5901 drop(broadcast);
5902 assert_eq!(announces(), (2, 2));
5903 }
5904
5905 #[tokio::test]
5907 async fn local_broadcast_wins_a_cost_tie() {
5908 let producer = origin(1).produce();
5909 let consumer = producer.consume();
5910 let mut announced = consumer.announced();
5911
5912 let server = producer
5913 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(2))
5914 .unwrap();
5915 announced.assert_next_active("room/alice");
5916 let _local = producer.publish("room/alice", Route::default().with_cost(2)).unwrap();
5917 assert!(announced.assert_next_active("room/alice").hops.is_empty());
5918
5919 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5920 assert_eq!(resolved.info().path.as_str(), "room/alice");
5921 for _ in 0..20 {
5922 tokio::task::yield_now().await;
5923 }
5924 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5925 }
5926
5927 #[tokio::test]
5930 async fn unannounce_ends_the_front() {
5931 let producer = origin(1).produce();
5932 let consumer = producer.consume();
5933
5934 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5935 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5936
5937 broadcast.unannounce();
5938 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5939 assert!(matches!(err, Error::Unroutable), "joined a retracted front: {err}");
5940 settle(|| resolved.is_closed()).await;
5941 assert!(!broadcast.consume().is_closed(), "the broadcast itself lives on");
5942
5943 broadcast.announce(Route::default()).unwrap();
5945 let again = consumer.request_broadcast("room/alice").await.expect("resolves again");
5946 assert!(!again.is_clone(&resolved));
5947 }
5948
5949 #[tokio::test]
5953 async fn unannounce_keeps_a_track_awaiting_its_info() {
5954 let producer = origin(1).produce();
5955 let consumer = producer.consume();
5956
5957 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5958 let mut dynamic = broadcast.dynamic();
5959 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5960 let track = resolved.track("video").unwrap();
5961 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
5962 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
5963 .await
5964 .expect("the front asked the source")
5965 .expect("request");
5966
5967 broadcast.unannounce();
5968 settle(|| resolved.is_closed()).await;
5969
5970 let source = request.accept(None);
5971 let mut group = source.append_group().unwrap();
5972 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
5973 group.finish().unwrap();
5974 source.finish().unwrap();
5975
5976 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
5977 let mut group = subscription.recv_group().await.unwrap().expect("the source's group");
5978 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late");
5979 assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly");
5980 }
5981
5982 async fn served_front() -> (Dynamic, broadcast::Producer, broadcast::Dynamic, broadcast::Consumer) {
5985 let producer = origin(1).produce();
5986 let consumer = producer.consume();
5987 let server = producer
5988 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
5989 .unwrap();
5990 let pending = consumer.request_broadcast("room/alice");
5991 let upstream = broadcast::Info::new().produce();
5992 let dynamic = upstream.dynamic();
5993 queued(&server).await.accept(&upstream);
5994 let resolved = pending.await.expect("resolves");
5995 (server, upstream, dynamic, resolved)
5996 }
5997
5998 #[tokio::test]
6003 async fn returning_reader_skips_a_warm_cache_the_copy_resolved_past() {
6004 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
6005 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6006 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
6007
6008 let track = resolved.track("audio").unwrap();
6009 let b = budget.clone();
6010 let subscribing = tokio::spawn(async move { track.subscribe(b).await });
6011 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6012 .await
6013 .expect("the front asked the source")
6014 .expect("request");
6015 let source = request.resolving_start().accept(None);
6016 for seq in 0..4u64 {
6017 let mut group = source.create_group(seq.into()).unwrap();
6018 group.write_frame(ms(seq * 20), b"old".as_ref()).unwrap();
6019 group.finish().unwrap();
6020 }
6021 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6022 subscription.recv_group().await.unwrap().expect("the live group");
6023 drop(subscription);
6024 tokio::time::timeout(Duration::from_secs(1), source.unused())
6025 .await
6026 .expect("parked")
6027 .expect("source open");
6028 drop(source);
6029
6030 let track = resolved.track("audio").unwrap();
6031 let subscribing = tokio::spawn(async move { track.subscribe(budget).await });
6032 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6033 .await
6034 .expect("the front asked the source again")
6035 .expect("request");
6036 let mut source = request.resolving_start().accept(None);
6037 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6038
6039 assert!(
6041 tokio::time::timeout(Duration::from_millis(50), subscription.recv_group())
6042 .await
6043 .is_err(),
6044 "the warm cache was served before the copy resolved its start"
6045 );
6046
6047 source.start_at(20).unwrap();
6049 let mut group = source.create_group(20u64.into()).unwrap();
6050 group.write_frame(ms(2000), b"new".as_ref()).unwrap();
6051 group.finish().unwrap();
6052 let group = subscription.recv_group().await.unwrap().expect("the live group");
6053 assert_eq!(group.sequence, 20, "a stale warm group was served");
6054 }
6055
6056 #[tokio::test]
6061 async fn returning_reader_replays_a_current_warm_cache() {
6062 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
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")
6069 .expect("request");
6070 let source = request.resolving_start().accept(None);
6071 let mut group = source.create_group(0u64.into()).unwrap();
6072 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6073 group.finish().unwrap();
6074 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6075 subscription.recv_group().await.unwrap().expect("the catalog");
6076 drop(subscription);
6077 tokio::time::timeout(Duration::from_secs(1), source.unused())
6078 .await
6079 .expect("parked")
6080 .expect("source open");
6081 drop(source);
6082
6083 let track = resolved.track("catalog").unwrap();
6084 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6085 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6086 .await
6087 .expect("the front asked the source again")
6088 .expect("request");
6089 let mut source = request.resolving_start().accept(None);
6090 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6091
6092 let reading = tokio::spawn(async move {
6095 let mut group = subscription.recv_group().await.unwrap().expect("the catalog");
6096 assert_eq!(group.sequence, 0);
6097 group.read_frame().await.unwrap().expect("the snapshot").payload
6098 });
6099 tokio::task::yield_now().await;
6100 assert_eq!(
6101 source.subscription().and_then(|sub| sub.start),
6102 Some(track::Position { group: 0, frame: 1 }),
6103 "the re-splice asked past the cached catalog"
6104 );
6105 source.start_at(0).unwrap();
6106 let mut tail = source.create_group(0u64.into()).unwrap();
6107 tail.start_at(1).unwrap();
6108 tail.finish().unwrap();
6109 let payload = tokio::time::timeout(Duration::from_secs(1), reading)
6110 .await
6111 .expect("the returning reader never got the catalog")
6112 .unwrap();
6113 assert_eq!(&payload[..], b"snapshot");
6114 }
6115
6116 #[tokio::test(start_paused = true)]
6120 async fn finished_track_is_forgotten_after_the_linger() {
6121 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6122
6123 let track = resolved.track("catalog").unwrap();
6124 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6125 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6126 .await
6127 .expect("the front asked the source")
6128 .expect("request");
6129 let source = request.resolving_start().accept(None);
6130 let mut group = source.create_group(0u64.into()).unwrap();
6131 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6132 group.finish().unwrap();
6133 source.finish().unwrap();
6134 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6135 assert_eq!(
6136 next_group(&mut subscription)
6137 .await
6138 .unwrap()
6139 .expect("the catalog")
6140 .sequence,
6141 0
6142 );
6143 assert!(next_group(&mut subscription).await.unwrap().is_none());
6144 drop(subscription);
6145 drop(source);
6146
6147 let mut subscription = resolved.track("catalog").unwrap().subscribe(None).await.unwrap();
6149 assert_eq!(
6150 next_group(&mut subscription)
6151 .await
6152 .unwrap()
6153 .expect("the catalog")
6154 .sequence,
6155 0
6156 );
6157 assert!(next_group(&mut subscription).await.unwrap().is_none());
6158 drop(subscription);
6159 assert!(
6160 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6161 .await
6162 .is_err(),
6163 "a finished track within the linger asked the source again"
6164 );
6165
6166 tokio::time::sleep(TRACK_IDLE_LINGER).await;
6168
6169 let track = resolved.track("catalog").unwrap();
6170 let _subscribing = tokio::spawn(async move { track.subscribe(None).await });
6171 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6172 .await
6173 .expect("the finished track outlived the linger")
6174 .expect("request");
6175 }
6176
6177 #[tokio::test]
6182 async fn returning_reader_continues_an_open_warm_group() {
6183 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6184
6185 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6186 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6187 .await
6188 .expect("frame")
6189 .unwrap()
6190 .expect("group ended");
6191 frame.payload.to_vec()
6192 }
6193
6194 let mut expect: Vec<&[u8]> = Vec::new();
6195 let mut floor: Option<track::Position> = None;
6196 for (round, payload) in [b"a".as_ref(), b"b", b"c"].into_iter().enumerate() {
6197 let track = resolved.track("log").unwrap();
6198 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6199 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6200 .await
6201 .expect("the front asked the source")
6202 .expect("request");
6203 let mut source = request.resolving_start().accept(None);
6204 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6205
6206 source.start_at(0).unwrap();
6208 let mut group = source.create_group(0u64.into()).unwrap();
6209 if let Some(floor) = floor {
6210 group.start_at(floor.frame).unwrap();
6211 }
6212 group.write_frame(crate::Timestamp::ZERO, payload).unwrap();
6213 expect.push(payload);
6214 source
6215 .insert_datagram(10, crate::Timestamp::ZERO, b"datagram".as_ref())
6216 .unwrap();
6217
6218 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6219 .await
6220 .expect("group 0")
6221 .unwrap()
6222 .expect("track ended");
6223 assert_eq!(reading.sequence, 0);
6224 for frame in &expect {
6225 assert_eq!(read(&mut reading).await, *frame, "round {round}");
6226 }
6227 assert_eq!(
6228 source.subscription().and_then(|sub| sub.start),
6229 floor,
6230 "round {round} asked for the wrong continuation"
6231 );
6232
6233 drop(reading);
6234 drop(subscription);
6235 tokio::time::timeout(Duration::from_secs(1), source.unused())
6236 .await
6237 .expect("parked")
6238 .expect("source open");
6239 drop(group);
6240 drop(source);
6241 floor = Some(track::Position {
6242 group: 0,
6243 frame: expect.len() as u64,
6244 });
6245 }
6246 }
6247
6248 #[tokio::test]
6255 async fn a_takeover_keeps_the_open_group_head_for_later_readers() {
6256 tokio::time::pause();
6257 for dies_first in [true, false] {
6258 takeover_keeps_the_open_group_head(dies_first).await;
6259 }
6260 }
6261
6262 async fn takeover_keeps_the_open_group_head(dies_first: bool) {
6263 let producer = origin(1).produce();
6264 let first_server = producer
6265 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6266 .unwrap();
6267 let pending = producer.consume().request_broadcast("room/alice");
6268 let upstream = broadcast::Info::new().produce();
6269 let mut dynamic = upstream.dynamic();
6270 queued(&first_server).await.accept(&upstream);
6271 let resolved = pending.await.unwrap();
6272
6273 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6274 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6275 .await
6276 .expect("frame")
6277 .unwrap()
6278 .expect("group ended");
6279 frame.payload.to_vec()
6280 }
6281
6282 let track = resolved.track("log").unwrap();
6284 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6285 let source = dynamic.requested_track().await.unwrap().accept(None);
6286 let mut group = source.create_group(0u64.into()).unwrap();
6287 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6288 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6289 let mut subscription = subscribing.await.unwrap().unwrap();
6290 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6291 assert_eq!(read(&mut reading).await, b"a");
6292 assert_eq!(read(&mut reading).await, b"b");
6293
6294 let mut first = Some((group, source, upstream, dynamic));
6297 let mut joining = None;
6298 if dies_first {
6299 drop(first.take());
6300 assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none());
6303 let track = resolved.track("log").unwrap();
6304 joining = Some(tokio::spawn(async move {
6305 let mut joined = track.subscribe(None).await.unwrap();
6306 let group = joined.recv_group().await.unwrap().expect("track ended");
6307 (joined, group)
6308 }));
6309 }
6310 let cost = if dies_first { 5 } else { 0 };
6311 let replacement_server = producer
6312 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(cost))
6313 .unwrap();
6314 let replacement = broadcast::Info::new().produce();
6315 let mut replacement_dynamic = replacement.dynamic();
6316 queued(&replacement_server).await.accept(&replacement);
6317 let mut next = tokio::time::timeout(Duration::from_secs(1), replacement_dynamic.requested_track())
6318 .await
6319 .expect("the front asked the replacement")
6320 .unwrap()
6321 .resolving_start()
6322 .accept(None);
6323 next.start_at(0).unwrap();
6324 let mut next_group = next.create_group(0u64.into()).unwrap();
6325 next_group.start_at(2).unwrap();
6326 next_group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6327 assert_eq!(read(&mut reading).await, b"c", "dies_first={dies_first}");
6328
6329 if let Some((group, ..)) = &mut first {
6331 let mut independent = group.consume();
6332 group.write_frame(crate::Timestamp::ZERO, b"x".as_ref()).unwrap();
6333 for expect in [b"a", b"b", b"x"] {
6334 assert_eq!(read(&mut independent).await, expect);
6335 }
6336 group.finish().unwrap();
6337 }
6338 drop(first);
6339
6340 let track = resolved.track("log").unwrap();
6341 let mut later = tokio::time::timeout(Duration::from_secs(1), track.subscribe(None))
6342 .await
6343 .expect("subscribe")
6344 .unwrap();
6345 let mut late = tokio::time::timeout(Duration::from_secs(1), later.recv_group())
6346 .await
6347 .unwrap_or_else(|_| panic!("dies_first={dies_first}: the later reader never got the group"))
6348 .unwrap()
6349 .expect("track ended");
6350 assert_eq!(late.sequence, 0);
6351 for expect in [b"a", b"b", b"c"] {
6352 assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}");
6353 }
6354 let mut joined = None;
6355 if let Some(joining) = joining {
6356 let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining)
6357 .await
6358 .expect("the reader joining mid-outage never got the group")
6359 .unwrap();
6360 assert_eq!(group.sequence, 0);
6361 for expect in [b"a", b"b", b"c"] {
6362 assert_eq!(read(&mut group).await, expect);
6363 }
6364 joined = Some((subscription, group));
6365 }
6366
6367 assert!(
6369 tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6370 .await
6371 .is_err(),
6372 "dies_first={dies_first}: the group was handed out twice"
6373 );
6374
6375 next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap();
6377 assert_eq!(read(&mut reading).await, b"d");
6378 assert_eq!(read(&mut late).await, b"d");
6379 if let Some((_, group)) = &mut joined {
6380 assert_eq!(read(group).await, b"d");
6381 }
6382 next_group.finish().unwrap();
6383 drop((
6384 joined,
6385 reading,
6386 late,
6387 subscription,
6388 later,
6389 next,
6390 replacement,
6391 replacement_server,
6392 first_server,
6393 ));
6394 }
6395
6396 #[tokio::test]
6401 async fn returning_reader_gets_a_live_open_warm_group() {
6402 use futures::FutureExt;
6403
6404 tokio::time::pause();
6405 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6406
6407 let track = resolved.track("video").unwrap();
6409 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6410 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6411 .await
6412 .expect("the front asked the source")
6413 .expect("request");
6414 let mut source = request.resolving_start().accept(None);
6415 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6416 source.start_at(0).unwrap();
6417 let mut group = source.create_group(0u64.into()).unwrap();
6418 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6419 let mut reading = subscription.recv_group().await.unwrap().expect("group 0");
6420 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6421 drop(reading);
6422 drop(subscription);
6423 tokio::time::timeout(Duration::from_secs(1), source.unused())
6424 .await
6425 .expect("parked")
6426 .expect("source open");
6427 drop(group);
6428 drop(source);
6429
6430 let track = resolved.track("video").unwrap();
6432 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6433 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6434 .await
6435 .expect("the front asked the source")
6436 .expect("request");
6437 let mut source = request.resolving_start().accept(None);
6438 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6439 source.start_at(0).unwrap();
6440 let mut group = source.create_group(0u64.into()).unwrap();
6441 group.start_at(1).unwrap();
6442 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6443
6444 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6445 .await
6446 .expect("group 0")
6447 .unwrap()
6448 .expect("track ended");
6449 assert_eq!(reading.sequence, 0);
6450 let early = reading.finished().now_or_never();
6451 assert!(
6452 early.is_none(),
6453 "the live group already resolved its end before a frame was read: {early:?}"
6454 );
6455
6456 group.finish().unwrap();
6458 assert_eq!(reading.finished().await.unwrap(), 0);
6459 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6460 assert_eq!(reading.finished().await.unwrap(), 1);
6461 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"b");
6462 assert_eq!(reading.finished().await.unwrap(), 2);
6463 }
6464
6465 #[tokio::test]
6466 async fn warm_head_survives_another_takeover_before_park() {
6467 tokio::time::pause();
6468 let producer = origin(1).produce();
6469 let _server = producer
6470 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6471 .unwrap();
6472 let pending = producer.consume().request_broadcast("room/alice");
6473 let upstream = broadcast::Info::new().produce();
6474 let mut dynamic = upstream.dynamic();
6475 queued(&_server).await.accept(&upstream);
6476 let resolved = pending.await.unwrap();
6477
6478 let track = resolved.track("log").unwrap();
6479 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6480 let source = dynamic.requested_track().await.unwrap().accept(None);
6481 let mut group = source.create_group(0u64.into()).unwrap();
6482 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6483 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6484 let mut subscription = subscribing.await.unwrap().unwrap();
6485 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6486 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6487 drop(reading);
6488 drop(subscription);
6489 source.unused().await.unwrap();
6490 drop(group);
6491 drop(source);
6492
6493 let track = resolved.track("log").unwrap();
6494 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6495 let mut resumed = dynamic.requested_track().await.unwrap().resolving_start().accept(None);
6496 resumed.start_at(0).unwrap();
6497 let mut subscription = subscribing.await.unwrap().unwrap();
6498 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6499 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6500
6501 let replacement_server = producer
6502 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
6503 .unwrap();
6504 let replacement = broadcast::Info::new().produce();
6505 let mut replacement_dynamic = replacement.dynamic();
6506 queued(&replacement_server).await.accept(&replacement);
6507 let mut source = replacement_dynamic
6508 .requested_track()
6509 .await
6510 .unwrap()
6511 .resolving_start()
6512 .accept(None);
6513 source.start_at(0).unwrap();
6514 let mut group = source.create_group(0u64.into()).unwrap();
6515 group.start_at(2).unwrap();
6516 group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6517 let mut continuation = reading.clone();
6518 continuation.start_at(2);
6519 let frame = tokio::time::timeout(Duration::from_secs(1), continuation.read_frame())
6520 .await
6521 .expect("replacement frame")
6522 .unwrap()
6523 .unwrap();
6524 assert_eq!(&frame.payload[..], b"c");
6525 drop(continuation);
6526 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"b");
6527 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"c");
6528 drop(reading);
6529 drop(subscription);
6530 source.unused().await.unwrap();
6531 drop(group);
6532 drop(source);
6533
6534 let track = resolved.track("log").unwrap();
6535 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6536 let mut next = replacement_dynamic
6537 .requested_track()
6538 .await
6539 .unwrap()
6540 .resolving_start()
6541 .accept(None);
6542 next.start_at(0).unwrap();
6543 let mut subscription = subscribing.await.unwrap().unwrap();
6544 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6545 for expected in [b"a", b"b", b"c"] {
6546 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], expected);
6547 }
6548 }
6549
6550 #[tokio::test]
6553 async fn unannounce_keeps_a_returning_reader_awaiting_its_info() {
6554 let producer = origin(1).produce();
6555 let consumer = producer.consume();
6556
6557 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6558 let mut dynamic = broadcast.dynamic();
6559 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6560 let track = resolved.track("video").unwrap();
6561 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6562 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6563 .await
6564 .expect("the front asked the source")
6565 .expect("request");
6566 let source = request.accept(None);
6567 let mut group = source.append_group().unwrap();
6568 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
6569 group.finish().unwrap();
6570 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6571 subscription.recv_group().await.unwrap().expect("the cached group");
6572 drop(subscription);
6573
6574 tokio::time::timeout(Duration::from_secs(1), source.unused())
6577 .await
6578 .expect("parked")
6579 .expect("source open");
6580 drop(source);
6581 let track = resolved.track("video").unwrap();
6582 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6583 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6584 .await
6585 .expect("the front asked the source again")
6586 .expect("request");
6587
6588 broadcast.unannounce();
6589 settle(|| resolved.is_closed()).await;
6590
6591 let source = request.accept(None);
6593 let mut group = source.create_group(1u64.into()).unwrap();
6594 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
6595 group.finish().unwrap();
6596 source.finish().unwrap();
6597
6598 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
6599 let mut payloads = Vec::new();
6600 while let Some(mut group) = subscription.recv_group().await.expect("ends cleanly") {
6601 payloads.push(group.read_frame().await.unwrap().unwrap().payload);
6602 }
6603 assert_eq!(payloads.last().map(|p| &p[..]), Some(&b"late"[..]));
6604 }
6605
6606 #[tokio::test]
6609 async fn reannounce_before_the_front_acts_keeps_it() {
6610 let producer = origin(1).produce();
6611 let consumer = producer.consume();
6612
6613 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6614 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6615
6616 broadcast.unannounce();
6617 broadcast.announce(Route::default()).unwrap();
6618 for _ in 0..20 {
6619 tokio::task::yield_now().await;
6620 }
6621 assert!(!resolved.is_closed(), "the front ended across a reannouncement");
6622 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6623 assert!(again.is_clone(&resolved));
6624 }
6625
6626 #[tokio::test]
6627 async fn local_broadcast_resolves_once_announced() {
6628 let producer = origin(1).produce();
6629 let consumer = producer.consume();
6630
6631 let broadcast = producer.create_broadcast("room/alice").unwrap();
6633 let err = consumer
6634 .request_broadcast("room/alice")
6635 .now_or_never()
6636 .expect("unroutable is synchronous")
6637 .err()
6638 .unwrap();
6639 assert!(matches!(err, Error::Unroutable));
6640
6641 broadcast.announce(Route::default()).unwrap();
6642 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6643 assert_eq!(resolved.info().path.as_str(), "room/alice");
6644 drop(broadcast);
6645
6646 let err = consumer
6648 .request_broadcast("room/bob")
6649 .now_or_never()
6650 .expect("unroutable is synchronous")
6651 .err()
6652 .unwrap();
6653 assert!(matches!(err, Error::Unroutable));
6654 }
6655
6656 #[test]
6657 fn create_broadcast_accepts_a_max_depth_path() {
6658 let producer = origin(1).produce();
6659 let path = vec!["a"; Path::MAX_PARTS].join("/");
6660 let _broadcast = producer.create_broadcast(path.as_str()).expect("max depth is allowed");
6661 let deeper = vec!["a"; Path::MAX_PARTS + 1].join("/");
6662 assert!(matches!(
6663 producer.create_broadcast(deeper.as_str()),
6664 Err(Error::BoundsExceeded(_))
6665 ));
6666 }
6667
6668 #[tokio::test]
6669 async fn duplicate_routes_aggregate_until_the_last_leaves() {
6670 let producer = origin(1).produce();
6671 let first = producer.dynamic("live", Route::default().with_cost(3)).unwrap();
6672 let second = producer.dynamic("live", Route::default().with_cost(1)).unwrap();
6673
6674 let mut announced = producer.consume().announced();
6675 let update = announced.next().now_or_never().expect("next").expect("no next");
6676 assert_eq!(update.prefix.as_str(), "live");
6677 assert_eq!(update.kind, AnnounceKind::Announced);
6678 assert_eq!(update.route.cost, Cost::new(1));
6679 announced.assert_next_wait();
6680
6681 drop(second);
6682 let update = announced.next().now_or_never().expect("next").expect("no next");
6683 assert_eq!(update.prefix.as_str(), "live");
6684 assert_eq!(update.kind, AnnounceKind::Updated);
6685 assert_eq!(update.route.cost, Cost::new(3));
6686
6687 drop(first);
6688 announced.assert_next_ended("live");
6689 announced.assert_next_wait();
6690 }
6691
6692 #[test]
6693 fn dynamic_may_cover_a_scope_but_disjoint_prefixes_are_refused() {
6694 let producer = origin(1).produce();
6695 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
6696 let _broad = scoped
6697 .dynamic("", Route::default())
6698 .expect("an overlapping prefix is accepted");
6699
6700 let _ok = scoped
6701 .dynamic("room/alice", Route::default())
6702 .expect("a contained prefix is accepted");
6703 assert!(matches!(
6704 scoped.dynamic("other", Route::default()),
6705 Err(Error::Unauthorized)
6706 ));
6707 }
6708
6709 #[tokio::test]
6710 async fn dynamic_route_keeps_its_producer_scope() {
6711 let producer = origin(1).produce();
6712 let scope = Patterns::from("*/chat".parse::<Pattern>().unwrap());
6713 let scoped = producer.scope("", &scope).unwrap();
6714 let dynamic = scoped.dynamic("", Route::default()).unwrap();
6715
6716 let mut matching = producer
6717 .consume()
6718 .scope("", &scopes(&["room/chat"]))
6719 .unwrap()
6720 .announced();
6721 matching.assert_next_active("");
6722 let mut outside = producer
6723 .consume()
6724 .scope("", &scopes(&["room/video"]))
6725 .unwrap()
6726 .announced();
6727 outside.assert_next_wait();
6728
6729 let refused = producer
6730 .consume()
6731 .request_broadcast("room/video")
6732 .now_or_never()
6733 .expect("an out-of-scope request must be refused synchronously");
6734 assert!(matches!(refused, Err(Error::Unroutable)));
6735 assert!(dynamic.requested_broadcast().now_or_never().is_none());
6736
6737 let _pending = producer.consume().request_broadcast("room/chat");
6738 let request = queued(&dynamic).await;
6739 assert_eq!(request.path().as_str(), "room/chat");
6740 }
6741
6742 #[tokio::test]
6743 async fn scoped_cursor_selects_among_the_routes_it_can_see() {
6744 let producer = origin(1).produce();
6745 let scoped = |pattern: &str| {
6746 producer
6747 .scope("", &Patterns::from(pattern.parse::<Pattern>().unwrap()))
6748 .unwrap()
6749 };
6750 let _chat = scoped("*/chat").dynamic("", Route::default().with_cost(1)).unwrap();
6751 let _video = scoped("*/video").dynamic("", Route::default().with_cost(5)).unwrap();
6752
6753 let mut video = producer
6754 .consume()
6755 .scope("", &scopes(&["room/video"]))
6756 .unwrap()
6757 .announced();
6758 assert_eq!(video.assert_next_active("").cost, Cost::new(5));
6759 }
6760
6761 #[tokio::test]
6762 async fn route_changes_preserve_each_cursors_visible_winner() {
6763 let producer = origin(1).produce();
6764 let consumer = producer.consume();
6765 let mut all = consumer.clone().announced();
6766 let mut excluded = consumer.clone().excluding(origin(7)).announced();
6767 let mut video = consumer
6768 .clone()
6769 .scope("", &scopes(&["room/live/video"]))
6770 .unwrap()
6771 .announced();
6772 let mut rooted = consumer
6773 .scope("room/live", &Patterns::from(Pattern::all()))
6774 .unwrap()
6775 .announced();
6776
6777 let chat = producer
6778 .scope("", &scopes(&["room/live/chat"]))
6779 .unwrap()
6780 .dynamic("room/live", Route::default().with_hops(hops(&[7])).with_cost(1))
6781 .unwrap();
6782 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(1));
6783 assert_eq!(rooted.assert_next_active("").cost, Cost::new(1));
6784 excluded.assert_next_wait();
6785 video.assert_next_wait();
6786
6787 let video_route = producer
6788 .scope("", &scopes(&["room/live/video"]))
6789 .unwrap()
6790 .dynamic("room/live", Route::default().with_hops(hops(&[8])).with_cost(5))
6791 .unwrap();
6792 all.assert_next_wait();
6793 rooted.assert_next_wait();
6794 assert_eq!(excluded.assert_next_active("room/live").cost, Cost::new(5));
6795 assert_eq!(video.assert_next_active("room/live").cost, Cost::new(5));
6796
6797 chat.update(Route::default().with_hops(hops(&[7])).with_cost(9))
6798 .unwrap();
6799 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(5));
6800 assert_eq!(rooted.assert_next_active("").cost, Cost::new(5));
6801 excluded.assert_next_wait();
6802 video.assert_next_wait();
6803
6804 drop(video_route);
6805 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(9));
6806 assert_eq!(rooted.assert_next_active("").cost, Cost::new(9));
6807 excluded.assert_next_ended("room/live");
6808 video.assert_next_ended("room/live");
6809 }
6810
6811 #[tokio::test]
6812 async fn root_cursor_keeps_the_more_specific_covering_route() {
6813 let producer = origin(1).produce();
6814 let broad = producer.dynamic("room", Route::default().with_cost(1)).unwrap();
6815 let narrow = producer.dynamic("room/live", Route::default().with_cost(9)).unwrap();
6816 let mut announced = producer
6817 .consume()
6818 .scope("room/live/video", &Patterns::from(Pattern::all()))
6819 .unwrap()
6820 .announced();
6821 assert_eq!(announced.assert_next_active("").cost, Cost::new(9));
6822
6823 broad.update(Route::default().with_cost(0)).unwrap();
6824 announced.assert_next_wait();
6825 narrow.update(Route::default().with_cost(8)).unwrap();
6826 assert_eq!(announced.assert_next_active("").cost, Cost::new(8));
6827 drop(narrow);
6828 assert_eq!(announced.assert_next_active("").cost, Cost::new(0));
6829 }
6830
6831 #[tokio::test]
6832 async fn dynamic_accepts_a_max_depth_prefix() {
6833 let producer = origin(1).produce();
6834 let path = (0..Path::MAX_PARTS)
6835 .map(|i| format!("s{i}"))
6836 .collect::<Vec<_>>()
6837 .join("/");
6838 let mut announced = producer.consume().announced();
6839
6840 let dynamic = producer.dynamic(&path, Route::default()).expect("max depth is allowed");
6841 announced.assert_next_active(&path);
6842
6843 let _pending = producer.consume().request_broadcast(&path);
6844 let request = queued(&dynamic).await;
6845 assert_eq!(request.path().as_str(), path);
6846 }
6847
6848 #[tokio::test]
6849 async fn dynamic_exclusion_skips_routes_through_the_subscriber() {
6850 let producer = origin(1).produce();
6851 let _server = producer
6852 .dynamic("live", Route::default().with_hops(hops(&[7])))
6853 .unwrap();
6854
6855 let mut excluded = producer.consume().excluding(origin(7)).announced();
6856 excluded.assert_next_wait();
6857
6858 let mut clean = producer.consume().excluding(origin(8)).announced();
6859 clean.assert_next_active("live");
6860 }
6861
6862 #[tokio::test]
6864 async fn announce_consumer_is_a_stream() {
6865 use futures::StreamExt;
6866 let producer = origin(1).produce();
6867 let server = producer.dynamic("live", Route::default()).unwrap();
6868 let mut announced = producer.consume().announced();
6869 let update = StreamExt::next(&mut announced)
6870 .now_or_never()
6871 .expect("next")
6872 .expect("no next");
6873 assert_eq!(update.prefix.as_str(), "live");
6874 assert_eq!(update.kind, AnnounceKind::Announced);
6875 assert!(StreamExt::next(&mut announced).now_or_never().is_none());
6876 drop(server);
6877 let update = StreamExt::next(&mut announced)
6878 .now_or_never()
6879 .expect("next")
6880 .expect("no next");
6881 assert_eq!(update.kind, AnnounceKind::Retracted);
6882 }
6883
6884 #[tokio::test]
6885 async fn dynamic_retracts() {
6886 let producer = origin(1).produce();
6887 let server = producer.dynamic("live", Route::default()).unwrap();
6888 let mut announced = producer.consume().announced();
6889 announced.assert_next_active("live");
6890
6891 drop(server);
6892 announced.assert_next_ended("live");
6893 }
6894
6895 #[test]
6896 fn charged_wildcard_cost_accumulates_across_hops() {
6897 let first = Cost::new(4).charged(1);
6898 let second = first.charged(2);
6899 assert_eq!(second, Cost { warm: 7, cold: 7 });
6900 }
6901
6902 #[tokio::test]
6903 async fn local_broadcast_is_invisible_until_announced() {
6904 let producer = origin(1).produce();
6905 let mut local = producer.consume().announced();
6906 let mut peer = producer.consume().excluding(Hop::UNKNOWN).announced();
6907 let broadcast = producer.create_broadcast("room/alice").unwrap();
6908 local.assert_next_wait();
6909 peer.assert_next_wait();
6910
6911 broadcast.announce(Route::default()).unwrap();
6912 local.assert_next_active("room/alice");
6913 peer.assert_next_active("room/alice");
6914
6915 drop(broadcast);
6916 local.assert_next_ended("room/alice");
6917 peer.assert_next_ended("room/alice");
6918 }
6919
6920 #[tokio::test]
6921 async fn served_route_materializes_on_demand() {
6922 let producer = origin(1).produce();
6923 let consumer = producer.consume();
6924
6925 let server = producer.dynamic("room", Route::default()).unwrap();
6926
6927 let pending = consumer.request_broadcast("room/alice");
6928 let request = queued(&server).await;
6929 assert_eq!(request.path().as_str(), "room/alice");
6930
6931 let source = broadcast::Info::new().produce();
6932 request.accept(&source);
6933
6934 let resolved = pending.await.expect("resolves");
6935 assert_eq!(resolved.info().path.as_str(), "room/alice");
6937
6938 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6940 assert!(again.is_clone(&resolved));
6941 }
6942
6943 #[tokio::test]
6944 async fn served_requests_coalesce() {
6945 let producer = origin(1).produce();
6946 let consumer = producer.consume();
6947 let server = producer.dynamic("room", Route::default()).unwrap();
6948
6949 let first = consumer.request_broadcast("room/alice");
6950 let second = consumer.request_broadcast("room/alice");
6951
6952 let request = queued(&server).await;
6953 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
6955
6956 let source = broadcast::Info::new().produce();
6957 request.accept(&source);
6958
6959 let first = first.await.expect("resolves");
6960 let second = second.await.expect("resolves");
6961 assert!(first.is_clone(&second));
6962 }
6963
6964 #[tokio::test]
6965 async fn retract_rejects_pending_requests() {
6966 let producer = origin(1).produce();
6967 let consumer = producer.consume();
6968 let server = producer.dynamic("room", Route::default()).unwrap();
6969
6970 let pending = consumer.request_broadcast("room/alice");
6971 drop(server);
6972
6973 let err = pending.await.err().unwrap();
6974 assert!(matches!(err, Error::Unroutable));
6975
6976 let err = consumer
6978 .request_broadcast("room/alice")
6979 .now_or_never()
6980 .expect("unroutable")
6981 .err()
6982 .unwrap();
6983 assert!(matches!(err, Error::Unroutable));
6984 }
6985
6986 #[tokio::test]
6987 async fn routed_broadcast_survives_serving_route_retraction() {
6988 let producer = origin(1).produce();
6989 let consumer = producer.consume();
6990
6991 let standby_server = producer.dynamic("room", Route::default()).unwrap();
6994 let second_server = producer.dynamic("room", Route::default()).unwrap();
6995 let incumbent_server = producer.dynamic("room", Route::default()).unwrap();
6996
6997 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
6998 assert!((&mut resolving).now_or_never().is_none());
6999
7000 drop(incumbent_server);
7006 assert!((&mut resolving).now_or_never().is_none());
7007 drop(second_server);
7008 assert!((&mut resolving).now_or_never().is_none());
7009
7010 let request = queued(&standby_server).await;
7011 let source = broadcast::Info::new().produce();
7012 request.accept(&source);
7013
7014 let resolved = resolving.await.expect("resolves via the standby");
7015 assert_eq!(resolved.info().path.as_str(), "room/alice");
7016 }
7017
7018 #[tokio::test]
7019 async fn split_horizon_skips_routes_through_the_requester() {
7020 let producer = origin(1).produce();
7021 let _server = producer
7022 .dynamic("room", Route::default().with_hops(hops(&[7])))
7023 .unwrap();
7024
7025 let excluded = producer.consume().excluding(origin(7));
7027 let err = excluded
7028 .request_broadcast("room/alice")
7029 .now_or_never()
7030 .expect("unroutable")
7031 .err()
7032 .unwrap();
7033 assert!(matches!(err, Error::Unroutable));
7034
7035 let clean = producer.consume().excluding(origin(8));
7037 let pending = clean.request_broadcast("room/alice");
7038 assert!(pending.now_or_never().is_none());
7039 }
7040
7041 #[tokio::test(start_paused = true)]
7045 async fn viewer_sessions_share_the_plain_front() {
7046 let producer = origin(1).produce();
7047 let server = producer
7049 .dynamic("room", Route::default().with_hops(hops(&[10])))
7050 .unwrap();
7051 let upstream = broadcast::Info::new().produce();
7052 let source = upstream.create_track("video", None).unwrap();
7053 let mut group = source.append_group().unwrap();
7054 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
7055
7056 let counts = || {
7057 let table = producer.shared.lock();
7058 let watches = table
7059 .routes
7060 .root
7061 .find(Path::new("room/alice").parts())
7062 .map_or(0, |node| node.watches.len());
7063 (table.fronts.len(), watches)
7064 };
7065
7066 let mut first = None;
7067 let mut front = None;
7068 for viewer in 0..1000 {
7069 let session = producer.consume().excluding(Hop::random());
7070 let pending = session.request_broadcast("room/alice");
7071 if viewer == 0 {
7072 queued(&server).await.accept(&upstream);
7073 }
7074 let resolved = pending.await.expect("resolves");
7075 let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap();
7076 next_group(&mut subscription).await.unwrap().expect("the live group");
7077 drop(subscription);
7078 first.get_or_insert_with(counts);
7079 front = Some(resolved);
7080 }
7081 assert_eq!(first, Some((1, 1)), "one front and one watch for the first viewer");
7082 assert_eq!(counts(), (1, 1), "viewers minted fronts of their own");
7083 drop(front);
7084 }
7085
7086 #[tokio::test]
7090 async fn a_hop_in_a_chain_gets_its_own_front() {
7091 let producer = origin(1).produce();
7092 let _incumbent = producer
7093 .dynamic("room", Route::default().with_hops(hops(&[10])))
7094 .unwrap();
7095 let fronts = || producer.shared.lock().fronts.len();
7096
7097 let peer = producer.consume().excluding(origin(7));
7098 let _local = producer.consume().request_broadcast("room/alice");
7099 let _peer = peer.request_broadcast("room/alice");
7100 assert_eq!(fronts(), 1, "a hop no chain names excludes nothing");
7101
7102 let _echo = producer
7104 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7105 .unwrap();
7106 let _filtered = peer.request_broadcast("room/alice");
7107 assert_eq!(fronts(), 2, "the peer still shares the plain front");
7108 let _viewer = producer.consume().excluding(origin(8)).request_broadcast("room/alice");
7109 assert_eq!(fronts(), 2, "a viewer left the plain front");
7110 }
7111
7112 #[tokio::test]
7116 async fn a_hairpin_through_the_plain_front_is_withdrawn() {
7117 let producer = origin(1).produce();
7118 let peer = producer.consume().excluding(origin(7));
7119 let mut view = peer.announced();
7120 let incumbent = producer
7121 .dynamic("room", Route::default().with_hops(hops(&[10])))
7122 .unwrap();
7123 view.assert_next_active("room");
7124
7125 let pending = peer.request_broadcast("room/alice");
7126 let source = broadcast::Info::new().produce();
7127 queued(&incumbent).await.accept(&source);
7128 let resolved = pending.await.expect("resolves through the incumbent");
7129
7130 let echo = producer
7132 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7133 .unwrap();
7134 drop(incumbent);
7135 drop(source);
7136
7137 let replacement = broadcast::Info::new().produce();
7138 queued(&echo).await.accept(&replacement);
7139 resolved.assert_not_closed();
7140 view.assert_next_ended("room");
7141 }
7142
7143 #[tokio::test]
7144 async fn routes_report_where_they_entered() {
7145 let producer = origin(1).produce();
7146 let peer = producer.clone().peer();
7147 let mut announced = producer.consume().announced();
7148
7149 let _ingest = producer
7150 .dynamic("client", Route::default().with_hops(hops(&[5])).with_via(origin(5)))
7151 .unwrap();
7152 let _gateway = producer.publish("gateway", Route::default()).unwrap();
7153 let _forwarded = peer
7154 .dynamic(
7155 "forwarded",
7156 Route::default().with_hops(hops(&[5, 7])).with_via(origin(7)),
7157 )
7158 .unwrap();
7159
7160 assert_eq!(announced.assert_next_active("client").source(), Source::Local);
7161 assert_eq!(
7162 announced.assert_next_active("forwarded").source(),
7163 Source::Peer(origin(7))
7164 );
7165 assert_eq!(announced.assert_next_active("gateway").source(), Source::Local);
7166
7167 let scoped = peer.scope("room", &Patterns::from(Pattern::all())).unwrap();
7169 let _nested = scoped.dynamic("x", Route::default().with_via(origin(8))).unwrap();
7170 assert_eq!(announced.assert_next_active("room/x").source(), Source::Peer(origin(8)));
7171 }
7172
7173 #[tokio::test(start_paused = true)]
7177 async fn try_next_delivers_a_held_update() {
7178 async fn step(by: Duration) {
7180 tokio::time::advance(by).await;
7181 for _ in 0..10 {
7182 tokio::task::yield_now().await;
7183 }
7184 }
7185
7186 let producer = origin(1).produce();
7187 let peer = producer.clone().peer();
7188 let mut announced = producer.consume().announced();
7189 step(Duration::from_millis(1)).await;
7190
7191 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7192 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7193 assert!(announced.try_next().unwrap().kind.is_active());
7194
7195 let _far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7196 drop(near);
7197 assert!(announced.try_next().is_none(), "the update is held");
7198 step(Duration::from_millis(1)).await;
7199 assert!(announced.try_next().is_none(), "the update is still held");
7200 step(DEFAULT_UPDATE_HOLD).await;
7201 let update = announced.try_next().expect("the hold passed");
7202 assert_eq!(update.kind, AnnounceKind::Updated);
7203 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7204 }
7205
7206 #[tokio::test(start_paused = true)]
7209 async fn a_changed_route_waits_out_the_hold() {
7210 let producer = origin(1).produce();
7211 let peer = producer.clone().peer();
7212 let mut announced = producer.consume().announced();
7213 tokio::time::sleep(Duration::from_millis(1)).await;
7215
7216 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7217 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7218 announced.assert_next_active("room");
7219
7220 let far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7222 let farther = peer.dynamic("room", route(&[9, 5, 6, 7], 7)).unwrap();
7223 let start = tokio::time::Instant::now();
7224 drop(near);
7225 drop(far);
7226 let update = announced.next().await.unwrap();
7227 assert_eq!(update.kind, AnnounceKind::Updated);
7228 assert_eq!(update.route.hops, hops(&[9, 5, 6, 7]));
7229 assert_eq!(start.elapsed(), DEFAULT_UPDATE_HOLD);
7230 announced.assert_next_wait();
7231
7232 drop(farther);
7234 announced.assert_next_ended("room");
7235 }
7236
7237 #[tokio::test(start_paused = true)]
7240 async fn a_retraction_quarantines_a_named_publisher() {
7241 let producer = origin(1).produce();
7242 let peer = producer.clone().peer();
7243 let mut announced = producer.consume().announced();
7244 tokio::time::sleep(Duration::from_millis(1)).await;
7245
7246 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7247
7248 let near = peer.dynamic("named", route(&[9, 2], 2)).unwrap();
7249 announced.assert_next_active("named");
7250 drop(near);
7251 announced.assert_next_ended("named");
7252 let start = tokio::time::Instant::now();
7253 let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap();
7254 let update = announced.next().await.unwrap();
7255 assert_eq!(update.kind, AnnounceKind::Announced);
7256 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7257 assert!(start.elapsed() >= RETRACT_QUARANTINE);
7258
7259 let near = peer.dynamic("anonymous", route(&[0, 2], 2)).unwrap();
7260 announced.assert_next_active("anonymous");
7261 drop(near);
7262 announced.assert_next_ended("anonymous");
7263 let _other = peer.dynamic("anonymous", route(&[0, 3, 4], 4)).unwrap();
7264 announced.assert_next_active("anonymous");
7265
7266 let queued = announced.anchoring.len();
7269 for chain in [[9, 2], [0, 2], [9, 2], [0, 2]] {
7270 let churn = peer.dynamic("churn", route(&chain, 2)).unwrap();
7271 announced.assert_next_active("churn");
7272 drop(churn);
7273 announced.assert_next_ended("churn");
7274 }
7275 assert_eq!(announced.anchoring.len(), queued + 1);
7276
7277 let gone = peer.dynamic("gone", route(&[9, 2], 2)).unwrap();
7280 announced.assert_next_active("gone");
7281 drop(gone);
7282 announced.assert_next_ended("gone");
7283 assert!(!announced.retracted.is_empty());
7284 for _ in 0..2 {
7285 announced.assert_next_wait();
7286 tokio::time::sleep(RETRACT_QUARANTINE).await;
7287 }
7288 announced.assert_next_wait();
7289 assert!(announced.retracted.is_empty(), "the ended quarantine is still held");
7290 }
7291
7292 #[tokio::test]
7296 async fn withdrawn_peer_hides_routes_through_it() {
7297 let producer = origin(1).produce();
7298 let peer = producer.clone().peer();
7299 let mut announced = producer.consume().announced();
7300
7301 let direct = || Route::default().with_hops(hops(&[9, 2])).with_via(origin(2));
7303 let first = peer.dynamic("room", direct()).unwrap();
7304 let relayed = peer
7305 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])).with_via(origin(3)))
7306 .unwrap();
7307 announced.assert_next_active("room");
7308 announced.assert_next_wait();
7309
7310 first.withdrawn();
7312 announced.assert_next_ended("room");
7313 announced.assert_next_wait();
7314
7315 let second = peer.dynamic("room", direct()).unwrap();
7317 announced.assert_next_active("room");
7318 assert!(producer.shared.lock().withdrawn.is_empty());
7319
7320 second.withdrawn();
7322 drop(relayed);
7323 announced.assert_next_ended("room");
7324 assert!(producer.shared.lock().withdrawn.is_empty());
7325 }
7326
7327 #[tokio::test]
7329 async fn old_session_withdrawal_keeps_newer_route() {
7330 for restart in [false, true] {
7331 let producer = origin(1).produce();
7332 let peer = producer.clone().peer();
7333 let mut announced = producer.consume().announced();
7334 let route = Route::default().with_hops(hops(&[9, 2]));
7335 let first = peer.dynamic("room", route.clone()).unwrap();
7336 let second = peer.dynamic("room", route.clone()).unwrap();
7337 announced.assert_next_active("room");
7338 announced.assert_next_wait();
7339 let remaining = if restart {
7340 first.update(route).unwrap();
7342 second.withdrawn();
7343 first
7344 } else {
7345 first.withdrawn();
7346 second
7347 };
7348 announced.assert_next_wait();
7349 assert!(producer.shared.lock().withdrawn.is_empty());
7350 remaining.withdrawn();
7351 announced.assert_next_ended("room");
7352 assert!(producer.shared.lock().withdrawn.is_empty());
7353 }
7354 }
7355
7356 #[tokio::test]
7357 async fn withdrawn_route_no_longer_covers_requests() {
7358 let producer = origin(1).produce();
7359 let peer = producer.clone().peer();
7360 let direct = peer.dynamic("room", Route::default().with_hops(hops(&[9, 2]))).unwrap();
7361 let _relayed = peer
7362 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])))
7363 .unwrap();
7364 let path = Path::new("room/video");
7365 let id = producer
7366 .shared
7367 .read()
7368 .routes
7369 .covering(&path)
7370 .find(|entry| entry.hops.iter().last() == Some(&origin(3)))
7371 .unwrap()
7372 .id;
7373 assert!(producer.shared.read().routes.covers(&path, id));
7374 direct.withdrawn();
7375 assert!(!producer.shared.read().routes.covers(&path, id));
7376 }
7377
7378 #[tokio::test]
7381 async fn source_change_is_an_update() {
7382 let producer = origin(1).produce();
7383 let peer = producer.clone().peer();
7384 let mut announced = producer.consume().announced();
7385
7386 let route = Route::default().with_hops(hops(&[7])).with_via(origin(7));
7387 let _forwarded = peer.dynamic("room", route.clone()).unwrap();
7388 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7389
7390 let local = producer.dynamic("room", route).unwrap();
7392 let update = announced.next().now_or_never().expect("next blocked").expect("no next");
7393 assert_eq!(update.kind, AnnounceKind::Updated);
7394 assert_eq!(update.route.source(), Source::Local);
7395
7396 drop(local);
7397 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7398 }
7399
7400 #[tokio::test]
7401 async fn local_view_hides_peer_routes() {
7402 let producer = origin(1).produce();
7403 let peer = producer.clone().peer();
7404 let mut local = producer.consume().local().announced();
7405
7406 let _forwarded = peer
7407 .dynamic("remote", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7408 .unwrap();
7409 local.assert_next_wait();
7410
7411 let _shadow = peer
7415 .dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7416 .unwrap();
7417 let ingest = producer
7418 .dynamic(
7419 "both",
7420 Route::default().with_hops(hops(&[5])).with_via(origin(5)).with_cost(9),
7421 )
7422 .unwrap();
7423 assert_eq!(local.assert_next_active("both").source(), Source::Local);
7424 drop(ingest);
7425 local.assert_next_ended("both");
7426
7427 let err = producer
7430 .consume()
7431 .local()
7432 .request_broadcast("remote/alice")
7433 .now_or_never()
7434 .expect("unroutable")
7435 .err()
7436 .unwrap();
7437 assert!(matches!(err, Error::Unroutable));
7438 assert!(
7439 producer
7440 .consume()
7441 .request_broadcast("remote/alice")
7442 .now_or_never()
7443 .is_none()
7444 );
7445 }
7446
7447 #[tokio::test]
7451 async fn handler_rejection_is_final() {
7452 let producer = origin(1).produce();
7453 let consumer = producer.consume();
7454 let server = producer.dynamic("room", Route::default()).unwrap();
7455
7456 let pending = consumer.request_broadcast("room/alice");
7457 let request = queued(&server).await;
7458 request.reject(Error::Unroutable);
7459 let err = tokio::time::timeout(Duration::from_secs(5), pending)
7460 .await
7461 .expect("the front must give up, not spin")
7462 .err()
7463 .unwrap();
7464 assert!(matches!(err, Error::Unroutable));
7465
7466 let pending = consumer.request_broadcast("room/bob");
7468 let request = queued(&server).await;
7469 assert_eq!(request.path().as_str(), "room/bob");
7470 let served = broadcast::Info::new().produce();
7471 request.accept(&served);
7472 pending.await.expect("resolves");
7473 }
7474
7475 #[tokio::test]
7478 async fn routed_broadcast_waits_out_a_rejection() {
7479 let producer = origin(1).produce();
7480 let consumer = producer.consume();
7481 let server = producer.dynamic("room", Route::default()).unwrap();
7482
7483 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7484 assert!((&mut resolving).now_or_never().is_none());
7485 let request = queued(&server).await;
7486 request.reject(Error::Unroutable);
7487
7488 for _ in 0..20 {
7490 tokio::task::yield_now().await;
7491 }
7492 assert!((&mut resolving).now_or_never().is_none());
7493 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7494
7495 server.update(Route::default().with_cost(2)).unwrap();
7497 assert!((&mut resolving).now_or_never().is_none());
7498 let request = queued(&server).await;
7499 let served = broadcast::Info::new().produce();
7500 request.accept(&served);
7501 resolving.await.expect("resolves");
7502 }
7503
7504 #[tokio::test]
7507 async fn routed_broadcast_reports_teardown_as_closed() {
7508 let (producer, driver) = Producer::new(Config::new(origin(1)));
7509 let consumer = producer.consume();
7510 let _server = producer.dynamic("room", Route::default()).unwrap();
7511
7512 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7514 assert!((&mut resolving).now_or_never().is_none());
7515
7516 drop(driver);
7517
7518 let err = tokio::time::timeout(Duration::from_secs(5), resolving)
7519 .await
7520 .expect("teardown resolves the wait")
7521 .err()
7522 .unwrap();
7523 assert!(matches!(err, Error::Closed), "unexpected end: {err}");
7524 }
7525
7526 #[tokio::test]
7529 async fn routed_broadcast_wakes_for_a_local_broadcast() {
7530 let producer = origin(1).produce();
7531 let consumer = producer.consume();
7532 let server = producer.dynamic("room", Route::default()).unwrap();
7533
7534 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7535 assert!((&mut resolving).now_or_never().is_none());
7536 queued(&server).await.reject(Error::Unroutable);
7537 for _ in 0..20 {
7538 tokio::task::yield_now().await;
7539 }
7540 assert!((&mut resolving).now_or_never().is_none());
7541
7542 let _local = producer.publish("room/alice", Route::default()).unwrap();
7544 let resolved = resolving.await.expect("resolves locally");
7545 assert_eq!(resolved.info().path.as_str(), "room/alice");
7546 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7547 }
7548
7549 #[tokio::test]
7552 async fn late_track_on_a_served_front_replays() {
7553 let producer = origin(1).produce();
7554 let consumer = producer.consume();
7555 let server = producer.dynamic("room", Route::default()).unwrap();
7556
7557 let source = broadcast::Info::new().produce();
7558 for name in ["a", "b"] {
7559 let track = source.create_track(name, None).unwrap();
7560 let mut group = track.append_group().unwrap();
7561 group.write_frame(crate::Timestamp::ZERO, name.as_bytes()).unwrap();
7562 group.finish().unwrap();
7563 std::mem::forget(track);
7565 }
7566
7567 let pending = consumer.request_broadcast("room/alice");
7568 queued(&server).await.accept(&source);
7569 let resolved = pending.await.expect("resolves");
7570
7571 let budget = track::Subscription::default().with_max_age(Duration::from_secs(3600));
7572 for name in ["a", "b"] {
7573 let mut subscription = resolved
7574 .track(name)
7575 .unwrap()
7576 .subscribe(budget.clone())
7577 .await
7578 .expect("subscribe");
7579 let mut group = tokio::time::timeout(Duration::from_secs(5), subscription.recv_group())
7580 .await
7581 .expect("the late track must replay, not park")
7582 .expect("recv group")
7583 .expect("track ended early");
7584 let frame = group.read_frame().await.expect("read frame").expect("frame");
7585 assert_eq!(&frame.payload[..], name.as_bytes());
7586 }
7587 }
7588
7589 #[tokio::test]
7590 async fn most_specific_prefix_shadows() {
7591 let producer = origin(1).produce();
7592 let consumer = producer.consume();
7593
7594 let broad_server = producer.dynamic("", Route::default()).unwrap();
7595 let _narrow = producer.announce(".dash", Route::default()).unwrap();
7598
7599 let err = consumer
7600 .request_broadcast(".dash/pid")
7601 .now_or_never()
7602 .expect("unroutable")
7603 .err()
7604 .unwrap();
7605 assert!(matches!(err, Error::Unroutable));
7606
7607 let _pending = consumer.request_broadcast("room/alice");
7609 let request = queued(&broad_server).await;
7610 assert_eq!(request.path().as_str(), "room/alice");
7611 }
7612
7613 #[tokio::test]
7614 async fn root_dynamic_serves_any_path() {
7615 let producer = origin(1).produce();
7616 let consumer = producer.consume();
7617 let mut announced = consumer.announced();
7618 let dynamic = producer.dynamic("", Route::default()).unwrap();
7619 announced.assert_next_active("");
7621
7622 let pending = consumer.request_broadcast("anything/at/all");
7623 let request = queued(&dynamic).await;
7624 assert_eq!(request.path().as_str(), "anything/at/all");
7625
7626 let source = broadcast::Info::new().produce();
7627 request.accept(&source);
7628 let resolved = pending.await.expect("resolves");
7629 assert_eq!(resolved.info().path.as_str(), "anything/at/all");
7630
7631 drop(dynamic);
7633 announced.assert_next_ended("");
7634 let err = consumer
7635 .request_broadcast("something/else")
7636 .now_or_never()
7637 .expect("unroutable")
7638 .err()
7639 .unwrap();
7640 assert!(matches!(err, Error::Unroutable));
7641 }
7642
7643 #[tokio::test]
7649 async fn out_of_scope_request_never_reaches_the_dynamic_handler() {
7650 let producer = origin(1).produce();
7651 let dynamic = producer.dynamic("", Route::default()).unwrap();
7652 let scoped = producer.consume().scope("", &scopes(&["tenant-a"])).unwrap();
7653
7654 for path in ["tenant-b/live", "tenant-a-other/live"] {
7657 let refused = scoped
7658 .request_broadcast(path)
7659 .now_or_never()
7660 .expect("an out-of-scope request must be refused synchronously, not queued");
7661 assert!(matches!(refused, Err(Error::Unauthorized)));
7662 assert!(
7663 dynamic.requested_broadcast().now_or_never().is_none(),
7664 "the dynamic handler was asked to create a broadcast the requester may not read"
7665 );
7666 }
7667 }
7668
7669 #[tokio::test]
7670 async fn routed_waits_for_coverage() {
7671 let producer = origin(1).produce();
7672 let consumer = producer.consume();
7673
7674 let mut fut = consumer.routed("room/alice").boxed();
7675 assert!((&mut fut).now_or_never().is_none());
7676
7677 let _a = producer.announce("room", Route::default().with_cost(3)).unwrap();
7679 let route = fut.now_or_never().expect("covered").expect("routed");
7680 assert_eq!(route.cost, Cost::new(3));
7681
7682 consumer
7684 .routed("room/alice/cam")
7685 .now_or_never()
7686 .expect("covered")
7687 .expect("routed");
7688 }
7689
7690 #[tokio::test]
7691 async fn routed_ignores_deeper_routes() {
7692 let producer = origin(1).produce();
7693 let consumer = producer.consume();
7694
7695 let _deep = producer.announce("room/alice/cam", Route::default()).unwrap();
7697 let mut fut = consumer.routed("room/alice").boxed();
7698 assert!((&mut fut).now_or_never().is_none());
7699
7700 let _exact = producer.announce("room/alice", Route::default()).unwrap();
7701 fut.now_or_never().expect("covered").expect("routed");
7702 }
7703
7704 #[tokio::test]
7705 async fn routed_accepts_a_max_depth_path() {
7706 let producer = origin(1).produce();
7707 let consumer = producer.consume();
7708 let path = (0..Path::MAX_PARTS)
7709 .map(|i| format!("s{i}"))
7710 .collect::<Vec<_>>()
7711 .join("/");
7712 assert_eq!(Path::new(&path).parts().count(), Path::MAX_PARTS);
7713
7714 assert!(consumer.allowed().matches(&path));
7715
7716 let mut fut = consumer.routed(&path).boxed();
7717 assert!((&mut fut).now_or_never().is_none());
7718
7719 let _a = producer.announce("", Route::default()).unwrap();
7721 fut.now_or_never().expect("covered").expect("routed");
7722 }
7723
7724 #[tokio::test]
7725 async fn teardown_ends_everything() {
7726 let (producer, driver) = Producer::new(Config::new(origin(1)));
7727 let consumer = producer.consume();
7728 let _announcement = producer.announce("room", Route::default()).unwrap();
7729 let mut announced = consumer.announced();
7730 announced.assert_next_active("room");
7731
7732 let _server = producer.dynamic("served", Route::default()).unwrap();
7733 let pending = consumer.request_broadcast("served/path");
7734
7735 drop(driver);
7736
7737 announced.assert_next_active("served");
7739 assert!(announced.next().now_or_never().expect("ended").is_none());
7740
7741 assert!(pending.now_or_never().expect("rejected").is_err());
7743 assert!(matches!(producer.announce("x", Route::default()), Err(Error::Closed)));
7744 assert!(matches!(producer.create_broadcast("x"), Err(Error::Closed)));
7745 let err = consumer
7746 .request_broadcast("y")
7747 .now_or_never()
7748 .expect("closed")
7749 .err()
7750 .unwrap();
7751 assert!(matches!(err, Error::Closed));
7752
7753 let mut late = consumer.announced();
7755 assert!(late.next().now_or_never().expect("ended").is_none());
7756 }
7757
7758 struct ResumeRig {
7761 producer: Producer,
7762 resolved: broadcast::Consumer,
7763 subscription: track::Subscriber,
7764 incumbent_track: track::Producer,
7767 }
7768
7769 impl ResumeRig {
7770 async fn new(first: &[u64]) -> (Self, Dynamic, broadcast::Producer) {
7773 let producer = origin(1).produce();
7774 let consumer = producer.consume();
7775
7776 let server = producer
7777 .dynamic("room", Route::default().with_hops(hops(first)))
7778 .unwrap();
7779
7780 let pending = consumer.request_broadcast("room/alice");
7781 let request = queued(&server).await;
7782 let source = broadcast::Info::new().produce();
7783 let track = source.create_track("video", None).unwrap();
7784 let mut group = track.append_group().unwrap();
7785 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7786 group.finish().unwrap();
7787 request.accept(&source);
7788
7789 let resolved = pending.await.expect("resolves");
7790 let mut subscription = resolved
7791 .track("video")
7792 .unwrap()
7793 .subscribe(None)
7794 .await
7795 .expect("subscribe");
7796 let mut group = subscription
7797 .recv_group()
7798 .await
7799 .expect("recv group")
7800 .expect("track ended early");
7801 let frame = group.read_frame().await.expect("read frame").expect("frame");
7802 assert_eq!(&frame.payload[..], b"before");
7803
7804 (
7805 Self {
7806 producer,
7807 resolved,
7808 subscription,
7809 incumbent_track: track,
7810 },
7811 server,
7812 source,
7813 )
7814 }
7815
7816 fn standby(&self, first: &[u64]) -> Dynamic {
7819 self.producer
7820 .dynamic("room", Route::default().with_hops(hops(first)))
7821 .unwrap()
7822 }
7823 }
7824
7825 async fn assert_resumes(rig: &mut ResumeRig, server: &Dynamic) {
7830 let request = queued(server).await;
7831 let replacement = broadcast::Info::new().produce();
7832 let track = replacement.create_track("video", None).unwrap();
7833 let mut group = track.append_group().unwrap();
7836 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7837 group.finish().unwrap();
7838 request.accept(&replacement);
7839
7840 let mut group = track.append_group().unwrap();
7841 group.write_frame(crate::Timestamp::ZERO, b"resumed".as_ref()).unwrap();
7842 group.finish().unwrap();
7843
7844 let mut group = rig
7845 .subscription
7846 .recv_group()
7847 .await
7848 .expect("subscription survives the failover")
7849 .expect("track ended early");
7850 let frame = group.read_frame().await.expect("read frame").expect("frame");
7851 assert_eq!(&frame.payload[..], b"resumed");
7852 }
7853
7854 #[tokio::test]
7857 async fn driver_resolves_with_live_consumers() {
7858 let (producer, driver) = Producer::new(Config::new(origin(1)));
7859 let consumer = producer.consume();
7860 let run = crate::time::run(driver);
7861 drop(producer);
7862 tokio::time::timeout(Duration::from_secs(5), run)
7863 .await
7864 .expect("driver must finish once the producers are gone");
7865 drop(consumer);
7866 }
7867
7868 #[tokio::test]
7869 async fn remote_source_resumes_through_same_first_hop() {
7870 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7871 let standby_server = rig.standby(&[10, 20]);
7872
7873 drop(incumbent);
7875 drop(source);
7876
7877 assert_resumes(&mut rig, &standby_server).await;
7879 }
7880
7881 #[tokio::test]
7885 async fn incompatible_successor_is_refused() {
7886 for replacement in [
7887 track::Info::default().with_timescale(crate::Timescale::MICRO),
7888 track::Info::default().with_priority(7),
7889 track::Info::default().with_max_age(Duration::from_secs(7)),
7890 ] {
7891 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7892 let standby_server = rig.standby(&[10, 20]);
7893 drop(incumbent);
7894 drop(source);
7895
7896 let request = queued(&standby_server).await;
7899 let successor = broadcast::Info::new().produce();
7900 let track = successor.create_track("video", replacement).unwrap();
7901 let mut group = track.append_group().unwrap();
7902 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7903 group.finish().unwrap();
7904 request.accept(&successor);
7905
7906 assert!(
7907 matches!(rig.subscription.recv_group().await, Err(Error::Unsupported)),
7908 "the subscription must abort rather than resume onto incompatible metadata"
7909 );
7910
7911 let reopened = rig.resolved.track("video").unwrap();
7913 assert!(matches!(reopened.query().await, Err(Error::Unsupported)));
7914 assert!(matches!(reopened.subscribe(None).await, Err(Error::Unsupported)));
7915 }
7916 }
7917
7918 #[tokio::test]
7919 async fn different_first_hop_ends_the_subscription() {
7920 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7921 let rival_server = rig.standby(&[11]);
7923
7924 drop(incumbent);
7927 drop(source);
7928 rig.incumbent_track.abort(Error::Dropped).unwrap();
7929
7930 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
7932 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
7933
7934 let consumer = rig.producer.consume();
7936 let pending = consumer.request_broadcast("room/alice");
7937 let request = queued(&rival_server).await;
7938 let replacement = broadcast::Info::new().produce();
7939 request.accept(&replacement);
7940 pending.await.expect("re-request resolves through the rival");
7941 }
7942
7943 #[tokio::test]
7948 async fn a_first_hop_update_drains_the_old_publisher() {
7949 for first in [&[][..], &[0][..], &[10][..]] {
7950 let (mut rig, server, _source) = ResumeRig::new(first).await;
7951 server.update(Route::default().with_hops(hops(&[11]))).unwrap();
7952
7953 let mut group = rig.incumbent_track.append_group().unwrap();
7955 group.write_frame(crate::Timestamp::ZERO, b"draining".as_ref()).unwrap();
7956 group.finish().unwrap();
7957 let mut group = next_group(&mut rig.subscription)
7958 .await
7959 .expect("the in-flight subscription survives the update")
7960 .expect("track ended early");
7961 let frame = group.read_frame().await.expect("read frame").expect("frame");
7962 assert_eq!(&frame.payload[..], b"draining", "first hop {first:?}");
7963
7964 let consumer = rig.producer.consume();
7967 let pending = consumer.request_broadcast("room/alice");
7968 let request = queued(&server).await;
7969 let replacement = broadcast::Info::new().produce();
7970 let track = replacement
7971 .create_track("video", track::Info::default().with_priority(7))
7972 .unwrap();
7973 let mut group = track.append_group().unwrap();
7974 group.write_frame(crate::Timestamp::ZERO, b"new".as_ref()).unwrap();
7975 group.finish().unwrap();
7976 request.accept(&replacement);
7977
7978 let resolved = pending.await.expect("resolves through the new publisher");
7979 assert!(
7980 !resolved.is_clone(&rig.resolved),
7981 "first hop {first:?} joined the old front"
7982 );
7983 let mut fresh = resolved
7984 .track("video")
7985 .unwrap()
7986 .subscribe(None)
7987 .await
7988 .expect("the old publisher's track info does not apply");
7989 let mut group = next_group(&mut fresh)
7990 .await
7991 .expect("recv group")
7992 .expect("track ended early");
7993 let frame = group.read_frame().await.expect("read frame").expect("frame");
7994 assert_eq!(&frame.payload[..], b"new");
7995
7996 rig.incumbent_track.finish().unwrap();
7999 let end = next_group(&mut rig.subscription).await;
8000 assert!(
8001 !matches!(end, Ok(Some(_))),
8002 "first hop {first:?} spliced the new publisher into a live subscription"
8003 );
8004 }
8005 }
8006
8007 #[tokio::test]
8010 async fn a_first_hop_update_resumes_through_the_same_publisher() {
8011 let (mut rig, incumbent, _source) = ResumeRig::new(&[10]).await;
8012 let standby_server = rig.standby(&[10, 20]);
8013
8014 incumbent.update(Route::default().with_hops(hops(&[11]))).unwrap();
8015
8016 assert_resumes(&mut rig, &standby_server).await;
8017 assert!(
8018 incumbent.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8019 "the front never asks the new publisher"
8020 );
8021 }
8022
8023 #[tokio::test]
8024 async fn anonymous_routes_never_resume() {
8025 let (mut rig, incumbent, source) = ResumeRig::new(&[]).await;
8028 let _twin_server = rig.standby(&[]);
8029
8030 drop(incumbent);
8031 drop(source);
8032 rig.incumbent_track.abort(Error::Dropped).unwrap();
8033
8034 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
8035 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8036 }
8037
8038 #[tokio::test]
8045 async fn anonymous_handoff_serves_the_newcomer_immediately() {
8046 let producer = origin(1).produce();
8047
8048 let server_a = producer
8050 .dynamic("room", Route::default().with_hops(hops(&[10])))
8051 .unwrap();
8052
8053 let consumer = producer.consume().excluding(origin(30));
8056 let pending = consumer.request_broadcast("room/alice");
8057 let request = queued(&server_a).await;
8058 let source_a = broadcast::Info::new().produce();
8059 let track_a = source_a.create_track("video", None).unwrap();
8060 let mut group = track_a.append_group().unwrap();
8061 group.write_frame(crate::Timestamp::ZERO, b"from-a".as_ref()).unwrap();
8062 group.finish().unwrap();
8063 request.accept(&source_a);
8064
8065 let resolved_a = pending.await.expect("resolves");
8066 let mut sub_a = resolved_a
8067 .track("video")
8068 .unwrap()
8069 .subscribe(None)
8070 .await
8071 .expect("subscribe");
8072 let mut group = sub_a
8073 .recv_group()
8074 .await
8075 .expect("recv group")
8076 .expect("track ended early");
8077 assert_eq!(
8078 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8079 b"from-a"
8080 );
8081
8082 drop(track_a);
8085 drop(source_a);
8086 drop(server_a);
8087
8088 let err = sub_a.recv_group().await.err().expect("front closed");
8090 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8091
8092 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8096 settle(|| {
8097 matches!(
8098 consumer.request_broadcast("room/alice").now_or_never(),
8099 Some(Err(Error::Unroutable))
8100 )
8101 })
8102 .await;
8103
8104 let server_b = producer
8106 .dynamic("room", Route::default().with_hops(hops(&[20])))
8107 .unwrap();
8108 let pending = consumer.request_broadcast("room/alice");
8109 let request = queued(&server_b).await;
8110 let source_b = broadcast::Info::new().produce();
8111 let track_b = source_b.create_track("video", None).unwrap();
8112 let mut group = track_b.append_group().unwrap();
8113 group.write_frame(crate::Timestamp::ZERO, b"from-b".as_ref()).unwrap();
8114 group.finish().unwrap();
8115 request.accept(&source_b);
8116
8117 let resolved_b = pending.await.expect("B's front is served immediately");
8118 assert!(
8119 !resolved_b.is_clone(&resolved_a),
8120 "B must not splice into A's closed front"
8121 );
8122
8123 let mut sub_b = resolved_b
8124 .track("video")
8125 .unwrap()
8126 .subscribe(None)
8127 .await
8128 .expect("subscribe");
8129 let mut group = sub_b
8130 .recv_group()
8131 .await
8132 .expect("recv group")
8133 .expect("track ended early");
8134 assert_eq!(
8135 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8136 b"from-b"
8137 );
8138 }
8139
8140 #[tokio::test]
8141 async fn reprice_is_invisible_to_the_subscription() {
8142 let (rig, incumbent, source) = ResumeRig::new(&[10]).await;
8143
8144 incumbent
8147 .update(Route::default().with_hops(hops(&[10])).with_cost(9))
8148 .unwrap();
8149
8150 let track = source.create_track("audio", None).unwrap();
8151 let mut group = track.append_group().unwrap();
8152 group.write_frame(crate::Timestamp::ZERO, b"steady".as_ref()).unwrap();
8153 group.finish().unwrap();
8154
8155 let mut audio = rig
8156 .resolved
8157 .track("audio")
8158 .unwrap()
8159 .subscribe(None)
8160 .await
8161 .expect("subscribe survives the reprice");
8162 let mut group = audio
8163 .recv_group()
8164 .await
8165 .expect("recv group")
8166 .expect("track ended early");
8167 let frame = group.read_frame().await.expect("read frame").expect("frame");
8168 assert_eq!(&frame.payload[..], b"steady");
8169 }
8170
8171 #[tokio::test]
8172 async fn drain_reprice_migrates_before_the_session_dies() {
8173 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8174 let standby_server = rig.standby(&[10, 20]);
8175
8176 incumbent
8180 .update(Route::default().with_hops(hops(&[10])).with_cost(Cost::DRAIN))
8181 .unwrap();
8182
8183 assert_resumes(&mut rig, &standby_server).await;
8184
8185 drop(incumbent);
8187 drop(source);
8188 }
8189
8190 #[tokio::test]
8191 async fn local_sources_splice_newest_first() {
8192 let producer = origin(1).produce();
8193 let consumer = producer.consume();
8194
8195 let first = producer.publish("room/alice", Route::default()).unwrap();
8196 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8197
8198 let second = producer.publish("room/alice", Route::default()).unwrap();
8200 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
8201 assert!(again.is_clone(&resolved));
8202
8203 first.close();
8205 settle(|| consumer.get_broadcast("room/alice").is_some()).await;
8206 second.close();
8207 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8208
8209 let _third = producer.publish("room/alice", Route::default()).unwrap();
8211 assert!(consumer.get_broadcast("room/alice").is_some());
8212 }
8213
8214 #[tokio::test]
8222 async fn a_finished_broadcast_concludes_in_flight_subscriptions() {
8223 let producer = origin(1).produce();
8224 let consumer = producer.consume();
8225
8226 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8227 let track = broadcast.create_track("video", None).unwrap();
8228
8229 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8230 let mut subscription = resolved
8231 .track("video")
8232 .unwrap()
8233 .subscribe(None)
8234 .await
8235 .expect("subscribe");
8236 let mut group = track.append_group().unwrap();
8238 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8239 group.finish().unwrap();
8240 track.finish().unwrap();
8241 drop(track);
8242 broadcast.close();
8243
8244 let mut group = next_group(&mut subscription)
8245 .await
8246 .expect("a cleanly finished track was served as an error")
8247 .expect("the track ended before its last group");
8248 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8249 drop(group);
8250
8251 let end = next_group(&mut subscription)
8252 .await
8253 .expect("a cleanly finished track ended as an error");
8254 assert!(end.is_none(), "a group followed the final one");
8255 }
8256
8257 #[tokio::test]
8264 async fn a_retracted_route_concludes_in_flight_subscriptions() {
8265 let producer = origin(1).produce();
8266 let consumer = producer.consume();
8267 let server = producer
8268 .dynamic("room", Route::default().with_hops(hops(&[10])))
8269 .unwrap();
8270
8271 let pending = consumer.request_broadcast("room/alice");
8272 let request = queued(&server).await;
8273 let source = broadcast::Info::new().produce();
8274 let track = source.create_track("video", None).unwrap();
8275 request.accept(&source);
8276
8277 let resolved = pending.await.expect("resolves");
8278 let mut subscription = resolved
8279 .track("video")
8280 .unwrap()
8281 .subscribe(None)
8282 .await
8283 .expect("subscribe");
8284
8285 source.close();
8288 drop(server);
8289 settle(|| resolved.is_closed()).await;
8290 let mut group = track.append_group().unwrap();
8291 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8292 group.finish().unwrap();
8293 track.finish().unwrap();
8294 drop(track);
8295
8296 let mut group = next_group(&mut subscription)
8297 .await
8298 .expect("a retracted route's track was served as an error")
8299 .expect("the track ended before its last group");
8300 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8301 drop(group);
8302
8303 let end = next_group(&mut subscription)
8304 .await
8305 .expect("a cleanly finished track ended as an error");
8306 assert!(end.is_none(), "a group followed the final one");
8307 }
8308
8309 #[tokio::test]
8312 async fn a_closed_source_is_not_requested_again_from_its_standing_route() {
8313 let producer = origin(1).produce();
8314 let server = producer
8315 .dynamic("room", Route::default().with_hops(hops(&[10])))
8316 .unwrap();
8317 let pending = producer.consume().request_broadcast("room/alice");
8318 let source = broadcast::Info::new().produce();
8319 queued(&server).await.accept(&source);
8320 let resolved = pending.await.unwrap();
8321
8322 source.close();
8323 settle(|| resolved.is_closed()).await;
8324 assert!(
8325 server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8326 "the closed source was requested again"
8327 );
8328 }
8329
8330 #[tokio::test]
8335 async fn origin_front_drops_the_source_when_unused() {
8336 let producer = origin(1).produce();
8337 let consumer = producer.consume();
8338
8339 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8340 let track = broadcast.create_track("video", None).unwrap();
8341 let mut group = track.append_group().unwrap();
8342 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8343 group.finish().unwrap();
8344
8345 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8346 let mut subscription = resolved
8347 .track("video")
8348 .unwrap()
8349 .subscribe(None)
8350 .await
8351 .expect("subscribe");
8352 let mut group = subscription.recv_group().await.unwrap().unwrap();
8353 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8354 drop(group);
8355 drop(subscription);
8356
8357 tokio::time::timeout(Duration::from_secs(1), track.unused())
8358 .await
8359 .expect("source unused should resolve far below TRACK_IDLE_LINGER")
8360 .expect("source closed");
8361
8362 let mut again = resolved
8365 .track("video")
8366 .unwrap()
8367 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8368 .await
8369 .expect("resubscribe");
8370 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8371 .await
8372 .expect("cached group is still on the front")
8373 .expect("recv group")
8374 .expect("track ended early");
8375 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8376
8377 tokio::time::timeout(Duration::from_secs(1), track.used())
8378 .await
8379 .expect("returning reader re-splices the source")
8380 .expect("source closed");
8381
8382 let mut group = track.append_group().unwrap();
8383 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8384 group.finish().unwrap();
8385 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8386 .await
8387 .expect("groups past the cached edge come from the re-splice")
8388 .expect("recv group")
8389 .expect("track ended early");
8390 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8391 }
8392
8393 #[tokio::test]
8403 async fn resumed_reader_skips_warm_groups_behind_the_new_edge() {
8404 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
8405 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
8406 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
8407 let write = |source: &track::Producer, sequence: u64, millis: u64| {
8408 let mut group = source.create_group(sequence.into()).unwrap();
8409 group.write_frame(ms(millis), b"x".as_ref()).unwrap();
8410 group.finish().unwrap();
8411 };
8412 let drain = |subscription: &mut track::Subscriber| {
8413 let mut sequences = Vec::new();
8414 while let Poll::Ready(group) = subscription.poll_recv_group(&kio::Waiter::noop()) {
8415 sequences.push(group.unwrap().expect("track ended").sequence);
8416 }
8417 sequences
8418 };
8419
8420 let track = resolved.track("video").unwrap();
8421 let first = budget.clone();
8422 let subscribing = tokio::spawn(async move { track.subscribe(first).await });
8423 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8424 .await
8425 .expect("the front asked the source")
8426 .expect("request");
8427 let source = request.resolving_start().accept(None);
8428 for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] {
8429 write(&source, sequence, millis);
8430 }
8431 let mut subscription = subscribing.await.unwrap().expect("subscribe");
8432 next_group(&mut subscription).await.unwrap().expect("a cached group");
8433 drain(&mut subscription);
8434 drop(subscription);
8435
8436 tokio::time::timeout(Duration::from_secs(1), source.unused())
8437 .await
8438 .expect("parked")
8439 .expect("source open");
8440 drop(source);
8441
8442 let track = resolved.track("video").unwrap();
8443 let second = budget.clone();
8444 let subscribing = tokio::spawn(async move { track.subscribe(second).await });
8445 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8446 .await
8447 .expect("the front asked the source again")
8448 .expect("request");
8449 let mut source = request.resolving_start().accept(None);
8450 for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] {
8451 write(&source, sequence, millis);
8452 }
8453 source.start_at(3).unwrap();
8456 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
8457 settle(|| subscription.latest() == Some(22)).await;
8458
8459 assert_eq!(drain(&mut subscription), [3, 20, 21, 22]);
8462 }
8463
8464 #[tokio::test]
8469 async fn chained_front_drops_the_source_when_unused() {
8470 let leaf = origin(1).produce();
8471 let leaf_consumer = leaf.consume();
8472
8473 let broadcast = leaf.publish("room/alice", Route::default()).unwrap();
8474 let track = broadcast.create_track("video", None).unwrap();
8475 let mut group = track.append_group().unwrap();
8476 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8477 group.finish().unwrap();
8478
8479 let leaf_front = leaf_consumer.request_broadcast("room/alice").await.expect("resolves");
8482
8483 let mid = origin(2).produce();
8484 let mid_server = mid.dynamic("room", Route::default().with_hops(hops(&[10]))).unwrap();
8485 let mid_pending = mid.consume().request_broadcast("room/alice");
8486 queued(&mid_server).await.accept(&leaf_front);
8487 let mid_resolved = mid_pending.await.expect("mid resolves");
8488
8489 let edge = origin(3).produce();
8490 let edge_server = edge.dynamic("room", Route::default().with_hops(hops(&[20]))).unwrap();
8491 let edge_pending = edge.consume().request_broadcast("room/alice");
8492 queued(&edge_server).await.accept(&mid_resolved);
8493 let edge_resolved = edge_pending.await.expect("edge resolves");
8494
8495 let mut subscription = edge_resolved
8496 .track("video")
8497 .unwrap()
8498 .subscribe(None)
8499 .await
8500 .expect("subscribe");
8501 let mut group = subscription.recv_group().await.unwrap().unwrap();
8502 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8503 drop(group);
8504 drop(subscription);
8505
8506 tokio::time::timeout(Duration::from_secs(5), track.unused())
8507 .await
8508 .expect("chained unused should resolve far below TRACK_IDLE_LINGER")
8509 .expect("source closed");
8510
8511 let cached = edge_resolved.track("video").unwrap().cached_groups();
8512 assert_eq!(
8513 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8514 vec![0],
8515 "every front keeps the delivered groups after releasing its source"
8516 );
8517
8518 let mut subscription = edge_resolved
8520 .track("video")
8521 .unwrap()
8522 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8523 .await
8524 .expect("resubscribe");
8525 tokio::time::timeout(Duration::from_secs(5), track.used())
8526 .await
8527 .expect("resubscribe should reach the leaf")
8528 .expect("source open");
8529 let mut group = track.append_group().unwrap();
8530 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8531 group.finish().unwrap();
8532 let mut group = subscription.recv_group().await.unwrap().unwrap();
8533 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8534 drop(group);
8535 let mut group = subscription.recv_group().await.unwrap().unwrap();
8536 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8537 drop(group);
8538 drop(subscription);
8539
8540 tokio::time::timeout(Duration::from_secs(5), track.unused())
8541 .await
8542 .expect("second chained unused should resolve far below TRACK_IDLE_LINGER")
8543 .expect("source closed");
8544
8545 let cached = edge_resolved.track("video").unwrap().cached_groups();
8546 assert_eq!(
8547 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8548 vec![0, 1],
8549 "repeated demand keeps every complete group while releasing its source"
8550 );
8551
8552 let fetch = edge_resolved.track("video").unwrap().fetch_group(2, None);
8553 let mut fetch = std::pin::pin!(fetch);
8554 assert!(futures::poll!(fetch.as_mut()).is_pending(), "fetch should re-splice");
8555 tokio::time::timeout(Duration::from_secs(5), track.used())
8556 .await
8557 .expect("fetch should reach the leaf")
8558 .expect("source open");
8559 let mut group = track.append_group().unwrap();
8560 group.write_frame(crate::Timestamp::ZERO, b"fetched".as_ref()).unwrap();
8561 group.finish().unwrap();
8562 let mut group = tokio::time::timeout(Duration::from_secs(5), fetch)
8563 .await
8564 .expect("re-spliced source should answer the fetch")
8565 .expect("fetch succeeds");
8566 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"fetched");
8567 }
8568
8569 #[tokio::test]
8573 async fn incompatible_local_source_keeps_the_incumbent() {
8574 let producer = origin(1).produce();
8575 let consumer = producer.consume();
8576
8577 let first = producer.publish("room/alice", Route::default()).unwrap();
8578 let track = first.create_track("video", None).unwrap();
8579 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8580 let mut subscription = resolved
8581 .track("video")
8582 .unwrap()
8583 .subscribe(None)
8584 .await
8585 .expect("subscribe");
8586 let mut group = track.append_group().unwrap();
8587 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8588 group.finish().unwrap();
8589 let mut group = subscription.recv_group().await.unwrap().unwrap();
8590 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"before");
8591
8592 let second = producer.publish("room/alice", Route::default()).unwrap();
8594 let _incompatible = second
8595 .create_track("video", track::Info::default().with_timescale(crate::Timescale::MICRO))
8596 .unwrap();
8597 for _ in 0..10 {
8598 tokio::task::yield_now().await;
8599 }
8600
8601 let mut group = track.append_group().unwrap();
8603 group.write_frame(crate::Timestamp::ZERO, b"still".as_ref()).unwrap();
8604 group.finish().unwrap();
8605 let mut group = subscription.recv_group().await.unwrap().unwrap();
8606 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"still");
8607
8608 drop(track);
8610 first.close();
8611 assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported)));
8612 }
8613
8614 #[tokio::test]
8615 async fn multiple_scopes_present_one_broad_prefix() {
8616 let producer = origin(1).produce();
8617 let _a = producer.announce("", Route::default()).unwrap();
8618
8619 let consumer = producer.consume().scope("", &scopes(&["alpha", "beta"])).unwrap();
8620 let mut announced = consumer.announced();
8621 announced.assert_next_active("");
8622 announced.assert_next_wait();
8623 }
8624
8625 #[test]
8626 fn scope_accepts_every_pattern_union() {
8627 let producer = origin(1).produce();
8628
8629 let root = producer.scope("", &Patterns::from(Pattern::all())).unwrap();
8631 assert_eq!(root.allowed(), Patterns::from(Pattern::all()));
8632
8633 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8635 assert_eq!(scoped.allowed(), scopes(&["room"]));
8636
8637 let multi = producer.scope("", &scopes(&["room", "room/chat", "anon"])).unwrap();
8639 assert_eq!(multi.allowed(), scopes(&["room", "anon"]));
8640
8641 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
8643 assert_eq!(consumer.allowed(), scopes(&["room"]));
8644
8645 for text in ["room", "", "*room", "room/*", "*", "**/room", "room/**/chat", "*.hang"] {
8646 let union = Patterns::from(text.parse::<Pattern>().unwrap());
8647 assert_eq!(producer.scope("", &union).expect(text).allowed(), union, "{text}");
8648 assert_eq!(
8649 producer.consume().scope("", &union).expect(text).allowed(),
8650 union,
8651 "{text}"
8652 );
8653 }
8654
8655 let mixed: Patterns = ["room/**".parse().unwrap(), "other".parse().unwrap()]
8656 .into_iter()
8657 .collect();
8658 assert_eq!(producer.scope("", &mixed).unwrap().allowed(), mixed);
8659 }
8660
8661 #[test]
8662 fn route_table_prunes_to_empty() {
8663 let producer = origin(1).produce();
8664 let consumer = producer.consume();
8665
8666 let cursor = consumer
8669 .scope("", &scopes(&["room/a", "other/deep/head"]))
8670 .unwrap()
8671 .announced();
8672 let route = producer.announce("room/a/b/c", Route::default()).unwrap();
8673 {
8674 let table = producer.shared.lock();
8675 assert!(table.routes.root.find(Path::new("room/a/b/c").parts()).is_some());
8676 assert!(table.routes.root.find(Path::new("other/deep/head").parts()).is_some());
8677 assert_eq!(table.routes.root.cursors_below, 2);
8678 }
8679
8680 drop(route);
8681 drop(cursor);
8682 let table = producer.shared.lock();
8683 assert!(table.routes.root.is_empty());
8684 assert_eq!(table.routes.root.cursors_below, 0);
8685 }
8686
8687 #[test]
8691 fn a_published_broadcast_keeps_the_driver_running() {
8692 let (producer, mut driver) = Producer::new(Config::new(origin(1)));
8693 let waiter = kio::Waiter::noop();
8694 let broadcast = producer.create_broadcast("room/a").unwrap();
8695 drop(producer);
8696 assert!(
8697 driver.poll(Instant::now(), &waiter).is_ok(),
8698 "the broadcast is lifecycle work"
8699 );
8700 drop(broadcast);
8701 assert!(matches!(driver.poll(Instant::now(), &waiter), Err(Error::Closed)));
8702 }
8703
8704 #[tokio::test]
8707 async fn a_dynamic_keeps_the_driver_running() {
8708 let (producer, driver) = Producer::new(Config::new(origin(1)));
8709 let run = tokio::spawn(crate::time::run(driver));
8710 let consumer = producer.consume();
8711 let server = producer.dynamic("room", Route::default()).unwrap();
8712 drop(producer);
8713
8714 let pending = consumer.request_broadcast("room/alice");
8715 let request = queued(&server).await;
8716 let source = broadcast::Info::new().produce();
8717 request.accept(&source);
8718 let resolved = pending.await.expect("the dynamic still serves");
8719
8720 drop(resolved);
8721 drop(source);
8722 drop(server);
8723 tokio::time::timeout(Duration::from_secs(5), run)
8724 .await
8725 .expect("driver must finish once the dynamic is gone")
8726 .unwrap();
8727 drop(consumer);
8728 }
8729
8730 #[test]
8731 fn watch_wakes_only_for_covering_changes() {
8732 let producer = origin(1).produce();
8733 let waiter = kio::Waiter::noop();
8734 let watch = producer.shared.lock().watch(&producer.shared, &Path::new("room/a"));
8735 let seen = watch.seen();
8736
8737 let _other = producer.announce("other", Route::default()).unwrap();
8739 let _below = producer.announce("room/a/b", Route::default()).unwrap();
8740 assert!(watch.poll_changed(&waiter, seen).is_pending());
8741
8742 let above = producer.announce("room", Route::default()).unwrap();
8744 assert!(watch.poll_changed(&waiter, seen).is_ready());
8745 let seen = watch.seen();
8746 drop(above);
8747 assert!(watch.poll_changed(&waiter, seen).is_ready());
8748 let seen = watch.seen();
8749
8750 let _beside = producer.create_broadcast("room/b").unwrap();
8752 assert!(watch.poll_changed(&waiter, seen).is_pending());
8753 let _here = producer.create_broadcast("room/a").unwrap();
8754 assert!(watch.poll_changed(&waiter, seen).is_ready());
8755
8756 drop(watch);
8758 let table = producer.shared.lock();
8759 let node = table
8760 .routes
8761 .root
8762 .find(Path::new("room/a").parts())
8763 .expect("route below keeps the node");
8764 assert!(node.watches.is_empty());
8765 assert_eq!(table.routes.root.watches_below, 0);
8766 }
8767
8768 #[test]
8769 fn a_discarded_front_task_unregisters_its_watch() {
8770 let (producer, _driver) = Producer::new(Config {
8771 hop: origin(1),
8772 ..Default::default()
8773 });
8774 let consumer = producer.consume();
8775 let _served = producer.dynamic("room", Route::default()).unwrap();
8776 drop(producer);
8781 let _pending = consumer.request_broadcast("room/a");
8782 }
8783
8784 #[test]
8785 fn create_broadcast_refuses_a_path_no_pattern_can_spell() {
8786 let producer = origin(1).produce();
8787
8788 assert!(matches!(
8792 producer.create_broadcast("room/*"),
8793 Err(Error::InvalidPath(_))
8794 ));
8795 assert!(matches!(
8796 producer.announce("room/**", Route::default()),
8797 Err(Error::InvalidPath(_))
8798 ));
8799 }
8800
8801 #[test]
8802 fn scope_empty_union_grants_nothing() {
8803 let producer = origin(1).produce();
8804
8805 assert!(matches!(producer.scope("", &Patterns::new()), Err(Error::Unauthorized)));
8807 assert!(matches!(
8808 producer.consume().scope("", &Patterns::new()),
8809 Err(Error::Unauthorized)
8810 ));
8811 }
8812
8813 #[test]
8814 fn scope_nests_and_rebases_roots() {
8815 let producer = origin(1).produce();
8816
8817 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8819 let nested = scoped.scope("", &scopes(&["room/chat"])).unwrap();
8820 assert_eq!(nested.allowed(), scopes(&["room/chat"]));
8821
8822 assert!(matches!(
8824 scoped.scope("", &scopes(&["other"])),
8825 Err(Error::Unauthorized)
8826 ));
8827
8828 let rooted = nested.scope("room/chat", &Patterns::from(Pattern::all())).unwrap();
8830 assert_eq!(rooted.allowed(), scopes(&[""]));
8831
8832 let broadcast = nested.create_broadcast("room/chat/live").unwrap();
8834 assert!(producer.consume().get_broadcast("room/chat/live").is_some());
8835 broadcast.close();
8836 }
8837
8838 #[test]
8839 fn scope_intersects_and_rebases_arbitrary_grants() {
8840 let producer = origin(1).produce();
8841 let rooms = producer
8842 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
8843 .unwrap();
8844 let chats = rooms
8845 .scope("", &Patterns::from("*/chat".parse::<Pattern>().unwrap()))
8846 .unwrap();
8847 assert_eq!(chats.allowed(), Patterns::from("room/chat".parse::<Pattern>().unwrap()));
8848
8849 let exact = producer
8850 .scope("", &Patterns::from("room/alice".parse::<Pattern>().unwrap()))
8851 .unwrap();
8852 let rooted = exact.scope("room", &Patterns::from(Pattern::all())).unwrap();
8853 assert_eq!(rooted.allowed(), Patterns::from("alice".parse::<Pattern>().unwrap()));
8854 assert!(matches!(
8855 exact.scope("room/bob", &Patterns::from(Pattern::all())),
8856 Err(Error::Unauthorized)
8857 ));
8858
8859 let broadcast = exact.create_broadcast("room/alice").unwrap();
8860 assert!(matches!(
8861 exact.create_broadcast("room/alice/cam"),
8862 Err(Error::Unauthorized)
8863 ));
8864 assert!(producer.consume().get_broadcast("room/alice").is_some());
8865 drop(broadcast);
8866 }
8867
8868 #[tokio::test]
8869 async fn wildcard_scope_filters_announcements_and_reports_captures() {
8870 let producer = origin(1).produce();
8871 let consumer = producer
8872 .consume()
8873 .scope("", &Patterns::from("room/*/chat".parse::<Pattern>().unwrap()))
8874 .unwrap();
8875 let mut announced = consumer.announced();
8876
8877 let alice = producer.create_broadcast("room/alice/chat").unwrap();
8878 alice.announce(Route::default()).unwrap();
8879 let update = announced.try_next().expect("alice's chat");
8880 assert_eq!(update.prefix.as_str(), "room/alice/chat");
8881 assert_eq!(update.captures, Some(vec!["alice".parse::<Pattern>().unwrap()]));
8882
8883 let audio = producer.create_broadcast("room/alice/audio").unwrap();
8884 audio.announce(Route::default()).unwrap();
8885 announced.assert_next_wait();
8886
8887 let broad = producer.announce("room", Route::default()).unwrap();
8888 let update = announced.try_next().expect("overlapping broad route");
8889 assert_eq!(update.prefix.as_str(), "room");
8890 assert_eq!(update.captures, None, "an overlap does not pin the wildcard");
8891
8892 drop(broad);
8893 drop(audio);
8894 drop(alice);
8895 }
8896
8897 #[tokio::test]
8898 async fn local_broadcast_wins_announcement_ties() {
8899 let producer = origin(1).produce();
8900 let remote = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
8901 let local = producer.create_broadcast("room/alice").unwrap();
8902 local.announce(Route::default()).unwrap();
8903
8904 let mut announced = producer.consume().announced();
8905 let update = announced.try_next().expect("one winning route");
8906 assert_eq!(update.prefix.as_str(), "room/alice");
8907 assert_eq!(update.route.cost, Cost::default());
8908 announced.assert_next_wait();
8909
8910 drop(local);
8911 drop(remote);
8912 }
8913
8914 #[test]
8919 fn cost_charge_saturates() {
8920 assert_eq!(Cost { warm: 4, cold: 6 }.charged(5), Cost { warm: 9, cold: 11 });
8921 assert_eq!(Cost::new(u64::MAX).charged(10), Cost::new(MAX_COST));
8922
8923 assert_eq!(Cost::UNKNOWN.charged(3).cold, MAX_COST);
8926 }
8927
8928 fn expiring_origin(expiry: Duration) -> Producer {
8930 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
8931 Config {
8932 pool,
8933 ..Config::default()
8934 }
8935 .produce()
8936 }
8937
8938 #[tokio::test(start_paused = true)]
8942 async fn stalled_publisher_open_group_is_reclaimed() {
8943 let expiry = Duration::from_secs(1);
8944 let origin = expiring_origin(expiry);
8945 let broadcast = origin.create_broadcast("test").unwrap();
8946 let track = broadcast.create_track("video", None).unwrap();
8947
8948 let mut stalled = track.append_group().unwrap();
8949 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
8950 let _successor = track.append_group().unwrap();
8954
8955 let mut reading = stalled.consume();
8956 assert!(reading.read_frame().await.unwrap().is_some());
8957
8958 crate::model::clock::advance(expiry * 2);
8960
8961 let reclaimed = tokio::time::timeout(Duration::from_secs(60), reading.read_frame()).await;
8964 assert!(
8965 matches!(reclaimed, Ok(Err(Error::Old))),
8966 "the sweep must reclaim an idle open group and surface the gap, got {reclaimed:?}"
8967 );
8968 }
8969
8970 #[tokio::test(start_paused = true)]
8973 async fn sweep_respects_a_disabled_expiry() {
8974 let origin = Config {
8975 pool: cache::Pool::unbounded(),
8976 ..Config::default()
8977 }
8978 .produce();
8979 let broadcast = origin.create_broadcast("test").unwrap();
8980 let track = broadcast.create_track("video", None).unwrap();
8981
8982 let mut stalled = track.append_group().unwrap();
8983 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
8984 let _successor = track.append_group().unwrap();
8985
8986 let mut reading = stalled.consume();
8987 assert!(reading.read_frame().await.unwrap().is_some());
8988
8989 crate::model::clock::advance(Duration::from_secs(3600));
8990 tokio::time::advance(Duration::from_secs(3600)).await;
8991
8992 assert!(
8993 reading.read_frame().now_or_never().is_none(),
8994 "a pool without an expiry window never reclaims"
8995 );
8996 }
8997
8998 #[test]
9001 fn drain_cost_is_encodable() {
9002 use crate::coding::Encode;
9003
9004 let mut buf = Vec::new();
9005 Cost::DRAIN
9006 .encode(&mut buf, crate::lite::Version::Lite06)
9007 .expect("a draining route is still forwarded, so its cost must encode");
9008 }
9009}