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>) {
657 (
658 entry.is_anonymous(),
659 entry.cost,
660 !entry.local,
661 entry.hops.len(),
662 match entry.peer {
663 true => fnv_key(prefix.as_str(), entry.hops.iter().copied()),
664 false => 0,
665 },
666 Reverse(entry.id),
667 )
668}
669
670type RouteMeta = (Hops, Cost, Source);
672
673fn earliest(a: Option<Instant>, b: Option<Instant>) -> Option<Instant> {
675 match (a, b) {
676 (Some(a), Some(b)) => Some(a.min(b)),
677 (a, b) => a.or(b),
678 }
679}
680
681#[derive(Clone, Copy)]
685struct Quarantine {
686 held: Held,
687 publisher: Option<Hop>,
688 rank: (Cost, usize),
689}
690
691impl Quarantine {
692 fn holds(&self, meta: &RouteMeta) -> bool {
696 self.publisher.is_some_and(|hop| meta.0.iter().next() == Some(&hop)) && (meta.1, meta.0.len()) > self.rank
697 }
698}
699
700#[derive(Clone, Copy, PartialEq)]
707enum Held {
708 Anchoring(Instant),
709 Until(Instant),
710}
711
712type AnnounceMeta = (RouteMeta, Option<Vec<Pattern>>);
719
720enum PendingUpdate {
721 Announce(AnnounceMeta),
722 Unannounce(AnnounceMeta),
723 UnannounceAnnounce { old: AnnounceMeta, new: AnnounceMeta },
724}
725
726#[derive(Default)]
731struct OriginConsumerState {
732 pending: BTreeMap<PathOwned, PendingUpdate>,
733 delivered: BTreeSet<PathOwned>,
738 ended: bool,
741}
742
743impl OriginConsumerState {
744 fn apply_announce(&mut self, prefix: PathOwned, meta: RouteMeta, captures: Option<Vec<Pattern>>) {
745 let meta = (meta, captures);
746 let new = match self.pending.remove(&prefix) {
747 None | Some(PendingUpdate::Announce(_)) => PendingUpdate::Announce(meta),
749 Some(PendingUpdate::Unannounce(old) | PendingUpdate::UnannounceAnnounce { old, .. }) => {
751 PendingUpdate::UnannounceAnnounce { old, new: meta }
752 }
753 };
754 self.pending.insert(prefix, new);
755 }
756
757 fn apply_unannounce(&mut self, prefix: PathOwned, last: RouteMeta, captures: Option<Vec<Pattern>>) {
758 let last = (last, captures);
759 match self.pending.remove(&prefix) {
760 Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => {}
763 None | Some(PendingUpdate::Announce(_) | PendingUpdate::Unannounce(_)) => {
766 self.pending.insert(prefix, PendingUpdate::Unannounce(last));
767 }
768 Some(PendingUpdate::UnannounceAnnounce { old, .. }) => {
771 self.pending.insert(prefix, PendingUpdate::Unannounce(old));
772 }
773 }
774 }
775
776 fn is_update(&self, prefix: &PathOwned, update: &PendingUpdate) -> bool {
779 matches!(update, PendingUpdate::Announce(_)) && self.delivered.contains(prefix)
780 }
781
782 fn scan(
786 &self,
787 held: &mut HashMap<PathOwned, Held>,
788 retracted: &HashMap<PathOwned, Quarantine>,
789 now: Option<Instant>,
790 hold: Duration,
791 ) -> Result<PathOwned, Option<Instant>> {
792 let mut wake = None;
793 for (prefix, update) in &self.pending {
794 let Some(now) = now.filter(|_| !hold.is_zero() && !self.ended) else {
795 return Ok(prefix.clone());
796 };
797 if !self.is_update(prefix, update) {
798 let quarantine = match update {
799 PendingUpdate::Announce((meta, _)) => retracted.get(prefix).filter(|q| q.holds(meta)),
800 _ => None,
801 };
802 let until = match quarantine.map(|q| q.held) {
803 Some(Held::Anchoring(seen)) => seen + Duration::from_nanos(1),
804 Some(Held::Until(until)) if now < until => until,
805 _ => return Ok(prefix.clone()),
806 };
807 wake = Some(wake.map_or(until, |w: Instant| w.min(until)));
808 continue;
809 }
810 let until = match held.get(prefix).copied() {
813 None => {
814 held.insert(prefix.clone(), Held::Anchoring(now));
815 now + Duration::from_nanos(1)
816 }
817 Some(Held::Anchoring(seen)) if now <= seen => seen + Duration::from_nanos(1),
818 Some(Held::Anchoring(_)) => {
819 held.insert(prefix.clone(), Held::Until(now + hold));
820 now + hold
821 }
822 Some(Held::Until(until)) if until <= now => return Ok(prefix.clone()),
823 Some(Held::Until(until)) => until,
824 };
825 wake = Some(wake.map_or(until, |w: Instant| w.min(until)));
826 }
827 Err(wake)
828 }
829
830 fn take_prefix(&mut self, prefix: PathOwned) -> Option<AnnounceUpdate> {
832 let ((meta, captures), kind) = match self.pending.remove(&prefix).unwrap() {
833 PendingUpdate::Announce(meta) => {
834 let kind = match self.delivered.insert(prefix.clone()) {
836 true => AnnounceKind::Announced,
837 false => AnnounceKind::Updated,
838 };
839 (meta, kind)
840 }
841 PendingUpdate::Unannounce(meta) => {
842 self.delivered.remove(&prefix);
843 (meta, AnnounceKind::Retracted)
844 }
845 PendingUpdate::UnannounceAnnounce { old, new } => {
846 self.delivered.remove(&prefix);
849 self.pending.insert(prefix.clone(), PendingUpdate::Announce(new));
850 (old, AnnounceKind::Retracted)
851 }
852 };
853 Some(AnnounceUpdate {
854 prefix,
855 captures,
856 route: Route {
857 hops: meta.0,
858 cost: meta.1,
859 via: Hop::UNKNOWN,
860 source: meta.2,
861 },
862 kind,
863 })
864 }
865}
866
867struct RouteEntry {
869 id: u64,
870 prefix: PathOwned,
871 scope: Patterns,
874 hops: Hops,
875 cost: Cost,
876 via: Hop,
880 local: bool,
883 peer: bool,
886 server: Option<kio::Shared<ServeState>>,
890 source: Option<broadcast::Consumer>,
894 advertised: bool,
898 stale: bool,
902 claim: Pattern,
908}
909
910impl RouteEntry {
911 fn live(&self) -> bool {
913 self.advertised && !self.stale
914 }
915
916 fn is_anonymous(&self) -> bool {
917 self.hops.iter().any(|hop| *hop == Hop::UNKNOWN)
918 }
919
920 fn entered(&self) -> Source {
922 match self.peer {
923 true => Source::Peer(self.via),
924 false => Source::Local,
925 }
926 }
927
928 fn serves(&self, path: &Path) -> bool {
932 self.server.is_some() || (self.source.is_some() && self.prefix == *path)
933 }
934
935 fn qualifies(&self, pin: Pin) -> bool {
937 match pin {
938 Pin::Any => true,
939 Pin::Local => self.local,
940 Pin::Publisher(first) => self.hops.iter().next() == Some(&first),
941 Pin::Route(id) => self.id == id && self.hops.iter().next().is_none_or(|first| *first == Hop::UNKNOWN),
943 }
944 }
945
946 fn visible_to(&self, exclude: Option<Hop>) -> bool {
951 match exclude {
952 Some(peer) if peer != Hop::UNKNOWN => self.via != peer && !self.hops.contains(&peer),
953 _ => true,
954 }
955 }
956
957 fn overlaps(&self, allowed: &Patterns) -> bool {
959 self.scope.iter().any(|scope| {
960 scope
961 .intersect(&self.claim)
962 .is_ok_and(|scoped| scoped.iter().any(|restriction| allowed.overlaps(restriction)))
963 })
964 }
965}
966
967fn prefix_claim(prefix: &Path) -> Result<Pattern, InvalidPattern> {
969 if prefix.parts().count() == Path::MAX_PARTS {
970 Pattern::literal(prefix.as_str())
971 } else {
972 Pattern::subtree(prefix.as_str())
973 }
974}
975
976#[derive(Default)]
981struct ServeState {
982 requests: Requests<PathOwned, kio::Producer<PendingBroadcast>>,
985
986 served: WeakCache<PathOwned, broadcast::WeakConsumer>,
992
993 closed: bool,
996}
997
998type FrontKey = (PathOwned, Horizon);
1005
1006#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
1008struct Horizon {
1009 exclude: Option<Hop>,
1013 local: bool,
1015}
1016
1017impl Horizon {
1018 fn admits(&self, entry: &RouteEntry) -> bool {
1020 !(self.local && entry.peer) && entry.visible_to(self.exclude)
1021 }
1022
1023 fn effective(self, routes: &RouteTable, path: &Path) -> Self {
1033 match self.exclude {
1034 Some(peer) if routes.covering(path).any(|entry| !entry.visible_to(Some(peer))) => self,
1035 _ => Self { exclude: None, ..self },
1036 }
1037 }
1038}
1039
1040#[derive(Clone)]
1043struct RemoteFront {
1044 request: kio::Producer<PendingBroadcast>,
1048 broadcast: broadcast::WeakConsumer,
1051 pin: kio::Lock<Pin>,
1054}
1055
1056type CursorRoute = (u64, RouteMeta, bool, Option<Vec<Pattern>>);
1058
1059impl WeakEntry for RemoteFront {
1060 fn is_closed(&self) -> bool {
1061 self.broadcast.is_closed()
1062 }
1063
1064 fn same_channel(&self, other: &Self) -> bool {
1065 self.broadcast.same_channel(&other.broadcast)
1066 }
1067}
1068
1069struct TableCursor {
1072 root: PathOwned,
1074 allowed: Patterns,
1076 heads: Vec<PathOwned>,
1080 horizon: Horizon,
1082 hidden: Hidden,
1084 state: kio::Producer<OriginConsumerState>,
1087 under: PathOwned,
1090 holes: Vec<PathOwned>,
1093 named: Option<(Mount, Patterns)>,
1096 current: HashMap<PathOwned, CursorRoute>,
1101}
1102
1103impl TableCursor {
1104 fn presented(&self, prefix: &Path, claim: &Pattern) -> Option<PathOwned> {
1110 if !self.allowed.overlaps(claim) {
1111 return None;
1112 }
1113
1114 if let Some(relative) = prefix.strip_prefix(&self.root) {
1115 return Some(relative.to_owned());
1116 }
1117 self.root.has_prefix(prefix).then(PathOwned::default)
1118 }
1119
1120 fn captures(&self, prefix: &Path) -> Option<Vec<Pattern>> {
1123 let (literal, allowed) = match &self.named {
1124 Some((mount, allowed)) => (Pattern::literal(mount.name(prefix)?.as_str()).ok()?, allowed),
1125 None => (Pattern::literal(prefix.as_str()).ok()?, &self.allowed),
1126 };
1127 allowed
1128 .iter()
1129 .filter_map(|allowed| {
1130 allowed
1131 .captures(&literal)
1132 .map(|captures| (allowed.specificity(), captures))
1133 })
1134 .max_by_key(|(specificity, _)| *specificity)
1135 .map(|(_, captures)| captures)
1136 }
1137
1138 fn visible(&self, entry: &RouteEntry) -> bool {
1142 entry.live()
1143 && self.horizon.admits(entry)
1144 && entry.overlaps(&self.allowed)
1145 && self.discovers(&entry.prefix)
1146 && !self.holes.iter().any(|hole| entry.prefix.has_prefix(hole))
1147 && self.named.as_ref().is_none_or(|(mount, _)| mount.names(&entry.prefix))
1148 }
1149
1150 fn update(&mut self, presented: &PathOwned, best: Option<&RouteEntry>) {
1152 match best {
1153 Some(entry) => {
1154 let meta = (entry.hops.clone(), entry.cost, entry.entered());
1155 let served = entry.server.is_some();
1156 let captures = self.captures(&entry.prefix);
1157 let previous = self
1158 .current
1159 .insert(presented.clone(), (entry.id, meta.clone(), served, captures.clone()));
1160 match previous {
1161 Some((_, prev, prev_served, prev_captures))
1168 if prev == meta && prev_served == served && prev_captures == captures => {}
1169 Some((_, prev, _, prev_captures)) if prev_captures != captures => {
1172 if let Ok(mut state) = self.state.write() {
1173 state.apply_unannounce(self.under.join(presented), prev, prev_captures);
1174 state.apply_announce(self.under.join(presented), meta, captures);
1175 }
1176 }
1177 _ => {
1178 if let Ok(mut state) = self.state.write() {
1179 state.apply_announce(self.under.join(presented), meta, captures);
1180 }
1181 }
1182 }
1183 }
1184 None => {
1185 if let Some((_, last, _, captures)) = self.current.remove(presented)
1186 && let Ok(mut state) = self.state.write()
1187 {
1188 state.apply_unannounce(self.under.join(presented), last, captures);
1189 }
1190 }
1191 }
1192 }
1193
1194 fn discovers(&self, prefix: &Path) -> bool {
1196 self.hidden.discovers(&self.heads, prefix)
1197 }
1198}
1199
1200#[derive(Clone)]
1203struct OriginScope {
1204 allowed: Patterns,
1207 mounts: Arc<[Mount]>,
1209}
1210
1211#[derive(Clone, Debug)]
1214struct Mount {
1215 at: PathOwned,
1216 target: PathOwned,
1217}
1218
1219impl Mount {
1220 fn resolve(&self, path: &Path) -> Option<PathOwned> {
1223 let resolved = self.target.join(path.strip_prefix(&self.at)?);
1224 (resolved.parts().count() <= Path::MAX_PARTS).then_some(resolved)
1225 }
1226
1227 fn name(&self, path: &Path) -> Option<PathOwned> {
1230 Some(self.at.join(path.strip_prefix(&self.target)?))
1231 }
1232
1233 fn names(&self, path: &Path) -> bool {
1237 path.strip_prefix(&self.target)
1238 .is_none_or(|rest| self.at.parts().count() + rest.parts().count() <= Path::MAX_PARTS)
1239 }
1240
1241 fn translate(&self, patterns: &Patterns) -> Patterns {
1248 let target = self.target.as_str();
1249 patterns
1250 .rebase(self.at.as_str())
1251 .iter()
1252 .filter_map(|member| {
1253 member
1254 .rooted(target)
1255 .or_else(|_| {
1256 let segments = member
1257 .segments()
1258 .iter()
1259 .filter(|segment| **segment != Segment::Globstar);
1260 Pattern::new(segments.cloned())?.rooted(target)
1261 })
1262 .ok()
1263 })
1264 .collect()
1265 }
1266
1267 fn translate_head(&self, head: &Path) -> Option<PathOwned> {
1270 match head.strip_prefix(&self.at) {
1271 Some(rest) => Some(self.target.join(rest)),
1272 None => self.at.has_prefix(head).then(|| self.target.clone()),
1273 }
1274 }
1275}
1276
1277impl OriginScope {
1278 fn empty() -> Self {
1280 Self {
1281 allowed: Patterns::new(),
1282 mounts: Arc::from([]),
1283 }
1284 }
1285
1286 fn narrow(&self, patterns: &Patterns) -> Option<Self> {
1288 let allowed = self.allowed.intersect(patterns).ok()?;
1289 if allowed.is_empty() {
1290 None
1291 } else {
1292 Some(Self {
1293 allowed,
1294 mounts: self.mounts.clone(),
1295 })
1296 }
1297 }
1298
1299 fn mount(&self, path: &Path) -> Option<&Mount> {
1301 self.mounts.iter().find(|mount| path.has_prefix(&mount.at))
1302 }
1303
1304 fn resolve<'a>(&self, path: &'a Path<'a>) -> Option<Path<'a>> {
1307 match self.mount(path) {
1308 Some(mount) => mount.resolve(path),
1309 None => Some(path.borrow()),
1310 }
1311 }
1312
1313 fn publishes(&self, prefix: &Path) -> bool {
1316 self.mount(prefix).is_none()
1317 }
1318
1319 fn permits(&self, path: &Path) -> bool {
1321 self.allowed.matches(path.as_str())
1322 }
1323
1324 fn relative(&self, root: &Path) -> Patterns {
1326 self.allowed.rebase(root.as_str())
1327 }
1328}
1329
1330impl Default for OriginScope {
1331 fn default() -> Self {
1332 Self {
1333 allowed: Patterns::from(Pattern::all()),
1334 mounts: Arc::from([]),
1335 }
1336 }
1337}
1338
1339pub(crate) fn interest_prefixes(allowed: &Patterns) -> Vec<PathOwned> {
1342 let mut heads: Vec<PathOwned> = allowed
1343 .iter()
1344 .map(|pattern| Path::new(pattern.head()).to_owned())
1345 .collect();
1346 heads.sort();
1347 heads.dedup();
1348 let covered = heads.clone();
1349 heads.retain(|head| !covered.iter().any(|other| other != head && head.has_prefix(other)));
1350 heads
1351}
1352
1353#[derive(Clone, Debug, Default, PartialEq, Eq)]
1360pub(crate) struct Hidden {
1361 include: bool,
1363 from: Option<PathOwned>,
1365 beyond: Option<Vec<PathOwned>>,
1368}
1369
1370impl Hidden {
1371 fn discovers(&self, heads: &[PathOwned], prefix: &Path) -> bool {
1373 (self.include || !hides(self.from.as_ref().map(std::slice::from_ref).unwrap_or(heads), prefix))
1374 && self.beyond.as_ref().is_none_or(|outer| hides(outer, prefix))
1375 }
1376
1377 fn translate(&self, mount: &Mount) -> Self {
1381 let beyond = self.beyond.as_ref().and_then(|outer| {
1382 (!hides(outer, &mount.at)).then(|| outer.iter().filter_map(|head| mount.translate_head(head)).collect())
1383 });
1384 Self {
1385 include: self.include,
1386 from: self.from.as_ref().and_then(|head| mount.translate_head(head)),
1387 beyond,
1388 }
1389 }
1390}
1391
1392fn hides(heads: &[PathOwned], prefix: &Path) -> bool {
1395 heads
1396 .iter()
1397 .any(|head| prefix.strip_prefix(head).is_some_and(|below| below.is_hidden()))
1398}
1399
1400#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1402pub enum AnnounceKind {
1403 Announced,
1405 Updated,
1407 Retracted,
1409}
1410
1411impl AnnounceKind {
1412 pub fn is_active(self) -> bool {
1414 !matches!(self, Self::Retracted)
1415 }
1416}
1417
1418#[derive(Clone, Debug)]
1426pub struct AnnounceUpdate {
1427 pub prefix: PathOwned,
1429 pub captures: Option<Vec<Pattern>>,
1432 pub route: Route,
1435 pub kind: AnnounceKind,
1437}
1438
1439#[derive(Clone)]
1441pub struct Producer {
1442 hop: Hop,
1445
1446 scope: OriginScope,
1448
1449 root: PathOwned,
1451
1452 shared: kio::Shared<OriginState>,
1455
1456 pool: cache::Pool,
1459
1460 cache_duration: Duration,
1463
1464 default_max_age: Duration,
1467
1468 stats: stats::Session,
1472
1473 peer: bool,
1476
1477 tasks: Tasks,
1481
1482 timers: Clock,
1484}
1485
1486impl Producer {
1487 pub fn new(config: Config) -> (Self, Driver) {
1494 let (tasks, set) = TaskSet::new();
1495 let scope = OriginScope::default();
1496 let shared = kio::Shared::new(OriginState::new(config.update_hold));
1497 let timers = Clock::default();
1498 let pool = config.pool.clone();
1499 let producer = Self {
1500 hop: config.hop,
1501 scope: scope.clone(),
1502 root: PathOwned::default(),
1503 shared: shared.clone(),
1504 pool: config.pool,
1505 cache_duration: config.cache_duration,
1506 default_max_age: config.default_max_age,
1507 stats: stats::Session::default(),
1508 peer: false,
1509 tasks,
1510 timers: timers.clone(),
1511 };
1512 let driver = Driver {
1513 state: DriverState {
1514 set,
1515 shared,
1516 done: false,
1517 },
1518 timers,
1519 pool,
1520 };
1521 (producer, driver)
1522 }
1523
1524 pub fn with_stats(mut self, session: stats::Session) -> Self {
1528 self.stats = session;
1529 self
1530 }
1531
1532 pub fn peer(mut self) -> Self {
1540 self.peer = true;
1541 self
1542 }
1543
1544 pub fn config(&self) -> Config {
1546 Config {
1547 hop: self.hop,
1548 pool: self.pool.clone(),
1549 cache_duration: self.cache_duration,
1550 default_max_age: self.default_max_age,
1551 update_hold: self.shared.lock().update_hold,
1552 }
1553 }
1554
1555 pub fn hop(&self) -> Hop {
1557 self.hop
1558 }
1559
1560 pub(crate) fn default_max_age(&self) -> Duration {
1563 self.default_max_age
1564 }
1565
1566 pub(crate) fn empty(hop: Hop) -> Self {
1571 let (tasks, _) = TaskSet::new();
1574 Self {
1575 hop,
1576 scope: OriginScope::empty(),
1577 root: PathOwned::default(),
1578 shared: kio::Shared::new(OriginState::new(DEFAULT_UPDATE_HOLD)),
1579 pool: cache::Pool::default(),
1580 cache_duration: Duration::MAX,
1581 default_max_age: track::DEFAULT_MAX_AGE,
1582 stats: stats::Session::default(),
1583 peer: false,
1584 tasks,
1585 timers: Clock::default(),
1586 }
1587 }
1588
1589 pub fn create_broadcast(&self, path: impl AsPath) -> Result<broadcast::Producer, Error> {
1620 let path = path.as_path();
1621
1622 let full = self.root.join(&path).to_owned();
1623 if !self.scope.permits(&full) || !self.scope.publishes(&full) {
1624 return Err(Error::Unauthorized);
1625 }
1626 if full.parts().count() > Path::MAX_PARTS {
1630 return Err(BoundsExceeded.into());
1631 }
1632 let claim = prefix_claim(&full)?;
1635
1636 let ingress = self.stats.ingress(&full);
1638
1639 let announcing = Announcing {
1644 hop: self.hop,
1645 shared: self.shared.clone(),
1646 requested: full.clone(),
1647 prefixes: vec![(full.clone(), claim)],
1648 scope: self.scope.allowed.clone(),
1649 local: true,
1650 peer: self.peer,
1651 stats: self.stats.clone(),
1652 };
1653 let info = broadcast::Info {
1654 pool: self.pool.clone(),
1655 cache_duration: self.cache_duration,
1656 path: full,
1657 };
1658 let source = info.produce().with_stats(ingress.clone());
1659 let entry = announcing.announce(
1660 Route::default(),
1661 Serving {
1662 server: None,
1663 source: Some(source.consume()),
1664 advertised: false,
1665 },
1666 )?;
1667 Ok(source.with_announcer(Announcer {
1668 entry,
1669 ingress,
1670 _keepalive: self.tasks.keepalive(),
1671 }))
1672 }
1673
1674 pub fn publish(&self, path: impl AsPath, route: Route) -> Result<broadcast::Producer, Error> {
1676 let broadcast = self.create_broadcast(path)?;
1677 broadcast.announce(route)?;
1678 Ok(broadcast)
1679 }
1680
1681 pub(crate) fn create_source(&self, path: impl AsPath) -> broadcast::Producer {
1687 let path = path.as_path();
1688 let full = self.root.join(&path).to_owned();
1689 let ingress = self.stats.ingress(&full);
1690 broadcast::Info {
1691 pool: self.pool.clone(),
1692 cache_duration: self.cache_duration,
1693 path: full,
1694 }
1695 .produce()
1696 .with_stats(ingress)
1697 }
1698
1699 #[cfg(test)]
1708 pub(crate) fn announce(&self, prefix: impl AsPath, route: Route) -> Result<AnnounceProducer, Error> {
1709 Announcing::new(self, prefix)?.announce(
1710 route,
1711 Serving {
1712 server: None,
1713 source: None,
1714 advertised: true,
1715 },
1716 )
1717 }
1718
1719 pub fn dynamic(&self, prefix: impl AsPath, route: Route) -> Result<Dynamic, Error> {
1741 let announcing = Announcing::new(self, prefix)?;
1742 let serve = kio::Shared::<ServeState>::default();
1743 serve.lock().requests.add_handler();
1744 let announcement = announcing.announce(
1745 route,
1746 Serving {
1747 server: Some(serve.clone()),
1748 source: None,
1749 advertised: true,
1750 },
1751 )?;
1752 Ok(Dynamic {
1753 announcement,
1754 state: serve,
1755 _keepalive: self.tasks.keepalive(),
1756 })
1757 }
1758
1759 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Producer, Error> {
1766 let root = self.root.join(root).to_owned();
1767 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
1768 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
1769 Ok(Producer {
1770 hop: self.hop,
1771 scope,
1772 root,
1773 shared: self.shared.clone(),
1774 pool: self.pool.clone(),
1775 cache_duration: self.cache_duration,
1776 default_max_age: self.default_max_age,
1777 stats: self.stats.clone(),
1778 peer: self.peer,
1779 tasks: self.tasks.clone(),
1780 timers: self.timers.clone(),
1781 })
1782 }
1783
1784 pub fn mount(&self, at: impl AsPath, target: impl AsPath) -> Result<Producer, Error> {
1803 let at = self.root.join(at).to_owned();
1804 let target = self.root.join(target).to_owned();
1805 if [&at, &target]
1806 .into_iter()
1807 .any(|path| path.parts().count() > Path::MAX_PARTS)
1808 {
1809 return Err(BoundsExceeded.into());
1810 }
1811 Pattern::literal(at.as_str())?;
1813 if !self
1814 .scope
1815 .allowed
1816 .covers(&Patterns::from(Pattern::subtree(target.as_str())?))
1817 {
1818 return Err(Error::Unauthorized);
1819 }
1820 let overlaps = |a: &Path, b: &Path| a.has_prefix(b) || b.has_prefix(a);
1824 if overlaps(&at, &target)
1825 || self
1826 .scope
1827 .mounts
1828 .iter()
1829 .any(|mount| overlaps(&at, &mount.at) || overlaps(&target, &mount.at) || overlaps(&at, &mount.target))
1830 {
1831 return Err(Error::Duplicate);
1832 }
1833 let mounts = self
1834 .scope
1835 .mounts
1836 .iter()
1837 .cloned()
1838 .chain([Mount { at, target }])
1839 .collect();
1840 Ok(Producer {
1841 scope: OriginScope {
1842 allowed: self.scope.allowed.clone(),
1843 mounts,
1844 },
1845 ..self.clone()
1846 })
1847 }
1848
1849 pub fn consume(&self) -> Consumer {
1854 Consumer::from_producer(self, stats::Session::default())
1857 }
1858
1859 pub fn root(&self) -> &Path<'_> {
1861 &self.root
1862 }
1863
1864 pub fn allowed(&self) -> Patterns {
1866 self.scope.relative(&self.root)
1867 }
1868
1869 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
1871 self.root.join(path)
1872 }
1873}
1874
1875struct Announcing {
1879 hop: Hop,
1880 shared: kio::Shared<OriginState>,
1881 requested: PathOwned,
1883 prefixes: Vec<(PathOwned, Pattern)>,
1887 scope: Patterns,
1889 local: bool,
1890 peer: bool,
1892 stats: stats::Session,
1893}
1894
1895impl Announcing {
1896 fn new(producer: &Producer, prefix: impl AsPath) -> Result<Self, Error> {
1898 let requested = producer.root.join(prefix.as_path()).to_owned();
1899 if requested.parts().count() > Path::MAX_PARTS {
1900 return Err(BoundsExceeded.into());
1901 }
1902 let claim = prefix_claim(&requested)?;
1903 if !producer.scope.allowed.overlaps(&claim) || !producer.scope.publishes(&requested) {
1904 return Err(Error::Unauthorized);
1905 }
1906 Ok(Self {
1907 hop: producer.hop,
1908 shared: producer.shared.clone(),
1909 requested: requested.clone(),
1910 prefixes: vec![(requested, claim)],
1911 scope: producer.scope.allowed.clone(),
1912 local: false,
1913 peer: producer.peer,
1914 stats: producer.stats.clone(),
1915 })
1916 }
1917
1918 fn announce(&self, route: Route, serving: Serving) -> Result<AnnounceProducer, Error> {
1919 debug_assert!(
1920 !route.hops.contains(&self.hop),
1921 "announce called with a looping hop chain",
1922 );
1923
1924 let via = route.via;
1925
1926 let mut shared = self.shared.lock();
1927 if shared.closed {
1928 return Err(Error::Closed);
1929 }
1930
1931 let mut entries = Vec::with_capacity(self.prefixes.len());
1932 for (prefix, claim) in &self.prefixes {
1933 shared.reannounced(prefix, &route.hops);
1934 let id = shared.next_route;
1935 shared.next_route += 1;
1936 let stale = shared.withdrawn_through(prefix, &route.hops);
1937 shared.routes.insert(RouteEntry {
1938 id,
1939 prefix: prefix.clone(),
1940 scope: self.scope.clone(),
1941 hops: route.hops.clone(),
1942 cost: route.cost,
1943 via,
1944 local: self.local,
1945 peer: self.peer,
1946 server: serving.server.clone(),
1947 source: serving.source.clone(),
1948 advertised: serving.advertised,
1949 stale,
1950 claim: claim.clone(),
1951 });
1952 shared.sync_route(prefix, claim);
1953 entries.push((prefix.clone(), id));
1954 }
1955 drop(shared);
1956
1957 let guard = serving
1959 .advertised
1960 .then(|| self.stats.ingress(&self.requested).announce());
1961
1962 Ok(AnnounceProducer {
1963 shared: self.shared.clone(),
1964 entries,
1965 guard,
1966 })
1967 }
1968}
1969
1970struct Serving {
1972 server: Option<kio::Shared<ServeState>>,
1973 source: Option<broadcast::Consumer>,
1974 advertised: bool,
1975}
1976
1977pub(crate) struct Announcer {
1984 entry: AnnounceProducer,
1985 ingress: stats::Scope,
1987 _keepalive: Keepalive,
1991}
1992
1993impl Announcer {
1994 pub(crate) fn announce(&mut self, route: Route) -> Result<(), Error> {
1996 self.entry.update(route)?;
1997 if self.entry.guard.is_none() {
1998 self.entry.guard = Some(self.ingress.announce());
1999 }
2000 Ok(())
2001 }
2002
2003 pub(crate) fn withdraw(&mut self) {
2005 self.entry.withdraw();
2006 self.entry.guard = None;
2007 }
2008}
2009
2010#[must_use = "dropping an announcement retracts the route"]
2017pub(crate) struct AnnounceProducer {
2018 shared: kio::Shared<OriginState>,
2019 entries: Vec<(PathOwned, u64)>,
2022 guard: Option<stats::Announce>,
2024}
2025
2026impl AnnounceProducer {
2027 pub fn update(&self, route: Route) -> Result<(), Error> {
2035 let mut shared = self.shared.lock();
2036 if shared.closed {
2037 return Err(Error::Closed);
2038 }
2039 for (prefix, id) in &self.entries {
2040 shared.reannounced(prefix, &route.hops);
2041 let stale = shared.withdrawn_through(prefix, &route.hops);
2042 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2044 return Err(Error::Closed);
2045 };
2046 if entry.hops.iter().next() != route.hops.iter().next()
2049 && let Some(server) = &entry.server
2050 {
2051 drop(std::mem::take(&mut server.lock().served));
2052 }
2053 entry.hops = route.hops.clone();
2054 entry.stale = stale;
2055 entry.cost = route.cost;
2056 entry.via = route.via;
2057 entry.advertised = true;
2058 let claim = entry.claim.clone();
2059 shared.sync_route(prefix, &claim);
2060 shared.prune_withdrawn(prefix);
2061 }
2062 Ok(())
2063 }
2064
2065 fn withdraw(&self) {
2070 let mut shared = self.shared.lock();
2071 for (prefix, id) in &self.entries {
2072 let Some(entry) = shared.routes.entry_mut(prefix, *id) else {
2073 continue;
2074 };
2075 if !entry.advertised {
2076 continue;
2077 }
2078 entry.advertised = false;
2079 let claim = entry.claim.clone();
2080 shared.sync_route(prefix, &claim);
2081 }
2082 }
2083
2084 fn retract(&self, withdrawn: bool) {
2087 let mut shared = self.shared.lock();
2088 for (prefix, id) in &self.entries {
2089 let Some(entry) = shared.routes.remove(prefix, *id) else {
2090 continue;
2091 };
2092 if let Some(server) = &entry.server {
2095 let mut server = server.lock();
2096 server.closed = true;
2097 for producer in server.requests.drain_all() {
2098 if let Ok(mut request) = producer.write() {
2099 request.resolved.get_or_insert(Err(Error::Unroutable));
2100 }
2101 }
2102 }
2103 if withdrawn
2108 && let Some(&peer) = entry.hops.iter().last()
2109 && peer != Hop::UNKNOWN
2110 && !shared
2111 .routes
2112 .at(prefix)
2113 .any(|other| other.live() && other.hops.iter().last() == Some(&peer))
2114 {
2115 shared.withdrawn.entry(prefix.clone()).or_default().insert(peer);
2116 shared.restale(prefix);
2117 }
2118 shared.sync_route(&entry.prefix, &entry.claim);
2119 shared.prune_withdrawn(&entry.prefix);
2120 }
2121 }
2122}
2123
2124impl Drop for AnnounceProducer {
2125 fn drop(&mut self) {
2126 self.retract(false);
2127 }
2128}
2129
2130#[must_use = "poll the driver or the origin makes no progress"]
2145pub struct Driver {
2146 state: DriverState,
2147 timers: Clock,
2149 pool: cache::Pool,
2152}
2153
2154struct DriverState {
2156 set: TaskSet,
2158 shared: kio::Shared<OriginState>,
2161 done: bool,
2163}
2164
2165impl Driver {
2166 pub fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2173 self.timers.advance(now);
2174 self.timers.register_driver(waiter);
2175 let result = self.state.poll(waiter);
2176 let gc = self.pool.gc(now);
2177 if result.is_ready() {
2178 return Err(Error::Closed);
2179 }
2180 Ok(self.timers.timeout().into_iter().chain(gc).min())
2181 }
2182}
2183
2184impl crate::time::Driver for Driver {
2185 fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result<Option<Instant>, Error> {
2186 self.poll(now, waiter)
2187 }
2188}
2189
2190impl DriverState {
2191 fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> {
2192 if !self.done {
2196 ready!(self.set.poll(waiter));
2197 self.done = true;
2198 }
2199 Poll::Ready(())
2200 }
2201
2202 fn teardown(&mut self) {
2206 drop(std::mem::replace(&mut self.set, TaskSet::owned()));
2209
2210 let (servers, cursors, fronts) = {
2215 let mut shared = self.shared.lock();
2216 shared.closed = true;
2217 shared.routes.poke_all();
2219 let servers: Vec<_> = shared
2220 .routes
2221 .entries()
2222 .filter_map(|entry| entry.server.clone())
2223 .collect();
2224 let cursors: Vec<_> = shared.cursors.values().map(|cursor| cursor.state.clone()).collect();
2225 let fronts: Vec<_> = shared.fronts.values().map(|front| front.request.clone()).collect();
2226 (servers, cursors, fronts)
2227 };
2228
2229 for producer in fronts {
2232 if let Ok(mut request) = producer.write() {
2233 request.resolved.get_or_insert(Err(Error::Dropped));
2234 }
2235 }
2236 for server in servers {
2240 let mut server = server.lock();
2241 server.closed = true;
2242 for producer in server.requests.drain_all() {
2243 if let Ok(mut request) = producer.write() {
2244 request.resolved.get_or_insert(Err(Error::Dropped));
2245 }
2246 }
2247 }
2248
2249 for state in cursors {
2252 if let Ok(mut state) = state.write() {
2253 state.ended = true;
2254 }
2255 }
2256 }
2257}
2258
2259impl Drop for DriverState {
2260 fn drop(&mut self) {
2261 self.teardown();
2262 }
2263}
2264
2265const TRACK_IDLE_LINGER: Duration = Duration::from_secs(30);
2279
2280struct WarmCopy {
2286 track: track::Producer,
2287 _dynamic: track::Dynamic,
2288 edge: Option<WarmGroup>,
2291}
2292
2293impl Drop for WarmCopy {
2294 fn drop(&mut self) {
2295 let _ = self.track.finish();
2296 }
2297}
2298
2299struct WarmGroup(group::Producer);
2306
2307impl Drop for WarmGroup {
2308 fn drop(&mut self) {
2309 if !self.0.is_finished() {
2310 let _ = self.0.clone().abort(Error::Cancel);
2311 }
2312 }
2313}
2314
2315struct HeldGroup(group::Producer);
2320
2321impl Drop for HeldGroup {
2322 fn drop(&mut self) {
2323 if !self.0.is_finished() {
2324 let _ = self.0.abort_if_last(Error::Cancel);
2325 }
2326 }
2327}
2328
2329fn warm_copy(source: &track::Consumer, head: Option<&WarmGroup>, held: Option<&group::Producer>) -> Option<WarmCopy> {
2337 let info = source.cached_info()?;
2338 let mut track = track::Producer::new(Arc::new(source.broadcast().clone()), source.name(), info);
2339 let head = head.map(|head| &head.0);
2340 let mut groups = source.cached_groups();
2341 if let Some(held) = held
2342 && !groups.iter().any(|(group, _)| group.sequence == held.sequence)
2343 {
2344 groups.push((held.clone(), true));
2345 }
2346 let latest = groups
2348 .iter()
2349 .filter(|(_, visible)| *visible)
2350 .map(|(group, _)| group.sequence)
2351 .max();
2352 let mut edge = None;
2353 let mut first = None;
2354
2355 if let Some(head) = head
2359 && !groups.iter().any(|(group, _)| group.sequence == head.sequence)
2360 {
2361 let is_latest = latest.is_none_or(|latest| head.sequence >= latest);
2362 if head.is_finished() {
2363 let _ = track.adopt_group(head.clone(), true);
2364 first = Some(head.sequence);
2365 if is_latest {
2366 edge = Some(WarmGroup(head.clone()));
2367 }
2368 } else if is_latest {
2369 edge = warm_rebuild(&track, head, None);
2370 first = edge.as_ref().map(|_| head.sequence);
2371 }
2372 }
2373
2374 for (group, visible) in groups {
2375 let finished = group.is_finished();
2376 let whole = group.live_first_frame() == Some(0);
2377 let is_latest = visible && Some(group.sequence) == latest;
2378 let warm = if finished && whole {
2379 let _ = track.adopt_group(group.clone(), visible);
2380 Some(WarmGroup(group))
2381 } else if (finished || is_latest) && (whole || head.is_some_and(|head| head.sequence == group.sequence)) {
2382 warm_rebuild(&track, &group, head)
2388 } else {
2389 None
2391 };
2392 if visible && let Some(warm) = &warm {
2393 first = Some(first.map_or(warm.0.sequence, |first: u64| first.min(warm.0.sequence)));
2394 }
2395 if is_latest {
2396 edge = warm;
2397 }
2398 }
2399 track.start_at(first).ok()?;
2403 let dynamic = track.dynamic();
2404 Some(WarmCopy {
2405 track,
2406 _dynamic: dynamic,
2407 edge,
2408 })
2409}
2410
2411fn warm_rebuild(track: &track::Producer, live: &group::Producer, head: Option<&group::Producer>) -> Option<WarmGroup> {
2414 let mut start = live.live_first_frame()? as u64;
2416 let mut tail = live.consume();
2417 tail.start_at(start);
2418 let mut frames = Vec::new();
2419 if start > 0
2420 && let Some(head) = head.filter(|head| head.sequence == live.sequence)
2421 && let Some(head_start) = head.live_first_frame()
2422 {
2423 let mut head = head.consume();
2424 head.start_at(head_start as u64);
2425 while head.index() < start {
2426 match head.poll_read_frame(&kio::Waiter::noop()) {
2427 Poll::Ready(Ok(Some(frame))) => frames.push(frame),
2428 _ => break,
2429 }
2430 }
2431 match head.index() == start {
2432 true => start = head_start as u64,
2433 false => frames.clear(),
2435 }
2436 }
2437 while let Poll::Ready(Ok(Some(frame))) = tail.poll_read_frame(&kio::Waiter::noop()) {
2438 frames.push(frame);
2439 }
2440 if frames.is_empty() {
2441 return None;
2442 }
2443
2444 let rebuilt = WarmGroup(
2446 track
2447 .create_group(group::Info {
2448 sequence: live.sequence,
2449 })
2450 .ok()?,
2451 );
2452 let mut writer = rebuilt.0.clone();
2453 if start > 0 {
2454 writer.start_at(start).ok()?;
2455 }
2456 for frame in frames {
2457 writer.write_frame(frame.timestamp, frame.payload).ok()?;
2458 }
2459 if live.is_finished() {
2460 writer.finish().ok()?;
2461 }
2462 Some(rebuilt)
2463}
2464
2465struct FrontTask {
2467 shared: kio::Shared<OriginState>,
2469 broadcast: broadcast::Producer,
2471 path: PathOwned,
2473 horizon: Horizon,
2475 watch: Watch,
2477 request: kio::Producer<PendingBroadcast>,
2479 pin: kio::Lock<Pin>,
2481 timers: Clock,
2482}
2483
2484struct TrackIo {
2487 resume: super::resume::Producer,
2488 staged: Option<(u64, track::Consumer)>,
2490 query: Option<(u64, track::Consumer, track::Querying)>,
2492 copy: Option<(u64, track::Consumer)>,
2494 edge: Option<track::Position>,
2499 warm: Option<WarmCopy>,
2502 head: Option<WarmGroup>,
2505 prior: Option<(u64, WarmCopy)>,
2510 held: Option<HeldGroup>,
2515 dead: Option<track::Consumer>,
2518 used: bool,
2520}
2521
2522impl TrackIo {
2523 fn end(&mut self) {
2526 self.staged = None;
2527 self.query = None;
2528 self.copy = None;
2529 self.warm = None;
2530 self.head = None;
2531 self.prior = None;
2532 self.held = None;
2533 self.dead = None;
2534 }
2535
2536 fn hold(&mut self, group: group::Producer) {
2539 if self.prior.as_ref().is_some_and(|(edge, _)| group.sequence > *edge) {
2540 self.prior = None;
2541 }
2542 self.held = Some(HeldGroup(group));
2543 }
2544
2545 fn bury(&mut self) {
2548 let Some((_, copy)) = self.copy.take() else { return };
2549 let after = self.held.as_ref().map(|held| held.0.sequence);
2550 if let Poll::Ready(group) = copy.poll_latest_group(after, &kio::Waiter::noop()) {
2551 self.hold(group);
2552 }
2553 self.dead = Some(copy);
2554 }
2555
2556 fn retire(&mut self, copy: &track::Consumer) -> Result<(), Error> {
2562 let Some(held) = self.held.take() else {
2563 return Ok(());
2564 };
2565 let Some(warm) = warm_copy(copy, self.head.as_ref(), Some(&held.0)) else {
2566 return Ok(());
2567 };
2568 self.head = None;
2569 self.resume.park(&warm.track)?;
2570 self.warm = Some(warm);
2571 Ok(())
2572 }
2573}
2574
2575async fn run_front(task: FrontTask) {
2579 let FrontTask {
2580 shared,
2581 broadcast,
2582 path,
2583 horizon,
2584 watch,
2585 request,
2586 pin,
2587 timers,
2588 } = task;
2589
2590 enum Step {
2592 Assigned(Arc<str>, super::resume::Producer),
2593 Resolved(u64, Result<broadcast::Consumer, Error>),
2594 SourceClosed(u64),
2595 Info(Arc<str>, u64, Result<track::Info, Error>),
2596 Ended(Arc<str>, u64, Result<(), Error>),
2597 Held(Arc<str>, group::Producer),
2598 Demand(Arc<str>),
2599 Deadline,
2600 Table,
2601 }
2602
2603 let mut front = Front::new(TRACK_IDLE_LINGER);
2604 let mut sources: HashMap<u64, broadcast::Consumer> = HashMap::new();
2605 let mut next_source = 0u64;
2606 let mut upstream: Option<(u64, kio::Consumer<PendingBroadcast>)> = None;
2608 let mut tracks: HashMap<Arc<str>, TrackIo> = HashMap::new();
2609 let mut deadline = crate::runtime::Deadline::new(&timers);
2610 let mut seen = 0;
2612 let mut events: VecDeque<Event> = VecDeque::new();
2613
2614 let select = |front: &mut Front, sources: &HashMap<u64, broadcast::Consumer>, seen: &mut u64| -> Event {
2617 let table = shared.read();
2618 if table.closed {
2619 return Event::Closed;
2620 }
2621 *seen = watch.seen();
2623 front.retain_routes(|route| table.routes.covers(&path.as_path(), route));
2624 let best = table
2625 .best_route(&path.as_path(), horizon, front.pin(), front.refused_routes())
2626 .map(|entry| Candidate {
2627 route: entry.id,
2628 first: entry.hops.iter().next().copied(),
2629 local: entry.local,
2630 });
2631 let serving_closing = front
2632 .serving()
2633 .and_then(|id| sources.get(&id))
2634 .is_some_and(|source| source.is_closing());
2635 Event::Selected { best, serving_closing }
2636 };
2637
2638 events.push_back(select(&mut front, &sources, &mut seen));
2639
2640 loop {
2641 while let Some(event) = events.pop_front() {
2642 for action in front.step(event) {
2643 match action {
2644 Action::Reselect => events.push_back(select(&mut front, &sources, &mut seen)),
2645 Action::Request { route } => {
2646 let found = {
2648 let table = shared.read();
2649 table
2650 .routes
2651 .covering(&path.as_path())
2652 .find(|entry| entry.id == route && entry.live())
2653 .map(|entry| {
2654 (
2655 Candidate {
2656 route,
2657 first: entry.hops.iter().next().copied(),
2658 local: entry.local,
2659 },
2660 entry.source.clone(),
2661 entry.server.clone(),
2662 )
2663 })
2664 };
2665 let Some((candidate, source, server)) = found else {
2666 events.push_back(Event::Resolved {
2667 route,
2668 result: Err(Refusal {
2669 err: Error::Unroutable,
2670 standing: false,
2671 }),
2672 });
2673 continue;
2674 };
2675 front.identify(candidate);
2676 *pin.lock() = front.pin();
2677 if let Some(source) = source {
2678 let id = next_source;
2679 next_source += 1;
2680 sources.insert(id, source);
2681 events.push_back(Event::Resolved { route, result: Ok(id) });
2682 continue;
2683 }
2684 let Some(server) = server else {
2685 events.push_back(Event::Resolved {
2686 route,
2687 result: Err(Refusal {
2688 err: Error::Unroutable,
2689 standing: true,
2690 }),
2691 });
2692 continue;
2693 };
2694 let mut serve = server.lock();
2695 if serve.closed {
2696 drop(serve);
2699 events.push_back(Event::Resolved {
2700 route,
2701 result: Err(Refusal {
2702 err: Error::Unroutable,
2703 standing: true,
2704 }),
2705 });
2706 continue;
2707 }
2708 if let Some(weak) = serve.served.get(&path) {
2711 drop(serve);
2712 let id = next_source;
2713 next_source += 1;
2714 sources.insert(id, weak.consume());
2715 events.push_back(Event::Resolved { route, result: Ok(id) });
2716 continue;
2717 }
2718 let pending = match serve.requests.join(&path) {
2719 Some(producer) => producer.consume(),
2720 None => {
2721 let producer = kio::Producer::<PendingBroadcast>::default();
2722 let consumer = producer.consume();
2723 match serve.requests.insert(path.clone(), producer) {
2724 Ok(()) => consumer,
2725 Err(_) => {
2728 drop(serve);
2729 events.push_back(Event::Resolved {
2730 route,
2731 result: Err(Refusal {
2732 err: Error::Unroutable,
2733 standing: true,
2734 }),
2735 });
2736 continue;
2737 }
2738 }
2739 }
2740 };
2741 upstream = Some((route, pending));
2742 }
2743 Action::Detach { source } => {
2744 sources.remove(&source);
2745 for io in tracks.values_mut() {
2748 if io.copy.as_ref().is_some_and(|(s, _)| *s == source) {
2749 io.bury();
2750 }
2751 if io.query.as_ref().is_some_and(|(s, ..)| *s == source) {
2752 io.query = None;
2753 }
2754 if io.staged.as_ref().is_some_and(|(s, _)| *s == source) {
2755 io.staged = None;
2756 }
2757 }
2758 }
2759 Action::Resolve => {
2760 if let Ok(mut pending) = request.write() {
2761 pending.resolved.get_or_insert(Ok(broadcast.consume()));
2762 }
2763 }
2764 Action::Query { track: name, source } => {
2765 let Some(io) = tracks.get_mut(&name) else { continue };
2766 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
2767 match sources.get(&source).map(|s| s.track(&name)) {
2768 Some(Ok(copy)) => {
2769 let query = copy.query().into_inner();
2772 io.query = Some((source, copy, query));
2773 }
2774 Some(Err(err)) => events.push_back(Event::TrackInfo {
2775 track: name,
2776 source,
2777 closing,
2778 result: Err(err),
2779 }),
2780 None => {}
2781 }
2782 }
2783 Action::Splice { track: name, source } => {
2784 let Some(io) = tracks.get_mut(&name) else { continue };
2785 let Some((staged, copy)) = io.staged.take() else {
2786 continue;
2787 };
2788 if staged != source {
2789 continue;
2790 }
2791 let outgoing = io.copy.take().map(|(_, copy)| copy).or_else(|| io.dead.take());
2792 let retired = match outgoing {
2793 Some(outgoing) => io.retire(&outgoing),
2794 None => Ok(()),
2795 };
2796 if let Err(err) = retired.and_then(|()| io.resume.takeover(©)) {
2797 let _ = io.resume.abort(err);
2801 tracks.remove(&name);
2802 continue;
2803 }
2804 if let Some(mut warm) = io.warm.take()
2805 && let Some(head) = warm.edge.take()
2806 {
2807 io.prior = Some((head.0.sequence, warm));
2808 io.head = Some(head);
2809 }
2810 io.edge = io.resume.resume_position();
2813 io.held = None;
2814 io.copy = Some((source, copy));
2815 }
2816 Action::Park { track: name } => {
2817 let Some(io) = tracks.get_mut(&name) else { continue };
2818 let Some((_, copy)) = io.copy.take() else { continue };
2819 let warm = warm_copy(©, io.head.as_ref(), io.held.as_ref().map(|held| &held.0));
2823 drop(copy);
2824 io.head = None;
2825 io.prior = None;
2826 io.held = None;
2827 io.dead = None;
2828 let parked = match &warm {
2829 Some(warm) => io.resume.park(&warm.track),
2830 None => io.resume.release(),
2831 };
2832 if parked.is_err() {
2833 tracks.remove(&name);
2834 continue;
2835 }
2836 io.warm = warm;
2837 }
2838 Action::Forget { track: name } => {
2839 if let Some(io) = tracks.get_mut(&name)
2844 && !broadcast.forget_spliced(&name, &io.resume)
2845 {
2846 if !io.used {
2847 io.used = true;
2848 events.push_back(Event::Used { track: name });
2849 }
2850 continue;
2851 }
2852 tracks.remove(&name);
2853 events.push_back(Event::Forgotten { track: name });
2854 }
2855 Action::Finish { track: name } => {
2856 if let Some(io) = tracks.get_mut(&name) {
2857 io.end();
2858 let _ = io.resume.finish();
2859 }
2860 }
2861 Action::Abort { track: name, err } => {
2862 if let Some(io) = tracks.get_mut(&name) {
2863 tracing::debug!(name = %name, %err, "aborting track");
2864 io.end();
2865 let _ = io.resume.abort(err);
2866 }
2867 }
2868 Action::Arm { at } => deadline.set(at),
2869 Action::End { err } => {
2870 if let Ok(mut pending) = request.write() {
2871 pending.resolved.get_or_insert(Err(err.clone()));
2872 }
2873 broadcast.close();
2880 broadcast.release_spliced(err.clone());
2881 for (_, mut io) in tracks.drain() {
2882 let waiting = io.staged.take().map(|(_, copy)| copy);
2886 let waiting = waiting.or_else(|| io.query.take().map(|(_, copy, _)| copy));
2887 if let Some(copy) = waiting
2888 && io.resume.is_used()
2889 {
2890 if io.resume.takeover(©).is_err() {
2891 continue;
2892 }
2893 io.warm = None;
2894 }
2895 if !io.resume.is_used() || !io.resume.is_spliced() || io.warm.is_some() {
2897 let _ = io.resume.abort(err.clone());
2898 }
2899 }
2900 return;
2901 }
2902 }
2903 }
2904 }
2905
2906 let step = kio::wait(|waiter| {
2907 if let Poll::Ready((name, resume)) = broadcast.poll_spliced_assigned(waiter) {
2908 return Poll::Ready(Step::Assigned(name, resume));
2909 }
2910 if let Some((route, pending)) = &upstream
2911 && let Poll::Ready(result) = pending.poll(waiter, |p| match &p.resolved {
2912 Some(result) => Poll::Ready(result.clone()),
2913 None => Poll::Pending,
2914 }) {
2915 return Poll::Ready(Step::Resolved(
2916 *route,
2917 match result {
2918 Ok(resolved) => resolved,
2919 Err(_closed) => Err(Error::Unroutable),
2922 },
2923 ));
2924 }
2925 if let Some(id) = front.serving()
2926 && let Some(source) = sources.get(&id)
2927 && source.poll_closed(waiter).is_ready()
2928 {
2929 return Poll::Ready(Step::SourceClosed(id));
2930 }
2931 for (name, io) in &tracks {
2932 if let Some((source, _, query)) = &io.query
2933 && let Poll::Ready(result) = query.poll(waiter)
2934 {
2935 return Poll::Ready(Step::Info(name.clone(), *source, result));
2936 }
2937 if let Some((source, copy)) = &io.copy
2938 && let Poll::Ready(result) = copy.poll_complete(waiter)
2939 {
2940 return Poll::Ready(Step::Ended(name.clone(), *source, result));
2941 }
2942 if let Some((_, copy)) = &io.copy
2943 && let Poll::Ready(group) =
2944 copy.poll_latest_group(io.held.as_ref().map(|held| held.0.sequence), waiter)
2945 {
2946 return Poll::Ready(Step::Held(name.clone(), group));
2947 }
2948 let edge = match io.used {
2950 true => io.resume.poll_unused(waiter),
2951 false => io.resume.poll_used(waiter),
2952 };
2953 if edge.is_ready() {
2954 return Poll::Ready(Step::Demand(name.clone()));
2955 }
2956 }
2957 if deadline.poll(waiter).is_ready() {
2958 return Poll::Ready(Step::Deadline);
2959 }
2960 watch.poll_changed(waiter, seen).map(|()| Step::Table)
2961 })
2962 .await;
2963
2964 let event = match step {
2965 Step::Assigned(name, resume) => {
2966 tracks.insert(
2967 name.clone(),
2968 TrackIo {
2969 resume,
2970 staged: None,
2971 query: None,
2972 copy: None,
2973 edge: None,
2974 warm: None,
2975 head: None,
2976 prior: None,
2977 held: None,
2978 dead: None,
2979 used: false,
2980 },
2981 );
2982 Event::TrackAssigned {
2983 track: name,
2984 now: timers.now(),
2985 }
2986 }
2987 Step::Resolved(route, result) => {
2988 upstream = None;
2989 match result {
2990 Ok(source) => {
2991 let id = next_source;
2992 next_source += 1;
2993 sources.insert(id, source);
2994 Event::Resolved { route, result: Ok(id) }
2995 }
2996 Err(err) => {
2997 let standing =
3001 !matches!(err, Error::Unroutable) || shared.read().routes.covers(&path.as_path(), route);
3002 Event::Resolved {
3003 route,
3004 result: Err(Refusal { err, standing }),
3005 }
3006 }
3007 }
3008 }
3009 Step::SourceClosed(source) => Event::SourceClosed { source },
3010 Step::Info(name, source, result) => {
3011 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3012 let Some(io) = tracks.get_mut(&name) else { continue };
3013 let Some((_, copy, _)) = io.query.take() else { continue };
3014 let result = match result {
3017 Ok(info) => match copy.poll_complete(&kio::Waiter::noop()) {
3018 Poll::Ready(Err(err)) => Err(err),
3019 _ => Ok(info),
3020 },
3021 Err(err) => Err(err),
3022 };
3023 if result.is_ok() && io.used {
3026 io.staged = Some((source, copy));
3027 }
3028 Event::TrackInfo {
3029 track: name,
3030 source,
3031 closing,
3032 result,
3033 }
3034 }
3035 Step::Ended(name, source, result) => {
3036 let closing = sources.get(&source).is_some_and(|s| s.is_closing());
3037 let Some(io) = tracks.get_mut(&name) else { continue };
3038 io.bury();
3039 let delivered = io.resume.resume_position() != io.edge;
3040 Event::TrackEnded {
3041 track: name,
3042 source,
3043 closing,
3044 result,
3045 delivered,
3046 }
3047 }
3048 Step::Held(name, group) => {
3049 if let Some(io) = tracks.get_mut(&name) {
3050 io.hold(group);
3051 }
3052 continue;
3053 }
3054 Step::Demand(name) => {
3055 let Some(io) = tracks.get_mut(&name) else { continue };
3056 io.used = io.resume.is_used();
3057 if !io.used {
3058 io.query = None;
3061 io.staged = None;
3062 }
3063 match io.used {
3064 true => Event::Used { track: name },
3065 false => Event::Unused {
3066 track: name,
3067 now: timers.now(),
3068 },
3069 }
3070 }
3071 Step::Deadline => {
3072 deadline.set(None);
3075 Event::Deadline { now: timers.now() }
3076 }
3077 Step::Table => select(&mut front, &sources, &mut seen),
3078 };
3079 events.push_back(event);
3080 }
3081}
3082
3083#[derive(Default)]
3088struct RouteTable {
3089 root: RouteNode,
3090}
3091
3092#[derive(Default)]
3095struct RouteNode {
3096 entries: Vec<RouteEntry>,
3098 cursors: Vec<ConsumerId>,
3100 cursors_below: usize,
3103 watches: Vec<(u64, kio::Producer<Watched>)>,
3106 watches_below: usize,
3109 children: HashMap<String, RouteNode>,
3110}
3111
3112#[derive(Default)]
3115struct Watched {
3116 generation: u64,
3117}
3118
3119struct Watch {
3125 shared: kio::Shared<OriginState>,
3126 path: PathOwned,
3127 id: u64,
3128 signal: kio::Consumer<Watched>,
3129}
3130
3131impl Watch {
3132 fn seen(&self) -> u64 {
3136 self.signal.read().generation
3137 }
3138
3139 fn poll_changed(&self, waiter: &kio::Waiter, seen: u64) -> Poll<()> {
3141 self.signal
3142 .poll(waiter, |watched| match watched.generation != seen {
3143 true => Poll::Ready(()),
3144 false => Poll::Pending,
3145 })
3146 .map(|_| ())
3147 }
3148}
3149
3150impl Drop for Watch {
3151 fn drop(&mut self) {
3152 self.shared.lock().routes.remove_watch(&self.path, self.id);
3153 }
3154}
3155
3156#[derive(Clone, Copy)]
3158struct Below {
3159 cursors: usize,
3160 watches: usize,
3161}
3162
3163impl Below {
3164 const NONE: Self = Self { cursors: 0, watches: 0 };
3165 const CURSOR: Self = Self { cursors: 1, watches: 0 };
3166 const WATCH: Self = Self { cursors: 0, watches: 1 };
3167}
3168
3169impl RouteNode {
3170 fn is_empty(&self) -> bool {
3172 self.entries.is_empty() && self.cursors.is_empty() && self.watches.is_empty() && self.children.is_empty()
3173 }
3174
3175 fn find<'a>(&self, mut parts: impl Iterator<Item = &'a str>) -> Option<&Self> {
3177 match parts.next() {
3178 None => Some(self),
3179 Some(part) => self.children.get(part)?.find(parts),
3180 }
3181 }
3182
3183 fn reach<'a>(&mut self, mut parts: impl Iterator<Item = &'a str>, below: Below) -> &mut Self {
3186 self.cursors_below += below.cursors;
3187 self.watches_below += below.watches;
3188 match parts.next() {
3189 None => self,
3190 Some(part) => self.children.entry(part.to_string()).or_default().reach(parts, below),
3191 }
3192 }
3193
3194 fn edit<'a, R>(
3198 &mut self,
3199 mut parts: impl Iterator<Item = &'a str>,
3200 below: Below,
3201 f: impl FnOnce(&mut Self) -> R,
3202 ) -> Option<R> {
3203 let result = match parts.next() {
3204 None => f(self),
3205 Some(part) => {
3206 let child = self.children.get_mut(part)?;
3207 let result = child.edit(parts, below, f)?;
3208 if child.is_empty() {
3209 self.children.remove(part);
3210 }
3211 result
3212 }
3213 };
3214 self.cursors_below -= below.cursors;
3215 self.watches_below -= below.watches;
3216 Some(result)
3217 }
3218
3219 fn poke(&self) {
3221 for (_, watch) in &self.watches {
3222 if let Ok(mut watched) = watch.write() {
3223 watched.generation += 1;
3224 }
3225 }
3226 }
3227
3228 fn poke_below(&self) {
3231 if self.watches_below == 0 {
3232 return;
3233 }
3234 self.poke();
3235 for child in self.children.values() {
3236 child.poke_below();
3237 }
3238 }
3239
3240 fn walk<'a>(&'a self, visit: &mut impl FnMut(&'a Self)) {
3242 visit(self);
3243 for child in self.children.values() {
3244 child.walk(visit);
3245 }
3246 }
3247
3248 fn collect_cursors(&self, out: &mut Vec<ConsumerId>) {
3250 if self.cursors_below == 0 {
3251 return;
3252 }
3253 out.extend(&self.cursors);
3254 for child in self.children.values() {
3255 child.collect_cursors(out);
3256 }
3257 }
3258}
3259
3260impl RouteTable {
3261 fn split(&self, path: &Path) -> (Vec<&RouteNode>, Option<&RouteNode>) {
3264 let mut above = Vec::new();
3265 let mut node = &self.root;
3266 for part in path.parts() {
3267 above.push(node);
3268 match node.children.get(part) {
3269 Some(child) => node = child,
3270 None => return (above, None),
3271 }
3272 }
3273 (above, Some(node))
3274 }
3275
3276 fn covering(&self, path: &Path) -> impl Iterator<Item = &RouteEntry> {
3278 let (above, at) = self.split(path);
3279 above.into_iter().chain(at).flat_map(|node| node.entries.iter())
3280 }
3281
3282 fn covers(&self, path: &Path, id: u64) -> bool {
3284 self.covering(path).any(|entry| entry.id == id && entry.live())
3285 }
3286
3287 fn at(&self, prefix: &Path) -> impl Iterator<Item = &RouteEntry> {
3289 self.root
3290 .find(prefix.parts())
3291 .into_iter()
3292 .flat_map(|node| node.entries.iter())
3293 }
3294
3295 fn at_mut(&mut self, prefix: &Path) -> impl Iterator<Item = &mut RouteEntry> {
3297 let mut node = Some(&mut self.root);
3298 for part in prefix.parts() {
3299 node = node.and_then(|node| node.children.get_mut(part));
3300 }
3301 node.into_iter().flat_map(|node| node.entries.iter_mut())
3302 }
3303
3304 fn entries(&self) -> impl Iterator<Item = &RouteEntry> {
3306 let mut nodes = Vec::new();
3307 self.root.walk(&mut |node| nodes.push(node));
3308 nodes.into_iter().flat_map(|node| node.entries.iter())
3309 }
3310
3311 fn insert(&mut self, entry: RouteEntry) {
3313 let node = self.root.reach(entry.prefix.parts(), Below::NONE);
3314 node.entries.push(entry);
3315 }
3316
3317 fn entry_mut(&mut self, prefix: &Path, id: u64) -> Option<&mut RouteEntry> {
3319 let mut node = &mut self.root;
3320 for part in prefix.parts() {
3321 node = node.children.get_mut(part)?;
3322 }
3323 node.entries.iter_mut().find(|entry| entry.id == id)
3324 }
3325
3326 fn remove(&mut self, prefix: &Path, id: u64) -> Option<RouteEntry> {
3328 self.root
3329 .edit(prefix.parts(), Below::NONE, |node| {
3330 let index = node.entries.iter().position(|entry| entry.id == id)?;
3331 Some(node.entries.swap_remove(index))
3332 })
3333 .flatten()
3334 }
3335
3336 fn add_cursor(&mut self, head: &Path, id: ConsumerId) {
3338 self.root.reach(head.parts(), Below::CURSOR).cursors.push(id);
3339 }
3340
3341 fn remove_cursor(&mut self, head: &Path, id: ConsumerId) {
3344 self.root.edit(head.parts(), Below::CURSOR, |node| {
3345 node.cursors.retain(|cursor| *cursor != id)
3346 });
3347 }
3348
3349 fn add_watch(&mut self, path: &Path, id: u64) -> kio::Consumer<Watched> {
3351 let producer = kio::Producer::<Watched>::default();
3352 let consumer = producer.consume();
3353 self.root.reach(path.parts(), Below::WATCH).watches.push((id, producer));
3354 consumer
3355 }
3356
3357 fn remove_watch(&mut self, path: &Path, id: u64) {
3360 self.root.edit(path.parts(), Below::WATCH, |node| {
3361 node.watches.retain(|(watch, _)| *watch != id)
3362 });
3363 }
3364
3365 fn poke_below(&self, prefix: &Path) {
3367 if let (_, Some(node)) = self.split(prefix) {
3368 node.poke_below();
3369 }
3370 }
3371
3372 fn poke_all(&self) {
3374 self.root.walk(&mut |node| node.poke());
3375 }
3376
3377 fn cursors_touching(&self, prefix: &Path) -> Vec<ConsumerId> {
3381 let (above, at) = self.split(prefix);
3382 let mut cursors: Vec<ConsumerId> = above.iter().flat_map(|node| node.cursors.iter().copied()).collect();
3383 if let Some(node) = at {
3384 node.collect_cursors(&mut cursors);
3385 }
3386 cursors.sort_unstable();
3388 cursors.dedup();
3389 cursors
3390 }
3391}
3392
3393struct OriginState {
3400 update_hold: Duration,
3403
3404 routes: RouteTable,
3407 next_route: u64,
3408 next_watch: u64,
3409
3410 cursors: HashMap<ConsumerId, TableCursor>,
3414
3415 fronts: WeakCache<FrontKey, RemoteFront>,
3421
3422 withdrawn: HashMap<PathOwned, HashSet<Hop>>,
3425
3426 closed: bool,
3429}
3430
3431impl OriginState {
3432 fn new(update_hold: Duration) -> Self {
3433 Self {
3434 update_hold,
3435 routes: RouteTable::default(),
3436 next_route: 0,
3437 next_watch: 0,
3438 cursors: HashMap::new(),
3439 fronts: WeakCache::default(),
3440 withdrawn: HashMap::new(),
3441 closed: false,
3442 }
3443 }
3444
3445 fn withdrawn_through(&self, prefix: &Path, hops: &Hops) -> bool {
3447 if self.withdrawn.is_empty() {
3448 return false;
3449 }
3450 self.withdrawn
3451 .get(prefix)
3452 .is_some_and(|peers| hops.iter().any(|hop| peers.contains(hop)))
3453 }
3454
3455 fn restale(&mut self, prefix: &Path) {
3457 let peers = self.withdrawn.get(prefix);
3458 let mut changed = None;
3459 for entry in self.routes.at_mut(prefix) {
3460 let stale = peers.is_some_and(|peers| entry.hops.iter().any(|hop| peers.contains(hop)));
3461 if entry.stale != stale {
3462 entry.stale = stale;
3463 changed = Some(entry.claim.clone());
3464 }
3465 }
3466 if let Some(claim) = changed {
3467 self.sync_route(prefix, &claim);
3468 }
3469 }
3470
3471 fn reannounced(&mut self, prefix: &PathOwned, hops: &Hops) {
3475 if let Some(sender) = hops.iter().last()
3476 && self.withdrawn.get_mut(prefix).is_some_and(|peers| peers.remove(sender))
3477 {
3478 self.restale(prefix);
3479 self.prune_withdrawn(prefix);
3480 }
3481 }
3482
3483 fn prune_withdrawn(&mut self, prefix: &PathOwned) {
3485 if self.withdrawn.is_empty() {
3486 return;
3487 }
3488 let Some(peers) = self.withdrawn.get_mut(prefix) else {
3489 return;
3490 };
3491 if peers.len() <= 1 {
3492 peers.retain(|peer| self.routes.at(prefix).any(|entry| entry.hops.contains(peer)));
3494 } else {
3495 let mut unreferenced = peers.clone();
3496 for entry in self.routes.at(prefix) {
3497 if unreferenced.is_empty() {
3498 break;
3499 }
3500 for hop in entry.hops.iter() {
3501 unreferenced.remove(hop);
3502 }
3503 }
3504 peers.retain(|peer| !unreferenced.contains(peer));
3505 }
3506 if peers.is_empty() {
3507 self.withdrawn.remove(prefix);
3508 }
3509 }
3510
3511 fn sync_route(&mut self, prefix: &Path, claim: &Pattern) {
3516 let routes = &self.routes;
3518 let mut candidates = None;
3519 for id in routes.cursors_touching(prefix) {
3520 let Some(cursor) = self.cursors.get_mut(&id) else {
3521 continue;
3522 };
3523 if let Some(presented) = cursor.presented(prefix, claim) {
3524 if presented.is_empty() {
3525 Self::sync_cursor(routes, cursor, &presented);
3527 } else {
3528 let candidates = candidates.get_or_insert_with(|| {
3531 let mut entries: Vec<_> = routes
3532 .at(prefix)
3533 .map(|entry| (route_order(prefix, entry), entry))
3534 .collect();
3535 entries.sort_unstable_by_key(|(order, _)| *order);
3536 entries
3537 });
3538 let best = candidates
3539 .iter()
3540 .map(|(_, entry)| *entry)
3541 .find(|entry| cursor.visible(entry));
3542 cursor.update(&presented, best);
3543 }
3544 }
3545 }
3546 routes.poke_below(prefix);
3548 }
3549
3550 fn watch(&mut self, shared: &kio::Shared<OriginState>, path: &Path) -> Watch {
3552 let id = self.next_watch;
3553 self.next_watch += 1;
3554 let signal = self.routes.add_watch(path, id);
3555 Watch {
3556 shared: shared.clone(),
3557 path: path.to_owned(),
3558 id,
3559 signal,
3560 }
3561 }
3562
3563 fn sync_cursor(routes: &RouteTable, cursor: &mut TableCursor, presented: &PathOwned) {
3566 let candidates: Vec<&RouteEntry> = match presented.is_empty() {
3572 true => routes
3573 .covering(&cursor.root)
3574 .filter(|entry| cursor.visible(entry))
3575 .collect(),
3576 false => {
3577 let absolute = cursor.root.join(presented);
3578 routes.at(&absolute).filter(|entry| cursor.visible(entry)).collect()
3579 }
3580 };
3581 let most = candidates.iter().map(|entry| entry.prefix.len()).max();
3582 let best = most.and_then(|most| {
3583 candidates
3584 .into_iter()
3585 .filter(|entry| entry.prefix.len() == most)
3586 .min_by_key(|entry| route_order(&entry.prefix, entry))
3587 });
3588
3589 cursor.update(presented, best);
3590 }
3591
3592 fn register_cursor(&mut self, id: ConsumerId, mut cursor: TableCursor) {
3594 let mut presented: BTreeSet<PathOwned> = BTreeSet::new();
3597 for head in &cursor.heads {
3598 let (above, at) = self.routes.split(head);
3599 let mut nodes = above;
3600 if let Some(node) = at {
3601 node.walk(&mut |node| nodes.push(node));
3602 }
3603 for entry in nodes.into_iter().flat_map(|node| node.entries.iter()) {
3604 if let Some(p) = cursor.presented(&entry.prefix, &entry.claim) {
3605 presented.insert(p);
3606 }
3607 }
3608 }
3609 for p in &presented {
3610 Self::sync_cursor(&self.routes, &mut cursor, p);
3611 }
3612 for head in &cursor.heads {
3613 self.routes.add_cursor(head, id);
3614 }
3615 self.cursors.insert(id, cursor);
3616 }
3617
3618 fn best_route(&self, path: &Path, horizon: Horizon, pin: Pin, refused: &HashSet<u64>) -> Option<&RouteEntry> {
3633 let (above, at) = self.routes.split(path);
3637 let mut best = None;
3638 for node in above.into_iter().chain(at) {
3639 let mut candidates = node
3640 .entries
3641 .iter()
3642 .filter(|entry| entry.live())
3643 .filter(|entry| entry.scope.matches(path.as_str()))
3644 .filter(|entry| horizon.admits(entry))
3645 .filter(|entry| entry.qualifies(pin))
3646 .filter(|entry| !refused.contains(&entry.id))
3647 .peekable();
3648 if candidates.peek().is_some() {
3649 best = candidates
3650 .filter(|entry| entry.serves(path))
3651 .min_by_key(|entry| route_order(&entry.prefix, entry));
3652 }
3653 }
3654 best
3655 }
3656}
3657
3658#[derive(Default)]
3665struct PendingBroadcast {
3666 resolved: Option<Result<broadcast::Consumer, Error>>,
3667}
3668
3669#[must_use = "dropping an origin::Dynamic retracts the route"]
3684pub struct Dynamic {
3685 announcement: AnnounceProducer,
3687 state: kio::Shared<ServeState>,
3688 _keepalive: Keepalive,
3692}
3693
3694impl Dynamic {
3695 pub(crate) fn withdrawn(self) {
3698 self.announcement.retract(true);
3699 }
3700
3701 pub fn update(&self, route: Route) -> Result<(), Error> {
3711 self.announcement.update(route)
3712 }
3713
3714 pub fn poll_requested_broadcast(&self, waiter: &kio::Waiter) -> Poll<Result<Request, Error>> {
3719 let mut state = ready!(self.state.poll(waiter, |state| {
3720 if state.closed || state.requests.has_queued() {
3721 Poll::Ready(())
3722 } else {
3723 Poll::Pending
3724 }
3725 }));
3726
3727 if state.closed {
3729 return Poll::Ready(Err(Error::Closed));
3730 }
3731
3732 let path = state.requests.pop().expect("predicate guaranteed a request");
3733 let producer = state.requests.get(&path).expect("popped key must be pending").clone();
3739 Poll::Ready(Ok(Request {
3740 path,
3741 producer,
3742 home: self.state.clone(),
3743 }))
3744 }
3745
3746 pub async fn requested_broadcast(&self) -> Result<Request, Error> {
3752 kio::wait(|waiter| self.poll_requested_broadcast(waiter)).await
3753 }
3754}
3755
3756impl ServeState {
3757 fn resolve(
3768 shared: &kio::Shared<Self>,
3769 path: &PathOwned,
3770 producer: &kio::Producer<PendingBroadcast>,
3771 result: Result<broadcast::Consumer, Error>,
3772 ) {
3773 let mut state = shared.lock();
3774 if state.closed {
3775 return;
3776 }
3777 let resolved = match result {
3778 Ok(broadcast) => {
3779 let existing = state.served.insert(path.clone(), broadcast.weak());
3783 Ok(existing.map(|weak| weak.consume()).unwrap_or(broadcast))
3784 }
3785 Err(err) => Err(err),
3786 };
3787 state.requests.remove_if(path, |p| p.same_channel(producer));
3788 if let Ok(mut pending) = producer.write() {
3789 pending.resolved.get_or_insert(resolved);
3790 drop(state);
3791 }
3792 }
3793
3794 fn forget(shared: &kio::Shared<Self>, path: &PathOwned, producer: &kio::Producer<PendingBroadcast>) {
3796 shared.lock().requests.remove_if(path, |p| p.same_channel(producer));
3797 }
3798}
3799
3800pub struct Request {
3807 path: PathOwned,
3809
3810 producer: kio::Producer<PendingBroadcast>,
3813
3814 home: kio::Shared<ServeState>,
3817}
3818
3819impl Request {
3820 pub fn path(&self) -> &Path<'_> {
3822 &self.path
3823 }
3824
3825 pub fn accept(self, broadcast: impl Consume<broadcast::Consumer>) {
3831 let broadcast = broadcast.consume();
3832 ServeState::resolve(&self.home, &self.path, &self.producer, Ok(broadcast));
3833 }
3835
3836 pub fn reject(self, err: Error) {
3838 ServeState::resolve(&self.home, &self.path, &self.producer, Err(err));
3839 }
3840}
3841
3842impl Drop for Request {
3843 fn drop(&mut self) {
3844 ServeState::forget(&self.home, &self.path, &self.producer);
3852 }
3853}
3854
3855pub struct Requesting {
3862 inner: RequestState,
3863 path: PathOwned,
3867 stats: stats::Scope,
3870}
3871
3872enum RequestState {
3873 Failed(Error),
3876 Pending(kio::Consumer<PendingBroadcast>),
3878}
3879
3880impl Requesting {
3881 fn failed(error: Error) -> Self {
3882 Self::new(RequestState::Failed(error))
3883 }
3884
3885 fn queued(consumer: kio::Consumer<PendingBroadcast>) -> Self {
3886 Self::new(RequestState::Pending(consumer))
3887 }
3888
3889 pub fn is_queued(&self) -> bool {
3898 matches!(self.inner, RequestState::Pending(_))
3899 }
3900
3901 fn new(inner: RequestState) -> Self {
3902 Self {
3903 inner,
3904 path: PathOwned::default(),
3905 stats: stats::Scope::default(),
3906 }
3907 }
3908
3909 fn with_path(mut self, path: PathOwned) -> Self {
3910 self.path = path;
3911 self
3912 }
3913
3914 fn with_stats(mut self, scope: stats::Scope) -> Self {
3916 self.stats = scope;
3917 self
3918 }
3919
3920 fn hand_out(&self, broadcast: broadcast::Consumer) -> broadcast::Consumer {
3922 broadcast.with_path(self.path.clone()).with_stats(self.stats.clone())
3923 }
3924
3925 pub fn poll_ok(&self, waiter: &kio::Waiter) -> Poll<Result<broadcast::Consumer, Error>> {
3927 match &self.inner {
3928 RequestState::Failed(error) => Poll::Ready(Err(error.clone())),
3929 RequestState::Pending(consumer) => Poll::Ready(
3930 match ready!(consumer.poll(waiter, |state| match &state.resolved {
3931 Some(result) => Poll::Ready(result.clone()),
3932 None => Poll::Pending,
3933 })) {
3934 Ok(result) => result.map(|broadcast| self.hand_out(broadcast)),
3935 Err(_closed) => Err(Error::Unroutable),
3937 },
3938 ),
3939 }
3940 }
3941}
3942
3943impl kio::Pollable for Requesting {
3944 type Output = Result<broadcast::Consumer, Error>;
3945
3946 fn poll(&self, waiter: &kio::Waiter) -> Poll<Self::Output> {
3947 self.poll_ok(waiter)
3948 }
3949}
3950
3951pub trait Consume<T> {
3959 fn consume(&self) -> T;
3961}
3962
3963impl<T, U: Consume<T>> Consume<T> for &U {
3964 fn consume(&self) -> T {
3965 (**self).consume()
3966 }
3967}
3968
3969impl Consume<Consumer> for Producer {
3970 fn consume(&self) -> Consumer {
3971 Consumer::from_producer(self, stats::Session::default())
3975 }
3976}
3977
3978impl Consume<Consumer> for Consumer {
3979 fn consume(&self) -> Consumer {
3980 self.clone()
3981 }
3982}
3983
3984impl Consume<broadcast::Consumer> for broadcast::Producer {
3985 fn consume(&self) -> broadcast::Consumer {
3986 self.consume()
3988 }
3989}
3990
3991impl Consume<broadcast::Consumer> for broadcast::Consumer {
3992 fn consume(&self) -> broadcast::Consumer {
3993 self.clone()
3994 }
3995}
3996
3997impl Consume<track::Consumer> for track::Producer {
3998 fn consume(&self) -> track::Consumer {
3999 self.consume()
4000 }
4001}
4002
4003impl Consume<track::Consumer> for track::Consumer {
4004 fn consume(&self) -> track::Consumer {
4005 self.clone()
4006 }
4007}
4008
4009#[derive(Clone)]
4015pub struct Consumer {
4016 hop: Hop,
4018 scope: OriginScope,
4019
4020 root: PathOwned,
4022
4023 shared: kio::Shared<OriginState>,
4026
4027 stats: stats::Session,
4031
4032 horizon: Horizon,
4037
4038 hidden: Hidden,
4040
4041 pool: cache::Pool,
4044 cache_duration: Duration,
4045
4046 tasks: TasksWeak,
4050
4051 timers: Clock,
4054}
4055
4056impl Consumer {
4057 fn from_producer(producer: &Producer, stats: stats::Session) -> Self {
4058 Self {
4059 hop: producer.hop,
4060 scope: producer.scope.clone(),
4061 root: producer.root.clone(),
4062 shared: producer.shared.clone(),
4063 stats,
4064 horizon: Horizon::default(),
4065 hidden: Hidden::default(),
4066 pool: producer.pool.clone(),
4067 cache_duration: producer.cache_duration,
4068 tasks: producer.tasks.downgrade(),
4069 timers: producer.timers.clone(),
4070 }
4071 }
4072
4073 pub fn hop(&self) -> Hop {
4075 self.hop
4076 }
4077
4078 pub(crate) fn excluding(mut self, peer: Hop) -> Self {
4086 self.horizon.exclude = Some(peer);
4087 self
4088 }
4089
4090 pub fn local(mut self) -> Self {
4097 self.horizon.local = true;
4098 self
4099 }
4100
4101 pub fn with_hidden(mut self, hidden: bool) -> Self {
4106 self.hidden.include = hidden;
4107 self
4108 }
4109
4110 pub(crate) fn beyond(mut self, outer: &Consumer) -> Self {
4113 self.hidden.beyond = Some(
4114 outer
4115 .hidden
4116 .from
4117 .clone()
4118 .map(|head| vec![head])
4119 .unwrap_or_else(|| interest_prefixes(&outer.scope.allowed)),
4120 );
4121 self
4122 }
4123
4124 pub(crate) fn discovery(mut self, hidden: bool) -> Self {
4126 self.hidden.include = hidden;
4127 self.hidden.from = Some(self.root.clone());
4128 self
4129 }
4130
4131 pub(crate) fn includes_hidden(&self) -> bool {
4133 self.hidden.include
4134 }
4135
4136 pub fn with_stats(mut self, session: stats::Session) -> Self {
4140 self.stats = session;
4141 self
4142 }
4143
4144 fn untagged(&self) -> Self {
4148 Self {
4149 stats: stats::Session::default(),
4150 ..self.clone()
4151 }
4152 }
4153
4154 pub(crate) fn empty(&self) -> Self {
4159 Self {
4160 scope: OriginScope::empty(),
4161 ..self.clone()
4162 }
4163 }
4164
4165 pub fn announced(&self) -> AnnounceConsumer {
4174 let state = kio::Producer::<OriginConsumerState>::default();
4175 let cursor = |root: PathOwned,
4176 allowed: Patterns,
4177 hidden: Hidden,
4178 mount: Option<&Mount>,
4179 under: PathOwned,
4180 holes: Vec<PathOwned>| TableCursor {
4181 root,
4182 heads: interest_prefixes(&allowed),
4183 allowed,
4184 horizon: self.horizon,
4185 hidden,
4186 state: state.clone(),
4187 under,
4188 holes,
4189 named: mount.map(|mount| (mount.clone(), self.scope.allowed.clone())),
4190 current: HashMap::new(),
4191 };
4192
4193 if let Some(mount) = self.scope.mount(&self.root) {
4196 let Some(root) = mount.resolve(&self.root) else {
4198 return AnnounceConsumer::new(
4199 self.root.clone(),
4200 Vec::new(),
4201 state,
4202 self.stats.clone(),
4203 &self.shared,
4204 self.timers.clone(),
4205 );
4206 };
4207 let cursors = vec![cursor(
4208 root,
4209 mount.translate(&self.scope.allowed),
4210 self.hidden.translate(mount),
4211 Some(mount),
4212 PathOwned::default(),
4213 Vec::new(),
4214 )];
4215 return AnnounceConsumer::new(
4216 self.root.clone(),
4217 cursors,
4218 state,
4219 self.stats.clone(),
4220 &self.shared,
4221 self.timers.clone(),
4222 );
4223 }
4224
4225 let heads = interest_prefixes(&self.scope.allowed);
4228 let mut holes = Vec::new();
4229 let mut cursors = Vec::new();
4230 for mount in self.scope.mounts.iter() {
4231 let Some(under) = mount.at.strip_prefix(&self.root) else {
4232 continue;
4233 };
4234 holes.push(mount.at.clone());
4235 let allowed = mount.translate(&self.scope.allowed);
4236 if allowed.is_empty()
4239 || !(self.hidden.include
4240 || !hides(
4241 self.hidden.from.as_ref().map(std::slice::from_ref).unwrap_or(&heads),
4242 &mount.at,
4243 )) {
4244 continue;
4245 }
4246 cursors.push(cursor(
4247 mount.target.clone(),
4248 allowed,
4249 self.hidden.translate(mount),
4250 Some(mount),
4251 under.to_owned(),
4252 Vec::new(),
4253 ));
4254 }
4255 cursors.insert(
4256 0,
4257 cursor(
4258 self.root.clone(),
4259 self.scope.allowed.clone(),
4260 self.hidden.clone(),
4261 None,
4262 PathOwned::default(),
4263 holes,
4264 ),
4265 );
4266 AnnounceConsumer::new(
4267 self.root.clone(),
4268 cursors,
4269 state,
4270 self.stats.clone(),
4271 &self.shared,
4272 self.timers.clone(),
4273 )
4274 }
4275
4276 pub fn consume(&self) -> Self {
4278 self.clone()
4279 }
4280
4281 #[cfg(test)]
4284 pub(crate) fn get_broadcast(&self, path: impl AsPath) -> Option<broadcast::Consumer> {
4285 let full = self.root.join(path).to_owned();
4286 if !self.scope.permits(&full) {
4287 return None;
4288 }
4289 let full = self.scope.resolve(&full)?;
4290 let table = self.shared.lock();
4291 table
4292 .routes
4293 .at(&full)
4294 .filter(|entry| entry.local)
4295 .min_by_key(|entry| route_order(&entry.prefix, entry))
4296 .and_then(|entry| entry.source.clone())
4297 }
4298
4299 pub async fn routed(&self, path: impl AsPath) -> Option<Route> {
4309 let path = path.as_path();
4310
4311 let consumer = match Pattern::subtree(path.as_str()) {
4316 Ok(subtree) => self.scope("", &Patterns::from(subtree)).ok()?,
4317 Err(InvalidPattern::TooManySegments) => self.clone(),
4318 Err(_) => return None,
4319 };
4320
4321 if !consumer.allowed().matches(path.as_str()) {
4325 return None;
4326 }
4327
4328 let mut announced = consumer.untagged().with_hidden(true).announced();
4332 loop {
4333 let update = announced.next().await?;
4334 if update.kind.is_active() && path.has_prefix(&update.prefix) {
4335 return Some(update.route);
4336 }
4337 }
4338 }
4339
4340 pub async fn routed_broadcast(&self, path: impl AsPath) -> Result<broadcast::Consumer, Error> {
4354 let path = path.as_path();
4355
4356 if !self.allowed().matches(path.as_str()) {
4359 return Err(Error::Unauthorized);
4360 }
4361 loop {
4362 let (watch, seen) = {
4371 let mut table = self.shared.lock();
4372 if table.closed {
4373 return Err(Error::Closed);
4374 }
4375 let named = self.root.join(&path);
4376 let resolved = self.scope.resolve(&named).ok_or(BoundsExceeded)?;
4377 let watch = table.watch(&self.shared, &resolved);
4378 let seen = watch.seen();
4379 (watch, seen)
4380 };
4381 match self.request_broadcast(&path).await {
4382 Ok(broadcast) => return Ok(broadcast),
4383 Err(Error::Unroutable) => {
4384 kio::wait(|waiter| watch.poll_changed(waiter, seen)).await;
4385 }
4386 Err(Error::Dropped) if self.shared.lock().closed => return Err(Error::Closed),
4389 Err(err) => return Err(err),
4390 }
4391 }
4392 }
4393
4394 pub fn scope(&self, root: impl AsPath, patterns: &Patterns) -> Result<Consumer, Error> {
4401 let root = self.root.join(root).to_owned();
4402 let rooted = patterns.rooted(root.as_str()).map_err(|_| BoundsExceeded)?;
4403 let scope = self.scope.narrow(&rooted).ok_or(Error::Unauthorized)?;
4404 Ok(Consumer {
4405 scope,
4406 root,
4407 ..self.clone()
4408 })
4409 }
4410
4411 pub fn request_broadcast(&self, path: impl AsPath) -> kio::Pending<Requesting> {
4434 let path = path.as_path();
4435
4436 let named = self.root.join(&path).to_owned();
4440 let scope = self.stats.egress(&named);
4441 let requested = path.to_owned();
4445
4446 if !self.scope.permits(&named) {
4448 return kio::Pending::new(Requesting::failed(Error::Unauthorized));
4449 }
4450
4451 let Some(absolute) = self.scope.resolve(&named).map(|path| path.to_owned()) else {
4454 return kio::Pending::new(Requesting::failed(BoundsExceeded.into()));
4455 };
4456
4457 let mut state = self.shared.lock();
4458
4459 if state.closed {
4461 return kio::Pending::new(Requesting::failed(Error::Closed));
4462 }
4463
4464 let horizon = self.horizon.effective(&state.routes, &absolute.as_path());
4465
4466 if state
4470 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4471 .is_none()
4472 {
4473 return kio::Pending::new(Requesting::failed(Error::Unroutable));
4474 }
4475
4476 let key = (absolute.clone(), horizon);
4484 if let Some(front) = state.fronts.get(&key) {
4485 let pin = *front.pin.lock();
4486 let current = state
4487 .best_route(&absolute.as_path(), horizon, Pin::Any, &HashSet::new())
4488 .is_some_and(|entry| entry.qualifies(pin));
4489 if current {
4490 let pending = Requesting::queued(front.request.consume())
4491 .with_path(requested)
4492 .with_stats(scope);
4493 return kio::Pending::new(pending);
4494 }
4495 state.fronts.remove(&key);
4496 }
4497
4498 let broadcast = broadcast::Producer::new_spliced(broadcast::Info {
4503 pool: self.pool.clone(),
4504 cache_duration: self.cache_duration,
4505 path: absolute.clone(),
4506 });
4507 let request = kio::Producer::<PendingBroadcast>::default();
4508 let consumer = request.consume();
4509 let watch = state.watch(&self.shared, &absolute);
4510 let pin = kio::Lock::new(Pin::Any);
4511 state.fronts.insert(
4512 key,
4513 RemoteFront {
4514 request: request.clone(),
4515 broadcast: broadcast.consume().weak(),
4516 pin: pin.clone(),
4517 },
4518 );
4519 drop(state);
4522 self.tasks.push(run_front(FrontTask {
4523 shared: self.shared.clone(),
4524 broadcast,
4525 path: absolute,
4526 horizon,
4527 watch,
4528 request,
4529 pin,
4530 timers: self.timers.clone(),
4531 }));
4532 kio::Pending::new(Requesting::queued(consumer).with_path(requested).with_stats(scope))
4533 }
4534
4535 pub fn root(&self) -> &Path<'_> {
4537 &self.root
4538 }
4539
4540 pub fn allowed(&self) -> Patterns {
4542 self.scope.relative(&self.root)
4543 }
4544
4545 pub fn absolute(&self, path: impl AsPath) -> Path<'_> {
4547 self.root.join(path)
4548 }
4549}
4550
4551pub struct AnnounceConsumer {
4556 ids: Vec<ConsumerId>,
4559 shared: kio::Shared<OriginState>,
4560 root: PathOwned,
4561
4562 state: kio::Producer<OriginConsumerState>,
4565
4566 stats: stats::Session,
4569
4570 guards: HashMap<PathOwned, stats::Announce>,
4574
4575 park: kio::Park,
4578
4579 update_hold: Duration,
4582 held: HashMap<PathOwned, Held>,
4583 retracted: HashMap<PathOwned, Quarantine>,
4586 anchoring: VecDeque<(Instant, PathOwned)>,
4587 expiring: VecDeque<(Instant, PathOwned)>,
4588 timers: Clock,
4589 hold: crate::runtime::Deadline<Clock>,
4590}
4591
4592impl AnnounceConsumer {
4593 fn new(
4594 root: PathOwned,
4595 cursors: Vec<TableCursor>,
4596 state: kio::Producer<OriginConsumerState>,
4597 stats: stats::Session,
4598 shared: &kio::Shared<OriginState>,
4599 timers: Clock,
4600 ) -> Self {
4601 let mut ids = Vec::with_capacity(cursors.len());
4602 let update_hold;
4603 {
4604 let mut table = shared.lock();
4605 update_hold = table.update_hold;
4606 if table.closed {
4607 if let Ok(mut state) = state.write() {
4609 state.ended = true;
4610 }
4611 } else {
4612 for cursor in cursors {
4613 let id = ConsumerId::new();
4614 table.register_cursor(id, cursor);
4615 ids.push(id);
4616 }
4617 }
4618 }
4619
4620 Self {
4621 ids,
4622 shared: shared.clone(),
4623 root,
4624 state,
4625 stats,
4626 guards: HashMap::new(),
4627 park: kio::Park::default(),
4628 update_hold,
4629 held: HashMap::new(),
4630 retracted: HashMap::new(),
4631 anchoring: VecDeque::new(),
4632 expiring: VecDeque::new(),
4633 hold: crate::runtime::Deadline::new(&timers),
4634 timers,
4635 }
4636 }
4637
4638 fn quarantine(&mut self, update: &AnnounceUpdate, now: Option<Instant>) {
4641 if update.kind.is_active() {
4642 return;
4643 }
4644 let Some(now) = now.filter(|_| !self.update_hold.is_zero()) else {
4645 return;
4646 };
4647 let publisher = update
4649 .route
4650 .hops
4651 .iter()
4652 .next()
4653 .copied()
4654 .filter(|hop| *hop != Hop::UNKNOWN);
4655 let quarantine = Quarantine {
4657 held: Held::Anchoring(now),
4658 publisher,
4659 rank: (update.route.cost, update.route.hops.len()),
4660 };
4661 let replaced = self.retracted.insert(update.prefix.clone(), quarantine);
4662 if !replaced.is_some_and(|q| q.held == Held::Anchoring(now)) {
4665 self.anchoring.push_back((now, update.prefix.clone()));
4666 }
4667 }
4668
4669 fn settle(&mut self, now: Option<Instant>) -> Option<Instant> {
4672 let now = now?;
4673 while let Some((seen, prefix)) = self.anchoring.front() {
4676 if *seen >= now {
4677 break;
4678 }
4679 if let Some(quarantine) = self.retracted.get_mut(prefix)
4680 && quarantine.held == Held::Anchoring(*seen)
4681 {
4682 quarantine.held = Held::Until(now + RETRACT_QUARANTINE);
4683 self.expiring.push_back((now + RETRACT_QUARANTINE, prefix.clone()));
4684 }
4685 self.anchoring.pop_front();
4686 }
4687 while let Some((until, prefix)) = self.expiring.front() {
4688 if *until > now {
4689 break;
4690 }
4691 if self
4692 .retracted
4693 .get(prefix)
4694 .is_some_and(|q| q.held == Held::Until(*until))
4695 {
4696 self.retracted.remove(prefix);
4697 }
4698 self.expiring.pop_front();
4699 }
4700 let anchoring = self.anchoring.front().map(|(seen, _)| *seen + Duration::from_nanos(1));
4701 earliest(anchoring, self.expiring.front().map(|(until, _)| *until))
4702 }
4703
4704 fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate {
4706 let absolute = self.root.join(&update.prefix).to_owned();
4707 if update.kind.is_active() {
4708 let scope = self.stats.egress(&absolute);
4709 self.guards
4710 .entry(update.prefix.clone())
4711 .or_insert_with(|| scope.announce());
4712 } else {
4713 self.guards.remove(&update.prefix);
4714 }
4715 update
4716 }
4717
4718 pub async fn next(&mut self) -> Option<AnnounceUpdate> {
4726 kio::wait(|waiter| self.poll_next(waiter)).await
4727 }
4728
4729 pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Option<AnnounceUpdate>> {
4735 loop {
4736 let now = self.timers.try_now();
4737 let settling = self.settle(now);
4738 let hold = self.update_hold;
4739 let held = &mut self.held;
4740 let retracted = &self.retracted;
4741 let mut ready = None;
4742 let mut wake = None;
4743 let update = match self.state.poll(waiter, |state| {
4744 if state.pending.is_empty() {
4745 return match state.ended {
4746 true => Poll::Ready(()),
4747 false => Poll::Pending,
4748 };
4749 }
4750 match state.scan(held, retracted, now, hold) {
4751 Ok(prefix) => {
4752 ready = Some(prefix);
4753 Poll::Ready(())
4754 }
4755 Err(at) => {
4756 wake = at;
4757 Poll::Pending
4758 }
4759 }
4760 }) {
4761 Poll::Ready(Ok(mut state)) => match ready {
4762 Some(prefix) => state.take_prefix(prefix),
4763 None => {
4764 state.close();
4767 None
4768 }
4769 },
4770 Poll::Ready(Err(_)) => None,
4772 Poll::Pending => {
4773 self.hold.set(earliest(wake, settling));
4774 ready!(self.hold.poll(waiter));
4775 continue;
4776 }
4777 };
4778 let Some(update) = update else {
4779 return Poll::Ready(None);
4780 };
4781 self.held.remove(&update.prefix);
4782 self.quarantine(&update, now);
4783 return Poll::Ready(Some(self.hand_out(update)));
4784 }
4785 }
4786
4787 pub fn try_next(&mut self) -> Option<AnnounceUpdate> {
4792 let now = self.timers.try_now();
4793 let settling = self.settle(now);
4794 let mut state = self.state.write().ok()?;
4795 let prefix = match state.scan(&mut self.held, &self.retracted, now, self.update_hold) {
4796 Ok(prefix) => prefix,
4797 Err(wake) => {
4798 let wake = earliest(wake, settling);
4799 drop(state);
4802 self.hold.set(wake);
4803 return None;
4804 }
4805 };
4806 self.held.remove(&prefix);
4807 let update = state.take_prefix(prefix)?;
4808 drop(state);
4809 self.quarantine(&update, now);
4810 Some(self.hand_out(update))
4811 }
4812
4813 pub fn is_closed(&self) -> bool {
4815 let state = self.state.read();
4816 state.is_closed() || state.ended
4817 }
4818
4819 pub fn root(&self) -> &Path<'_> {
4821 &self.root
4822 }
4823
4824 pub fn absolute(&self, prefix: impl AsPath) -> Path<'_> {
4826 self.root.join(prefix)
4827 }
4828}
4829
4830impl futures::Stream for AnnounceConsumer {
4831 type Item = AnnounceUpdate;
4832
4833 fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Option<Self::Item>> {
4834 let this = self.get_mut();
4835 let waiter = this.park.hold(cx).clone();
4836 this.poll_next(&waiter)
4837 }
4838}
4839
4840impl Drop for AnnounceConsumer {
4841 fn drop(&mut self) {
4842 let mut shared = self.shared.lock();
4843 for id in &self.ids {
4844 if let Some(cursor) = shared.cursors.remove(id) {
4845 for head in &cursor.heads {
4846 shared.routes.remove_cursor(head, *id);
4847 }
4848 }
4849 }
4850 }
4851}
4852
4853#[cfg(test)]
4854use futures::FutureExt;
4855
4856#[cfg(test)]
4857#[allow(missing_docs)] impl AnnounceConsumer {
4859 pub fn assert_next_active(&mut self, expected: impl AsPath) -> Route {
4861 let expected = expected.as_path();
4862 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4863 assert_eq!(update.prefix, expected, "wrong prefix");
4864 assert!(update.kind.is_active(), "should be an active route");
4865 update.route
4866 }
4867
4868 pub fn assert_try_next_active(&mut self, expected: impl AsPath) -> Route {
4870 let expected = expected.as_path();
4871 let update = self.try_next().expect("no next");
4872 assert_eq!(update.prefix, expected, "wrong prefix");
4873 assert!(update.kind.is_active(), "should be an active route");
4874 update.route
4875 }
4876
4877 pub fn assert_next_ended(&mut self, expected: impl AsPath) {
4879 let expected = expected.as_path();
4880 let update = self.next().now_or_never().expect("next blocked").expect("no next");
4881 assert_eq!(update.prefix, expected, "wrong prefix");
4882 assert_eq!(update.kind, AnnounceKind::Retracted, "should be a retraction");
4883 }
4884
4885 pub fn assert_next_wait(&mut self) {
4886 if let Some(res) = self.next().now_or_never() {
4887 panic!("next should block: got {:?}", res.map(|u| u.prefix));
4888 }
4889 }
4890}
4891
4892#[cfg(test)]
4896pub(crate) trait ProduceTest {
4897 fn produce(self) -> Producer;
4898}
4899
4900#[cfg(test)]
4901impl ProduceTest for Config {
4902 fn produce(self) -> Producer {
4903 let (producer, driver) = Producer::new(self);
4904 if tokio::runtime::Handle::try_current().is_ok() {
4905 tokio::spawn(crate::time::run(driver));
4906 } else {
4907 std::mem::forget(driver);
4910 }
4911 producer
4912 }
4913}
4914
4915#[cfg(test)]
4916impl ProduceTest for Hop {
4917 fn produce(self) -> Producer {
4918 Config::new(self).produce()
4919 }
4920}
4921
4922#[cfg(test)]
4923mod tests {
4924 use super::*;
4925 use futures::FutureExt;
4926
4927 fn origin(id: u64) -> Hop {
4928 Hop::new(id).unwrap()
4929 }
4930
4931 fn hops(ids: &[u64]) -> Hops {
4932 let mut list = Hops::new();
4933 for &id in ids {
4934 list.push(if id == 0 { Hop::UNKNOWN } else { origin(id) }).unwrap();
4935 }
4936 list
4937 }
4938
4939 fn scopes(prefixes: &[&str]) -> Patterns {
4941 prefixes
4942 .iter()
4943 .map(|prefix| Pattern::subtree(prefix).unwrap())
4944 .collect()
4945 }
4946
4947 #[test]
4948 fn default_config_mints_a_real_hop() {
4949 let config = Config::default();
4950 assert_ne!(config.hop, Hop::UNKNOWN);
4951 let (producer, _driver) = Producer::new(config.clone());
4952 assert_eq!(producer.hop(), config.hop);
4953 assert_eq!(producer.consume().hop(), config.hop);
4954 }
4955
4956 #[test]
4957 fn random_hops_fit_legacy_lite_clients() {
4958 for _ in 0..32 {
4959 assert!(Hop::random().id() < 1u64 << 53);
4960 }
4961 }
4962
4963 async fn settle(mut check: impl FnMut() -> bool) {
4966 for _ in 0..100 {
4967 if check() {
4968 return;
4969 }
4970 tokio::task::yield_now().await;
4971 }
4972 panic!("condition never settled");
4973 }
4974
4975 async fn queued(server: &Dynamic) -> Request {
4977 let mut request = None;
4978 settle(|| match server.poll_requested_broadcast(&kio::Waiter::noop()) {
4979 Poll::Ready(Ok(popped)) => {
4980 request = Some(popped);
4981 true
4982 }
4983 _ => false,
4984 })
4985 .await;
4986 request.unwrap()
4987 }
4988
4989 async fn next_group(subscription: &mut crate::track::Subscriber) -> Result<Option<crate::group::Consumer>, Error> {
4991 let mut next = None;
4992 settle(|| match subscription.poll_recv_group(&kio::Waiter::noop()) {
4993 Poll::Ready(result) => {
4994 next = Some(result);
4995 true
4996 }
4997 Poll::Pending => false,
4998 })
4999 .await;
5000 next.unwrap()
5001 }
5002
5003 #[tokio::test]
5004 async fn announce_and_retract() {
5005 let producer = origin(1).produce();
5006 let consumer = producer.consume();
5007 let mut announced = consumer.announced();
5008 announced.assert_next_wait();
5009
5010 let announcement = producer.announce("room/alice", Route::default()).unwrap();
5011 let route = announced.assert_next_active("room/alice");
5012 assert!(route.hops.is_empty());
5013 assert_eq!(route.cost, Cost::default());
5014 announced.assert_next_wait();
5015
5016 drop(announcement);
5017 announced.assert_next_ended("room/alice");
5018 announced.assert_next_wait();
5019 }
5020
5021 #[tokio::test]
5024 async fn hidden_routes_need_an_opt_in() {
5025 let producer = origin(1).produce();
5026 let consumer = producer.consume();
5027 let _visible = producer.announce("room/alice", Route::default()).unwrap();
5028 let _stats = producer.announce(".stats/node", Route::default()).unwrap();
5029 let _nested = producer.announce("room/.internal", Route::default()).unwrap();
5030 let _suffix = producer.announce("room/catalog.pro", Route::default()).unwrap();
5032
5033 let mut announced = consumer.announced();
5034 announced.assert_next_active("room/alice");
5035 announced.assert_next_active("room/catalog.pro");
5036 announced.assert_next_wait();
5037
5038 let mut announced = consumer.clone().with_hidden(true).announced();
5039 announced.assert_next_active(".stats/node");
5040 announced.assert_next_active("room/.internal");
5041 announced.assert_next_active("room/alice");
5042 announced.assert_next_active("room/catalog.pro");
5043 announced.assert_next_wait();
5044
5045 let mut announced = consumer
5047 .scope(".stats", &Patterns::from(Pattern::all()))
5048 .unwrap()
5049 .announced();
5050 announced.assert_next_active("node");
5051 announced.assert_next_wait();
5052 let mut announced = consumer.scope("", &scopes(&["room/.internal"])).unwrap().announced();
5053 announced.assert_next_active("room/.internal");
5054 announced.assert_next_wait();
5055
5056 let mut announced = consumer.clone().with_hidden(true).beyond(&consumer).announced();
5059 announced.assert_next_active(".stats/node");
5060 announced.assert_next_active("room/.internal");
5061 announced.assert_next_wait();
5062 let room = consumer.scope("", &scopes(&["room"])).unwrap().beyond(&consumer);
5063 let mut announced = room.announced();
5064 announced.assert_next_wait();
5065 let stats = consumer.scope("", &scopes(&[".stats"])).unwrap().beyond(&consumer);
5066 let mut announced = stats.announced();
5067 announced.assert_next_active(".stats/node");
5068 announced.assert_next_wait();
5069 }
5070
5071 #[tokio::test]
5073 async fn hidden_broadcast_resolves_by_path() {
5074 let producer = origin(1).produce();
5075 let consumer = producer.consume();
5076 let broadcast = producer.create_broadcast(".stats/node").unwrap();
5077 broadcast.announce(Route::default()).unwrap();
5078
5079 consumer.announced().assert_next_wait();
5080 let resolved = consumer.request_broadcast(".stats/node").await.expect("resolves");
5081 assert_eq!(resolved.info().path.as_str(), ".stats/node");
5082 }
5083
5084 fn mounted(producer: &Producer, pid: &str, patterns: &[&str]) -> Consumer {
5086 let patterns: Patterns = patterns.iter().map(|pattern| pattern.parse().unwrap()).collect();
5087 producer
5088 .mount(format!("{pid}/.svc"), format!(".svc/{pid}"))
5089 .unwrap()
5090 .scope(pid, &patterns)
5091 .unwrap()
5092 .consume()
5093 }
5094
5095 #[tokio::test]
5098 async fn mount_resolves_on_the_target_front() {
5099 let producer = origin(1).produce();
5100 let server = producer.dynamic(".svc", Route::default()).unwrap();
5101 let project = mounted(&producer, "p1", &["**"]);
5102
5103 let through = project.request_broadcast(".svc/foo");
5104 let direct = producer.consume().request_broadcast(".svc/p1/foo");
5105
5106 let request = queued(&server).await;
5107 assert_eq!(request.path().as_str(), ".svc/p1/foo");
5108 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5109 let source = broadcast::Info::new().produce();
5110 request.accept(&source);
5111
5112 let through = through.await.expect("resolves");
5113 let direct = direct.await.expect("resolves");
5114 assert!(through.is_clone(&direct));
5115 assert_eq!(through.info().path.as_str(), ".svc/foo");
5117 }
5118
5119 #[tokio::test]
5122 async fn mount_presents_target_routes_under_the_mount() {
5123 let producer = origin(1).produce();
5124 let _claim = producer.announce(".svc", Route::default()).unwrap();
5125 let foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5126 let _other = producer.publish(".svc/p2/bar", Route::default()).unwrap();
5127 let _cam = producer.publish("p1/cam", Route::default()).unwrap();
5128 let _shadowed = producer.publish("p1/.svc/forged", Route::default()).unwrap();
5129 let project = mounted(&producer, "p1", &["**"]);
5130
5131 let mut announced = project.announced();
5133 announced.assert_next_active("cam");
5134 announced.assert_next_wait();
5135
5136 let mut announced = project.clone().with_hidden(true).announced();
5137 announced.assert_next_active(".svc");
5138 announced.assert_next_active(".svc/foo");
5139 announced.assert_next_active("cam");
5140 announced.assert_next_wait();
5141
5142 let mut inside = project
5144 .scope(".svc", &Patterns::from(Pattern::all()))
5145 .unwrap()
5146 .announced();
5147 inside.assert_next_active("");
5148 inside.assert_next_active("foo");
5149 inside.assert_next_wait();
5150
5151 drop(foo);
5152 announced.assert_next_ended(".svc/foo");
5153 inside.assert_next_ended("foo");
5154 announced.assert_next_wait();
5155
5156 let err = project.request_broadcast(".svc/forged").await.err().unwrap();
5158 assert!(matches!(err, Error::NotFound | Error::Unroutable), "{err:?}");
5159 }
5160
5161 #[tokio::test]
5164 async fn mount_authorizes_the_named_path() {
5165 let producer = origin(1).produce();
5166 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5167 let _bar = producer.publish(".svc/p1/bar", Route::default()).unwrap();
5168
5169 let granted = mounted(&producer, "p1", &["foo", ".svc/foo"]);
5170 granted.request_broadcast(".svc/foo").await.expect("granted");
5171 let refused = granted
5172 .request_broadcast(".svc/bar")
5173 .now_or_never()
5174 .expect("refused at once");
5175 assert!(matches!(refused, Err(Error::Unauthorized)));
5176 let mut announced = granted.clone().with_hidden(true).announced();
5177 announced.assert_next_active(".svc/foo");
5178 announced.assert_next_wait();
5179
5180 let narrower = mounted(&producer, "p1", &["foo"]);
5181 let refused = narrower
5182 .request_broadcast(".svc/foo")
5183 .now_or_never()
5184 .expect("refused at once");
5185 assert!(matches!(refused, Err(Error::Unauthorized)));
5186 narrower.clone().with_hidden(true).announced().assert_next_wait();
5187
5188 let other = mounted(&producer, "p2", &["**"]);
5190 other.clone().with_hidden(true).announced().assert_next_wait();
5191 let err = other.request_broadcast(".svc/foo").await.err().unwrap();
5192 assert!(matches!(err, Error::Unroutable));
5193 }
5194
5195 #[tokio::test]
5198 async fn mount_captures_the_named_path() {
5199 let producer = origin(1).produce();
5200 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5201 let project = mounted(&producer, "p1", &["**"]).with_hidden(true);
5202
5203 let update = project.announced().try_next().expect("foo");
5204 assert_eq!(update.prefix.as_str(), ".svc/foo");
5205 assert_eq!(update.captures, Some(vec![".svc/foo".parse::<Pattern>().unwrap()]));
5206
5207 let inside = project.scope(".svc", &Patterns::from(Pattern::all())).unwrap();
5208 let update = inside.announced().try_next().expect("foo");
5209 assert_eq!(update.prefix.as_str(), "foo");
5210 assert_eq!(update.captures, Some(vec!["foo".parse::<Pattern>().unwrap()]));
5211 }
5212
5213 #[tokio::test]
5216 async fn mount_keeps_a_max_depth_target() {
5217 let producer = origin(1).produce();
5218 let deep = vec!["d"; Path::MAX_PARTS].join("/");
5219 let _leaf = producer.publish(deep.as_str(), Route::default()).unwrap();
5220 let project = producer
5221 .mount("p1/.svc", deep.as_str())
5222 .unwrap()
5223 .scope("p1", &Patterns::from(Pattern::all()))
5224 .unwrap()
5225 .consume()
5226 .with_hidden(true);
5227
5228 let mut announced = project.announced();
5229 announced.assert_next_active(".svc");
5230 announced.assert_next_wait();
5231
5232 let err = project.request_broadcast(".svc/x").await.err().unwrap();
5234 assert!(matches!(err, Error::BoundsExceeded(_)), "{err:?}");
5235 let inside = project.scope(".svc/x", &Patterns::from(Pattern::all())).unwrap();
5236 inside.announced().assert_next_wait();
5237 }
5238
5239 #[tokio::test]
5242 async fn mount_bounds_the_named_path() {
5243 let producer = origin(1).produce();
5244 let _near = producer.publish("t/x", Route::default()).unwrap();
5245 let deep = format!("t/{}", vec!["d"; Path::MAX_PARTS - 1].join("/"));
5246 let _deep = producer.publish(deep.as_str(), Route::default()).unwrap();
5247 let project = producer
5248 .mount("p1/a/b", "t")
5249 .unwrap()
5250 .scope("p1", &Patterns::from(Pattern::all()))
5251 .unwrap()
5252 .consume();
5253
5254 let mut announced = project.announced();
5255 announced.assert_next_active("a/b/x");
5256 announced.assert_next_wait();
5257
5258 let over = vec!["d"; Path::MAX_PARTS + 1].join("/");
5259 assert!(matches!(
5260 producer.mount(over.as_str(), "t"),
5261 Err(Error::BoundsExceeded(_))
5262 ));
5263 }
5264
5265 #[tokio::test]
5267 async fn mount_is_read_only() {
5268 let producer = origin(1).produce();
5269 let project = producer
5270 .mount("p1/.svc", ".svc/p1")
5271 .unwrap()
5272 .scope("p1", &Patterns::from(Pattern::all()))
5273 .unwrap();
5274 assert!(matches!(project.create_broadcast(".svc/foo"), Err(Error::Unauthorized)));
5275 assert!(matches!(
5276 project.dynamic(".svc", Route::default()),
5277 Err(Error::Unauthorized)
5278 ));
5279 assert!(matches!(
5280 project.dynamic(".svc/foo", Route::default()),
5281 Err(Error::Unauthorized)
5282 ));
5283 project.publish("cam", Route::default()).unwrap();
5284 }
5285
5286 #[test]
5288 fn mount_never_widens_a_scope() {
5289 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5290 let project = producer.scope("", &scopes(&["p1"])).unwrap();
5291 assert!(matches!(project.mount("p1/.svc", ".svc/p1"), Err(Error::Unauthorized)));
5292
5293 let mounted = producer.mount("p1/.svc", ".svc/p1").unwrap();
5294 assert!(matches!(mounted.mount("p1/.svc/x", ".other"), Err(Error::Duplicate)));
5295 assert!(matches!(mounted.mount("p1", ".other"), Err(Error::Duplicate)));
5296 assert!(matches!(mounted.mount("p2", "p1/.svc/x"), Err(Error::Duplicate)));
5298 assert!(matches!(mounted.mount("p2", "p1"), Err(Error::Duplicate)));
5299 assert!(matches!(mounted.mount(".svc/p1/x", ".other"), Err(Error::Duplicate)));
5301 assert!(matches!(mounted.mount(".svc", ".other"), Err(Error::Duplicate)));
5302 mounted.mount("p1/.other", ".other/p1").unwrap();
5303 mounted.mount("p2/.svc", ".svc/p1").unwrap();
5305 for (at, target) in [("a", "a/b"), ("a/b", "a"), ("a", "a")] {
5307 assert!(
5308 matches!(producer.mount(at, target), Err(Error::Duplicate)),
5309 "{at} -> {target}"
5310 );
5311 }
5312 }
5313
5314 #[test]
5316 fn mount_refuses_a_wildcard_mount_point() {
5317 let (producer, _driver) = Producer::new(Config::new(origin(1)));
5318 for at in ["*", "p1/*", "p1/**", "p1/a*"] {
5319 assert!(
5320 matches!(producer.mount(at, ".svc/p1"), Err(Error::InvalidPath(_))),
5321 "{at}"
5322 );
5323 }
5324 }
5325
5326 #[tokio::test]
5329 async fn mount_egress_counts_under_the_named_path() {
5330 let registry = stats::Registry::new(stats::Config::new());
5331 let producer = origin(1).produce();
5332 let _foo = producer.publish(".svc/p1/foo", Route::default()).unwrap();
5333 let project = mounted(&producer, "p1", &["**"])
5334 .with_stats(registry.tier(stats::Tier::default()).session("p1"))
5335 .with_hidden(true);
5336
5337 let mut announced = project.announced();
5338 announced.assert_next_active(".svc/foo");
5339 project.request_broadcast(".svc/foo").await.expect("resolves");
5340
5341 let mut report = stats::Report::default();
5342 registry.report(&mut report);
5343 let paths: Vec<_> = report
5344 .traffic
5345 .iter()
5346 .map(|entry| entry.path.as_str().to_string())
5347 .collect();
5348 assert_eq!(paths, ["p1/.svc/foo"]);
5349 }
5350
5351 #[tokio::test]
5353 async fn hidden_route_announced_later_stays_hidden() {
5354 let producer = origin(1).produce();
5355 let consumer = producer.consume();
5356 let mut announced = consumer.announced();
5357 let mut opted = consumer.clone().with_hidden(true).announced();
5358
5359 let hidden = producer.announce(".stats/node", Route::default()).unwrap();
5360 announced.assert_next_wait();
5361 opted.assert_next_active(".stats/node");
5362
5363 drop(hidden);
5364 announced.assert_next_wait();
5365 opted.assert_next_ended(".stats/node");
5366 }
5367
5368 #[tokio::test]
5369 async fn broadcast_announces_its_own_path() {
5370 let producer = origin(1).produce();
5371 let consumer = producer.consume();
5372 let mut announced = consumer.announced();
5373 let mut peer = consumer.clone().excluding(Hop::UNKNOWN).announced();
5374
5375 let broadcast = producer.create_broadcast("room/alice").unwrap();
5377 announced.assert_next_wait();
5378 peer.assert_next_wait();
5379
5380 broadcast.announce(Route::default().with_cost(3)).unwrap();
5381 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(3));
5382 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(3));
5383
5384 broadcast.announce(Route::default().with_cost(1)).unwrap();
5386 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(1));
5387 assert_eq!(peer.assert_next_active("room/alice").cost, Cost::new(1));
5388
5389 broadcast.unannounce();
5391 announced.assert_next_ended("room/alice");
5392 peer.assert_next_ended("room/alice");
5393 broadcast.unannounce();
5394 announced.assert_next_wait();
5395 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5396 assert!(matches!(err, Error::Unroutable));
5397
5398 broadcast.announce(Route::default()).unwrap();
5400 announced.assert_next_active("room/alice");
5401 peer.assert_next_active("room/alice");
5402 broadcast.close();
5403 announced.assert_next_ended("room/alice");
5404 peer.assert_next_ended("room/alice");
5405 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5406 announced.assert_next_wait();
5407 }
5408
5409 #[tokio::test]
5410 async fn broadcast_announcement_retracts_with_the_last_producer() {
5411 let producer = origin(1).produce();
5412 let consumer = producer.consume();
5413 let mut announced = consumer.announced();
5414
5415 let broadcast = producer.create_broadcast("room/alice").unwrap();
5416 let clone = broadcast.clone();
5417 broadcast.announce(Route::default()).unwrap();
5418 announced.assert_next_active("room/alice");
5419
5420 drop(broadcast);
5422 announced.assert_next_wait();
5423 drop(clone);
5424 announced.assert_next_ended("room/alice");
5425 }
5426
5427 #[tokio::test]
5428 async fn publish_creates_and_announces_together() {
5429 let producer = origin(1).produce();
5430 let mut announced = producer.consume().announced();
5431 let _broadcast = producer.publish("room/alice", Route::default()).unwrap();
5432 announced.assert_next_active("room/alice");
5433 }
5434
5435 #[tokio::test]
5436 async fn standalone_broadcast_cannot_announce() {
5437 let broadcast = broadcast::Info::new().produce();
5438 assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed)));
5439 broadcast.unannounce();
5441 }
5442
5443 #[tokio::test]
5444 async fn announce_replays_to_late_cursor() {
5445 let producer = origin(1).produce();
5446 let _a = producer.announce("room/alice", Route::default()).unwrap();
5447 let _b = producer.announce("room/bob", Route::default()).unwrap();
5448
5449 let mut announced = producer.consume().announced();
5450 announced.assert_next_active("room/alice");
5452 announced.assert_next_active("room/bob");
5453 announced.assert_next_wait();
5454 }
5455
5456 #[tokio::test]
5457 async fn announce_keeps_its_prefix_under_a_producer_scope() {
5458 let producer = origin(1).produce();
5459 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
5460
5461 let _a = scoped.announce("", Route::default()).unwrap();
5463 let mut announced = producer.consume().announced();
5464 announced.assert_next_active("");
5465
5466 assert!(matches!(
5468 scoped.announce("other", Route::default()),
5469 Err(Error::Unauthorized)
5470 ));
5471 }
5472
5473 #[tokio::test]
5474 async fn cursor_keeps_an_overlapping_prefix_above_its_scope() {
5475 let producer = origin(1).produce();
5476 let _a = producer.announce("", Route::default()).unwrap();
5477
5478 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
5479 let mut announced = consumer.announced();
5480 announced.assert_next_active("");
5481 }
5482
5483 #[tokio::test]
5484 async fn cursor_root_strips_prefix() {
5485 let producer = origin(1).produce();
5486 let _a = producer.announce("room/alice", Route::default()).unwrap();
5487
5488 let consumer = producer
5489 .consume()
5490 .scope("room", &Patterns::from(Pattern::all()))
5491 .unwrap();
5492 let mut announced = consumer.announced();
5493 announced.assert_next_active("alice");
5494 }
5495
5496 #[tokio::test]
5497 async fn best_route_wins_and_fails_over() {
5498 let producer = origin(1).produce();
5499 let mut announced = producer.consume().announced();
5500
5501 let expensive = producer
5502 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5503 .unwrap();
5504 let route = announced.assert_next_active("room");
5505 assert_eq!(route.cost, Cost::new(5));
5506
5507 let cheap = producer
5509 .announce("room", Route::default().with_hops(hops(&[20])).with_cost(1))
5510 .unwrap();
5511 let route = announced.assert_next_active("room");
5512 assert_eq!(route.cost, Cost::new(1));
5513
5514 drop(cheap);
5516 let route = announced.assert_next_active("room");
5517 assert_eq!(route.cost, Cost::new(5));
5518
5519 drop(expensive);
5521 announced.assert_next_ended("room");
5522 }
5523
5524 #[tokio::test]
5525 async fn identical_reannounce_is_invisible() {
5526 let producer = origin(1).produce();
5527 let mut announced = producer.consume().announced();
5528
5529 let old = producer
5530 .announce("room", Route::default().with_hops(hops(&[10])))
5531 .unwrap();
5532 let first = announced.assert_next_active("room");
5533 assert_eq!(first.hops.as_slice(), hops(&[10]).as_slice());
5534
5535 let _new = producer
5539 .announce("room", Route::default().with_hops(hops(&[10])))
5540 .unwrap();
5541 announced.assert_next_wait();
5542
5543 drop(old);
5545 announced.assert_next_wait();
5546 }
5547
5548 #[tokio::test]
5549 async fn exclude_hides_routes_through_the_peer() {
5550 let producer = origin(1).produce();
5551 let _a = producer
5552 .announce("room", Route::default().with_hops(hops(&[7])))
5553 .unwrap();
5554
5555 let mut hidden = producer.consume().excluding(origin(7)).announced();
5556 hidden.assert_next_wait();
5557
5558 let mut visible = producer.consume().excluding(origin(8)).announced();
5559 visible.assert_next_active("room");
5560 }
5561
5562 #[tokio::test]
5563 async fn exclude_matches_via_when_the_chain_is_anonymous() {
5564 let producer = origin(1).produce();
5565 let assigned = origin(777);
5566 let _echoed = producer
5567 .announce("echoed", Route::default().with_hops(hops(&[0])).with_via(assigned))
5568 .unwrap();
5569 let _local = producer
5570 .announce("local", Route::default().with_hops(hops(&[10])))
5571 .unwrap();
5572
5573 let mut hidden = producer.consume().excluding(assigned).announced();
5574 hidden.assert_next_active("local");
5575 hidden.assert_next_wait();
5576 }
5577
5578 #[tokio::test]
5579 async fn anonymous_route_loses_to_identified_at_any_cost() {
5580 let producer = origin(1).produce();
5581 let mut announced = producer.consume().announced();
5582
5583 let _anonymous = producer
5584 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5585 .unwrap();
5586 let route = announced.assert_next_active("room");
5587 assert!(route.is_anonymous());
5588 assert_eq!(route.cost, Cost::new(1));
5589
5590 let _identified = producer
5591 .announce("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5592 .unwrap();
5593 let route = announced.assert_next_active("room");
5594 assert!(!route.is_anonymous());
5595 assert_eq!(route.cost, Cost::new(5));
5596 }
5597
5598 #[tokio::test]
5601 async fn a_stamped_route_ranks_below_an_identified_one() {
5602 let producer = origin(1).produce();
5603 let mut announced = producer.consume().announced();
5604
5605 let mut stamped = Hops::new();
5606 stamped.stamp(origin(5)).unwrap();
5607 assert_eq!(stamped.as_slice(), &[origin(5), Hop::UNKNOWN]);
5608
5609 let _legacy = producer
5610 .announce("room", Route::default().with_hops(stamped).with_cost(0))
5611 .unwrap();
5612 let route = announced.assert_next_active("room");
5613 assert!(route.is_anonymous());
5614
5615 let _identified = producer
5616 .announce("room", Route::default().with_hops(hops(&[10, 11])).with_cost(5))
5617 .unwrap();
5618 let route = announced.assert_next_active("room");
5619 assert_eq!(route.hops.as_slice(), hops(&[10, 11]).as_slice());
5620 }
5621
5622 #[test]
5623 fn stamping_keeps_the_leading_zero_and_names_nothing_else() {
5624 let mut chain = hops(&[0, 7]);
5625 chain.stamp(origin(5)).unwrap();
5626 assert_eq!(chain.as_slice(), hops(&[5, 0, 7]).as_slice());
5627
5628 let mut named = hops(&[7, 0]);
5630 named.stamp(origin(5)).unwrap();
5631 assert_eq!(named.as_slice(), hops(&[7, 0]).as_slice());
5632
5633 let mut full = Hops::try_from(vec![Hop::UNKNOWN; MAX_HOPS]).unwrap();
5635 assert_eq!(full.stamp(origin(5)), Err(InvalidHop::TooMany));
5636 }
5637
5638 #[tokio::test]
5639 async fn anonymous_routes_order_by_cost() {
5640 let producer = origin(1).produce();
5641 let mut announced = producer.consume().announced();
5642
5643 let expensive = producer
5644 .announce("room", Route::default().with_hops(hops(&[0])).with_cost(5))
5645 .unwrap();
5646 let route = announced.assert_next_active("room");
5647 assert_eq!(route.cost, Cost::new(5));
5648
5649 let _cheap = producer
5650 .announce("room", Route::default().with_hops(hops(&[0, 7])).with_cost(1))
5651 .unwrap();
5652 let route = announced.assert_next_active("room");
5653 assert!(route.is_anonymous());
5654 assert_eq!(route.cost, Cost::new(1));
5655
5656 drop(expensive);
5657 announced.assert_next_wait();
5658 }
5659
5660 #[tokio::test]
5661 async fn anonymous_chain_from_identified_peer_still_ranks_last() {
5662 let producer = origin(1).produce();
5663 let mut announced = producer.consume().announced();
5664
5665 let _anonymous = producer
5666 .announce(
5667 "room",
5668 Route::default()
5669 .with_hops(hops(&[0, 7]))
5670 .with_cost(1)
5671 .with_via(origin(7)),
5672 )
5673 .unwrap();
5674 announced.assert_next_active("room");
5675
5676 let _identified = producer
5677 .announce("room", Route::default().with_hops(hops(&[10, 20])).with_cost(5))
5678 .unwrap();
5679 let route = announced.assert_next_active("room");
5680 assert!(!route.is_anonymous());
5681 assert_eq!(route.cost, Cost::new(5));
5682 }
5683
5684 #[tokio::test]
5685 async fn request_prefers_identified_over_cheaper_anonymous() {
5686 let producer = origin(1).produce();
5687 let consumer = producer.consume();
5688
5689 let anonymous = producer
5690 .dynamic("room", Route::default().with_hops(hops(&[0])).with_cost(1))
5691 .unwrap();
5692 let identified = producer
5693 .dynamic("room", Route::default().with_hops(hops(&[10])).with_cost(5))
5694 .unwrap();
5695
5696 let _pending = consumer.request_broadcast("room/alice");
5697 let request = queued(&identified).await;
5698 assert_eq!(request.path().as_str(), "room/alice");
5699 assert!(
5700 anonymous.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
5701 "the cheaper anonymous route must not serve"
5702 );
5703 }
5704
5705 #[tokio::test]
5706 async fn update_reprices_in_place() {
5707 let producer = origin(1).produce();
5708 let mut announced = producer.consume().announced();
5709
5710 let announcement = producer.announce("room", Route::default()).unwrap();
5711 announced.assert_next_active("room");
5712
5713 announcement.update(Route::default().with_cost(9)).unwrap();
5714 let route = announced.assert_next_active("room");
5715 assert_eq!(route.cost, Cost::new(9));
5716 }
5717
5718 #[tokio::test]
5719 async fn retract_after_undelivered_reprice_still_delivered() {
5720 let producer = origin(1).produce();
5721 let mut announced = producer.consume().announced();
5722
5723 let announcement = producer.announce("room", Route::default()).unwrap();
5724 announced.assert_next_active("room");
5725
5726 announcement.update(Route::default().with_cost(9)).unwrap();
5730 drop(announcement);
5731 announced.assert_next_ended("room");
5732 announced.assert_next_wait();
5733 }
5734
5735 #[tokio::test]
5736 async fn scoped_cursor_advertises_most_specific_covering_route() {
5737 let producer = origin(1).produce();
5738 let _broad = producer.announce("room", Route::default().with_cost(1)).unwrap();
5741 let _narrow = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
5742
5743 let consumer = producer
5744 .consume()
5745 .scope("room/alice", &Patterns::from(Pattern::all()))
5746 .unwrap();
5747 let mut announced = consumer.announced();
5748 let route = announced.assert_next_active("");
5749 assert_eq!(route.cost, Cost::new(9));
5750 announced.assert_next_wait();
5751 }
5752
5753 #[tokio::test]
5754 async fn capture_change_retracts_before_reannouncing_a_presented_prefix() {
5755 let producer = origin(1).produce();
5756 let _broad = producer.announce("room", Route::default()).unwrap();
5757 let exact = producer.announce("room/alice", Route::default()).unwrap();
5758 let consumer = producer
5759 .consume()
5760 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
5761 .unwrap()
5762 .scope("room/alice", &Patterns::from(Pattern::all()))
5763 .unwrap();
5764 let mut announced = consumer.announced();
5765
5766 let first = announced.next().now_or_never().expect("next").expect("announce");
5767 assert_eq!(first.prefix.as_str(), "");
5768 assert_eq!(first.kind, AnnounceKind::Announced);
5769 assert_eq!(first.captures, Some(Vec::new()));
5770
5771 drop(exact);
5772 let retracted = announced.next().now_or_never().expect("next").expect("retract");
5773 assert_eq!(retracted.prefix.as_str(), "");
5774 assert_eq!(retracted.kind, AnnounceKind::Retracted);
5775 assert_eq!(retracted.captures, Some(Vec::new()));
5776 let replacement = announced.next().now_or_never().expect("next").expect("announce");
5777 assert_eq!(replacement.prefix.as_str(), "");
5778 assert_eq!(replacement.kind, AnnounceKind::Announced);
5779 assert_eq!(replacement.captures, None);
5780 }
5781
5782 #[tokio::test]
5783 async fn routed_broadcast_resolves_once_announced() {
5784 let producer = origin(1).produce();
5785 let consumer = producer.consume();
5786
5787 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
5789 assert!((&mut resolving).now_or_never().is_none());
5790
5791 let broadcast = producer.create_broadcast("room/alice").unwrap();
5793 for _ in 0..20 {
5794 tokio::task::yield_now().await;
5795 }
5796 assert!((&mut resolving).now_or_never().is_none());
5797
5798 broadcast.announce(Route::default()).unwrap();
5799 let resolved = resolving.await.expect("resolves once announced");
5800 assert_eq!(resolved.info().path.as_str(), "room/alice");
5801 drop(broadcast);
5802 }
5803
5804 #[tokio::test]
5807 async fn cheaper_remote_route_beats_a_local_broadcast() {
5808 let producer = origin(1).produce();
5809 let consumer = producer.consume();
5810 let mut announced = consumer.announced();
5811
5812 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5813 assert_eq!(announced.assert_next_active("room/alice").cost, Cost::new(5));
5814
5815 let server = producer
5816 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5817 .unwrap();
5818 let route = announced.assert_next_active("room/alice");
5819 assert_eq!(route.cost, Cost::new(1));
5820 assert_eq!(route.hops, hops(&[10]));
5821
5822 let pending = consumer.request_broadcast("room/alice");
5824 let request = queued(&server).await;
5825 let upstream = broadcast::Info::new().produce();
5826 request.accept(&upstream);
5827 pending.await.expect("resolves through the cheaper route");
5828 }
5829
5830 #[tokio::test]
5834 async fn cheaper_route_after_a_front_wins_new_requests() {
5835 let producer = origin(1).produce();
5836 let consumer = producer.consume();
5837
5838 let _local = producer.publish("room/alice", Route::default().with_cost(5)).unwrap();
5839 let first = consumer
5840 .request_broadcast("room/alice")
5841 .await
5842 .expect("resolves locally");
5843
5844 let server = producer
5845 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(1))
5846 .unwrap();
5847
5848 let pending = consumer.request_broadcast("room/alice");
5849 let request = queued(&server).await;
5850 let upstream = broadcast::Info::new().produce();
5851 request.accept(&upstream);
5852 let second = pending.await.expect("resolves through the cheaper route");
5853
5854 assert!(!first.is_closed(), "the old front must keep serving its readers");
5855 assert!(!first.is_clone(&second), "the newcomer must not join the old front");
5856 }
5857
5858 #[tokio::test]
5862 async fn local_broadcast_wins_a_tie_with_a_hopless_route() {
5863 let producer = origin(1).produce();
5864 let consumer = producer.consume();
5865
5866 let _local = producer.publish("room/alice", Route::default()).unwrap();
5867 let server = producer.dynamic("room/alice", Route::default()).unwrap();
5868
5869 let resolved = tokio::time::timeout(Duration::from_secs(1), consumer.request_broadcast("room/alice"))
5871 .await
5872 .expect("the newer hopless route won the tie")
5873 .expect("resolves");
5874 assert_eq!(resolved.info().path.as_str(), "room/alice");
5875 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5876 }
5877
5878 #[tokio::test]
5881 async fn announce_stats_follow_the_advertisement() {
5882 let registry = stats::Registry::new(stats::Config::new());
5883 let producer = origin(1)
5884 .produce()
5885 .with_stats(registry.tier(stats::Tier::default()).session("root"));
5886 let announces = || {
5887 registry
5888 .snapshot()
5889 .traffic()
5890 .into_iter()
5891 .find(|(_, role, _)| *role == stats::Role::Subscriber)
5892 .map(|(_, _, traffic)| (traffic.announces_started, traffic.announces_ended))
5893 .unwrap_or_default()
5894 };
5895
5896 let broadcast = producer.create_broadcast("room/alice").unwrap();
5897 assert_eq!(announces(), (0, 0), "a hidden broadcast is not announced");
5898 broadcast.announce(Route::default()).unwrap();
5899 broadcast
5900 .announce(Route {
5901 cost: Cost::new(3),
5902 ..Route::default()
5903 })
5904 .unwrap();
5905 assert_eq!(announces(), (1, 0), "a re-price is not another announce");
5906 broadcast.unannounce();
5907 assert_eq!(announces(), (1, 1));
5908 broadcast.announce(Route::default()).unwrap();
5909 drop(broadcast);
5910 assert_eq!(announces(), (2, 2));
5911 }
5912
5913 #[tokio::test]
5915 async fn local_broadcast_wins_a_cost_tie() {
5916 let producer = origin(1).produce();
5917 let consumer = producer.consume();
5918 let mut announced = consumer.announced();
5919
5920 let server = producer
5921 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(2))
5922 .unwrap();
5923 announced.assert_next_active("room/alice");
5924 let _local = producer.publish("room/alice", Route::default().with_cost(2)).unwrap();
5925 assert!(announced.assert_next_active("room/alice").hops.is_empty());
5926
5927 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5928 assert_eq!(resolved.info().path.as_str(), "room/alice");
5929 for _ in 0..20 {
5930 tokio::task::yield_now().await;
5931 }
5932 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
5933 }
5934
5935 #[tokio::test]
5938 async fn unannounce_ends_the_front() {
5939 let producer = origin(1).produce();
5940 let consumer = producer.consume();
5941
5942 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5943 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5944
5945 broadcast.unannounce();
5946 let err = consumer.request_broadcast("room/alice").await.err().unwrap();
5947 assert!(matches!(err, Error::Unroutable), "joined a retracted front: {err}");
5948 settle(|| resolved.is_closed()).await;
5949 assert!(!broadcast.consume().is_closed(), "the broadcast itself lives on");
5950
5951 broadcast.announce(Route::default()).unwrap();
5953 let again = consumer.request_broadcast("room/alice").await.expect("resolves again");
5954 assert!(!again.is_clone(&resolved));
5955 }
5956
5957 #[tokio::test]
5961 async fn unannounce_keeps_a_track_awaiting_its_info() {
5962 let producer = origin(1).produce();
5963 let consumer = producer.consume();
5964
5965 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
5966 let mut dynamic = broadcast.dynamic();
5967 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
5968 let track = resolved.track("video").unwrap();
5969 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
5970 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
5971 .await
5972 .expect("the front asked the source")
5973 .expect("request");
5974
5975 broadcast.unannounce();
5976 settle(|| resolved.is_closed()).await;
5977
5978 let source = request.accept(None);
5979 let mut group = source.append_group().unwrap();
5980 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
5981 group.finish().unwrap();
5982 source.finish().unwrap();
5983
5984 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
5985 let mut group = subscription.recv_group().await.unwrap().expect("the source's group");
5986 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"late");
5987 assert!(matches!(subscription.recv_group().await, Ok(None)), "ends cleanly");
5988 }
5989
5990 async fn served_front() -> (Dynamic, broadcast::Producer, broadcast::Dynamic, broadcast::Consumer) {
5993 let producer = origin(1).produce();
5994 let consumer = producer.consume();
5995 let server = producer
5996 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
5997 .unwrap();
5998 let pending = consumer.request_broadcast("room/alice");
5999 let upstream = broadcast::Info::new().produce();
6000 let dynamic = upstream.dynamic();
6001 queued(&server).await.accept(&upstream);
6002 let resolved = pending.await.expect("resolves");
6003 (server, upstream, dynamic, resolved)
6004 }
6005
6006 #[tokio::test]
6011 async fn returning_reader_skips_a_warm_cache_the_copy_resolved_past() {
6012 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
6013 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6014 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
6015
6016 let track = resolved.track("audio").unwrap();
6017 let b = budget.clone();
6018 let subscribing = tokio::spawn(async move { track.subscribe(b).await });
6019 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6020 .await
6021 .expect("the front asked the source")
6022 .expect("request");
6023 let source = request.resolving_start().accept(None);
6024 for seq in 0..4u64 {
6025 let mut group = source.create_group(seq.into()).unwrap();
6026 group.write_frame(ms(seq * 20), b"old".as_ref()).unwrap();
6027 group.finish().unwrap();
6028 }
6029 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6030 subscription.recv_group().await.unwrap().expect("the live group");
6031 drop(subscription);
6032 tokio::time::timeout(Duration::from_secs(1), source.unused())
6033 .await
6034 .expect("parked")
6035 .expect("source open");
6036 drop(source);
6037
6038 let track = resolved.track("audio").unwrap();
6039 let subscribing = tokio::spawn(async move { track.subscribe(budget).await });
6040 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6041 .await
6042 .expect("the front asked the source again")
6043 .expect("request");
6044 let mut source = request.resolving_start().accept(None);
6045 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6046
6047 assert!(
6049 tokio::time::timeout(Duration::from_millis(50), subscription.recv_group())
6050 .await
6051 .is_err(),
6052 "the warm cache was served before the copy resolved its start"
6053 );
6054
6055 source.start_at(20).unwrap();
6057 let mut group = source.create_group(20u64.into()).unwrap();
6058 group.write_frame(ms(2000), b"new".as_ref()).unwrap();
6059 group.finish().unwrap();
6060 let group = subscription.recv_group().await.unwrap().expect("the live group");
6061 assert_eq!(group.sequence, 20, "a stale warm group was served");
6062 }
6063
6064 #[tokio::test]
6069 async fn returning_reader_replays_a_current_warm_cache() {
6070 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6071
6072 let track = resolved.track("catalog").unwrap();
6073 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6074 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6075 .await
6076 .expect("the front asked the source")
6077 .expect("request");
6078 let source = request.resolving_start().accept(None);
6079 let mut group = source.create_group(0u64.into()).unwrap();
6080 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6081 group.finish().unwrap();
6082 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6083 subscription.recv_group().await.unwrap().expect("the catalog");
6084 drop(subscription);
6085 tokio::time::timeout(Duration::from_secs(1), source.unused())
6086 .await
6087 .expect("parked")
6088 .expect("source open");
6089 drop(source);
6090
6091 let track = resolved.track("catalog").unwrap();
6092 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6093 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6094 .await
6095 .expect("the front asked the source again")
6096 .expect("request");
6097 let mut source = request.resolving_start().accept(None);
6098 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
6099
6100 let reading = tokio::spawn(async move {
6103 let mut group = subscription.recv_group().await.unwrap().expect("the catalog");
6104 assert_eq!(group.sequence, 0);
6105 group.read_frame().await.unwrap().expect("the snapshot").payload
6106 });
6107 tokio::task::yield_now().await;
6108 assert_eq!(
6109 source.subscription().and_then(|sub| sub.start),
6110 Some(track::Position { group: 0, frame: 1 }),
6111 "the re-splice asked past the cached catalog"
6112 );
6113 source.start_at(0).unwrap();
6114 let mut tail = source.create_group(0u64.into()).unwrap();
6115 tail.start_at(1).unwrap();
6116 tail.finish().unwrap();
6117 let payload = tokio::time::timeout(Duration::from_secs(1), reading)
6118 .await
6119 .expect("the returning reader never got the catalog")
6120 .unwrap();
6121 assert_eq!(&payload[..], b"snapshot");
6122 }
6123
6124 #[tokio::test(start_paused = true)]
6128 async fn finished_track_is_forgotten_after_the_linger() {
6129 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6130
6131 let track = resolved.track("catalog").unwrap();
6132 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6133 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6134 .await
6135 .expect("the front asked the source")
6136 .expect("request");
6137 let source = request.resolving_start().accept(None);
6138 let mut group = source.create_group(0u64.into()).unwrap();
6139 group.write_frame(crate::Timestamp::ZERO, b"snapshot".as_ref()).unwrap();
6140 group.finish().unwrap();
6141 source.finish().unwrap();
6142 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6143 assert_eq!(
6144 next_group(&mut subscription)
6145 .await
6146 .unwrap()
6147 .expect("the catalog")
6148 .sequence,
6149 0
6150 );
6151 assert!(next_group(&mut subscription).await.unwrap().is_none());
6152 drop(subscription);
6153 drop(source);
6154
6155 let mut subscription = resolved.track("catalog").unwrap().subscribe(None).await.unwrap();
6157 assert_eq!(
6158 next_group(&mut subscription)
6159 .await
6160 .unwrap()
6161 .expect("the catalog")
6162 .sequence,
6163 0
6164 );
6165 assert!(next_group(&mut subscription).await.unwrap().is_none());
6166 drop(subscription);
6167 assert!(
6168 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6169 .await
6170 .is_err(),
6171 "a finished track within the linger asked the source again"
6172 );
6173
6174 tokio::time::sleep(TRACK_IDLE_LINGER).await;
6176
6177 let track = resolved.track("catalog").unwrap();
6178 let _subscribing = tokio::spawn(async move { track.subscribe(None).await });
6179 tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6180 .await
6181 .expect("the finished track outlived the linger")
6182 .expect("request");
6183 }
6184
6185 #[tokio::test]
6190 async fn returning_reader_continues_an_open_warm_group() {
6191 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6192
6193 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6194 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6195 .await
6196 .expect("frame")
6197 .unwrap()
6198 .expect("group ended");
6199 frame.payload.to_vec()
6200 }
6201
6202 let mut expect: Vec<&[u8]> = Vec::new();
6203 let mut floor: Option<track::Position> = None;
6204 for (round, payload) in [b"a".as_ref(), b"b", b"c"].into_iter().enumerate() {
6205 let track = resolved.track("log").unwrap();
6206 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6207 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6208 .await
6209 .expect("the front asked the source")
6210 .expect("request");
6211 let mut source = request.resolving_start().accept(None);
6212 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6213
6214 source.start_at(0).unwrap();
6216 let mut group = source.create_group(0u64.into()).unwrap();
6217 if let Some(floor) = floor {
6218 group.start_at(floor.frame).unwrap();
6219 }
6220 group.write_frame(crate::Timestamp::ZERO, payload).unwrap();
6221 expect.push(payload);
6222 source
6223 .insert_datagram(10, crate::Timestamp::ZERO, b"datagram".as_ref())
6224 .unwrap();
6225
6226 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6227 .await
6228 .expect("group 0")
6229 .unwrap()
6230 .expect("track ended");
6231 assert_eq!(reading.sequence, 0);
6232 for frame in &expect {
6233 assert_eq!(read(&mut reading).await, *frame, "round {round}");
6234 }
6235 assert_eq!(
6236 source.subscription().and_then(|sub| sub.start),
6237 floor,
6238 "round {round} asked for the wrong continuation"
6239 );
6240
6241 drop(reading);
6242 drop(subscription);
6243 tokio::time::timeout(Duration::from_secs(1), source.unused())
6244 .await
6245 .expect("parked")
6246 .expect("source open");
6247 drop(group);
6248 drop(source);
6249 floor = Some(track::Position {
6250 group: 0,
6251 frame: expect.len() as u64,
6252 });
6253 }
6254 }
6255
6256 #[tokio::test]
6263 async fn a_takeover_keeps_the_open_group_head_for_later_readers() {
6264 tokio::time::pause();
6265 for dies_first in [true, false] {
6266 takeover_keeps_the_open_group_head(dies_first).await;
6267 }
6268 }
6269
6270 async fn takeover_keeps_the_open_group_head(dies_first: bool) {
6271 let producer = origin(1).produce();
6272 let first_server = producer
6273 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6274 .unwrap();
6275 let pending = producer.consume().request_broadcast("room/alice");
6276 let upstream = broadcast::Info::new().produce();
6277 let mut dynamic = upstream.dynamic();
6278 queued(&first_server).await.accept(&upstream);
6279 let resolved = pending.await.unwrap();
6280
6281 async fn read(group: &mut group::Consumer) -> Vec<u8> {
6282 let frame = tokio::time::timeout(Duration::from_secs(1), group.read_frame())
6283 .await
6284 .expect("frame")
6285 .unwrap()
6286 .expect("group ended");
6287 frame.payload.to_vec()
6288 }
6289
6290 let track = resolved.track("log").unwrap();
6292 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6293 let source = dynamic.requested_track().await.unwrap().accept(None);
6294 let mut group = source.create_group(0u64.into()).unwrap();
6295 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6296 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6297 let mut subscription = subscribing.await.unwrap().unwrap();
6298 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6299 assert_eq!(read(&mut reading).await, b"a");
6300 assert_eq!(read(&mut reading).await, b"b");
6301
6302 let mut first = Some((group, source, upstream, dynamic));
6305 let mut joining = None;
6306 if dies_first {
6307 drop(first.take());
6308 assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none());
6311 let track = resolved.track("log").unwrap();
6312 joining = Some(tokio::spawn(async move {
6313 let mut joined = track.subscribe(None).await.unwrap();
6314 let group = joined.recv_group().await.unwrap().expect("track ended");
6315 (joined, group)
6316 }));
6317 }
6318 let cost = if dies_first { 5 } else { 0 };
6319 let replacement_server = producer
6320 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(cost))
6321 .unwrap();
6322 let replacement = broadcast::Info::new().produce();
6323 let mut replacement_dynamic = replacement.dynamic();
6324 queued(&replacement_server).await.accept(&replacement);
6325 let mut next = tokio::time::timeout(Duration::from_secs(1), replacement_dynamic.requested_track())
6326 .await
6327 .expect("the front asked the replacement")
6328 .unwrap()
6329 .resolving_start()
6330 .accept(None);
6331 next.start_at(0).unwrap();
6332 let mut next_group = next.create_group(0u64.into()).unwrap();
6333 next_group.start_at(2).unwrap();
6334 next_group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6335 assert_eq!(read(&mut reading).await, b"c", "dies_first={dies_first}");
6336
6337 if let Some((group, ..)) = &mut first {
6339 let mut independent = group.consume();
6340 group.write_frame(crate::Timestamp::ZERO, b"x".as_ref()).unwrap();
6341 for expect in [b"a", b"b", b"x"] {
6342 assert_eq!(read(&mut independent).await, expect);
6343 }
6344 group.finish().unwrap();
6345 }
6346 drop(first);
6347
6348 let track = resolved.track("log").unwrap();
6349 let mut later = tokio::time::timeout(Duration::from_secs(1), track.subscribe(None))
6350 .await
6351 .expect("subscribe")
6352 .unwrap();
6353 let mut late = tokio::time::timeout(Duration::from_secs(1), later.recv_group())
6354 .await
6355 .unwrap_or_else(|_| panic!("dies_first={dies_first}: the later reader never got the group"))
6356 .unwrap()
6357 .expect("track ended");
6358 assert_eq!(late.sequence, 0);
6359 for expect in [b"a", b"b", b"c"] {
6360 assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}");
6361 }
6362 let mut joined = None;
6363 if let Some(joining) = joining {
6364 let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining)
6365 .await
6366 .expect("the reader joining mid-outage never got the group")
6367 .unwrap();
6368 assert_eq!(group.sequence, 0);
6369 for expect in [b"a", b"b", b"c"] {
6370 assert_eq!(read(&mut group).await, expect);
6371 }
6372 joined = Some((subscription, group));
6373 }
6374
6375 assert!(
6377 tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6378 .await
6379 .is_err(),
6380 "dies_first={dies_first}: the group was handed out twice"
6381 );
6382
6383 next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap();
6385 assert_eq!(read(&mut reading).await, b"d");
6386 assert_eq!(read(&mut late).await, b"d");
6387 if let Some((_, group)) = &mut joined {
6388 assert_eq!(read(group).await, b"d");
6389 }
6390 next_group.finish().unwrap();
6391 drop((
6392 joined,
6393 reading,
6394 late,
6395 subscription,
6396 later,
6397 next,
6398 replacement,
6399 replacement_server,
6400 first_server,
6401 ));
6402 }
6403
6404 #[tokio::test]
6409 async fn returning_reader_gets_a_live_open_warm_group() {
6410 use futures::FutureExt;
6411
6412 tokio::time::pause();
6413 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
6414
6415 let track = resolved.track("video").unwrap();
6417 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6418 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6419 .await
6420 .expect("the front asked the source")
6421 .expect("request");
6422 let mut source = request.resolving_start().accept(None);
6423 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6424 source.start_at(0).unwrap();
6425 let mut group = source.create_group(0u64.into()).unwrap();
6426 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6427 let mut reading = subscription.recv_group().await.unwrap().expect("group 0");
6428 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6429 drop(reading);
6430 drop(subscription);
6431 tokio::time::timeout(Duration::from_secs(1), source.unused())
6432 .await
6433 .expect("parked")
6434 .expect("source open");
6435 drop(group);
6436 drop(source);
6437
6438 let track = resolved.track("video").unwrap();
6440 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6441 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6442 .await
6443 .expect("the front asked the source")
6444 .expect("request");
6445 let mut source = request.resolving_start().accept(None);
6446 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6447 source.start_at(0).unwrap();
6448 let mut group = source.create_group(0u64.into()).unwrap();
6449 group.start_at(1).unwrap();
6450 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6451
6452 let mut reading = tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
6453 .await
6454 .expect("group 0")
6455 .unwrap()
6456 .expect("track ended");
6457 assert_eq!(reading.sequence, 0);
6458 let early = reading.finished().now_or_never();
6459 assert!(
6460 early.is_none(),
6461 "the live group already resolved its end before a frame was read: {early:?}"
6462 );
6463
6464 group.finish().unwrap();
6466 assert_eq!(reading.finished().await.unwrap(), 0);
6467 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"a");
6468 assert_eq!(reading.finished().await.unwrap(), 1);
6469 assert_eq!(reading.read_frame().await.unwrap().unwrap().payload.as_ref(), b"b");
6470 assert_eq!(reading.finished().await.unwrap(), 2);
6471 }
6472
6473 #[tokio::test]
6474 async fn warm_head_survives_another_takeover_before_park() {
6475 tokio::time::pause();
6476 let producer = origin(1).produce();
6477 let _server = producer
6478 .dynamic("room/alice", Route::default().with_hops(hops(&[10])).with_cost(5))
6479 .unwrap();
6480 let pending = producer.consume().request_broadcast("room/alice");
6481 let upstream = broadcast::Info::new().produce();
6482 let mut dynamic = upstream.dynamic();
6483 queued(&_server).await.accept(&upstream);
6484 let resolved = pending.await.unwrap();
6485
6486 let track = resolved.track("log").unwrap();
6487 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6488 let source = dynamic.requested_track().await.unwrap().accept(None);
6489 let mut group = source.create_group(0u64.into()).unwrap();
6490 group.write_frame(crate::Timestamp::ZERO, b"a".as_ref()).unwrap();
6491 group.write_frame(crate::Timestamp::ZERO, b"b".as_ref()).unwrap();
6492 let mut subscription = subscribing.await.unwrap().unwrap();
6493 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6494 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6495 drop(reading);
6496 drop(subscription);
6497 source.unused().await.unwrap();
6498 drop(group);
6499 drop(source);
6500
6501 let track = resolved.track("log").unwrap();
6502 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6503 let mut resumed = dynamic.requested_track().await.unwrap().resolving_start().accept(None);
6504 resumed.start_at(0).unwrap();
6505 let mut subscription = subscribing.await.unwrap().unwrap();
6506 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6507 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"a");
6508
6509 let replacement_server = producer
6510 .dynamic("room/alice", Route::default().with_hops(hops(&[10])))
6511 .unwrap();
6512 let replacement = broadcast::Info::new().produce();
6513 let mut replacement_dynamic = replacement.dynamic();
6514 queued(&replacement_server).await.accept(&replacement);
6515 let mut source = replacement_dynamic
6516 .requested_track()
6517 .await
6518 .unwrap()
6519 .resolving_start()
6520 .accept(None);
6521 source.start_at(0).unwrap();
6522 let mut group = source.create_group(0u64.into()).unwrap();
6523 group.start_at(2).unwrap();
6524 group.write_frame(crate::Timestamp::ZERO, b"c".as_ref()).unwrap();
6525 let mut continuation = reading.clone();
6526 continuation.start_at(2);
6527 let frame = tokio::time::timeout(Duration::from_secs(1), continuation.read_frame())
6528 .await
6529 .expect("replacement frame")
6530 .unwrap()
6531 .unwrap();
6532 assert_eq!(&frame.payload[..], b"c");
6533 drop(continuation);
6534 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"b");
6535 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], b"c");
6536 drop(reading);
6537 drop(subscription);
6538 source.unused().await.unwrap();
6539 drop(group);
6540 drop(source);
6541
6542 let track = resolved.track("log").unwrap();
6543 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6544 let mut next = replacement_dynamic
6545 .requested_track()
6546 .await
6547 .unwrap()
6548 .resolving_start()
6549 .accept(None);
6550 next.start_at(0).unwrap();
6551 let mut subscription = subscribing.await.unwrap().unwrap();
6552 let mut reading = subscription.recv_group().await.unwrap().unwrap();
6553 for expected in [b"a", b"b", b"c"] {
6554 assert_eq!(&reading.read_frame().await.unwrap().unwrap().payload[..], expected);
6555 }
6556 }
6557
6558 #[tokio::test]
6561 async fn unannounce_keeps_a_returning_reader_awaiting_its_info() {
6562 let producer = origin(1).produce();
6563 let consumer = producer.consume();
6564
6565 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6566 let mut dynamic = broadcast.dynamic();
6567 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6568 let track = resolved.track("video").unwrap();
6569 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6570 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6571 .await
6572 .expect("the front asked the source")
6573 .expect("request");
6574 let source = request.accept(None);
6575 let mut group = source.append_group().unwrap();
6576 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
6577 group.finish().unwrap();
6578 let mut subscription = subscribing.await.unwrap().expect("subscribe");
6579 subscription.recv_group().await.unwrap().expect("the cached group");
6580 drop(subscription);
6581
6582 tokio::time::timeout(Duration::from_secs(1), source.unused())
6585 .await
6586 .expect("parked")
6587 .expect("source open");
6588 drop(source);
6589 let track = resolved.track("video").unwrap();
6590 let subscribing = tokio::spawn(async move { track.subscribe(None).await });
6591 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
6592 .await
6593 .expect("the front asked the source again")
6594 .expect("request");
6595
6596 broadcast.unannounce();
6597 settle(|| resolved.is_closed()).await;
6598
6599 let source = request.accept(None);
6601 let mut group = source.create_group(1u64.into()).unwrap();
6602 group.write_frame(crate::Timestamp::ZERO, b"late".as_ref()).unwrap();
6603 group.finish().unwrap();
6604 source.finish().unwrap();
6605
6606 let mut subscription = subscribing.await.unwrap().expect("subscribe survives the retraction");
6607 let mut payloads = Vec::new();
6608 while let Some(mut group) = subscription.recv_group().await.expect("ends cleanly") {
6609 payloads.push(group.read_frame().await.unwrap().unwrap().payload);
6610 }
6611 assert_eq!(payloads.last().map(|p| &p[..]), Some(&b"late"[..]));
6612 }
6613
6614 #[tokio::test]
6617 async fn reannounce_before_the_front_acts_keeps_it() {
6618 let producer = origin(1).produce();
6619 let consumer = producer.consume();
6620
6621 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
6622 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6623
6624 broadcast.unannounce();
6625 broadcast.announce(Route::default()).unwrap();
6626 for _ in 0..20 {
6627 tokio::task::yield_now().await;
6628 }
6629 assert!(!resolved.is_closed(), "the front ended across a reannouncement");
6630 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6631 assert!(again.is_clone(&resolved));
6632 }
6633
6634 #[tokio::test]
6635 async fn local_broadcast_resolves_once_announced() {
6636 let producer = origin(1).produce();
6637 let consumer = producer.consume();
6638
6639 let broadcast = producer.create_broadcast("room/alice").unwrap();
6641 let err = consumer
6642 .request_broadcast("room/alice")
6643 .now_or_never()
6644 .expect("unroutable is synchronous")
6645 .err()
6646 .unwrap();
6647 assert!(matches!(err, Error::Unroutable));
6648
6649 broadcast.announce(Route::default()).unwrap();
6650 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
6651 assert_eq!(resolved.info().path.as_str(), "room/alice");
6652 drop(broadcast);
6653
6654 let err = consumer
6656 .request_broadcast("room/bob")
6657 .now_or_never()
6658 .expect("unroutable is synchronous")
6659 .err()
6660 .unwrap();
6661 assert!(matches!(err, Error::Unroutable));
6662 }
6663
6664 #[test]
6665 fn create_broadcast_accepts_a_max_depth_path() {
6666 let producer = origin(1).produce();
6667 let path = vec!["a"; Path::MAX_PARTS].join("/");
6668 let _broadcast = producer.create_broadcast(path.as_str()).expect("max depth is allowed");
6669 let deeper = vec!["a"; Path::MAX_PARTS + 1].join("/");
6670 assert!(matches!(
6671 producer.create_broadcast(deeper.as_str()),
6672 Err(Error::BoundsExceeded(_))
6673 ));
6674 }
6675
6676 #[tokio::test]
6677 async fn duplicate_routes_aggregate_until_the_last_leaves() {
6678 let producer = origin(1).produce();
6679 let first = producer.dynamic("live", Route::default().with_cost(3)).unwrap();
6680 let second = producer.dynamic("live", Route::default().with_cost(1)).unwrap();
6681
6682 let mut announced = producer.consume().announced();
6683 let update = announced.next().now_or_never().expect("next").expect("no next");
6684 assert_eq!(update.prefix.as_str(), "live");
6685 assert_eq!(update.kind, AnnounceKind::Announced);
6686 assert_eq!(update.route.cost, Cost::new(1));
6687 announced.assert_next_wait();
6688
6689 drop(second);
6690 let update = announced.next().now_or_never().expect("next").expect("no next");
6691 assert_eq!(update.prefix.as_str(), "live");
6692 assert_eq!(update.kind, AnnounceKind::Updated);
6693 assert_eq!(update.route.cost, Cost::new(3));
6694
6695 drop(first);
6696 announced.assert_next_ended("live");
6697 announced.assert_next_wait();
6698 }
6699
6700 #[test]
6701 fn dynamic_may_cover_a_scope_but_disjoint_prefixes_are_refused() {
6702 let producer = origin(1).produce();
6703 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
6704 let _broad = scoped
6705 .dynamic("", Route::default())
6706 .expect("an overlapping prefix is accepted");
6707
6708 let _ok = scoped
6709 .dynamic("room/alice", Route::default())
6710 .expect("a contained prefix is accepted");
6711 assert!(matches!(
6712 scoped.dynamic("other", Route::default()),
6713 Err(Error::Unauthorized)
6714 ));
6715 }
6716
6717 #[tokio::test]
6718 async fn dynamic_route_keeps_its_producer_scope() {
6719 let producer = origin(1).produce();
6720 let scope = Patterns::from("*/chat".parse::<Pattern>().unwrap());
6721 let scoped = producer.scope("", &scope).unwrap();
6722 let dynamic = scoped.dynamic("", Route::default()).unwrap();
6723
6724 let mut matching = producer
6725 .consume()
6726 .scope("", &scopes(&["room/chat"]))
6727 .unwrap()
6728 .announced();
6729 matching.assert_next_active("");
6730 let mut outside = producer
6731 .consume()
6732 .scope("", &scopes(&["room/video"]))
6733 .unwrap()
6734 .announced();
6735 outside.assert_next_wait();
6736
6737 let refused = producer
6738 .consume()
6739 .request_broadcast("room/video")
6740 .now_or_never()
6741 .expect("an out-of-scope request must be refused synchronously");
6742 assert!(matches!(refused, Err(Error::Unroutable)));
6743 assert!(dynamic.requested_broadcast().now_or_never().is_none());
6744
6745 let _pending = producer.consume().request_broadcast("room/chat");
6746 let request = queued(&dynamic).await;
6747 assert_eq!(request.path().as_str(), "room/chat");
6748 }
6749
6750 #[tokio::test]
6751 async fn scoped_cursor_selects_among_the_routes_it_can_see() {
6752 let producer = origin(1).produce();
6753 let scoped = |pattern: &str| {
6754 producer
6755 .scope("", &Patterns::from(pattern.parse::<Pattern>().unwrap()))
6756 .unwrap()
6757 };
6758 let _chat = scoped("*/chat").dynamic("", Route::default().with_cost(1)).unwrap();
6759 let _video = scoped("*/video").dynamic("", Route::default().with_cost(5)).unwrap();
6760
6761 let mut video = producer
6762 .consume()
6763 .scope("", &scopes(&["room/video"]))
6764 .unwrap()
6765 .announced();
6766 assert_eq!(video.assert_next_active("").cost, Cost::new(5));
6767 }
6768
6769 #[tokio::test]
6770 async fn route_changes_preserve_each_cursors_visible_winner() {
6771 let producer = origin(1).produce();
6772 let consumer = producer.consume();
6773 let mut all = consumer.clone().announced();
6774 let mut excluded = consumer.clone().excluding(origin(7)).announced();
6775 let mut video = consumer
6776 .clone()
6777 .scope("", &scopes(&["room/live/video"]))
6778 .unwrap()
6779 .announced();
6780 let mut rooted = consumer
6781 .scope("room/live", &Patterns::from(Pattern::all()))
6782 .unwrap()
6783 .announced();
6784
6785 let chat = producer
6786 .scope("", &scopes(&["room/live/chat"]))
6787 .unwrap()
6788 .dynamic("room/live", Route::default().with_hops(hops(&[7])).with_cost(1))
6789 .unwrap();
6790 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(1));
6791 assert_eq!(rooted.assert_next_active("").cost, Cost::new(1));
6792 excluded.assert_next_wait();
6793 video.assert_next_wait();
6794
6795 let video_route = producer
6796 .scope("", &scopes(&["room/live/video"]))
6797 .unwrap()
6798 .dynamic("room/live", Route::default().with_hops(hops(&[8])).with_cost(5))
6799 .unwrap();
6800 all.assert_next_wait();
6801 rooted.assert_next_wait();
6802 assert_eq!(excluded.assert_next_active("room/live").cost, Cost::new(5));
6803 assert_eq!(video.assert_next_active("room/live").cost, Cost::new(5));
6804
6805 chat.update(Route::default().with_hops(hops(&[7])).with_cost(9))
6806 .unwrap();
6807 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(5));
6808 assert_eq!(rooted.assert_next_active("").cost, Cost::new(5));
6809 excluded.assert_next_wait();
6810 video.assert_next_wait();
6811
6812 drop(video_route);
6813 assert_eq!(all.assert_next_active("room/live").cost, Cost::new(9));
6814 assert_eq!(rooted.assert_next_active("").cost, Cost::new(9));
6815 excluded.assert_next_ended("room/live");
6816 video.assert_next_ended("room/live");
6817 }
6818
6819 #[tokio::test]
6820 async fn root_cursor_keeps_the_more_specific_covering_route() {
6821 let producer = origin(1).produce();
6822 let broad = producer.dynamic("room", Route::default().with_cost(1)).unwrap();
6823 let narrow = producer.dynamic("room/live", Route::default().with_cost(9)).unwrap();
6824 let mut announced = producer
6825 .consume()
6826 .scope("room/live/video", &Patterns::from(Pattern::all()))
6827 .unwrap()
6828 .announced();
6829 assert_eq!(announced.assert_next_active("").cost, Cost::new(9));
6830
6831 broad.update(Route::default().with_cost(0)).unwrap();
6832 announced.assert_next_wait();
6833 narrow.update(Route::default().with_cost(8)).unwrap();
6834 assert_eq!(announced.assert_next_active("").cost, Cost::new(8));
6835 drop(narrow);
6836 assert_eq!(announced.assert_next_active("").cost, Cost::new(0));
6837 }
6838
6839 #[tokio::test]
6840 async fn dynamic_accepts_a_max_depth_prefix() {
6841 let producer = origin(1).produce();
6842 let path = (0..Path::MAX_PARTS)
6843 .map(|i| format!("s{i}"))
6844 .collect::<Vec<_>>()
6845 .join("/");
6846 let mut announced = producer.consume().announced();
6847
6848 let dynamic = producer.dynamic(&path, Route::default()).expect("max depth is allowed");
6849 announced.assert_next_active(&path);
6850
6851 let _pending = producer.consume().request_broadcast(&path);
6852 let request = queued(&dynamic).await;
6853 assert_eq!(request.path().as_str(), path);
6854 }
6855
6856 #[tokio::test]
6857 async fn dynamic_exclusion_skips_routes_through_the_subscriber() {
6858 let producer = origin(1).produce();
6859 let _server = producer
6860 .dynamic("live", Route::default().with_hops(hops(&[7])))
6861 .unwrap();
6862
6863 let mut excluded = producer.consume().excluding(origin(7)).announced();
6864 excluded.assert_next_wait();
6865
6866 let mut clean = producer.consume().excluding(origin(8)).announced();
6867 clean.assert_next_active("live");
6868 }
6869
6870 #[tokio::test]
6872 async fn announce_consumer_is_a_stream() {
6873 use futures::StreamExt;
6874 let producer = origin(1).produce();
6875 let server = producer.dynamic("live", Route::default()).unwrap();
6876 let mut announced = producer.consume().announced();
6877 let update = StreamExt::next(&mut announced)
6878 .now_or_never()
6879 .expect("next")
6880 .expect("no next");
6881 assert_eq!(update.prefix.as_str(), "live");
6882 assert_eq!(update.kind, AnnounceKind::Announced);
6883 assert!(StreamExt::next(&mut announced).now_or_never().is_none());
6884 drop(server);
6885 let update = StreamExt::next(&mut announced)
6886 .now_or_never()
6887 .expect("next")
6888 .expect("no next");
6889 assert_eq!(update.kind, AnnounceKind::Retracted);
6890 }
6891
6892 #[tokio::test]
6893 async fn dynamic_retracts() {
6894 let producer = origin(1).produce();
6895 let server = producer.dynamic("live", Route::default()).unwrap();
6896 let mut announced = producer.consume().announced();
6897 announced.assert_next_active("live");
6898
6899 drop(server);
6900 announced.assert_next_ended("live");
6901 }
6902
6903 #[test]
6904 fn charged_wildcard_cost_accumulates_across_hops() {
6905 let first = Cost::new(4).charged(1);
6906 let second = first.charged(2);
6907 assert_eq!(second, Cost { warm: 7, cold: 7 });
6908 }
6909
6910 #[tokio::test]
6911 async fn local_broadcast_is_invisible_until_announced() {
6912 let producer = origin(1).produce();
6913 let mut local = producer.consume().announced();
6914 let mut peer = producer.consume().excluding(Hop::UNKNOWN).announced();
6915 let broadcast = producer.create_broadcast("room/alice").unwrap();
6916 local.assert_next_wait();
6917 peer.assert_next_wait();
6918
6919 broadcast.announce(Route::default()).unwrap();
6920 local.assert_next_active("room/alice");
6921 peer.assert_next_active("room/alice");
6922
6923 drop(broadcast);
6924 local.assert_next_ended("room/alice");
6925 peer.assert_next_ended("room/alice");
6926 }
6927
6928 #[tokio::test]
6929 async fn served_route_materializes_on_demand() {
6930 let producer = origin(1).produce();
6931 let consumer = producer.consume();
6932
6933 let server = producer.dynamic("room", Route::default()).unwrap();
6934
6935 let pending = consumer.request_broadcast("room/alice");
6936 let request = queued(&server).await;
6937 assert_eq!(request.path().as_str(), "room/alice");
6938
6939 let source = broadcast::Info::new().produce();
6940 request.accept(&source);
6941
6942 let resolved = pending.await.expect("resolves");
6943 assert_eq!(resolved.info().path.as_str(), "room/alice");
6945
6946 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
6948 assert!(again.is_clone(&resolved));
6949 }
6950
6951 #[tokio::test]
6952 async fn served_requests_coalesce() {
6953 let producer = origin(1).produce();
6954 let consumer = producer.consume();
6955 let server = producer.dynamic("room", Route::default()).unwrap();
6956
6957 let first = consumer.request_broadcast("room/alice");
6958 let second = consumer.request_broadcast("room/alice");
6959
6960 let request = queued(&server).await;
6961 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
6963
6964 let source = broadcast::Info::new().produce();
6965 request.accept(&source);
6966
6967 let first = first.await.expect("resolves");
6968 let second = second.await.expect("resolves");
6969 assert!(first.is_clone(&second));
6970 }
6971
6972 #[tokio::test]
6973 async fn retract_rejects_pending_requests() {
6974 let producer = origin(1).produce();
6975 let consumer = producer.consume();
6976 let server = producer.dynamic("room", Route::default()).unwrap();
6977
6978 let pending = consumer.request_broadcast("room/alice");
6979 drop(server);
6980
6981 let err = pending.await.err().unwrap();
6982 assert!(matches!(err, Error::Unroutable));
6983
6984 let err = consumer
6986 .request_broadcast("room/alice")
6987 .now_or_never()
6988 .expect("unroutable")
6989 .err()
6990 .unwrap();
6991 assert!(matches!(err, Error::Unroutable));
6992 }
6993
6994 #[tokio::test]
6995 async fn routed_broadcast_survives_serving_route_retraction() {
6996 let producer = origin(1).produce();
6997 let consumer = producer.consume();
6998
6999 let standby_server = producer.dynamic("room", Route::default()).unwrap();
7002 let second_server = producer.dynamic("room", Route::default()).unwrap();
7003 let incumbent_server = producer.dynamic("room", Route::default()).unwrap();
7004
7005 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7006 assert!((&mut resolving).now_or_never().is_none());
7007
7008 drop(incumbent_server);
7014 assert!((&mut resolving).now_or_never().is_none());
7015 drop(second_server);
7016 assert!((&mut resolving).now_or_never().is_none());
7017
7018 let request = queued(&standby_server).await;
7019 let source = broadcast::Info::new().produce();
7020 request.accept(&source);
7021
7022 let resolved = resolving.await.expect("resolves via the standby");
7023 assert_eq!(resolved.info().path.as_str(), "room/alice");
7024 }
7025
7026 #[tokio::test]
7027 async fn split_horizon_skips_routes_through_the_requester() {
7028 let producer = origin(1).produce();
7029 let _server = producer
7030 .dynamic("room", Route::default().with_hops(hops(&[7])))
7031 .unwrap();
7032
7033 let excluded = producer.consume().excluding(origin(7));
7035 let err = excluded
7036 .request_broadcast("room/alice")
7037 .now_or_never()
7038 .expect("unroutable")
7039 .err()
7040 .unwrap();
7041 assert!(matches!(err, Error::Unroutable));
7042
7043 let clean = producer.consume().excluding(origin(8));
7045 let pending = clean.request_broadcast("room/alice");
7046 assert!(pending.now_or_never().is_none());
7047 }
7048
7049 #[tokio::test(start_paused = true)]
7053 async fn viewer_sessions_share_the_plain_front() {
7054 let producer = origin(1).produce();
7055 let server = producer
7057 .dynamic("room", Route::default().with_hops(hops(&[10])))
7058 .unwrap();
7059 let upstream = broadcast::Info::new().produce();
7060 let source = upstream.create_track("video", None).unwrap();
7061 let mut group = source.append_group().unwrap();
7062 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
7063
7064 let counts = || {
7065 let table = producer.shared.lock();
7066 let watches = table
7067 .routes
7068 .root
7069 .find(Path::new("room/alice").parts())
7070 .map_or(0, |node| node.watches.len());
7071 (table.fronts.len(), watches)
7072 };
7073
7074 let mut first = None;
7075 let mut front = None;
7076 for viewer in 0..1000 {
7077 let session = producer.consume().excluding(Hop::random());
7078 let pending = session.request_broadcast("room/alice");
7079 if viewer == 0 {
7080 queued(&server).await.accept(&upstream);
7081 }
7082 let resolved = pending.await.expect("resolves");
7083 let mut subscription = resolved.track("video").unwrap().subscribe(None).await.unwrap();
7084 next_group(&mut subscription).await.unwrap().expect("the live group");
7085 drop(subscription);
7086 first.get_or_insert_with(counts);
7087 front = Some(resolved);
7088 }
7089 assert_eq!(first, Some((1, 1)), "one front and one watch for the first viewer");
7090 assert_eq!(counts(), (1, 1), "viewers minted fronts of their own");
7091 drop(front);
7092 }
7093
7094 #[tokio::test]
7098 async fn a_hop_in_a_chain_gets_its_own_front() {
7099 let producer = origin(1).produce();
7100 let _incumbent = producer
7101 .dynamic("room", Route::default().with_hops(hops(&[10])))
7102 .unwrap();
7103 let fronts = || producer.shared.lock().fronts.len();
7104
7105 let peer = producer.consume().excluding(origin(7));
7106 let _local = producer.consume().request_broadcast("room/alice");
7107 let _peer = peer.request_broadcast("room/alice");
7108 assert_eq!(fronts(), 1, "a hop no chain names excludes nothing");
7109
7110 let _echo = producer
7112 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7113 .unwrap();
7114 let _filtered = peer.request_broadcast("room/alice");
7115 assert_eq!(fronts(), 2, "the peer still shares the plain front");
7116 let _viewer = producer.consume().excluding(origin(8)).request_broadcast("room/alice");
7117 assert_eq!(fronts(), 2, "a viewer left the plain front");
7118 }
7119
7120 #[tokio::test]
7124 async fn a_hairpin_through_the_plain_front_is_withdrawn() {
7125 let producer = origin(1).produce();
7126 let peer = producer.consume().excluding(origin(7));
7127 let mut view = peer.announced();
7128 let incumbent = producer
7129 .dynamic("room", Route::default().with_hops(hops(&[10])))
7130 .unwrap();
7131 view.assert_next_active("room");
7132
7133 let pending = peer.request_broadcast("room/alice");
7134 let source = broadcast::Info::new().produce();
7135 queued(&incumbent).await.accept(&source);
7136 let resolved = pending.await.expect("resolves through the incumbent");
7137
7138 let echo = producer
7140 .dynamic("room", Route::default().with_hops(hops(&[10, 7])))
7141 .unwrap();
7142 drop(incumbent);
7143 drop(source);
7144
7145 let replacement = broadcast::Info::new().produce();
7146 queued(&echo).await.accept(&replacement);
7147 resolved.assert_not_closed();
7148 view.assert_next_ended("room");
7149 }
7150
7151 #[tokio::test]
7152 async fn routes_report_where_they_entered() {
7153 let producer = origin(1).produce();
7154 let peer = producer.clone().peer();
7155 let mut announced = producer.consume().announced();
7156
7157 let _ingest = producer
7158 .dynamic("client", Route::default().with_hops(hops(&[5])).with_via(origin(5)))
7159 .unwrap();
7160 let _gateway = producer.publish("gateway", Route::default()).unwrap();
7161 let _forwarded = peer
7162 .dynamic(
7163 "forwarded",
7164 Route::default().with_hops(hops(&[5, 7])).with_via(origin(7)),
7165 )
7166 .unwrap();
7167
7168 assert_eq!(announced.assert_next_active("client").source(), Source::Local);
7169 assert_eq!(
7170 announced.assert_next_active("forwarded").source(),
7171 Source::Peer(origin(7))
7172 );
7173 assert_eq!(announced.assert_next_active("gateway").source(), Source::Local);
7174
7175 let scoped = peer.scope("room", &Patterns::from(Pattern::all())).unwrap();
7177 let _nested = scoped.dynamic("x", Route::default().with_via(origin(8))).unwrap();
7178 assert_eq!(announced.assert_next_active("room/x").source(), Source::Peer(origin(8)));
7179 }
7180
7181 #[tokio::test(start_paused = true)]
7185 async fn try_next_delivers_a_held_update() {
7186 async fn step(by: Duration) {
7188 tokio::time::advance(by).await;
7189 for _ in 0..10 {
7190 tokio::task::yield_now().await;
7191 }
7192 }
7193
7194 let producer = origin(1).produce();
7195 let peer = producer.clone().peer();
7196 let mut announced = producer.consume().announced();
7197 step(Duration::from_millis(1)).await;
7198
7199 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7200 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7201 assert!(announced.try_next().unwrap().kind.is_active());
7202
7203 let _far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7204 drop(near);
7205 assert!(announced.try_next().is_none(), "the update is held");
7206 step(Duration::from_millis(1)).await;
7207 assert!(announced.try_next().is_none(), "the update is still held");
7208 step(DEFAULT_UPDATE_HOLD).await;
7209 let update = announced.try_next().expect("the hold passed");
7210 assert_eq!(update.kind, AnnounceKind::Updated);
7211 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7212 }
7213
7214 #[tokio::test(start_paused = true)]
7217 async fn a_changed_route_waits_out_the_hold() {
7218 let producer = origin(1).produce();
7219 let peer = producer.clone().peer();
7220 let mut announced = producer.consume().announced();
7221 tokio::time::sleep(Duration::from_millis(1)).await;
7223
7224 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7225 let near = peer.dynamic("room", route(&[9, 2], 2)).unwrap();
7226 announced.assert_next_active("room");
7227
7228 let far = peer.dynamic("room", route(&[9, 3, 4], 4)).unwrap();
7230 let farther = peer.dynamic("room", route(&[9, 5, 6, 7], 7)).unwrap();
7231 let start = tokio::time::Instant::now();
7232 drop(near);
7233 drop(far);
7234 let update = announced.next().await.unwrap();
7235 assert_eq!(update.kind, AnnounceKind::Updated);
7236 assert_eq!(update.route.hops, hops(&[9, 5, 6, 7]));
7237 assert_eq!(start.elapsed(), DEFAULT_UPDATE_HOLD);
7238 announced.assert_next_wait();
7239
7240 drop(farther);
7242 announced.assert_next_ended("room");
7243 }
7244
7245 #[tokio::test(start_paused = true)]
7248 async fn a_retraction_quarantines_a_named_publisher() {
7249 let producer = origin(1).produce();
7250 let peer = producer.clone().peer();
7251 let mut announced = producer.consume().announced();
7252 tokio::time::sleep(Duration::from_millis(1)).await;
7253
7254 let route = |chain: &[u64], via| Route::default().with_hops(hops(chain)).with_via(origin(via));
7255
7256 let near = peer.dynamic("named", route(&[9, 2], 2)).unwrap();
7257 announced.assert_next_active("named");
7258 drop(near);
7259 announced.assert_next_ended("named");
7260 let start = tokio::time::Instant::now();
7261 let _stale = peer.dynamic("named", route(&[9, 3, 4], 4)).unwrap();
7262 let update = announced.next().await.unwrap();
7263 assert_eq!(update.kind, AnnounceKind::Announced);
7264 assert_eq!(update.route.hops, hops(&[9, 3, 4]));
7265 assert!(start.elapsed() >= RETRACT_QUARANTINE);
7266
7267 let near = peer.dynamic("anonymous", route(&[0, 2], 2)).unwrap();
7268 announced.assert_next_active("anonymous");
7269 drop(near);
7270 announced.assert_next_ended("anonymous");
7271 let _other = peer.dynamic("anonymous", route(&[0, 3, 4], 4)).unwrap();
7272 announced.assert_next_active("anonymous");
7273
7274 let queued = announced.anchoring.len();
7277 for chain in [[9, 2], [0, 2], [9, 2], [0, 2]] {
7278 let churn = peer.dynamic("churn", route(&chain, 2)).unwrap();
7279 announced.assert_next_active("churn");
7280 drop(churn);
7281 announced.assert_next_ended("churn");
7282 }
7283 assert_eq!(announced.anchoring.len(), queued + 1);
7284
7285 let gone = peer.dynamic("gone", route(&[9, 2], 2)).unwrap();
7288 announced.assert_next_active("gone");
7289 drop(gone);
7290 announced.assert_next_ended("gone");
7291 assert!(!announced.retracted.is_empty());
7292 for _ in 0..2 {
7293 announced.assert_next_wait();
7294 tokio::time::sleep(RETRACT_QUARANTINE).await;
7295 }
7296 announced.assert_next_wait();
7297 assert!(announced.retracted.is_empty(), "the ended quarantine is still held");
7298 }
7299
7300 #[tokio::test]
7304 async fn withdrawn_peer_hides_routes_through_it() {
7305 let producer = origin(1).produce();
7306 let peer = producer.clone().peer();
7307 let mut announced = producer.consume().announced();
7308
7309 let direct = || Route::default().with_hops(hops(&[9, 2])).with_via(origin(2));
7311 let first = peer.dynamic("room", direct()).unwrap();
7312 let relayed = peer
7313 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])).with_via(origin(3)))
7314 .unwrap();
7315 announced.assert_next_active("room");
7316 announced.assert_next_wait();
7317
7318 first.withdrawn();
7320 announced.assert_next_ended("room");
7321 announced.assert_next_wait();
7322
7323 let second = peer.dynamic("room", direct()).unwrap();
7325 announced.assert_next_active("room");
7326 assert!(producer.shared.lock().withdrawn.is_empty());
7327
7328 second.withdrawn();
7330 drop(relayed);
7331 announced.assert_next_ended("room");
7332 assert!(producer.shared.lock().withdrawn.is_empty());
7333 }
7334
7335 #[tokio::test]
7337 async fn old_session_withdrawal_keeps_newer_route() {
7338 for restart in [false, true] {
7339 let producer = origin(1).produce();
7340 let peer = producer.clone().peer();
7341 let mut announced = producer.consume().announced();
7342 let route = Route::default().with_hops(hops(&[9, 2]));
7343 let first = peer.dynamic("room", route.clone()).unwrap();
7344 let second = peer.dynamic("room", route.clone()).unwrap();
7345 announced.assert_next_active("room");
7346 announced.assert_next_wait();
7347 let remaining = if restart {
7348 first.update(route).unwrap();
7350 second.withdrawn();
7351 first
7352 } else {
7353 first.withdrawn();
7354 second
7355 };
7356 announced.assert_next_wait();
7357 assert!(producer.shared.lock().withdrawn.is_empty());
7358 remaining.withdrawn();
7359 announced.assert_next_ended("room");
7360 assert!(producer.shared.lock().withdrawn.is_empty());
7361 }
7362 }
7363
7364 #[tokio::test(start_paused = true)]
7368 async fn a_reconnect_under_a_new_hop_wins_over_the_old_session() {
7369 for (old, new) in [
7370 (&[10][..], &[11][..]),
7371 (&[11], &[10]),
7372 (&[10, 0], &[11, 0]),
7373 (&[11, 0], &[10, 0]),
7374 ] {
7375 let producer = origin(1).produce();
7376 let mut announced = producer.consume().announced();
7377 tokio::time::sleep(Duration::from_millis(1)).await;
7378
7379 let _old = producer.dynamic("room", Route::default().with_hops(hops(old))).unwrap();
7380 assert_eq!(announced.assert_next_active("room").hops, hops(old));
7381
7382 let _new = producer.dynamic("room", Route::default().with_hops(hops(new))).unwrap();
7383 let update = tokio::time::timeout(DEFAULT_UPDATE_HOLD * 2, announced.next())
7384 .await
7385 .unwrap_or_else(|_| panic!("{new:?} never took over from {old:?}"))
7386 .unwrap();
7387 assert_eq!(update.kind, AnnounceKind::Updated);
7388 assert_eq!(update.route.hops, hops(new));
7389 }
7390 }
7391
7392 #[tokio::test]
7393 async fn withdrawn_route_no_longer_covers_requests() {
7394 let producer = origin(1).produce();
7395 let peer = producer.clone().peer();
7396 let direct = peer.dynamic("room", Route::default().with_hops(hops(&[9, 2]))).unwrap();
7397 let _relayed = peer
7398 .dynamic("room", Route::default().with_hops(hops(&[9, 2, 3])))
7399 .unwrap();
7400 let path = Path::new("room/video");
7401 let id = producer
7402 .shared
7403 .read()
7404 .routes
7405 .covering(&path)
7406 .find(|entry| entry.hops.iter().last() == Some(&origin(3)))
7407 .unwrap()
7408 .id;
7409 assert!(producer.shared.read().routes.covers(&path, id));
7410 direct.withdrawn();
7411 assert!(!producer.shared.read().routes.covers(&path, id));
7412 }
7413
7414 #[tokio::test]
7417 async fn source_change_is_an_update() {
7418 let producer = origin(1).produce();
7419 let peer = producer.clone().peer();
7420 let mut announced = producer.consume().announced();
7421
7422 let route = Route::default().with_hops(hops(&[7])).with_via(origin(7));
7423 let _forwarded = peer.dynamic("room", route.clone()).unwrap();
7424 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7425
7426 let local = producer.dynamic("room", route).unwrap();
7428 let update = announced.next().now_or_never().expect("next blocked").expect("no next");
7429 assert_eq!(update.kind, AnnounceKind::Updated);
7430 assert_eq!(update.route.source(), Source::Local);
7431
7432 drop(local);
7433 assert_eq!(announced.assert_next_active("room").source(), Source::Peer(origin(7)));
7434 }
7435
7436 #[tokio::test]
7437 async fn local_view_hides_peer_routes() {
7438 let producer = origin(1).produce();
7439 let peer = producer.clone().peer();
7440 let mut local = producer.consume().local().announced();
7441
7442 let _forwarded = peer
7443 .dynamic("remote", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7444 .unwrap();
7445 local.assert_next_wait();
7446
7447 let _shadow = peer
7451 .dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
7452 .unwrap();
7453 let ingest = producer
7454 .dynamic(
7455 "both",
7456 Route::default().with_hops(hops(&[5])).with_via(origin(5)).with_cost(9),
7457 )
7458 .unwrap();
7459 assert_eq!(local.assert_next_active("both").source(), Source::Local);
7460 drop(ingest);
7461 local.assert_next_ended("both");
7462
7463 let err = producer
7466 .consume()
7467 .local()
7468 .request_broadcast("remote/alice")
7469 .now_or_never()
7470 .expect("unroutable")
7471 .err()
7472 .unwrap();
7473 assert!(matches!(err, Error::Unroutable));
7474 assert!(
7475 producer
7476 .consume()
7477 .request_broadcast("remote/alice")
7478 .now_or_never()
7479 .is_none()
7480 );
7481 }
7482
7483 #[tokio::test]
7487 async fn handler_rejection_is_final() {
7488 let producer = origin(1).produce();
7489 let consumer = producer.consume();
7490 let server = producer.dynamic("room", Route::default()).unwrap();
7491
7492 let pending = consumer.request_broadcast("room/alice");
7493 let request = queued(&server).await;
7494 request.reject(Error::Unroutable);
7495 let err = tokio::time::timeout(Duration::from_secs(5), pending)
7496 .await
7497 .expect("the front must give up, not spin")
7498 .err()
7499 .unwrap();
7500 assert!(matches!(err, Error::Unroutable));
7501
7502 let pending = consumer.request_broadcast("room/bob");
7504 let request = queued(&server).await;
7505 assert_eq!(request.path().as_str(), "room/bob");
7506 let served = broadcast::Info::new().produce();
7507 request.accept(&served);
7508 pending.await.expect("resolves");
7509 }
7510
7511 #[tokio::test]
7514 async fn routed_broadcast_waits_out_a_rejection() {
7515 let producer = origin(1).produce();
7516 let consumer = producer.consume();
7517 let server = producer.dynamic("room", Route::default()).unwrap();
7518
7519 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7520 assert!((&mut resolving).now_or_never().is_none());
7521 let request = queued(&server).await;
7522 request.reject(Error::Unroutable);
7523
7524 for _ in 0..20 {
7526 tokio::task::yield_now().await;
7527 }
7528 assert!((&mut resolving).now_or_never().is_none());
7529 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7530
7531 server.update(Route::default().with_cost(2)).unwrap();
7533 assert!((&mut resolving).now_or_never().is_none());
7534 let request = queued(&server).await;
7535 let served = broadcast::Info::new().produce();
7536 request.accept(&served);
7537 resolving.await.expect("resolves");
7538 }
7539
7540 #[tokio::test]
7543 async fn routed_broadcast_reports_teardown_as_closed() {
7544 let (producer, driver) = Producer::new(Config::new(origin(1)));
7545 let consumer = producer.consume();
7546 let _server = producer.dynamic("room", Route::default()).unwrap();
7547
7548 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7550 assert!((&mut resolving).now_or_never().is_none());
7551
7552 drop(driver);
7553
7554 let err = tokio::time::timeout(Duration::from_secs(5), resolving)
7555 .await
7556 .expect("teardown resolves the wait")
7557 .err()
7558 .unwrap();
7559 assert!(matches!(err, Error::Closed), "unexpected end: {err}");
7560 }
7561
7562 #[tokio::test]
7565 async fn routed_broadcast_wakes_for_a_local_broadcast() {
7566 let producer = origin(1).produce();
7567 let consumer = producer.consume();
7568 let server = producer.dynamic("room", Route::default()).unwrap();
7569
7570 let mut resolving = Box::pin(consumer.routed_broadcast("room/alice"));
7571 assert!((&mut resolving).now_or_never().is_none());
7572 queued(&server).await.reject(Error::Unroutable);
7573 for _ in 0..20 {
7574 tokio::task::yield_now().await;
7575 }
7576 assert!((&mut resolving).now_or_never().is_none());
7577
7578 let _local = producer.publish("room/alice", Route::default()).unwrap();
7580 let resolved = resolving.await.expect("resolves locally");
7581 assert_eq!(resolved.info().path.as_str(), "room/alice");
7582 assert!(server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending());
7583 }
7584
7585 #[tokio::test]
7588 async fn late_track_on_a_served_front_replays() {
7589 let producer = origin(1).produce();
7590 let consumer = producer.consume();
7591 let server = producer.dynamic("room", Route::default()).unwrap();
7592
7593 let source = broadcast::Info::new().produce();
7594 for name in ["a", "b"] {
7595 let track = source.create_track(name, None).unwrap();
7596 let mut group = track.append_group().unwrap();
7597 group.write_frame(crate::Timestamp::ZERO, name.as_bytes()).unwrap();
7598 group.finish().unwrap();
7599 std::mem::forget(track);
7601 }
7602
7603 let pending = consumer.request_broadcast("room/alice");
7604 queued(&server).await.accept(&source);
7605 let resolved = pending.await.expect("resolves");
7606
7607 let budget = track::Subscription::default().with_max_age(Duration::from_secs(3600));
7608 for name in ["a", "b"] {
7609 let mut subscription = resolved
7610 .track(name)
7611 .unwrap()
7612 .subscribe(budget.clone())
7613 .await
7614 .expect("subscribe");
7615 let mut group = tokio::time::timeout(Duration::from_secs(5), subscription.recv_group())
7616 .await
7617 .expect("the late track must replay, not park")
7618 .expect("recv group")
7619 .expect("track ended early");
7620 let frame = group.read_frame().await.expect("read frame").expect("frame");
7621 assert_eq!(&frame.payload[..], name.as_bytes());
7622 }
7623 }
7624
7625 #[tokio::test]
7626 async fn most_specific_prefix_shadows() {
7627 let producer = origin(1).produce();
7628 let consumer = producer.consume();
7629
7630 let broad_server = producer.dynamic("", Route::default()).unwrap();
7631 let _narrow = producer.announce(".dash", Route::default()).unwrap();
7634
7635 let err = consumer
7636 .request_broadcast(".dash/pid")
7637 .now_or_never()
7638 .expect("unroutable")
7639 .err()
7640 .unwrap();
7641 assert!(matches!(err, Error::Unroutable));
7642
7643 let _pending = consumer.request_broadcast("room/alice");
7645 let request = queued(&broad_server).await;
7646 assert_eq!(request.path().as_str(), "room/alice");
7647 }
7648
7649 #[tokio::test]
7650 async fn root_dynamic_serves_any_path() {
7651 let producer = origin(1).produce();
7652 let consumer = producer.consume();
7653 let mut announced = consumer.announced();
7654 let dynamic = producer.dynamic("", Route::default()).unwrap();
7655 announced.assert_next_active("");
7657
7658 let pending = consumer.request_broadcast("anything/at/all");
7659 let request = queued(&dynamic).await;
7660 assert_eq!(request.path().as_str(), "anything/at/all");
7661
7662 let source = broadcast::Info::new().produce();
7663 request.accept(&source);
7664 let resolved = pending.await.expect("resolves");
7665 assert_eq!(resolved.info().path.as_str(), "anything/at/all");
7666
7667 drop(dynamic);
7669 announced.assert_next_ended("");
7670 let err = consumer
7671 .request_broadcast("something/else")
7672 .now_or_never()
7673 .expect("unroutable")
7674 .err()
7675 .unwrap();
7676 assert!(matches!(err, Error::Unroutable));
7677 }
7678
7679 #[tokio::test]
7685 async fn out_of_scope_request_never_reaches_the_dynamic_handler() {
7686 let producer = origin(1).produce();
7687 let dynamic = producer.dynamic("", Route::default()).unwrap();
7688 let scoped = producer.consume().scope("", &scopes(&["tenant-a"])).unwrap();
7689
7690 for path in ["tenant-b/live", "tenant-a-other/live"] {
7693 let refused = scoped
7694 .request_broadcast(path)
7695 .now_or_never()
7696 .expect("an out-of-scope request must be refused synchronously, not queued");
7697 assert!(matches!(refused, Err(Error::Unauthorized)));
7698 assert!(
7699 dynamic.requested_broadcast().now_or_never().is_none(),
7700 "the dynamic handler was asked to create a broadcast the requester may not read"
7701 );
7702 }
7703 }
7704
7705 #[tokio::test]
7706 async fn routed_waits_for_coverage() {
7707 let producer = origin(1).produce();
7708 let consumer = producer.consume();
7709
7710 let mut fut = consumer.routed("room/alice").boxed();
7711 assert!((&mut fut).now_or_never().is_none());
7712
7713 let _a = producer.announce("room", Route::default().with_cost(3)).unwrap();
7715 let route = fut.now_or_never().expect("covered").expect("routed");
7716 assert_eq!(route.cost, Cost::new(3));
7717
7718 consumer
7720 .routed("room/alice/cam")
7721 .now_or_never()
7722 .expect("covered")
7723 .expect("routed");
7724 }
7725
7726 #[tokio::test]
7727 async fn routed_ignores_deeper_routes() {
7728 let producer = origin(1).produce();
7729 let consumer = producer.consume();
7730
7731 let _deep = producer.announce("room/alice/cam", Route::default()).unwrap();
7733 let mut fut = consumer.routed("room/alice").boxed();
7734 assert!((&mut fut).now_or_never().is_none());
7735
7736 let _exact = producer.announce("room/alice", Route::default()).unwrap();
7737 fut.now_or_never().expect("covered").expect("routed");
7738 }
7739
7740 #[tokio::test]
7741 async fn routed_accepts_a_max_depth_path() {
7742 let producer = origin(1).produce();
7743 let consumer = producer.consume();
7744 let path = (0..Path::MAX_PARTS)
7745 .map(|i| format!("s{i}"))
7746 .collect::<Vec<_>>()
7747 .join("/");
7748 assert_eq!(Path::new(&path).parts().count(), Path::MAX_PARTS);
7749
7750 assert!(consumer.allowed().matches(&path));
7751
7752 let mut fut = consumer.routed(&path).boxed();
7753 assert!((&mut fut).now_or_never().is_none());
7754
7755 let _a = producer.announce("", Route::default()).unwrap();
7757 fut.now_or_never().expect("covered").expect("routed");
7758 }
7759
7760 #[tokio::test]
7761 async fn teardown_ends_everything() {
7762 let (producer, driver) = Producer::new(Config::new(origin(1)));
7763 let consumer = producer.consume();
7764 let _announcement = producer.announce("room", Route::default()).unwrap();
7765 let mut announced = consumer.announced();
7766 announced.assert_next_active("room");
7767
7768 let _server = producer.dynamic("served", Route::default()).unwrap();
7769 let pending = consumer.request_broadcast("served/path");
7770
7771 drop(driver);
7772
7773 announced.assert_next_active("served");
7775 assert!(announced.next().now_or_never().expect("ended").is_none());
7776
7777 assert!(pending.now_or_never().expect("rejected").is_err());
7779 assert!(matches!(producer.announce("x", Route::default()), Err(Error::Closed)));
7780 assert!(matches!(producer.create_broadcast("x"), Err(Error::Closed)));
7781 let err = consumer
7782 .request_broadcast("y")
7783 .now_or_never()
7784 .expect("closed")
7785 .err()
7786 .unwrap();
7787 assert!(matches!(err, Error::Closed));
7788
7789 let mut late = consumer.announced();
7791 assert!(late.next().now_or_never().expect("ended").is_none());
7792 }
7793
7794 struct ResumeRig {
7797 producer: Producer,
7798 resolved: broadcast::Consumer,
7799 subscription: track::Subscriber,
7800 incumbent_track: track::Producer,
7803 }
7804
7805 impl ResumeRig {
7806 async fn new(first: &[u64]) -> (Self, Dynamic, broadcast::Producer) {
7809 let producer = origin(1).produce();
7810 let consumer = producer.consume();
7811
7812 let server = producer
7813 .dynamic("room", Route::default().with_hops(hops(first)))
7814 .unwrap();
7815
7816 let pending = consumer.request_broadcast("room/alice");
7817 let request = queued(&server).await;
7818 let source = broadcast::Info::new().produce();
7819 let track = source.create_track("video", None).unwrap();
7820 let mut group = track.append_group().unwrap();
7821 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7822 group.finish().unwrap();
7823 request.accept(&source);
7824
7825 let resolved = pending.await.expect("resolves");
7826 let mut subscription = resolved
7827 .track("video")
7828 .unwrap()
7829 .subscribe(None)
7830 .await
7831 .expect("subscribe");
7832 let mut group = subscription
7833 .recv_group()
7834 .await
7835 .expect("recv group")
7836 .expect("track ended early");
7837 let frame = group.read_frame().await.expect("read frame").expect("frame");
7838 assert_eq!(&frame.payload[..], b"before");
7839
7840 (
7841 Self {
7842 producer,
7843 resolved,
7844 subscription,
7845 incumbent_track: track,
7846 },
7847 server,
7848 source,
7849 )
7850 }
7851
7852 fn standby(&self, first: &[u64]) -> Dynamic {
7855 self.producer
7856 .dynamic("room", Route::default().with_hops(hops(first)))
7857 .unwrap()
7858 }
7859 }
7860
7861 async fn assert_resumes(rig: &mut ResumeRig, server: &Dynamic) {
7866 let request = queued(server).await;
7867 let replacement = broadcast::Info::new().produce();
7868 let track = replacement.create_track("video", None).unwrap();
7869 let mut group = track.append_group().unwrap();
7872 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7873 group.finish().unwrap();
7874 request.accept(&replacement);
7875
7876 let mut group = track.append_group().unwrap();
7877 group.write_frame(crate::Timestamp::ZERO, b"resumed".as_ref()).unwrap();
7878 group.finish().unwrap();
7879
7880 let mut group = rig
7881 .subscription
7882 .recv_group()
7883 .await
7884 .expect("subscription survives the failover")
7885 .expect("track ended early");
7886 let frame = group.read_frame().await.expect("read frame").expect("frame");
7887 assert_eq!(&frame.payload[..], b"resumed");
7888 }
7889
7890 #[tokio::test]
7893 async fn driver_resolves_with_live_consumers() {
7894 let (producer, driver) = Producer::new(Config::new(origin(1)));
7895 let consumer = producer.consume();
7896 let run = crate::time::run(driver);
7897 drop(producer);
7898 tokio::time::timeout(Duration::from_secs(5), run)
7899 .await
7900 .expect("driver must finish once the producers are gone");
7901 drop(consumer);
7902 }
7903
7904 #[tokio::test]
7905 async fn remote_source_resumes_through_same_first_hop() {
7906 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7907 let standby_server = rig.standby(&[10, 20]);
7908
7909 drop(incumbent);
7911 drop(source);
7912
7913 assert_resumes(&mut rig, &standby_server).await;
7915 }
7916
7917 #[tokio::test]
7921 async fn incompatible_successor_is_refused() {
7922 for replacement in [
7923 track::Info::default().with_timescale(crate::Timescale::MICRO),
7924 track::Info::default().with_priority(7),
7925 track::Info::default().with_max_age(Duration::from_secs(7)),
7926 ] {
7927 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7928 let standby_server = rig.standby(&[10, 20]);
7929 drop(incumbent);
7930 drop(source);
7931
7932 let request = queued(&standby_server).await;
7935 let successor = broadcast::Info::new().produce();
7936 let track = successor.create_track("video", replacement).unwrap();
7937 let mut group = track.append_group().unwrap();
7938 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
7939 group.finish().unwrap();
7940 request.accept(&successor);
7941
7942 assert!(
7943 matches!(rig.subscription.recv_group().await, Err(Error::Unsupported)),
7944 "the subscription must abort rather than resume onto incompatible metadata"
7945 );
7946
7947 let reopened = rig.resolved.track("video").unwrap();
7949 assert!(matches!(reopened.query().await, Err(Error::Unsupported)));
7950 assert!(matches!(reopened.subscribe(None).await, Err(Error::Unsupported)));
7951 }
7952 }
7953
7954 #[tokio::test]
7955 async fn different_first_hop_ends_the_subscription() {
7956 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
7957 let rival_server = rig.standby(&[11]);
7959
7960 drop(incumbent);
7963 drop(source);
7964 rig.incumbent_track.abort(Error::Dropped).unwrap();
7965
7966 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
7968 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
7969
7970 let consumer = rig.producer.consume();
7972 let pending = consumer.request_broadcast("room/alice");
7973 let request = queued(&rival_server).await;
7974 let replacement = broadcast::Info::new().produce();
7975 request.accept(&replacement);
7976 pending.await.expect("re-request resolves through the rival");
7977 }
7978
7979 #[tokio::test]
7984 async fn a_first_hop_update_drains_the_old_publisher() {
7985 for first in [&[][..], &[0][..], &[10][..]] {
7986 let (mut rig, server, _source) = ResumeRig::new(first).await;
7987 server.update(Route::default().with_hops(hops(&[11]))).unwrap();
7988
7989 let mut group = rig.incumbent_track.append_group().unwrap();
7991 group.write_frame(crate::Timestamp::ZERO, b"draining".as_ref()).unwrap();
7992 group.finish().unwrap();
7993 let mut group = next_group(&mut rig.subscription)
7994 .await
7995 .expect("the in-flight subscription survives the update")
7996 .expect("track ended early");
7997 let frame = group.read_frame().await.expect("read frame").expect("frame");
7998 assert_eq!(&frame.payload[..], b"draining", "first hop {first:?}");
7999
8000 let consumer = rig.producer.consume();
8003 let pending = consumer.request_broadcast("room/alice");
8004 let request = queued(&server).await;
8005 let replacement = broadcast::Info::new().produce();
8006 let track = replacement
8007 .create_track("video", track::Info::default().with_priority(7))
8008 .unwrap();
8009 let mut group = track.append_group().unwrap();
8010 group.write_frame(crate::Timestamp::ZERO, b"new".as_ref()).unwrap();
8011 group.finish().unwrap();
8012 request.accept(&replacement);
8013
8014 let resolved = pending.await.expect("resolves through the new publisher");
8015 assert!(
8016 !resolved.is_clone(&rig.resolved),
8017 "first hop {first:?} joined the old front"
8018 );
8019 let mut fresh = resolved
8020 .track("video")
8021 .unwrap()
8022 .subscribe(None)
8023 .await
8024 .expect("the old publisher's track info does not apply");
8025 let mut group = next_group(&mut fresh)
8026 .await
8027 .expect("recv group")
8028 .expect("track ended early");
8029 let frame = group.read_frame().await.expect("read frame").expect("frame");
8030 assert_eq!(&frame.payload[..], b"new");
8031
8032 rig.incumbent_track.finish().unwrap();
8035 let end = next_group(&mut rig.subscription).await;
8036 assert!(
8037 !matches!(end, Ok(Some(_))),
8038 "first hop {first:?} spliced the new publisher into a live subscription"
8039 );
8040 }
8041 }
8042
8043 #[tokio::test]
8046 async fn a_first_hop_update_resumes_through_the_same_publisher() {
8047 let (mut rig, incumbent, _source) = ResumeRig::new(&[10]).await;
8048 let standby_server = rig.standby(&[10, 20]);
8049
8050 incumbent.update(Route::default().with_hops(hops(&[11]))).unwrap();
8051
8052 assert_resumes(&mut rig, &standby_server).await;
8053 assert!(
8054 incumbent.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8055 "the front never asks the new publisher"
8056 );
8057 }
8058
8059 #[tokio::test]
8060 async fn anonymous_routes_never_resume() {
8061 let (mut rig, incumbent, source) = ResumeRig::new(&[]).await;
8064 let _twin_server = rig.standby(&[]);
8065
8066 drop(incumbent);
8067 drop(source);
8068 rig.incumbent_track.abort(Error::Dropped).unwrap();
8069
8070 let err = rig.subscription.recv_group().await.err().expect("subscription ends");
8071 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8072 }
8073
8074 #[tokio::test]
8081 async fn anonymous_handoff_serves_the_newcomer_immediately() {
8082 let producer = origin(1).produce();
8083
8084 let server_a = producer
8086 .dynamic("room", Route::default().with_hops(hops(&[10])))
8087 .unwrap();
8088
8089 let consumer = producer.consume().excluding(origin(30));
8092 let pending = consumer.request_broadcast("room/alice");
8093 let request = queued(&server_a).await;
8094 let source_a = broadcast::Info::new().produce();
8095 let track_a = source_a.create_track("video", None).unwrap();
8096 let mut group = track_a.append_group().unwrap();
8097 group.write_frame(crate::Timestamp::ZERO, b"from-a".as_ref()).unwrap();
8098 group.finish().unwrap();
8099 request.accept(&source_a);
8100
8101 let resolved_a = pending.await.expect("resolves");
8102 let mut sub_a = resolved_a
8103 .track("video")
8104 .unwrap()
8105 .subscribe(None)
8106 .await
8107 .expect("subscribe");
8108 let mut group = sub_a
8109 .recv_group()
8110 .await
8111 .expect("recv group")
8112 .expect("track ended early");
8113 assert_eq!(
8114 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8115 b"from-a"
8116 );
8117
8118 drop(track_a);
8121 drop(source_a);
8122 drop(server_a);
8123
8124 let err = sub_a.recv_group().await.err().expect("front closed");
8126 assert!(matches!(err, Error::Dropped), "unexpected end: {err}");
8127
8128 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8132 settle(|| {
8133 matches!(
8134 consumer.request_broadcast("room/alice").now_or_never(),
8135 Some(Err(Error::Unroutable))
8136 )
8137 })
8138 .await;
8139
8140 let server_b = producer
8142 .dynamic("room", Route::default().with_hops(hops(&[20])))
8143 .unwrap();
8144 let pending = consumer.request_broadcast("room/alice");
8145 let request = queued(&server_b).await;
8146 let source_b = broadcast::Info::new().produce();
8147 let track_b = source_b.create_track("video", None).unwrap();
8148 let mut group = track_b.append_group().unwrap();
8149 group.write_frame(crate::Timestamp::ZERO, b"from-b".as_ref()).unwrap();
8150 group.finish().unwrap();
8151 request.accept(&source_b);
8152
8153 let resolved_b = pending.await.expect("B's front is served immediately");
8154 assert!(
8155 !resolved_b.is_clone(&resolved_a),
8156 "B must not splice into A's closed front"
8157 );
8158
8159 let mut sub_b = resolved_b
8160 .track("video")
8161 .unwrap()
8162 .subscribe(None)
8163 .await
8164 .expect("subscribe");
8165 let mut group = sub_b
8166 .recv_group()
8167 .await
8168 .expect("recv group")
8169 .expect("track ended early");
8170 assert_eq!(
8171 &group.read_frame().await.expect("read frame").expect("frame").payload[..],
8172 b"from-b"
8173 );
8174 }
8175
8176 #[tokio::test]
8177 async fn reprice_is_invisible_to_the_subscription() {
8178 let (rig, incumbent, source) = ResumeRig::new(&[10]).await;
8179
8180 incumbent
8183 .update(Route::default().with_hops(hops(&[10])).with_cost(9))
8184 .unwrap();
8185
8186 let track = source.create_track("audio", None).unwrap();
8187 let mut group = track.append_group().unwrap();
8188 group.write_frame(crate::Timestamp::ZERO, b"steady".as_ref()).unwrap();
8189 group.finish().unwrap();
8190
8191 let mut audio = rig
8192 .resolved
8193 .track("audio")
8194 .unwrap()
8195 .subscribe(None)
8196 .await
8197 .expect("subscribe survives the reprice");
8198 let mut group = audio
8199 .recv_group()
8200 .await
8201 .expect("recv group")
8202 .expect("track ended early");
8203 let frame = group.read_frame().await.expect("read frame").expect("frame");
8204 assert_eq!(&frame.payload[..], b"steady");
8205 }
8206
8207 #[tokio::test]
8208 async fn drain_reprice_migrates_before_the_session_dies() {
8209 let (mut rig, incumbent, source) = ResumeRig::new(&[10]).await;
8210 let standby_server = rig.standby(&[10, 20]);
8211
8212 incumbent
8216 .update(Route::default().with_hops(hops(&[10])).with_cost(Cost::DRAIN))
8217 .unwrap();
8218
8219 assert_resumes(&mut rig, &standby_server).await;
8220
8221 drop(incumbent);
8223 drop(source);
8224 }
8225
8226 #[tokio::test]
8227 async fn local_sources_splice_newest_first() {
8228 let producer = origin(1).produce();
8229 let consumer = producer.consume();
8230
8231 let first = producer.publish("room/alice", Route::default()).unwrap();
8232 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8233
8234 let second = producer.publish("room/alice", Route::default()).unwrap();
8236 let again = consumer.request_broadcast("room/alice").await.expect("resolves");
8237 assert!(again.is_clone(&resolved));
8238
8239 first.close();
8241 settle(|| consumer.get_broadcast("room/alice").is_some()).await;
8242 second.close();
8243 settle(|| consumer.get_broadcast("room/alice").is_none()).await;
8244
8245 let _third = producer.publish("room/alice", Route::default()).unwrap();
8247 assert!(consumer.get_broadcast("room/alice").is_some());
8248 }
8249
8250 #[tokio::test]
8258 async fn a_finished_broadcast_concludes_in_flight_subscriptions() {
8259 let producer = origin(1).produce();
8260 let consumer = producer.consume();
8261
8262 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8263 let track = broadcast.create_track("video", None).unwrap();
8264
8265 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8266 let mut subscription = resolved
8267 .track("video")
8268 .unwrap()
8269 .subscribe(None)
8270 .await
8271 .expect("subscribe");
8272 let mut group = track.append_group().unwrap();
8274 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8275 group.finish().unwrap();
8276 track.finish().unwrap();
8277 drop(track);
8278 broadcast.close();
8279
8280 let mut group = next_group(&mut subscription)
8281 .await
8282 .expect("a cleanly finished track was served as an error")
8283 .expect("the track ended before its last group");
8284 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8285 drop(group);
8286
8287 let end = next_group(&mut subscription)
8288 .await
8289 .expect("a cleanly finished track ended as an error");
8290 assert!(end.is_none(), "a group followed the final one");
8291 }
8292
8293 #[tokio::test]
8300 async fn a_retracted_route_concludes_in_flight_subscriptions() {
8301 let producer = origin(1).produce();
8302 let consumer = producer.consume();
8303 let server = producer
8304 .dynamic("room", Route::default().with_hops(hops(&[10])))
8305 .unwrap();
8306
8307 let pending = consumer.request_broadcast("room/alice");
8308 let request = queued(&server).await;
8309 let source = broadcast::Info::new().produce();
8310 let track = source.create_track("video", None).unwrap();
8311 request.accept(&source);
8312
8313 let resolved = pending.await.expect("resolves");
8314 let mut subscription = resolved
8315 .track("video")
8316 .unwrap()
8317 .subscribe(None)
8318 .await
8319 .expect("subscribe");
8320
8321 source.close();
8324 drop(server);
8325 settle(|| resolved.is_closed()).await;
8326 let mut group = track.append_group().unwrap();
8327 group.write_frame(crate::Timestamp::ZERO, b"tail".as_ref()).unwrap();
8328 group.finish().unwrap();
8329 track.finish().unwrap();
8330 drop(track);
8331
8332 let mut group = next_group(&mut subscription)
8333 .await
8334 .expect("a retracted route's track was served as an error")
8335 .expect("the track ended before its last group");
8336 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"tail");
8337 drop(group);
8338
8339 let end = next_group(&mut subscription)
8340 .await
8341 .expect("a cleanly finished track ended as an error");
8342 assert!(end.is_none(), "a group followed the final one");
8343 }
8344
8345 #[tokio::test]
8348 async fn a_closed_source_is_not_requested_again_from_its_standing_route() {
8349 let producer = origin(1).produce();
8350 let server = producer
8351 .dynamic("room", Route::default().with_hops(hops(&[10])))
8352 .unwrap();
8353 let pending = producer.consume().request_broadcast("room/alice");
8354 let source = broadcast::Info::new().produce();
8355 queued(&server).await.accept(&source);
8356 let resolved = pending.await.unwrap();
8357
8358 source.close();
8359 settle(|| resolved.is_closed()).await;
8360 assert!(
8361 server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),
8362 "the closed source was requested again"
8363 );
8364 }
8365
8366 #[tokio::test]
8371 async fn origin_front_drops_the_source_when_unused() {
8372 let producer = origin(1).produce();
8373 let consumer = producer.consume();
8374
8375 let broadcast = producer.publish("room/alice", Route::default()).unwrap();
8376 let track = broadcast.create_track("video", None).unwrap();
8377 let mut group = track.append_group().unwrap();
8378 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8379 group.finish().unwrap();
8380
8381 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8382 let mut subscription = resolved
8383 .track("video")
8384 .unwrap()
8385 .subscribe(None)
8386 .await
8387 .expect("subscribe");
8388 let mut group = subscription.recv_group().await.unwrap().unwrap();
8389 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8390 drop(group);
8391 drop(subscription);
8392
8393 tokio::time::timeout(Duration::from_secs(1), track.unused())
8394 .await
8395 .expect("source unused should resolve far below TRACK_IDLE_LINGER")
8396 .expect("source closed");
8397
8398 let mut again = resolved
8401 .track("video")
8402 .unwrap()
8403 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8404 .await
8405 .expect("resubscribe");
8406 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8407 .await
8408 .expect("cached group is still on the front")
8409 .expect("recv group")
8410 .expect("track ended early");
8411 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8412
8413 tokio::time::timeout(Duration::from_secs(1), track.used())
8414 .await
8415 .expect("returning reader re-splices the source")
8416 .expect("source closed");
8417
8418 let mut group = track.append_group().unwrap();
8419 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8420 group.finish().unwrap();
8421 let mut group = tokio::time::timeout(Duration::from_secs(1), again.recv_group())
8422 .await
8423 .expect("groups past the cached edge come from the re-splice")
8424 .expect("recv group")
8425 .expect("track ended early");
8426 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8427 }
8428
8429 #[tokio::test]
8439 async fn resumed_reader_skips_warm_groups_behind_the_new_edge() {
8440 let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap();
8441 let (_server, _upstream, mut dynamic, resolved) = served_front().await;
8442 let budget = track::Subscription::default().with_max_age(Duration::from_millis(100));
8443 let write = |source: &track::Producer, sequence: u64, millis: u64| {
8444 let mut group = source.create_group(sequence.into()).unwrap();
8445 group.write_frame(ms(millis), b"x".as_ref()).unwrap();
8446 group.finish().unwrap();
8447 };
8448 let drain = |subscription: &mut track::Subscriber| {
8449 let mut sequences = Vec::new();
8450 while let Poll::Ready(group) = subscription.poll_recv_group(&kio::Waiter::noop()) {
8451 sequences.push(group.unwrap().expect("track ended").sequence);
8452 }
8453 sequences
8454 };
8455
8456 let track = resolved.track("video").unwrap();
8457 let first = budget.clone();
8458 let subscribing = tokio::spawn(async move { track.subscribe(first).await });
8459 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8460 .await
8461 .expect("the front asked the source")
8462 .expect("request");
8463 let source = request.resolving_start().accept(None);
8464 for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] {
8465 write(&source, sequence, millis);
8466 }
8467 let mut subscription = subscribing.await.unwrap().expect("subscribe");
8468 next_group(&mut subscription).await.unwrap().expect("a cached group");
8469 drain(&mut subscription);
8470 drop(subscription);
8471
8472 tokio::time::timeout(Duration::from_secs(1), source.unused())
8473 .await
8474 .expect("parked")
8475 .expect("source open");
8476 drop(source);
8477
8478 let track = resolved.track("video").unwrap();
8479 let second = budget.clone();
8480 let subscribing = tokio::spawn(async move { track.subscribe(second).await });
8481 let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track())
8482 .await
8483 .expect("the front asked the source again")
8484 .expect("request");
8485 let mut source = request.resolving_start().accept(None);
8486 for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] {
8487 write(&source, sequence, millis);
8488 }
8489 source.start_at(3).unwrap();
8492 let mut subscription = subscribing.await.unwrap().expect("resubscribe");
8493 settle(|| subscription.latest() == Some(22)).await;
8494
8495 assert_eq!(drain(&mut subscription), [3, 20, 21, 22]);
8498 }
8499
8500 #[tokio::test]
8505 async fn chained_front_drops_the_source_when_unused() {
8506 let leaf = origin(1).produce();
8507 let leaf_consumer = leaf.consume();
8508
8509 let broadcast = leaf.publish("room/alice", Route::default()).unwrap();
8510 let track = broadcast.create_track("video", None).unwrap();
8511 let mut group = track.append_group().unwrap();
8512 group.write_frame(crate::Timestamp::ZERO, b"cached".as_ref()).unwrap();
8513 group.finish().unwrap();
8514
8515 let leaf_front = leaf_consumer.request_broadcast("room/alice").await.expect("resolves");
8518
8519 let mid = origin(2).produce();
8520 let mid_server = mid.dynamic("room", Route::default().with_hops(hops(&[10]))).unwrap();
8521 let mid_pending = mid.consume().request_broadcast("room/alice");
8522 queued(&mid_server).await.accept(&leaf_front);
8523 let mid_resolved = mid_pending.await.expect("mid resolves");
8524
8525 let edge = origin(3).produce();
8526 let edge_server = edge.dynamic("room", Route::default().with_hops(hops(&[20]))).unwrap();
8527 let edge_pending = edge.consume().request_broadcast("room/alice");
8528 queued(&edge_server).await.accept(&mid_resolved);
8529 let edge_resolved = edge_pending.await.expect("edge resolves");
8530
8531 let mut subscription = edge_resolved
8532 .track("video")
8533 .unwrap()
8534 .subscribe(None)
8535 .await
8536 .expect("subscribe");
8537 let mut group = subscription.recv_group().await.unwrap().unwrap();
8538 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8539 drop(group);
8540 drop(subscription);
8541
8542 tokio::time::timeout(Duration::from_secs(5), track.unused())
8543 .await
8544 .expect("chained unused should resolve far below TRACK_IDLE_LINGER")
8545 .expect("source closed");
8546
8547 let cached = edge_resolved.track("video").unwrap().cached_groups();
8548 assert_eq!(
8549 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8550 vec![0],
8551 "every front keeps the delivered groups after releasing its source"
8552 );
8553
8554 let mut subscription = edge_resolved
8556 .track("video")
8557 .unwrap()
8558 .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600)))
8559 .await
8560 .expect("resubscribe");
8561 tokio::time::timeout(Duration::from_secs(5), track.used())
8562 .await
8563 .expect("resubscribe should reach the leaf")
8564 .expect("source open");
8565 let mut group = track.append_group().unwrap();
8566 group.write_frame(crate::Timestamp::ZERO, b"live".as_ref()).unwrap();
8567 group.finish().unwrap();
8568 let mut group = subscription.recv_group().await.unwrap().unwrap();
8569 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"cached");
8570 drop(group);
8571 let mut group = subscription.recv_group().await.unwrap().unwrap();
8572 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live");
8573 drop(group);
8574 drop(subscription);
8575
8576 tokio::time::timeout(Duration::from_secs(5), track.unused())
8577 .await
8578 .expect("second chained unused should resolve far below TRACK_IDLE_LINGER")
8579 .expect("source closed");
8580
8581 let cached = edge_resolved.track("video").unwrap().cached_groups();
8582 assert_eq!(
8583 cached.iter().map(|(group, _)| group.sequence).collect::<Vec<_>>(),
8584 vec![0, 1],
8585 "repeated demand keeps every complete group while releasing its source"
8586 );
8587
8588 let fetch = edge_resolved.track("video").unwrap().fetch_group(2, None);
8589 let mut fetch = std::pin::pin!(fetch);
8590 assert!(futures::poll!(fetch.as_mut()).is_pending(), "fetch should re-splice");
8591 tokio::time::timeout(Duration::from_secs(5), track.used())
8592 .await
8593 .expect("fetch should reach the leaf")
8594 .expect("source open");
8595 let mut group = track.append_group().unwrap();
8596 group.write_frame(crate::Timestamp::ZERO, b"fetched".as_ref()).unwrap();
8597 group.finish().unwrap();
8598 let mut group = tokio::time::timeout(Duration::from_secs(5), fetch)
8599 .await
8600 .expect("re-spliced source should answer the fetch")
8601 .expect("fetch succeeds");
8602 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"fetched");
8603 }
8604
8605 #[tokio::test]
8609 async fn incompatible_local_source_keeps_the_incumbent() {
8610 let producer = origin(1).produce();
8611 let consumer = producer.consume();
8612
8613 let first = producer.publish("room/alice", Route::default()).unwrap();
8614 let track = first.create_track("video", None).unwrap();
8615 let resolved = consumer.request_broadcast("room/alice").await.expect("resolves");
8616 let mut subscription = resolved
8617 .track("video")
8618 .unwrap()
8619 .subscribe(None)
8620 .await
8621 .expect("subscribe");
8622 let mut group = track.append_group().unwrap();
8623 group.write_frame(crate::Timestamp::ZERO, b"before".as_ref()).unwrap();
8624 group.finish().unwrap();
8625 let mut group = subscription.recv_group().await.unwrap().unwrap();
8626 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"before");
8627
8628 let second = producer.publish("room/alice", Route::default()).unwrap();
8630 let _incompatible = second
8631 .create_track("video", track::Info::default().with_timescale(crate::Timescale::MICRO))
8632 .unwrap();
8633 for _ in 0..10 {
8634 tokio::task::yield_now().await;
8635 }
8636
8637 let mut group = track.append_group().unwrap();
8639 group.write_frame(crate::Timestamp::ZERO, b"still".as_ref()).unwrap();
8640 group.finish().unwrap();
8641 let mut group = subscription.recv_group().await.unwrap().unwrap();
8642 assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"still");
8643
8644 drop(track);
8646 first.close();
8647 assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported)));
8648 }
8649
8650 #[tokio::test]
8651 async fn multiple_scopes_present_one_broad_prefix() {
8652 let producer = origin(1).produce();
8653 let _a = producer.announce("", Route::default()).unwrap();
8654
8655 let consumer = producer.consume().scope("", &scopes(&["alpha", "beta"])).unwrap();
8656 let mut announced = consumer.announced();
8657 announced.assert_next_active("");
8658 announced.assert_next_wait();
8659 }
8660
8661 #[test]
8662 fn scope_accepts_every_pattern_union() {
8663 let producer = origin(1).produce();
8664
8665 let root = producer.scope("", &Patterns::from(Pattern::all())).unwrap();
8667 assert_eq!(root.allowed(), Patterns::from(Pattern::all()));
8668
8669 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8671 assert_eq!(scoped.allowed(), scopes(&["room"]));
8672
8673 let multi = producer.scope("", &scopes(&["room", "room/chat", "anon"])).unwrap();
8675 assert_eq!(multi.allowed(), scopes(&["room", "anon"]));
8676
8677 let consumer = producer.consume().scope("", &scopes(&["room"])).unwrap();
8679 assert_eq!(consumer.allowed(), scopes(&["room"]));
8680
8681 for text in ["room", "", "*room", "room/*", "*", "**/room", "room/**/chat", "*.hang"] {
8682 let union = Patterns::from(text.parse::<Pattern>().unwrap());
8683 assert_eq!(producer.scope("", &union).expect(text).allowed(), union, "{text}");
8684 assert_eq!(
8685 producer.consume().scope("", &union).expect(text).allowed(),
8686 union,
8687 "{text}"
8688 );
8689 }
8690
8691 let mixed: Patterns = ["room/**".parse().unwrap(), "other".parse().unwrap()]
8692 .into_iter()
8693 .collect();
8694 assert_eq!(producer.scope("", &mixed).unwrap().allowed(), mixed);
8695 }
8696
8697 #[test]
8698 fn route_table_prunes_to_empty() {
8699 let producer = origin(1).produce();
8700 let consumer = producer.consume();
8701
8702 let cursor = consumer
8705 .scope("", &scopes(&["room/a", "other/deep/head"]))
8706 .unwrap()
8707 .announced();
8708 let route = producer.announce("room/a/b/c", Route::default()).unwrap();
8709 {
8710 let table = producer.shared.lock();
8711 assert!(table.routes.root.find(Path::new("room/a/b/c").parts()).is_some());
8712 assert!(table.routes.root.find(Path::new("other/deep/head").parts()).is_some());
8713 assert_eq!(table.routes.root.cursors_below, 2);
8714 }
8715
8716 drop(route);
8717 drop(cursor);
8718 let table = producer.shared.lock();
8719 assert!(table.routes.root.is_empty());
8720 assert_eq!(table.routes.root.cursors_below, 0);
8721 }
8722
8723 #[test]
8727 fn a_published_broadcast_keeps_the_driver_running() {
8728 let (producer, mut driver) = Producer::new(Config::new(origin(1)));
8729 let waiter = kio::Waiter::noop();
8730 let broadcast = producer.create_broadcast("room/a").unwrap();
8731 drop(producer);
8732 assert!(
8733 driver.poll(Instant::now(), &waiter).is_ok(),
8734 "the broadcast is lifecycle work"
8735 );
8736 drop(broadcast);
8737 assert!(matches!(driver.poll(Instant::now(), &waiter), Err(Error::Closed)));
8738 }
8739
8740 #[tokio::test]
8743 async fn a_dynamic_keeps_the_driver_running() {
8744 let (producer, driver) = Producer::new(Config::new(origin(1)));
8745 let run = tokio::spawn(crate::time::run(driver));
8746 let consumer = producer.consume();
8747 let server = producer.dynamic("room", Route::default()).unwrap();
8748 drop(producer);
8749
8750 let pending = consumer.request_broadcast("room/alice");
8751 let request = queued(&server).await;
8752 let source = broadcast::Info::new().produce();
8753 request.accept(&source);
8754 let resolved = pending.await.expect("the dynamic still serves");
8755
8756 drop(resolved);
8757 drop(source);
8758 drop(server);
8759 tokio::time::timeout(Duration::from_secs(5), run)
8760 .await
8761 .expect("driver must finish once the dynamic is gone")
8762 .unwrap();
8763 drop(consumer);
8764 }
8765
8766 #[test]
8767 fn watch_wakes_only_for_covering_changes() {
8768 let producer = origin(1).produce();
8769 let waiter = kio::Waiter::noop();
8770 let watch = producer.shared.lock().watch(&producer.shared, &Path::new("room/a"));
8771 let seen = watch.seen();
8772
8773 let _other = producer.announce("other", Route::default()).unwrap();
8775 let _below = producer.announce("room/a/b", Route::default()).unwrap();
8776 assert!(watch.poll_changed(&waiter, seen).is_pending());
8777
8778 let above = producer.announce("room", Route::default()).unwrap();
8780 assert!(watch.poll_changed(&waiter, seen).is_ready());
8781 let seen = watch.seen();
8782 drop(above);
8783 assert!(watch.poll_changed(&waiter, seen).is_ready());
8784 let seen = watch.seen();
8785
8786 let _beside = producer.create_broadcast("room/b").unwrap();
8788 assert!(watch.poll_changed(&waiter, seen).is_pending());
8789 let _here = producer.create_broadcast("room/a").unwrap();
8790 assert!(watch.poll_changed(&waiter, seen).is_ready());
8791
8792 drop(watch);
8794 let table = producer.shared.lock();
8795 let node = table
8796 .routes
8797 .root
8798 .find(Path::new("room/a").parts())
8799 .expect("route below keeps the node");
8800 assert!(node.watches.is_empty());
8801 assert_eq!(table.routes.root.watches_below, 0);
8802 }
8803
8804 #[test]
8805 fn a_discarded_front_task_unregisters_its_watch() {
8806 let (producer, _driver) = Producer::new(Config {
8807 hop: origin(1),
8808 ..Default::default()
8809 });
8810 let consumer = producer.consume();
8811 let _served = producer.dynamic("room", Route::default()).unwrap();
8812 drop(producer);
8817 let _pending = consumer.request_broadcast("room/a");
8818 }
8819
8820 #[test]
8821 fn create_broadcast_refuses_a_path_no_pattern_can_spell() {
8822 let producer = origin(1).produce();
8823
8824 assert!(matches!(
8828 producer.create_broadcast("room/*"),
8829 Err(Error::InvalidPath(_))
8830 ));
8831 assert!(matches!(
8832 producer.announce("room/**", Route::default()),
8833 Err(Error::InvalidPath(_))
8834 ));
8835 }
8836
8837 #[test]
8838 fn scope_empty_union_grants_nothing() {
8839 let producer = origin(1).produce();
8840
8841 assert!(matches!(producer.scope("", &Patterns::new()), Err(Error::Unauthorized)));
8843 assert!(matches!(
8844 producer.consume().scope("", &Patterns::new()),
8845 Err(Error::Unauthorized)
8846 ));
8847 }
8848
8849 #[test]
8850 fn scope_nests_and_rebases_roots() {
8851 let producer = origin(1).produce();
8852
8853 let scoped = producer.scope("", &scopes(&["room"])).unwrap();
8855 let nested = scoped.scope("", &scopes(&["room/chat"])).unwrap();
8856 assert_eq!(nested.allowed(), scopes(&["room/chat"]));
8857
8858 assert!(matches!(
8860 scoped.scope("", &scopes(&["other"])),
8861 Err(Error::Unauthorized)
8862 ));
8863
8864 let rooted = nested.scope("room/chat", &Patterns::from(Pattern::all())).unwrap();
8866 assert_eq!(rooted.allowed(), scopes(&[""]));
8867
8868 let broadcast = nested.create_broadcast("room/chat/live").unwrap();
8870 assert!(producer.consume().get_broadcast("room/chat/live").is_some());
8871 broadcast.close();
8872 }
8873
8874 #[test]
8875 fn scope_intersects_and_rebases_arbitrary_grants() {
8876 let producer = origin(1).produce();
8877 let rooms = producer
8878 .scope("", &Patterns::from("room/*".parse::<Pattern>().unwrap()))
8879 .unwrap();
8880 let chats = rooms
8881 .scope("", &Patterns::from("*/chat".parse::<Pattern>().unwrap()))
8882 .unwrap();
8883 assert_eq!(chats.allowed(), Patterns::from("room/chat".parse::<Pattern>().unwrap()));
8884
8885 let exact = producer
8886 .scope("", &Patterns::from("room/alice".parse::<Pattern>().unwrap()))
8887 .unwrap();
8888 let rooted = exact.scope("room", &Patterns::from(Pattern::all())).unwrap();
8889 assert_eq!(rooted.allowed(), Patterns::from("alice".parse::<Pattern>().unwrap()));
8890 assert!(matches!(
8891 exact.scope("room/bob", &Patterns::from(Pattern::all())),
8892 Err(Error::Unauthorized)
8893 ));
8894
8895 let broadcast = exact.create_broadcast("room/alice").unwrap();
8896 assert!(matches!(
8897 exact.create_broadcast("room/alice/cam"),
8898 Err(Error::Unauthorized)
8899 ));
8900 assert!(producer.consume().get_broadcast("room/alice").is_some());
8901 drop(broadcast);
8902 }
8903
8904 #[tokio::test]
8905 async fn wildcard_scope_filters_announcements_and_reports_captures() {
8906 let producer = origin(1).produce();
8907 let consumer = producer
8908 .consume()
8909 .scope("", &Patterns::from("room/*/chat".parse::<Pattern>().unwrap()))
8910 .unwrap();
8911 let mut announced = consumer.announced();
8912
8913 let alice = producer.create_broadcast("room/alice/chat").unwrap();
8914 alice.announce(Route::default()).unwrap();
8915 let update = announced.try_next().expect("alice's chat");
8916 assert_eq!(update.prefix.as_str(), "room/alice/chat");
8917 assert_eq!(update.captures, Some(vec!["alice".parse::<Pattern>().unwrap()]));
8918
8919 let audio = producer.create_broadcast("room/alice/audio").unwrap();
8920 audio.announce(Route::default()).unwrap();
8921 announced.assert_next_wait();
8922
8923 let broad = producer.announce("room", Route::default()).unwrap();
8924 let update = announced.try_next().expect("overlapping broad route");
8925 assert_eq!(update.prefix.as_str(), "room");
8926 assert_eq!(update.captures, None, "an overlap does not pin the wildcard");
8927
8928 drop(broad);
8929 drop(audio);
8930 drop(alice);
8931 }
8932
8933 #[tokio::test]
8934 async fn local_broadcast_wins_announcement_ties() {
8935 let producer = origin(1).produce();
8936 let remote = producer.announce("room/alice", Route::default().with_cost(9)).unwrap();
8937 let local = producer.create_broadcast("room/alice").unwrap();
8938 local.announce(Route::default()).unwrap();
8939
8940 let mut announced = producer.consume().announced();
8941 let update = announced.try_next().expect("one winning route");
8942 assert_eq!(update.prefix.as_str(), "room/alice");
8943 assert_eq!(update.route.cost, Cost::default());
8944 announced.assert_next_wait();
8945
8946 drop(local);
8947 drop(remote);
8948 }
8949
8950 #[test]
8955 fn cost_charge_saturates() {
8956 assert_eq!(Cost { warm: 4, cold: 6 }.charged(5), Cost { warm: 9, cold: 11 });
8957 assert_eq!(Cost::new(u64::MAX).charged(10), Cost::new(MAX_COST));
8958
8959 assert_eq!(Cost::UNKNOWN.charged(3).cold, MAX_COST);
8962 }
8963
8964 fn expiring_origin(expiry: Duration) -> Producer {
8966 let pool = cache::Pool::new(cache::Config::default().with_expiry(expiry));
8967 Config {
8968 pool,
8969 ..Config::default()
8970 }
8971 .produce()
8972 }
8973
8974 #[tokio::test(start_paused = true)]
8978 async fn stalled_publisher_open_group_is_reclaimed() {
8979 let expiry = Duration::from_secs(1);
8980 let origin = expiring_origin(expiry);
8981 let broadcast = origin.create_broadcast("test").unwrap();
8982 let track = broadcast.create_track("video", None).unwrap();
8983
8984 let mut stalled = track.append_group().unwrap();
8985 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
8986 let _successor = track.append_group().unwrap();
8990
8991 let mut reading = stalled.consume();
8992 assert!(reading.read_frame().await.unwrap().is_some());
8993
8994 crate::model::clock::advance(expiry * 2);
8996
8997 let reclaimed = tokio::time::timeout(Duration::from_secs(60), reading.read_frame()).await;
9000 assert!(
9001 matches!(reclaimed, Ok(Err(Error::Old))),
9002 "the sweep must reclaim an idle open group and surface the gap, got {reclaimed:?}"
9003 );
9004 }
9005
9006 #[tokio::test(start_paused = true)]
9009 async fn sweep_respects_a_disabled_expiry() {
9010 let origin = Config {
9011 pool: cache::Pool::unbounded(),
9012 ..Config::default()
9013 }
9014 .produce();
9015 let broadcast = origin.create_broadcast("test").unwrap();
9016 let track = broadcast.create_track("video", None).unwrap();
9017
9018 let mut stalled = track.append_group().unwrap();
9019 stalled.write_frame(crate::Timestamp::ZERO, b"x".as_slice()).unwrap();
9020 let _successor = track.append_group().unwrap();
9021
9022 let mut reading = stalled.consume();
9023 assert!(reading.read_frame().await.unwrap().is_some());
9024
9025 crate::model::clock::advance(Duration::from_secs(3600));
9026 tokio::time::advance(Duration::from_secs(3600)).await;
9027
9028 assert!(
9029 reading.read_frame().now_or_never().is_none(),
9030 "a pool without an expiry window never reclaims"
9031 );
9032 }
9033
9034 #[test]
9037 fn drain_cost_is_encodable() {
9038 use crate::coding::Encode;
9039
9040 let mut buf = Vec::new();
9041 Cost::DRAIN
9042 .encode(&mut buf, crate::lite::Version::Lite06)
9043 .expect("a draining route is still forwarded, so its cost must encode");
9044 }
9045}