rateguard 0.7.1

//! The background task: the node, the socket and the ticker.

use std::borrow::Cow;
use std::collections::BTreeSet;
use std::net::SocketAddr;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Weak};
use std::time::Duration;

use rateguard_core::boundary::{Action, Event, PeerId};
use rateguard_core::limiter::KeyRates;
use rateguard_core::node::Node;
use rateguard_proto::auth::{Auth, Rejected};
use rateguard_proto::{Address, Message, Update};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinSet;

use crate::classes::{Resolved, shadow_quota};
use crate::guard::Shared;
use crate::stats::RejectedDatagrams;
use crate::transport::Transport;

/// A datagram as the core gets it; see [`screen`].
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Screened<'a> {
    /// The sender: the member of the first record, which every message
    /// opens with (spec §3). Not the socket address, which a NAT on the
    /// way may have rewritten.
    pub from: PeerId,
    /// The message, without the records dropped.
    pub bytes: Cow<'a, [u8]>,
    /// How many records were dropped.
    pub dropped: u64,
}

/// Checks that every record names its member by the ID its address gives,
/// as every node derives its own: news is about the node at an address, and
/// an ID unhooked from it is either a made-up member or someone speaking
/// for another under a borrowed name. With gossip unsigned it is the one
/// check a node can make of news; a conforming peer, 0.7.0 included, never
/// makes up such a record.
///
/// `None`, for a malformed datagram, also when the sender's own record
/// fails the check: the message speaks for nobody then. A later record that
/// fails it is dropped alone, the rest re-encoded for the core. A peer
/// passes on news as it heard it, and a 0.7.0 one passes on a forger's;
/// dropping its whole datagram for that would lose its own record, its
/// demand and its answer to a probe. Re-encoding after `Auth::open` is
/// safe: the signature, if any, has been checked on the bytes as sent.
pub(crate) fn screen(bytes: &[u8]) -> Option<Screened<'_>> {
    let matches = |update: &Update| update.member == crate::key::peer_id(update.addr).get();
    let mut message = rateguard_proto::decode(bytes).ok()?;
    let opening = message.updates().first()?;
    if !matches(opening) {
        return None;
    }
    let from = PeerId::new(opening.member);
    if message.updates().iter().all(matches) {
        return Some(Screened {
            from,
            bytes: Cow::Borrowed(bytes),
            dropped: 0,
        });
    }
    let updates = match &mut message {
        Message::Ping { updates, .. }
        | Message::Ack { updates, .. }
        | Message::PingReq { updates, .. }
        | Message::Leave { updates } => updates,
        // A message of a later version this one cannot take apart: whole
        // or not at all, and with a record failing the check, not at all.
        _ => return None,
    };
    let before = updates.len();
    updates.retain(matches);
    let dropped = (before - updates.len()) as u64;
    let bytes = rateguard_proto::encode(&message).ok()?;
    Some(Screened {
        from,
        bytes: Cow::Owned(bytes),
        dropped,
    })
}

/// What a `Guard` asks of the background task, which alone owns the core.
#[derive(Debug)]
pub(crate) enum Query {
    /// The rates of the key with this hash, if the core tracks it.
    Key {
        hash: u64,
        reply: oneshot::Sender<Option<KeyRates>>,
    },
}

/// Runs the node until it is asked to leave or the last `Guard` is gone,
/// then leaves the cluster.
pub(crate) async fn run<T: Transport>(
    mut node: Node,
    transport: Arc<T>,
    mut names: Names,
    shared: Weak<Shared>,
    mut ticker: tokio::time::Interval,
    mut leave: mpsc::Receiver<oneshot::Sender<()>>,
    mut queries: mpsc::Receiver<Query>,
) {
    let Some((epoch, mut auth)) = shared.upgrade().map(|shared| (shared.epoch, shared.auth()))
    else {
        return;
    };
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    // Sending is given one tick: a transport that never completes a send
    // must not freeze the node, nor keep it alive once the last Guard is
    // gone.
    let budget = ticker.period();
    // A signed frame is FRAME_OVERHEAD bytes over its message, and the
    // largest message still fits framed in MAX_DATAGRAM (rateguard-proto's
    // `the_largest_message_fits_framed`): this buffer takes any datagram a
    // peer sends. A longer one is cut short, and refused.
    let mut buf = vec![0u8; rateguard_proto::MAX_DATAGRAM];
    let mut done = None;
    // Refusals since the last tick, for the report.
    let mut rejected = RejectedDatagrams::default();
    loop {
        // The shared state is held only while the node works, never across
        // an await: the last Guard dropped must be noticed.
        let actions = tokio::select! {
            received = transport.recv_from(&mut buf) => {
                let Some(shared) = shared.upgrade() else { break };
                match received {
                    Ok((len, _)) => {
                        let auth = shared.auth();
                        receive(&mut node, &auth, &buf[..len], shared.now(), &mut rejected)
                    }
                    Err(_) => Vec::new(),
                }
            }
            _ = ticker.tick() => {
                let Some(shared) = shared.upgrade() else { break };
                sync(&mut node, &shared, std::mem::take(&mut rejected))
            }
            found = names.next() => {
                names.update(&mut node, found);
                Vec::new()
            }
            // The branch is off once every sender is gone, which only
            // happens with the last Guard.
            Some(query) = queries.recv() => {
                match query {
                    Query::Key { hash, reply } => {
                        let _ = reply.send(node.limiter().key_rates(hash));
                    }
                }
                Vec::new()
            }
            // None: the last Guard is gone.
            asked = leave.recv() => {
                done = asked;
                break;
            }
        };
        // The last setting seen is kept for leaving: the news must reach
        // the peers signed even once the last Guard is gone.
        if let Some(shared) = shared.upgrade() {
            auth = shared.auth();
        }
        send(&node, &*transport, actions, &auth, budget).await;
    }

    // The quotas drop to the floor before the peers hear: from the moment
    // they do, they no longer count this node.
    if let Some(shared) = shared.upgrade() {
        hold_at_floor(&shared, node.limiter().config());
        shared.running.store(false, Ordering::Relaxed);
    }
    let now = epoch.elapsed().as_nanos() as u64;
    let actions = node.leave(now).to_vec();
    send(&node, &*transport, actions, &auth, budget).await;
    if let Some(done) = done {
        let _ = done.send(());
    }
}

/// How often the names among the seeds are looked up again once the node
/// has peers. While it has none, every reconnect interval: a fleet started
/// all at once may find nobody behind the name at first.
const RESOLVE_EVERY: Duration = Duration::from_secs(30);

/// The most time a round waits for a name: the names are looked up all at
/// once, so a stalled lookup holds a round up by this much at most, and a
/// timeout keeps that name's last addresses, just like a failed lookup.
/// Well above the system resolver's own 5 s retry, so that a lookup slow
/// for a lost packet still resolves rather than time out every round.
const RESOLVE_TIMEOUT: Duration = Duration::from_secs(20);

/// How many lookups in a row must find this node alone behind its names
/// before it takes the fleet to be itself alone: about 4 s at the default
/// reconnect interval. A fleet whose replicas become ready further apart
/// than that runs over the limit until they find each other.
const ALONE_AFTER: u32 = 3;

/// What each name among the seeds resolved to, in order, `None` where the
/// lookup failed.
type Found = Vec<Option<Vec<SocketAddr>>>;

/// Looks the names among the seeds up, now and then again, until the node
/// is gone. On its own task: a lookup may take seconds, the node's rounds
/// may not.
pub(crate) async fn resolve<T: Transport>(
    transport: Arc<T>,
    names: Vec<(String, u16)>,
    shared: Weak<Shared>,
    retry: Duration,
    found: mpsc::Sender<Found>,
) {
    loop {
        let mut pending = JoinSet::new();
        for (i, (host, port)) in names.iter().cloned().enumerate() {
            let transport = transport.clone();
            pending.spawn(async move {
                let lookup = tokio::time::timeout(RESOLVE_TIMEOUT, transport.resolve(&host, port));
                (i, lookup.await.ok().and_then(Result::ok))
            });
        }
        let lookups = async {
            let mut all = vec![None; names.len()];
            while let Some(done) = pending.join_next().await {
                if let Ok((i, addrs)) = done {
                    all[i] = addrs;
                }
            }
            all
        };
        // The node may stop meanwhile; the transport must go with it.
        let all = tokio::select! {
            all = lookups => all,
            _ = found.closed() => return,
        };
        if found.send(all).await.is_err() {
            return;
        }
        let Some(alone) = shared
            .upgrade()
            .map(|shared| shared.cluster_size.load(Ordering::Relaxed) <= 1)
        else {
            return;
        };
        let wait = if alone { retry } else { RESOLVE_EVERY };
        tokio::select! {
            _ = tokio::time::sleep(wait) => {}
            _ = found.closed() => return,
        }
    }
}

/// The seeds that come from names: added as their addresses appear,
/// removed as they go, so that the seed list follows the fleet rather than
/// grow with every instance it ever had.
#[derive(Debug)]
pub(crate) struct Names {
    local: PeerId,
    lookups: Option<mpsc::Receiver<Found>>,
    /// The last addresses of each name, kept through a failed lookup.
    last: Vec<Vec<SocketAddr>>,
    /// Seeds given as addresses: never removed.
    kept: BTreeSet<PeerId>,
    /// Seeds added from names.
    added: BTreeSet<PeerId>,
    /// Lookups in a row that found this node and nobody else.
    alone: u32,
}
impl Names {
    pub fn new(local: PeerId) -> Self {
        Self {
            local,
            lookups: None,
            last: Vec::new(),
            kept: BTreeSet::new(),
            added: BTreeSet::new(),
            alone: 0,
        }
    }

    /// A seed given as an address, which no lookup removes.
    pub fn keep(&mut self, seed: PeerId) {
        self.kept.insert(seed);
    }

    pub fn listen(&mut self, lookups: mpsc::Receiver<Found>) {
        self.lookups = Some(lookups);
    }

    // The next round of lookups; never, if there are no names. Cancel-safe.
    async fn next(&mut self) -> Found {
        if let Some(lookups) = &mut self.lookups
            && let Some(found) = lookups.recv().await
        {
            return found;
        }
        self.lookups = None;
        std::future::pending().await
    }

    fn update(&mut self, node: &mut Node, found: Found) {
        // A replica running alone finds itself behind its name and nobody
        // else, and would wait for peers for good. Nobody at all, or a
        // failed lookup, is no such sign: a fleet started at once may not
        // be listed yet.
        let only_me = found.iter().all(|found| {
            found.as_ref().is_some_and(|found| {
                found
                    .iter()
                    .all(|&address| crate::key::peer_id(address.into()) == self.local)
            })
        }) && found.iter().flatten().any(|found| !found.is_empty());
        self.alone = if only_me {
            self.alone.saturating_add(1)
        } else {
            0
        };
        if self.alone == ALONE_AFTER {
            node.expect_peers(false);
        }

        self.last.resize(found.len(), Vec::new());
        for (last, found) in self.last.iter_mut().zip(found) {
            if let Some(found) = found {
                *last = found;
            }
        }
        let mut current = BTreeSet::new();
        for &address in self.last.iter().flatten() {
            let address = Address::from(address);
            let id = crate::key::peer_id(address);
            // A name for the whole fleet resolves to this node too.
            if id == self.local || self.kept.contains(&id) {
                continue;
            }
            if current.insert(id) && !self.added.contains(&id) {
                node.add_seed(id, address);
            }
        }
        for &gone in self.added.difference(&current) {
            node.remove_seed(gone);
        }
        self.added = current;
    }
}

/// Waits for the background task. If it panicked, a bug, the node is gone
/// from its peers' view, and they hand its share out among themselves:
/// kept at its last quotas, it would admit that share a second time. It
/// drops every key to the floor `R × β / N` instead, the part of the limit
/// a cluster leaves to each member, and says it no longer runs.
pub(crate) async fn watch(
    task: tokio::task::JoinHandle<()>,
    shared: Weak<Shared>,
    config: rateguard_core::limiter::Config,
) {
    let Err(error) = task.await else { return };
    let Some(shared) = shared.upgrade() else {
        return;
    };
    if error.is_panic() {
        hold_at_floor(&shared, &config);
    }
    shared.running.store(false, Ordering::Relaxed);
}

// Every key to the floor `R × β / N`, the part of the limit a cluster
// leaves to each member: for a node its peers no longer count.
fn hold_at_floor(shared: &Shared, config: &rateguard_core::limiter::Config) {
    let n = shared.cluster_size.load(Ordering::Relaxed).max(1);
    let floor = |limit: rateguard_core::limiter::Limit| {
        let rate = limit.rate * config.floor_factor / n as f64;
        rateguard_core::gcra::Quota::per_second(rate, limit.burst)
    };
    let classes = shared.classes.snapshot();
    let key_limits = shared.key_limits.snapshot();
    let strictest = floor(classes.strictest());
    // Keys not seen yet too: a slot taken from now on is sized from these.
    let factor = config.floor_factor / n as f64;
    shared
        .fresh_factor
        .store(factor.to_bits(), Ordering::Relaxed);
    shared.free_quota.store(strictest.pack(), Ordering::Relaxed);
    let now = shared.now();
    shared.table.for_each(|key, slot| {
        let Resolved { limit, shadow, .. } =
            classes.resolve(&key_limits, slot.class.load(Ordering::Relaxed), key);
        let quota = floor(limit);
        slot.gcra.set_quota(quota, now);
        slot.shadow.set(shadow_quota(quota, limit, shadow), now);
    });
    shared
        .table
        .for_each_free(|slot| slot.gcra.set_quota(strictest, now));
    shared.table.overflow().gcra.set_quota(strictest, now);
}

// One tick: attempts in, the core's round, quotas out (spec §5.3).
fn sync(node: &mut Node, shared: &Arc<Shared>, rejected: RejectedDatagrams) -> Vec<Action> {
    let now = shared.now();
    let classes = shared.classes.snapshot();
    let key_limits = shared.key_limits.snapshot();
    let mut counts = Counts {
        rejected,
        ..Counts::default()
    };
    shared.table.for_each(|key, slot| {
        let tally = slot.take_counts();
        let Resolved { limit, known, .. } =
            classes.resolve(&key_limits, slot.class.load(Ordering::Relaxed), key);
        // A tracked key with no attempts this tick still learns a changed
        // limit: the quota the tick sets is computed from the limit the
        // core holds, and only `record` hands it over. An untracked idle
        // key stays untracked.
        if tally.attempts > 0
            || node
                .limiter()
                .limit(key)
                .is_some_and(|known_limit| known_limit != limit)
        {
            node.record(key, tally, limit, now);
        }
        counts.attempts += tally.attempts;
        counts.denied += tally.denials;
        counts.shadow_denied += tally.shadow_denials;
        if !known {
            counts.unknown_class += tally.attempts;
        }
    });
    // Not a key: counted, but no demand of anyone's. It has no shadow.
    let overflowed = shared.table.overflow().take_counts();
    counts.attempts += overflowed.attempts;
    counts.denied += overflowed.denials;
    counts.overflowed = overflowed.attempts;
    let actions = node.handle(Event::Tick, now).to_vec();
    report(node, shared, counts);

    // A slot nobody holds carries the strictest new key's quota of any
    // class; whoever takes it gives it its own class's.
    let new_key = node.new_key_quota_in(classes.strictest());
    shared
        .fresh_factor
        .store(node.new_key_factor().to_bits(), Ordering::Relaxed);
    shared.free_quota.store(new_key.pack(), Ordering::Relaxed);
    // A slot the node knows nothing of is freed, unless a check has come
    // since the counts were taken: a key that took its slot meanwhile, in
    // the middle of this tick, would lose it under its first check.
    // One kept for its debt, or for a check since, gets what a new key of
    // its class gets now, as the core treats a key it does not track: left
    // alone, a key the core dropped would keep the quota it had then.
    // Each shadow follows the quota just set, scaled; with no shadow in any
    // class, the classes are not looked up again.
    let shadows = classes.has_shadows();
    let resolve = |key: u64, slot: &crate::table::Slot| {
        classes.resolve(&key_limits, slot.class.load(Ordering::Relaxed), key)
    };
    shared.table.for_each(|key, slot| match node.quota(key) {
        Some(quota) => {
            slot.gcra.set_quota(quota, now);
            let shadow = if shadows {
                let Resolved { limit, shadow, .. } = resolve(key, slot);
                shadow_quota(quota, limit, shadow)
            } else {
                None
            };
            slot.shadow.set(shadow, now);
        }
        None if !slot.gcra.has_debt(now) && !slot.has_counts() => {
            shared.table.release(slot, key, new_key);
        }
        None => {
            let Resolved { limit, shadow, .. } = resolve(key, slot);
            let quota = node.new_key_quota_in(limit);
            slot.gcra.set_quota(quota, now);
            slot.shadow.set(shadow_quota(quota, limit, shadow), now);
        }
    });
    shared
        .table
        .for_each_free(|slot| slot.gcra.set_quota(new_key, now));
    shared.table.overflow().gcra.set_quota(new_key, now);
    shared
        .cluster_size
        .store(node.cluster_size(), Ordering::Relaxed);
    actions
}

// One tick's checks, from the key table.
#[derive(Debug, Default)]
struct Counts {
    attempts: u64,
    denied: u64,
    overflowed: u64,
    unknown_class: u64,
    shadow_denied: u64,
    /// Datagrams refused since the last tick.
    rejected: RejectedDatagrams,
}

// This tick's counts and the node's state into the report for
// Guard::stats.
fn report(node: &Node, shared: &Shared, counts: Counts) {
    use rateguard_core::membership::Membership;
    use rateguard_proto::Status;

    let members = node.members();
    let suspects = members
        .peers()
        .iter()
        .filter(|&&peer| members.status(peer) == Some(Status::Suspect))
        .count();
    let held = node.held();
    let mut report = shared.report();
    report.attempts += counts.attempts;
    report.last_round = node.last_round();
    let stats = &mut report.stats;
    stats.denied += counts.denied;
    stats.overflowed += counts.overflowed;
    stats.unknown_class += counts.unknown_class;
    stats.shadow_denied += counts.shadow_denied;
    let rejected = &mut stats.rejected_datagrams;
    rejected.unsigned += counts.rejected.unsigned;
    rejected.bad_tag += counts.rejected.bad_tag;
    rejected.malformed += counts.rejected.malformed;
    rejected.mismatched_records += counts.rejected.mismatched_records;
    let listed = |keys: Vec<(u64, KeyRates)>| -> Vec<crate::stats::DeniedKey> {
        keys.into_iter()
            .map(|(key_hash, rates)| crate::stats::DeniedKey {
                key_hash,
                denied_per_sec: rates.denied_per_sec,
                requests_per_sec: rates.requests_per_sec,
                shadow_denied_per_sec: rates.shadow_denied_per_sec,
            })
            .collect()
    };
    stats.most_denied = listed(node.most_denied(shared.most_denied));
    stats.most_shadow_denied = listed(node.most_shadow_denied(shared.most_denied));
    stats.cluster_size = node.cluster_size();
    stats.counted_size = node.counted_size();
    stats.suspects = suspects;
    stats.tracked_keys = node.limiter().tracked_keys();
    stats.hot_keys = node.limiter().hot_keys();
    stats.joining = held.joining;
    stats.learning = held.learning;
    stats.without_quorum = held.without_quorum;
    stats.members_full = held.members_full;
    stats.identity_conflicts = node.identity_conflicts();
    stats.members_refused = node.members_refused();

    #[cfg(feature = "metrics")]
    {
        let mut stats = stats.clone();
        stats.admitted = report.attempts.saturating_sub(stats.denied);
        stats.dry_run = shared.dry_run.load(Ordering::Relaxed);
        drop(report);
        stats.gossip_auth = shared.auth().mode();
        crate::metrics::publish(&stats, &shared.metrics_label);
    }
}

// One datagram in: opened first, so that nothing unsigned or forged
// reaches `sender` or the core, which see the message inside the frame
// only. What is refused is counted by reason and dropped.
fn receive(
    node: &mut Node,
    auth: &Auth,
    datagram: &[u8],
    now: u64,
    rejected: &mut RejectedDatagrams,
) -> Vec<Action> {
    let message = match auth.open(datagram) {
        Ok(message) => message,
        Err(Rejected::Unsigned) => {
            rejected.unsigned += 1;
            return Vec::new();
        }
        Err(Rejected::BadTag) => {
            rejected.bad_tag += 1;
            return Vec::new();
        }
        // Malformed, and whatever reason a later version adds: no message
        // came through either way.
        Err(_) => {
            rejected.malformed += 1;
            return Vec::new();
        }
    };
    match screen(message) {
        Some(screened) => {
            rejected.mismatched_records += screened.dropped;
            node.handle(
                Event::MessageReceived {
                    from: screened.from,
                    bytes: &screened.bytes,
                },
                now,
            )
            .to_vec()
        }
        None => {
            rejected.malformed += 1;
            Vec::new()
        }
    }
}

// The whole batch gets the budget, not each datagram: three PING-REQs on a
// hung transport would hold the node three ticks. A lost datagram, or one
// that took too long to leave, is the protocol's normal case. Every
// datagram is sealed with `auth`, read once for the batch.
async fn send<T: Transport>(
    node: &Node,
    transport: &T,
    actions: Vec<Action>,
    auth: &Auth,
    budget: Duration,
) {
    let batch = async {
        for action in actions {
            // Sending is the only action this runtime knows.
            let Action::SendTo { peer, bytes } = action else {
                continue;
            };
            if let Some(address) = node.address(peer) {
                let datagram = auth.seal(&bytes);
                let _ = transport.send_to(&datagram, address.into()).await;
            }
        }
    };
    let _ = tokio::time::timeout(budget, batch).await;
}

#[cfg(test)]
mod tests {
    use super::*;
    use rateguard_proto::auth::{FRAME_OVERHEAD, GossipAuth, Secret};
    use rateguard_proto::{Address, Message, Status, Update, encode};

    struct HungSend;
    impl Transport for HungSend {
        async fn send_to(&self, _: &[u8], _: std::net::SocketAddr) -> std::io::Result<usize> {
            std::future::pending().await
        }
        async fn recv_from(&self, _: &mut [u8]) -> std::io::Result<(usize, std::net::SocketAddr)> {
            std::future::pending().await
        }
    }

    // A probe gone unanswered sends three PING-REQs in one tick: on a hung
    // transport they must cost the node one tick, not three.
    #[tokio::test(start_paused = true)]
    async fn a_hung_transport_costs_one_budget_per_batch() {
        use rateguard_core::limiter::{Config, Limit};
        use rateguard_core::node::SwimConfig;

        let config = Config {
            limit: Limit {
                rate: 100.0,
                burst: 5,
            },
            alpha: 0.5,
            floor_factor: 0.05,
            cooldown: 5_000_000_000,
            hot_set_size: 4,
            max_tracked_keys: 8,
            demand_time_constant: 1_000_000_000,
        };
        let me = Address::V4([10, 0, 0, 1], 7946);
        let mut node = Node::new(config, SwimConfig::default(), PeerId::new(1), me, 1);
        let actions: Vec<Action> = (2..5)
            .map(|peer| {
                node.introduce(PeerId::new(peer), Address::V4([10, 0, 0, peer as u8], 7946));
                Action::SendTo {
                    peer: PeerId::new(peer),
                    bytes: vec![1],
                }
            })
            .collect();

        let budget = Duration::from_millis(50);
        let start = tokio::time::Instant::now();
        send(&node, &HungSend, actions, &Auth::off(), budget).await;
        assert_eq!(start.elapsed(), budget);
    }

    // What the builder shares with the request path, for a node driven by
    // hand: no background task to tick it meanwhile.
    fn shared_for(node: &Node) -> Arc<Shared> {
        let default = crate::Limit::per_second(100).burst(5).to_core().unwrap();
        let classes = crate::classes::Classes::new(
            crate::Limit::per_second(100).burst(5),
            default,
            Vec::new(),
            16,
        );
        let free = node.new_key_quota_in(classes.snapshot().strictest());
        Arc::new(Shared {
            table: crate::table::KeyTable::new(8, free),
            cluster_size: Default::default(),
            epoch: tokio::time::Instant::now(),
            running: std::sync::atomic::AtomicBool::new(true),
            leave: mpsc::channel(1).0,
            queries: mpsc::channel(1).0,
            most_denied: 10,
            dry_run: Default::default(),
            report: Default::default(),
            classes,
            key_limits: crate::key_limits::KeyLimits::new(16),
            fresh_factor: std::sync::atomic::AtomicU64::new(node.new_key_factor().to_bits()),
            free_quota: std::sync::atomic::AtomicU64::new(free.pack()),
            auth: std::sync::RwLock::new(Arc::new(Auth::off())),
            salt: None,
            #[cfg(feature = "metrics")]
            metrics_label: String::new(),
        })
    }

    // A slot whose key the node does not track, held while it is in debt,
    // is kept at the quota a new key of its class gets now: a key the node
    // dropped would otherwise keep the one it had, whatever the class's
    // limit, the cluster's size or the node's cap since.
    #[tokio::test]
    async fn a_slot_the_node_does_not_track_follows_the_quota_of_a_new_key() {
        let mut node = node_at(1);
        let shared = shared_for(&node);
        let (slot, _) = shared.table.claim(1000);
        for _ in 0..20 {
            slot.gcra.check(shared.now());
        }
        assert!(slot.gcra.has_debt(shared.now()));

        let stricter = crate::Limit::per_second(10).burst(5);
        shared.classes.set_default(stricter).unwrap();
        sync(&mut node, &shared, RejectedDatagrams::default());
        assert!(shared.table.holds(slot, 1000));
        assert_eq!(
            slot.gcra.quota(),
            node.new_key_quota_in(stricter.to_core().unwrap())
        );
    }

    fn node_at(host: u8) -> Node {
        use rateguard_core::limiter::{Config, Limit};
        use rateguard_core::node::SwimConfig;

        let config = Config {
            limit: Limit {
                rate: 100.0,
                burst: 5,
            },
            alpha: 0.5,
            floor_factor: 0.05,
            cooldown: 5_000_000_000,
            hot_set_size: 4,
            max_tracked_keys: 8,
            demand_time_constant: 1_000_000_000,
        };
        let me = Address::V4([10, 0, 0, host], 7946);
        Node::new(
            config,
            SwimConfig::default(),
            crate::key::peer_id(me),
            me,
            1,
        )
    }

    fn at(host: u8) -> SocketAddr {
        SocketAddr::from(([10, 0, 0, host], 7946))
    }

    fn id(host: u8) -> PeerId {
        crate::key::peer_id(Address::from(at(host)))
    }

    struct SlowLookup {
        calls: std::sync::atomic::AtomicUsize,
    }
    impl Transport for SlowLookup {
        async fn send_to(&self, _: &[u8], _: SocketAddr) -> std::io::Result<usize> {
            std::future::pending().await
        }
        async fn recv_from(&self, _: &mut [u8]) -> std::io::Result<(usize, SocketAddr)> {
            std::future::pending().await
        }
        async fn resolve(&self, host: &str, _: u16) -> std::io::Result<Vec<SocketAddr>> {
            match host {
                "slow.svc" if self.calls.fetch_add(1, Ordering::Relaxed) == 0 => Ok(vec![at(2)]),
                "slow.svc" | "dead.svc" => std::future::pending().await,
                "working.svc" => Ok(vec![at(3)]),
                _ => panic!("unexpected name: {host}"),
            }
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_hung_lookup_preserves_old_seeds_and_does_not_block_other_names() {
        let guard = crate::Guard::builder()
            .bind("10.0.0.1:7946")
            .limit(100)
            .spawn_on(HungSend)
            .unwrap();
        let (found, mut lookups) = mpsc::channel(1);
        let task = tokio::spawn(resolve(
            Arc::new(SlowLookup {
                calls: std::sync::atomic::AtomicUsize::new(0),
            }),
            vec![
                ("slow.svc".into(), 7946),
                ("dead.svc".into(), 7946),
                ("working.svc".into(), 7946),
            ],
            Arc::downgrade(&guard.shared),
            Duration::from_secs(2),
            found,
        ));
        let mut node = node_at(1);
        node.expect_peers(true);
        let mut names = Names::new(id(1));
        let first = lookups.recv().await.unwrap();
        assert_eq!(first, vec![Some(vec![at(2)]), None, Some(vec![at(3)])]);
        names.update(&mut node, first);

        // Each later round must finish despite two names hanging, and
        // within one timeout, not one per hung name. The first name's last
        // address remains a seed, and the working name is still looked up;
        // a timeout is not an empty DNS answer.
        for _ in 0..2 {
            let round = RESOLVE_TIMEOUT + Duration::from_secs(3);
            let next = tokio::time::timeout(round, lookups.recv())
                .await
                .expect("hung lookups must not hold the round past one timeout")
                .unwrap();
            assert_eq!(next, vec![None, None, Some(vec![at(3)])]);
            names.update(&mut node, next);
            for host in [2, 3] {
                assert_eq!(node.address(id(host)), Some(at(host).into()));
            }
        }

        drop(lookups);
        task.await.unwrap();
    }

    #[test]
    fn seeds_from_names_follow_the_lookups() {
        let mut node = node_at(1);
        let mut names = Names::new(id(1));
        node.add_seed(id(9), at(9).into());
        names.keep(id(9));

        // Two names; the first resolves to this node too.
        names.update(&mut node, vec![Some(vec![at(1), at(2)]), Some(vec![at(3)])]);
        for host in [2, 3, 9] {
            assert_eq!(node.address(id(host)), Some(at(host).into()), "{host}");
        }

        // 2 is gone, 4 has come, the second lookup failed: 3 is kept.
        names.update(&mut node, vec![Some(vec![at(4)]), None]);
        assert_eq!(node.address(id(2)), None);
        assert_eq!(node.address(id(3)), Some(at(3).into()));
        assert_eq!(node.address(id(4)), Some(at(4).into()));

        // A seed given as an address outlives a name that resolved to it.
        names.update(&mut node, vec![Some(vec![at(9)]), Some(Vec::new())]);
        names.update(&mut node, vec![Some(Vec::new()), Some(Vec::new())]);
        assert_eq!(node.address(id(9)), Some(at(9).into()));
        for host in [3, 4] {
            assert_eq!(node.address(id(host)), None, "{host}");
        }
    }

    #[test]
    fn a_node_alone_behind_its_name_stops_waiting_for_peers() {
        // What a node that takes itself for the whole fleet starts keys at.
        let alone = node_at(1).new_key_quota();
        let mut node = node_at(1);
        node.expect_peers(true);
        let mut names = Names::new(id(1));

        // Nobody listed yet, a failed lookup, another replica: none counts,
        // and the last starts the count again.
        for found in [
            vec![Some(vec![at(1)])],
            vec![Some(vec![at(1)])],
            vec![Some(Vec::new())],
            vec![None],
            vec![Some(vec![at(1), at(2)])],
            vec![Some(vec![at(1)])],
            vec![Some(vec![at(1)])],
        ] {
            names.update(&mut node, found);
            assert_ne!(node.new_key_quota(), alone, "still waiting");
        }
        names.update(&mut node, vec![Some(vec![at(1)])]);
        assert_eq!(node.new_key_quota(), alone, "a fleet of one");
    }

    fn auth(mode: GossipAuth, secret: [u8; 32]) -> Auth {
        Auth::new(mode, vec![Secret::new(secret).unwrap()]).unwrap()
    }

    // A ping from the node at `host`, as its core would encode it.
    fn ping_from(host: u8) -> Vec<u8> {
        let me = Address::from(at(host));
        encode(&Message::Ping {
            seq: 1,
            updates: vec![Update {
                member: crate::key::peer_id(me).get(),
                addr: me,
                incarnation: 0,
                status: Status::Alive,
            }],
            demand: Vec::new(),
        })
        .unwrap()
    }

    // Every datagram goes through `open` before anything reads it: a
    // refused one never reaches the core, and is counted by its reason.
    #[test]
    fn a_datagram_is_opened_before_the_core_sees_it() {
        let required = auth(GossipAuth::Required, [0xA1; 32]);
        let foreign = auth(GossipAuth::Required, [0xB2; 32]);
        let mut node = node_at(1);
        let mut rejected = RejectedDatagrams::default();

        let refused: [&[u8]; 4] = [
            &ping_from(2),
            &foreign.seal(&ping_from(2)),
            &[0xA6],
            &required.seal(&[0xde, 0xad]),
        ];
        for datagram in refused {
            let actions = receive(&mut node, &required, datagram, 0, &mut rejected);
            assert!(actions.is_empty(), "{datagram:?}");
        }
        assert_eq!(node.address(id(2)), None, "the core heard nobody");
        assert_eq!(
            (rejected.unsigned, rejected.bad_tag, rejected.malformed),
            (1, 1, 2)
        );

        let actions = receive(
            &mut node,
            &required,
            &required.seal(&ping_from(2)),
            0,
            &mut rejected,
        );
        assert!(!actions.is_empty(), "a valid ping is answered");
        assert_eq!(node.address(id(2)), Some(at(2).into()));
        assert_eq!(
            (rejected.unsigned, rejected.bad_tag, rejected.malformed),
            (1, 1, 2)
        );
    }

    // Every record must name its member by the ID its address gives. The
    // sender's own record failing it makes the whole datagram malformed,
    // and the core never sees it; news after it that fails is dropped
    // alone (see the test below).
    #[test]
    fn a_member_whose_id_does_not_follow_from_its_address_is_malformed() {
        let mut node = node_at(1);
        let mut rejected = RejectedDatagrams::default();
        let genuine = |host: u8| {
            let addr = Address::from(at(host));
            Update {
                member: crate::key::peer_id(addr).get(),
                addr,
                incarnation: 0,
                status: Status::Alive,
            }
        };
        let ping = |updates: Vec<Update>| {
            encode(&Message::Ping {
                seq: 1,
                updates,
                demand: Vec::new(),
            })
            .unwrap()
        };
        let wrong = |update: Update| Update {
            member: update.member ^ 1,
            ..update
        };

        for datagram in [
            ping(vec![wrong(genuine(2))]),
            ping(vec![wrong(genuine(2)), genuine(3)]),
            ping(vec![
                Update {
                    addr: at(4).into(),
                    ..genuine(2)
                },
                genuine(3),
            ]),
        ] {
            assert_eq!(screen(&datagram), None);
            let actions = receive(&mut node, &Auth::off(), &datagram, 0, &mut rejected);
            assert!(actions.is_empty());
        }
        assert_eq!(rejected.malformed, 3);
        assert_eq!(node.members().known(), 1, "the core heard nobody");

        // Later records failing the check: each dropped, and the rest taken.
        let datagram = ping(vec![
            genuine(2),
            wrong(genuine(3)),
            Update {
                addr: at(4).into(),
                ..genuine(3)
            },
        ]);
        let screened = screen(&datagram).unwrap();
        assert_eq!(screened.from, id(2));
        assert_eq!(screened.bytes, ping(vec![genuine(2)]));
        assert_eq!(screened.dropped, 2);

        let datagram = ping(vec![genuine(2), genuine(3)]);
        let screened = screen(&datagram).unwrap();
        assert!(
            matches!(screened.bytes, Cow::Borrowed(_)),
            "taken as sent when every record passes"
        );
        assert_eq!(screened.dropped, 0);
        let actions = receive(&mut node, &Auth::off(), &datagram, 0, &mut rejected);
        assert!(!actions.is_empty(), "a genuine ping is answered");
        assert_eq!(node.members().known(), 3);
        assert_eq!(rejected.malformed, 3);
        assert_eq!(rejected.mismatched_records, 0);
    }

    // A 0.7.0 peer passes on news as it heard it, a forger's record
    // included. Only that record goes: the peer's own record, its demand
    // and its answer still reach the core.
    #[test]
    fn a_forged_record_after_the_senders_is_dropped_alone() {
        use rateguard_proto::{DemandReport, KeyDemand};

        let mut node = node_at(1);
        let mut rejected = RejectedDatagrams::default();
        let genuine = |host: u8| {
            let addr = Address::from(at(host));
            Update {
                member: crate::key::peer_id(addr).get(),
                addr,
                incarnation: 0,
                status: Status::Alive,
            }
        };
        let forged = Update {
            member: 0xF0F0,
            ..genuine(4)
        };
        let datagram = encode(&Message::Ping {
            seq: 7,
            updates: vec![genuine(2), forged, genuine(3)],
            demand: vec![DemandReport {
                origin: id(2).get(),
                epoch: 1,
                round: 1,
                keys: vec![KeyDemand {
                    key_hash: 42,
                    demand: 300.0,
                    primary: true,
                }],
            }],
        })
        .unwrap();

        let actions = receive(&mut node, &Auth::off(), &datagram, 0, &mut rejected);
        assert!(!actions.is_empty(), "the ping is answered");
        assert_eq!(node.address(id(2)), Some(at(2).into()));
        assert_eq!(node.address(id(3)), Some(at(3).into()));
        assert_eq!(node.address(PeerId::new(0xF0F0)), None, "not learned");
        assert_eq!(node.members().known(), 3);
        assert_eq!(node.peer_demand().total(42), 300.0, "the demand is kept");
        assert_eq!(rejected.malformed, 0);
        assert_eq!(rejected.mismatched_records, 1);
    }

    // In Off a signed frame is stripped unchecked and not counted: a node
    // not signing yet keeps hearing the nodes that already do.
    #[test]
    fn off_takes_a_signed_frame_and_counts_nothing() {
        let mut node = node_at(1);
        let mut rejected = RejectedDatagrams::default();
        let frame = auth(GossipAuth::Required, [0xB2; 32]).seal(&ping_from(2));
        let actions = receive(&mut node, &Auth::off(), &frame, 0, &mut rejected);
        assert!(!actions.is_empty());
        assert_eq!(rejected, RejectedDatagrams::default());
    }

    #[derive(Default)]
    struct Recording(std::sync::Mutex<Vec<Vec<u8>>>);
    impl Transport for Recording {
        async fn send_to(&self, bytes: &[u8], _: SocketAddr) -> std::io::Result<usize> {
            self.0.lock().unwrap().push(bytes.to_vec());
            Ok(bytes.len())
        }
        async fn recv_from(&self, _: &mut [u8]) -> std::io::Result<(usize, SocketAddr)> {
            std::future::pending().await
        }
    }

    #[tokio::test]
    async fn what_is_sent_is_sealed_with_the_current_auth() {
        let mut node = node_at(1);
        node.introduce(id(2), at(2).into());
        let message = ping_from(1);
        let actions = || {
            vec![Action::SendTo {
                peer: id(2),
                bytes: message.clone(),
            }]
        };
        let budget = Duration::from_secs(1);
        let required = auth(GossipAuth::Required, [0xA1; 32]);

        let sent = Recording::default();
        send(&node, &sent, actions(), &Auth::off(), budget).await;
        send(&node, &sent, actions(), &required, budget).await;
        let sent = sent.0.into_inner().unwrap();
        assert_eq!(sent[0], message, "Off sends the bytes of 0.5");
        assert_eq!(required.open(&sent[1]), Ok(&message[..]));
        assert_eq!(sent[1].len(), message.len() + FRAME_OVERHEAD);
    }

    #[test]
    fn the_sender_is_the_first_record_not_the_socket_address() {
        let me = Address::V4([10, 0, 0, 9], 7946);
        let bytes = encode(&Message::Ping {
            seq: 1,
            updates: vec![Update {
                member: crate::key::peer_id(me).get(),
                addr: me,
                incarnation: 0,
                status: Status::Alive,
            }],
            demand: Vec::new(),
        })
        .unwrap();
        assert_eq!(screen(&bytes).unwrap().from, crate::key::peer_id(me));

        let empty = encode(&Message::Ping {
            seq: 1,
            updates: Vec::new(),
            demand: Vec::new(),
        })
        .unwrap();
        assert_eq!(screen(&empty), None);
        assert_eq!(screen(&[0xde, 0xad]), None);
    }
}