diff options
Diffstat (limited to 'src/transport/identity.rs')
| -rw-r--r-- | src/transport/identity.rs | 748 |
1 files changed, 748 insertions, 0 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs new file mode 100644 index 0000000..94faded --- /dev/null +++ b/src/transport/identity.rs @@ -0,0 +1,748 @@ +//! The identity layer: between the node and its transports, whose identities are +//! addresses (Mycelium's public keys), so that the rest of the node deals in node IDs +//! (`crate::identity`) only, whatever transport reaches a peer. A node may have several +//! transports; a peer is reached on one both have, so nodes on several transports +//! carry the traffic between the others. +//! +//! A node proves who it is to an address with WHO (which asks back) or ME (the answer): +//! its public node key, and a signature over its own address and the recipient's +//! (`signed`). The transport authenticates the sender's address, so a proof is good for +//! that pair of addresses only: replayed from another address, to another recipient or +//! on another transport, it fails. Once a node proved itself at an address, messages +//! from there are its. +//! +//! * A message from an address no node proved itself at is dropped, and the address is +//! asked WHO (at most every `ASK_INTERVAL`): after a restart, the first message from +//! a peer brings its proof back, and the peer's node retries what was lost. +//! * A seed (`resolve_seed`) is an address, or something one of the transports resolves +//! to one; it is asked WHO, and its ME says which node is there. +//! * Where other nodes are comes from the member list and HELLO_REPLYs +//! (`learn_addresses`): hints, not proofs. A message to a node goes to the addresses +//! it proved itself at, best first (`Paths::rank`, `Paths::send_first`: the next +//! when a send fails); with only hints, those are asked WHO, and the send is refused +//! until one answers (the node retries refused sends). A node proven on some +//! transports is asked WHO at its hints on the others too, now and then +//! (`OTHER_PATHS_ASK_INTERVAL`), so that each transport both have becomes a path. +//! +//! Addresses and the transports behind them are the paths layer's +//! (`transport::paths`): the books keep addresses as paths keep them. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, anyhow, bail}; +use async_trait::async_trait; +use tokio::sync::oneshot; +use tokio::time::Instant; +use tracing::{debug, info}; + +use crate::identity::{self, NodeKey}; +use crate::locks::MutexExt; +use crate::transport::firewall::Firewall; +use crate::transport::paths::{Gate, PathHandler, Paths, scheme_of}; +use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats}; +use crate::transport::traffic::Traffic; +use crate::transport::{ + Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, + ParsedMessage, Transport, +}; + +pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1"; +pub(crate) const ME_LABEL: &[u8] = b"continuum/me/1"; +/// An address is asked WHO at most this often, unless a seed is being resolved. +const ASK_INTERVAL: Duration = Duration::from_secs(5); +/// A node's addresses on transports it hasn't proven itself on are asked WHO at most +/// this often (`Book::unproven_hints`): over Tor, a WHO is a dial that takes seconds, +/// and an address that doesn't answer shouldn't be pressed. +const OTHER_PATHS_ASK_INTERVAL: Duration = Duration::from_secs(60); +/// How long resolving a seed waits for its ME. +const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10); +/// Addresses kept per node from hints: about one per transport. +const MAX_HINTS: usize = 8; +/// Addresses with a proven node, nodes with hints, and addresses asked, each at most: +/// anyone can send us proofs (keys are free), so the books are bounded. +const MAX_ENTRIES: usize = 4096; + +/// A send refused because the peer hasn't proven where it is yet (`send_encoded`): +/// expected when a peer known only from hints is first contacted, and retried. +#[derive(Debug, thiserror::Error)] +#[error("{0} hasn't proven where it is yet; asked where it is said to be")] +pub(crate) struct NotProven(String); + +/// Whether `error` is (or was caused by) a [`NotProven`]. +pub(crate) fn is_not_proven(error: &anyhow::Error) -> bool { + error.chain().any(|cause| cause.is::<NotProven>()) +} + +/// What a node under the identity layer needs (`node::NodeConfig::identity`). +#[derive(Clone)] +pub(crate) struct IdentityConfig { + pub(crate) key: Arc<NodeKey>, + /// The address scheme of the node's transport (`NodeConfig::transport`), e.g. + /// `mycelium`. + pub(crate) scheme: &'static str, + /// Further transports, by scheme, after the first in preference. + pub(crate) more: Vec<(&'static str, Arc<dyn Transport>)>, +} + +/// The bytes a WHO (`label`: `WHO_LABEL`) or ME proof signs: the label, then the +/// sender's and the recipient's address, each with its length. +pub(crate) fn signed(label: &[u8], from: &str, to: &str) -> Vec<u8> { + let mut bytes = label.to_vec(); + for address in [from, to] { + bytes.extend_from_slice(&(address.len() as u32).to_be_bytes()); + bytes.extend_from_slice(address.as_bytes()); + } + bytes +} + +pub(crate) struct IdentityTransport { + shared: Arc<Shared>, +} + +struct Shared { + paths: Arc<Paths>, + /// The debug firewall, by node ID; paths applies it (`Gate`). + firewall: Option<Arc<Firewall>>, + key: Arc<NodeKey>, + node_id: String, + /// WHO and ME are counted here, under the address (`traffic`). + traffic: Arc<Traffic>, + // A std mutex: every access is short and never spans an await. + book: Mutex<Book>, +} + +/// Who is where, by address. +#[derive(Default)] +struct Book { + /// The node that proved itself at an address, and when. + at: HashMap<String, (String, Instant)>, + /// The addresses each node last proved itself at, one per scheme. + proven: HashMap<String, Vec<String>>, + /// Where others say a node is. + hints: HashMap<String, Vec<String>>, + /// When each address was last asked WHO. + asked: HashMap<String, Instant>, + /// Seeds being resolved: who waits for a ME from where. + waiting: HashMap<String, Vec<oneshot::Sender<String>>>, + /// Since start (`stats`). + counts: IdentityCounts, +} + +impl Book { + /// `node` proved itself at `address`. Returns whether that is news. + fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool { + let news = self.at.get(address).is_none_or(|(known, _)| known != node); + // One address per node and scheme, and one node per address: a node that + // moved, or an address taken over by another node, leaves no stale entry. + let scheme = scheme_of(address); + let addresses = self.proven.entry(node.to_owned()).or_default(); + if let Some(slot) = addresses.iter_mut().find(|at| scheme_of(at) == scheme) { + if *slot != address { + let previous = std::mem::replace(slot, address.to_owned()); + self.at.remove(&previous); + self.counts.address_changes += 1; + } + } else { + addresses.push(address.to_owned()); + } + if let Some((previous, _)) = self.at.insert(address.to_owned(), (node.to_owned(), now)) + && previous != node + { + self.forget_proof(&previous, address); + } + if self.at.len() > MAX_ENTRIES + && let Some(oldest) = self + .at + .iter() + .min_by_key(|(_, (_, at))| *at) + .map(|(address, _)| address.clone()) + && let Some((node, _)) = self.at.remove(&oldest) + { + self.forget_proof(&node, &oldest); + } + for waiter in self.waiting.remove(address).into_iter().flatten() { + let _ = waiter.send(node.to_owned()); + } + news + } + + fn forget_proof(&mut self, node: &str, address: &str) { + if let Some(addresses) = self.proven.get_mut(node) { + addresses.retain(|at| at != address); + if addresses.is_empty() { + self.proven.remove(node); + } + } + } + + /// What the books hold now, and the counts. + fn stats(&self) -> IdentityStats { + let mut proven = std::collections::BTreeMap::new(); + for addresses in self.proven.values() { + for address in addresses { + *proven.entry(scheme_of(address).to_owned()).or_insert(0) += 1; + } + } + IdentityStats { + proven, + hinted: self + .hints + .keys() + .filter(|node| !self.proven.contains_key(*node)) + .count(), + counts: self.counts, + } + } + + /// Where `node` is said to be on transports it hasn't proven itself on yet. + fn unproven_hints(&self, node: &str) -> Vec<String> { + let proven = self.proven.get(node); + self.hints + .get(node) + .into_iter() + .flatten() + .filter(|hint| { + proven.is_none_or(|proven| { + !proven + .iter() + .any(|address| scheme_of(address) == scheme_of(hint)) + }) + }) + .cloned() + .collect() + } + + /// Whether to ask `address` WHO now (always, if `force`), noting that it was asked. + fn ask(&mut self, address: &str, now: Instant, force: bool) -> bool { + let interval = if force { Duration::ZERO } else { ASK_INTERVAL }; + self.ask_every(address, now, interval) + } + + /// Whether to ask `address` WHO now, if it wasn't within `interval`, noting that it + /// was asked. + fn ask_every(&mut self, address: &str, now: Instant, interval: Duration) -> bool { + if self + .asked + .get(address) + .is_some_and(|at| now.duration_since(*at) < interval) + { + return false; + } + if self.asked.len() >= MAX_ENTRIES { + // The longest interval: other paths' asks are kept until they are due again. + self.asked + .retain(|_, at| now.duration_since(*at) < OTHER_PATHS_ASK_INTERVAL); + } + self.asked.insert(address.to_owned(), now); + true + } +} + +impl IdentityTransport { + /// Over `transport`, whose addresses have `config.scheme`, and `config.more`. + /// `firewall`: the debug firewall, if any, which paths applies by address. + pub(crate) fn new( + transport: Arc<dyn Transport>, + config: IdentityConfig, + traffic: Arc<Traffic>, + firewall: Option<Arc<Firewall>>, + ) -> Self { + let node_id = config.key.id(); + let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| { + let gate: std::sync::Weak<dyn Gate> = shared.clone(); + Shared { + paths: Arc::new(Paths::new( + std::iter::once((config.scheme, transport)) + .chain(config.more) + .collect(), + Some(Arc::clone(&traffic)), + firewall.is_some().then_some(gate), + )), + firewall, + key: config.key, + node_id, + traffic, + book: Mutex::default(), + } + }); + Self { shared } + } +} + +/// The firewall's rules are by node ID: an address is blocked if the node proven at it +/// is. Addresses no node proved itself at aren't (nothing could name them). +impl Gate for Shared { + fn blocks_inbound(&self, address: &str) -> bool { + self.blocks(address, Firewall::blocks_inbound) + } + + fn blocks_outbound(&self, address: &str) -> bool { + self.blocks(address, Firewall::blocks_outbound) + } +} + +impl Shared { + fn blocks(&self, address: &str, rule: fn(&Firewall, &str) -> bool) -> bool { + let Some(firewall) = &self.firewall else { + return false; + }; + let node = self.lock().at.get(address).map(|(node, _)| node.clone()); + node.is_some_and(|node| rule(firewall, &node)) + } +} + +impl Shared { + fn lock(&self) -> MutexGuard<'_, Book> { + self.book.lock_or_recover() + } + + /// Our proof for `address` (as paths keep it), from our address on its transport. + fn proof(&self, label: &[u8], address: &str) -> Option<IdentityProof> { + let message = signed(label, &self.paths.own_address(address)?, address); + Some(IdentityProof { + key: self.key.public(), + signature: self.key.sign(&message), + }) + } + + /// Notes the node `proof` proves to be at `address` (as paths keep it); None if it + /// proves nothing. + fn prove(&self, address: &str, label: &[u8], proof: &IdentityProof) -> Option<String> { + let message = signed(label, address, &self.paths.own_address(address)?); + if !identity::verify(&proof.key, &message, &proof.signature) { + self.lock().counts.proofs_rejected += 1; + debug!(from = %address, "Ignoring an identity proof that doesn't hold"); + return None; + } + let node = identity::node_id_of(&proof.key); + if node == self.node_id { + debug!(from = %address, "Ignoring a proof of our own node key"); + return None; + } + if self.lock().bind(address, &node, Instant::now()) { + debug!(node = %node, address = %address, "Node proved its address"); + } + Some(node) + } + + /// Asks `address` WHO, unless it was asked recently (`force`: anyway). + fn ask(self: &Arc<Self>, address: &str, force: bool) { + if !self.lock().ask(address, Instant::now(), force) { + return; + } + self.send_who(address); + } + + /// Asks `node`'s addresses on transports it hasn't proven itself on WHO, now and then + /// (`OTHER_PATHS_ASK_INTERVAL`): its ME there proves another path to it, which the + /// paths layer then measures, and fails over to. + fn ask_other_paths(self: &Arc<Self>, node: &str) { + let due: Vec<String> = { + let mut book = self.lock(); + let now = Instant::now(); + book.unproven_hints(node) + .into_iter() + .filter(|address| book.ask_every(address, now, OTHER_PATHS_ASK_INTERVAL)) + .collect() + }; + for address in due { + self.send_who(&address); + } + } + + fn send_who(self: &Arc<Self>, address: &str) { + let shared = Arc::clone(self); + let address = address.to_owned(); + tokio::spawn(async move { shared.send_proof(MessageKind::Who, &address).await }); + } + + /// Sends a WHO or ME (`kind`) to `address`. + async fn send_proof(&self, kind: MessageKind, address: &str) { + let Some(address) = self.paths.canonical(address) else { + return; + }; + let address = address.as_str(); + let message = if kind == MessageKind::Who { + self.proof(WHO_LABEL, address).map(ParsedMessage::Who) + } else { + self.proof(ME_LABEL, address).map(ParsedMessage::Me) + }; + let Some(message) = message else { + return; + }; + let encoded = match self.paths.encode(address, &message) { + Ok(encoded) => encoded, + Err(error) => { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof"); + return; + } + }; + match self.paths.send(&encoded).await { + Ok(()) => self + .traffic + .sent(address, encoded.kind, encoded.payload_len), + Err(error) => { + self.traffic + .refused(address, encoded.kind, encoded.payload_len); + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send an identity proof"); + } + } + } + + /// A message from `address`. + async fn receive( + self: Arc<Self>, + address: String, + message: ParsedMessage, + payload_len: usize, + handler: &HandlerFn, + ) -> Result<()> { + match message { + ParsedMessage::Who(proof) => { + self.traffic + .received(&address, MessageKind::Who, payload_len); + if self.prove(&address, WHO_LABEL, &proof).is_some() { + let shared = Arc::clone(&self); + tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await }); + } + Ok(()) + } + ParsedMessage::Me(proof) => { + self.traffic + .received(&address, MessageKind::Me, payload_len); + self.prove(&address, ME_LABEL, &proof); + Ok(()) + } + message => { + let node = self.lock().at.get(&address).map(|(node, _)| node.clone()); + match node { + Some(node) => handler(node, message, payload_len).await, + None => { + self.lock().counts.dropped_unproven += 1; + debug!( + from = %address, + kind = ?message.kind(), + "Dropping a message from an address no node proved itself at; asking who is there" + ); + self.ask(&address, false); + Ok(()) + } + } + } + } + } + + /// Where to send to `peer`: an address it proved itself at, else one it is said to + /// be at; the most preferred transport's first. + fn destination(&self, peer: &str) -> Option<String> { + self.proven_at(peer) + .into_iter() + .next() + .or_else(|| self.hinted_at(peer).into_iter().next()) + } + + /// The addresses `peer` proved itself at, best first (`Paths::rank`). + fn proven_at(&self, peer: &str) -> Vec<String> { + let proven = self.lock().proven.get(peer).cloned().unwrap_or_default(); + self.paths.rank(&proven) + } + + /// Where `peer` is said to be, best first. + fn hinted_at(&self, peer: &str) -> Vec<String> { + let hints = self.lock().hints.get(peer).cloned().unwrap_or_default(); + self.paths.rank(&hints) + } +} + +#[async_trait] +impl Transport for IdentityTransport { + fn local_id(&self) -> &str { + &self.shared.node_id + } + + fn is_peer_id(&self, id: &str) -> bool { + identity::is_node_id(id) + } + + fn delivery_limit(&self) -> Duration { + self.shared.paths.delivery_limit() + } + + fn local_addresses(&self) -> Vec<String> { + self.shared.paths.local_addresses() + } + + /// Where `peer` proved itself, else where it is said to be, as others can use it. + fn addresses_of(&self, peer: &str) -> Vec<String> { + let addresses = { + let book = self.shared.lock(); + book.proven + .get(peer) + .or_else(|| book.hints.get(peer)) + .cloned() + .unwrap_or_default() + }; + let mut locations = Vec::new(); + for address in addresses { + for location in self.shared.paths.locations(&address) { + if !locations.contains(&location) { + locations.push(location); + } + } + } + locations + } + + fn learn_addresses(&self, peer: &str, addresses: &[String]) { + if peer == self.shared.node_id || !identity::is_node_id(peer) { + return; + } + let mut known: Vec<String> = Vec::new(); + for address in addresses { + if let Some(address) = self.shared.paths.canonical(address) + && !self.shared.paths.is_own(&address) + && !known.contains(&address) + { + known.push(address); + } + } + known.truncate(MAX_HINTS); + if known.is_empty() { + return; + } + let mut book = self.shared.lock(); + if book.hints.len() < MAX_ENTRIES || book.hints.contains_key(peer) { + book.hints.insert(peer.to_owned(), known); + } + } + + fn can_reach(&self, peer: &str) -> bool { + self.shared.destination(peer).is_some() + } + + fn transports(&self) -> Vec<String> { + self.shared.paths.transports() + } + + /// The address `encode` picks, the transport's connection there, and every path + /// to `peer` as paths measures it, best first. + fn destination(&self, peer: &str) -> Option<Destination> { + let proven = self.shared.proven_at(peer); + let (address, is_proven) = match proven.first() { + Some(address) => (address.clone(), true), + None => (self.shared.hinted_at(peer).into_iter().next()?, false), + }; + let connection = self.shared.paths.connection(&address); + Some(Destination { + address, + proven: is_proven, + connection, + paths: self.shared.paths.info(&proven), + }) + } + + /// The paths layer answers them (`Paths::receive`). + fn answers_pings(&self) -> bool { + true + } + + /// The best path's, where messages to `peer` go. + fn rtt(&self, peer: &str) -> Option<Duration> { + let address = self.shared.proven_at(peer).into_iter().next()?; + self.shared.paths.rtt(&address) + } + + /// The identity layer's, and each transport's. + fn stats(&self) -> TransportStats { + let mut stats = self.shared.paths.stats(); + stats.merge(TransportStats { + identity: Some(self.shared.lock().stats()), + ..TransportStats::default() + }); + stats + } + + fn discovered_seeds(&self) -> Vec<String> { + self.shared.paths.discovered_seeds() + } + + /// A node ID resolves to itself if this node knows where it is; anything else is + /// an address (`<scheme>:...`) or what one of the transports resolves to one (the + /// first that does), asked WHO. + async fn resolve_seed(&self, seed: &str) -> Result<String> { + if identity::is_node_id(seed) { + if self.shared.destination(seed).is_some() { + return Ok(seed.to_owned()); + } + bail!("a node ID alone doesn't say where that node is; give its address"); + } + let address = self.shared.paths.resolve(seed).await?; + if self.shared.paths.is_own(&address) { + bail!("{seed} is this node"); + } + let receiver = { + let mut book = self.shared.lock(); + if let Some((node, _)) = book.at.get(&address) { + return Ok(node.clone()); + } + let (sender, receiver) = oneshot::channel(); + let waiting = book.waiting.entry(address.clone()).or_default(); + waiting.retain(|waiter| !waiter.is_closed()); + waiting.push(sender); + receiver + }; + self.shared.ask(&address, true); + match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await { + Ok(Ok(node)) => { + info!(seed = %seed, node = %node, "Seed is node"); + Ok(node) + } + _ => { + self.shared.lock().counts.seeds_unanswered += 1; + Err(anyhow!( + "no node answered at {address} within {RESOLVE_TIMEOUT:?}" + )) + } + } + } + + /// For the address `destination` picks (a send there is refused until `peer` + /// proved itself there, `send_encoded`). + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let Some(address) = self.shared.destination(peer) else { + bail!("no address known for {peer} on this node's transports"); + }; + let mut encoded = self.shared.paths.encode(&address, message)?; + encoded.peer = peer.to_owned(); + Ok(encoded) + } + + /// To the addresses `peer` proved itself at, best first, the next when a send + /// fails (`Paths::send_first`); refused if it proved itself nowhere yet. + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let proven = self.shared.proven_at(&message.peer); + if !proven.is_empty() { + self.shared.ask_other_paths(&message.peer); + } + if proven.is_empty() { + let hints = { + let mut book = self.shared.lock(); + book.counts.refused_unproven += 1; + book.hints.get(&message.peer).cloned().unwrap_or_default() + }; + for address in hints { + self.shared.ask(&address, false); + } + return Err(NotProven(message.peer.clone()).into()); + } + self.shared + .paths + .send_first(message, &proven) + .await + .map(|_| ()) + } + + /// Every transport's consumer (`Paths::start_consumer`); returns when the first + /// one does. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let handler: Arc<HandlerFn> = Arc::from(handler); + let shared = Arc::clone(&self.shared); + let receive: Arc<PathHandler> = Arc::new(move |address, message, payload_len| { + let shared = Arc::clone(&shared); + let handler = Arc::clone(&handler); + Box::pin(async move { + shared + .receive(address, message, payload_len, &*handler) + .await + }) + }); + self.shared.paths.start_consumer(receive).await + } + + /// When the first transport exits. + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + self.shared.paths.wait_for_exit().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn signed_bytes_say_both_addresses() { + // A length before each address: "a:bc" to "d" isn't "a:b" to "cd". + assert_ne!( + signed(WHO_LABEL, "a:bc", "d"), + signed(WHO_LABEL, "a:b", "cd") + ); + assert_ne!(signed(WHO_LABEL, "a", "b"), signed(ME_LABEL, "a", "b")); + assert_ne!(signed(WHO_LABEL, "a", "b"), signed(WHO_LABEL, "b", "a")); + } + + #[test] + fn one_address_per_node_and_scheme_and_one_node_per_address() { + let now = Instant::now(); + let mut book = Book::default(); + assert!(book.bind("s:x", "n1", now)); + assert!(!book.bind("s:x", "n1", now)); + // n1 moved to s:y: s:x is no longer n1's. + assert!(book.bind("s:y", "n1", now)); + assert!(!book.at.contains_key("s:x")); + assert_eq!(book.proven["n1"], ["s:y"]); + // On another transport too. + assert!(book.bind("t:x", "n1", now)); + assert_eq!(book.proven["n1"], ["s:y", "t:x"]); + // n2 took over s:y: n1 is proven on t only. + assert!(book.bind("s:y", "n2", now)); + assert_eq!(book.proven["n1"], ["t:x"]); + assert_eq!(book.at["s:y"].0, "n2"); + // Only n1's move from s:x to s:y changed an address. + assert_eq!(book.counts.address_changes, 1); + let stats = book.stats(); + assert_eq!(stats.proven["s"], 1); + assert_eq!(stats.proven["t"], 1); + } + + #[test] + fn other_paths_are_the_hints_on_transports_not_proven() { + let now = Instant::now(); + let mut book = Book::default(); + book.hints + .insert("n1".into(), vec!["s:x".into(), "t:x".into(), "u:x".into()]); + // Proven nowhere: every hint. + assert_eq!(book.unproven_hints("n1"), ["s:x", "t:x", "u:x"]); + // Proven on s (even at another address): the others. + book.bind("s:y", "n1", now); + assert_eq!(book.unproven_hints("n1"), ["t:x", "u:x"]); + assert!(book.ask_every("t:x", now, OTHER_PATHS_ASK_INTERVAL)); + assert!(!book.ask_every("t:x", now + ASK_INTERVAL, OTHER_PATHS_ASK_INTERVAL)); + assert!(book.ask_every( + "t:x", + now + OTHER_PATHS_ASK_INTERVAL, + OTHER_PATHS_ASK_INTERVAL + )); + assert!(book.unproven_hints("n2").is_empty()); + } + + #[test] + fn asking_is_rate_limited() { + let now = Instant::now(); + let mut book = Book::default(); + assert!(book.ask("s:x", now, false)); + assert!(!book.ask("s:x", now, false)); + assert!(book.ask("s:x", now, true)); + assert!(book.ask("s:x", now + ASK_INTERVAL, false)); + } + + #[tokio::test] + async fn a_waiting_seed_learns_the_node() { + let mut book = Book::default(); + let (sender, receiver) = oneshot::channel(); + book.waiting.insert("s:x".into(), vec![sender]); + book.bind("s:x", "n1", Instant::now()); + assert_eq!(receiver.await.unwrap(), "n1"); + } +} |