didbot_lookup/

lib.rs

//! Looks up one named record in a repository this deployment does not
//! hold, and caches the answer.
//!
//! The policy gate decides a write from the subject alone: the account, its
//! kind, handle and depth, the client, the collection, the action and the
//! diff. Three rules the floor wants — an author's `app.bsky.feed.threadgate`
//! and `app.bsky.feed.postgate`, an `app.bsky.graph.block`, and a declared
//! `community.lexicon.preference.ai` — each need one more fact, and it is the
//! same fact for all three: the value of a public record in somebody else's
//! repository. This crate is that capability, built once, for the evaluators
//! that will ask for it.
//!
//! # Why its own crate
//!
//! It performs network I/O, and the two crates that would otherwise host it
//! cannot. `didbot-policy` is the policy core: no I/O, no async, no clock,
//! and [`Evaluator::evaluate`] takes a subject and nothing else precisely so
//! a predicate cannot reach the network from inside a decision.
//! `didbot-pds` holds no HTTP client and no async runtime either — the
//! operator poll's fetch lives in `didbot-serve` for exactly that reason.
//! Putting the reader in either would give both a `reqwest` dependency they
//! are deliberately without. So it is a crate of its own, depending on
//! `didbot-http` and `didbot-identity` and on neither policy crate, which
//! leaves an evaluator free to depend on it without `didbot-serve` coming
//! along.
//!
//! [`Evaluator::evaluate`]: https://docs.rs/didbot-policy
//!
//! # How an evaluator reaches it
//!
//! Two phases, because the decision channel is synchronous and the fetch is
//! not. `didbot-pds`'s write queue calls the gate inside a synchronous
//! critical section, so nothing under it can await.
//!
//! 1. **Prefetch.** The write path, which has a runtime, calls
//!    [`ForeignRecords::prefetch`] with the records this write needs. It
//!    resolves and fetches what the cache does not already hold, bounded by
//!    [`Limits::record_deadline`] per record and [`Limits::prefetch_deadline`]
//!    for the call.
//! 2. **Read.** An evaluator holding an `Arc<ForeignRecords>` from its own
//!    construction calls [`ForeignRecords::read`], which takes a lock, reads
//!    the cache and returns. It performs no I/O and does not await, so it is
//!    callable from inside a decision.
//!
//! That split is what keeps the regex and Cedar engines untouched: they hold
//! no reader, ask for no records, and their evaluation is unchanged.
//!
//! # What an unknown means
//!
//! [`Answer`] has three variants and the third one matters most. `Present`
//! and `Absent` are both things the repository said. [`Answer::Unknown`] is
//! everything else — not fetched yet, past the deadline, refused, a DID that
//! does not resolve, a server that did not answer — and this reader never
//! turns one into a refusal or into a permit. The evaluator that asked
//! decides what an unknown means for its own rule, because only it knows
//! whether the conservative reading of its rule is to deny or to stand
//! aside.
//!
//! That leaves `Staleness` and `EvalError` to the evaluator too. A reader
//! whose answers may be up to [`Limits::present_ttl`] old is a `BestEffort`
//! view, and an evaluator built on it declares a `max_lag` it can live with;
//! this crate reports no lag of its own and raises no `EvalError::Stale`,
//! because a cache entry inside its TTL is not a stale view — it is the
//! view, and the TTL is the propagation delay `plan/ai-preference.md` asks
//! to have written down rather than discovered.
//!
//! # Memory and network, declared
//!
//! - At most [`Limits::entries`] answers are held, each at most
//!   [`MAX_RECORD_BYTES`] of JSON, so the cache costs at most those
//!   multiplied together. At the defaults that is 1024 records of 64 KiB,
//!   64 MiB in the worst case and around a megabyte for records of the size
//!   a gate or a preference actually is.
//! - One reference costs at most two requests: the DID document, then
//!   `com.atproto.repo.getRecord` for one record, or
//!   `com.atproto.repo.listRecords` for a whole collection. Both read a
//!   bounded body, and a collection read takes at most
//!   [`MAX_COLLECTION_RECORDS`] records inside that same bound.
//! - One prefetch fetches at most [`Limits::per_write`] records, each
//!   bounded by [`Limits::record_deadline`], and the whole call by
//!   [`Limits::prefetch_deadline`].
//! - The cache is memory only. Nothing here writes to disk, and a restart
//!   begins with an empty cache — which costs one fetch per record and can
//!   never resurrect an answer somebody has since changed.

mod fetch;
mod record;

pub use fetch::{
    Fetch, FetchFuture, Fetched, LocalRepositories, Network, NoLocalRepositories, Unknown,
    MAX_COLLECTION_RECORDS, MAX_RECORD_BYTES,
};
pub use record::{RecordRef, RefError};

use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

// `tokio`'s clock, not `std`'s: a TTL and a deadline measured against it can
// be advanced by a test without waiting, and it is the same instant in a
// running server.
use tokio::time::Instant;

use serde_json::Value;

/// How many answers are held at once, absent configuration.
///
/// **Reasonable range: 64 to 16384.** See this crate's memory bound: the
/// cache costs this many times [`MAX_RECORD_BYTES`] in the worst case. A
/// thousand is a busy deployment's working set of reply gates over a TTL,
/// and small enough that the worst case is still a fraction of one process.
pub const DEFAULT_ENTRIES: usize = 1024;

/// How long a record this reader read is reused, absent configuration.
///
/// **Reasonable range: 60 to 3600 seconds.** Nothing tells this deployment
/// that a stranger changed a record — it consumes no firehose — so this TTL
/// *is* the propagation delay, and it is stated rather than discovered.
/// Fifteen minutes is short enough that a person who opens their replies
/// sees it take effect while they are still at the keyboard.
pub const DEFAULT_PRESENT_TTL: Duration = Duration::from_secs(15 * 60);

/// How long "that repository holds nothing at that key" is reused, absent
/// configuration.
///
/// **Reasonable range: 30 to 900 seconds.** Shorter than
/// [`DEFAULT_PRESENT_TTL`] on purpose: absence is the common answer — almost
/// nobody has written an AI preference, and most posts carry no reply gate —
/// and it is also the answer most likely to stop being true, because writing
/// the record is exactly what a person does when they decide to be left
/// alone. Holding an absence for as long as a value would make the moment
/// somebody asks to be left alone the slowest thing this reader notices.
pub const DEFAULT_ABSENT_TTL: Duration = Duration::from_secs(5 * 60);

/// How long an unknown is reused, absent configuration.
///
/// **Reasonable range: 10 to 300 seconds.** This is the negative cache: a
/// DID that does not resolve, a server that is down, an address this reader
/// refuses. Holding it stops one unreachable repository from costing a
/// fetch on every write that names it, and it is the shortest of the three
/// because it is the one that recovers by itself.
pub const DEFAULT_UNKNOWN_TTL: Duration = Duration::from_secs(60);

/// How long one record's fetch may take, absent configuration.
///
/// **Reasonable range: 250 to 5000 milliseconds.** This is the deadline in
/// `plan/policy.md`'s sense: what a write waits for somebody else's server
/// before it is told nothing. Shorter than
/// [`didbot_http::DEFAULT_TIMEOUT`], which is sized for a poll tick that
/// nothing is waiting behind; a write is.
pub const DEFAULT_RECORD_DEADLINE: Duration = Duration::from_millis(2000);

/// How long one prefetch may take in total, absent configuration.
///
/// **Reasonable range: 500 to 10000 milliseconds.** Bounds the whole call
/// rather than one record in it, so a write needing several records cannot
/// cost the sum of their deadlines. Records the budget runs out on are
/// answered as [`Unknown::Deadline`] without being fetched.
pub const DEFAULT_PREFETCH_DEADLINE: Duration = Duration::from_millis(4000);

/// How many records one write may need, absent configuration.
///
/// **Reasonable range: 1 to 64.** Bounds what a single write can cost
/// somebody else's server. The number is per write rather than per rule
/// because a rule about the accounts a record names asks the same question
/// of each of them: one post carrying a reply, a quote and several mentions
/// is already several repositories. Sixteen covers that post, and the reply
/// and quote gates a thread rule wants, with room over.
///
/// A write needing more records than this is refused before it is judged,
/// rather than judged on some of them; see [`MAX_NAMED_ACCOUNTS`].
pub const DEFAULT_PER_WRITE: usize = 16;

/// The most accounts one write may name while a policy that reads something
/// about each of them applies to it.
///
/// **Reasonable range: 1 to 64.** A write naming more is refused outright:
/// judging it on the first few would let a record built to name one more
/// person than this server checks name anybody at all. The count is of
/// accounts this deployment does not hold, since its own are read from its
/// own store. It is also held to [`Limits::per_write`] divided by the
/// lookups that apply, so every account named is read.
pub const MAX_NAMED_ACCOUNTS: usize = 16;

/// What a [`ForeignRecords`] is bounded by. Every field is a parameter a
/// deployment sets on its own, with its default and its reasonable range on
/// the constant it comes from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Limits {
    /// The most answers held at once; see [`DEFAULT_ENTRIES`].
    pub entries: usize,
    /// How long a record that was read is reused; see
    /// [`DEFAULT_PRESENT_TTL`].
    pub present_ttl: Duration,
    /// How long an absence is reused; see [`DEFAULT_ABSENT_TTL`].
    pub absent_ttl: Duration,
    /// How long an unknown is reused; see [`DEFAULT_UNKNOWN_TTL`].
    pub unknown_ttl: Duration,
    /// How long one record's fetch may take; see
    /// [`DEFAULT_RECORD_DEADLINE`].
    pub record_deadline: Duration,
    /// How long one prefetch may take in total; see
    /// [`DEFAULT_PREFETCH_DEADLINE`].
    pub prefetch_deadline: Duration,
    /// How many records one prefetch may fetch; see
    /// [`DEFAULT_PER_WRITE`].
    pub per_write: usize,
}

impl Default for Limits {
    fn default() -> Self {
        Self {
            entries: DEFAULT_ENTRIES,
            present_ttl: DEFAULT_PRESENT_TTL,
            absent_ttl: DEFAULT_ABSENT_TTL,
            unknown_ttl: DEFAULT_UNKNOWN_TTL,
            record_deadline: DEFAULT_RECORD_DEADLINE,
            prefetch_deadline: DEFAULT_PREFETCH_DEADLINE,
            per_write: DEFAULT_PER_WRITE,
        }
    }
}

/// What this reader knows about one reference.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Answer {
    /// The repository holds records there — see [`Fetched::Present`] for
    /// the shape, which is an array either way. An `Arc` because one cached
    /// value is handed to every evaluator that asks for it.
    Present(Arc<Value>),
    /// The repository answered, and holds nothing there.
    Absent,
    /// Neither was established; see [`Unknown`], and this crate's "What an
    /// unknown means".
    Unknown(Unknown),
}

impl Answer {
    /// A short word for a log line.
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Answer::Present(_) => "present",
            Answer::Absent => "absent",
            Answer::Unknown(unknown) => unknown.as_str(),
        }
    }
}

/// What this reader has done since the process started, for the operator
/// watching the numbers move.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Counts {
    /// Reads answered from a cache entry inside its TTL.
    pub hits: u64,
    /// Reads the cache held no fresh answer for.
    pub misses: u64,
    /// Fetches that established neither presence nor absence, not counting
    /// the ones in `refusals`.
    pub unknowns: u64,
    /// Fetches this reader refused before making a request: an address it
    /// will not reach, or a repository this deployment holds.
    pub refusals: u64,
}

/// One cached answer and when it stops being usable.
struct Entry {
    answer: Answer,
    expires: Instant,
}

/// The reader: a bounded cache of foreign records, over one [`Fetch`].
///
/// Deployment-wide, not per session: `plan/ai-preference.md`'s "one
/// resolver, one cache, deployment-wide" is the whole difference between
/// being a good guest on somebody else's server and being the traffic that
/// gets a deployment defederated.
pub struct ForeignRecords {
    fetcher: Arc<dyn Fetch>,
    limits: Limits,
    cache: Mutex<lru::LruCache<String, Entry>>,
    hits: AtomicU64,
    misses: AtomicU64,
    unknowns: AtomicU64,
    refusals: AtomicU64,
}

impl ForeignRecords {
    /// A reader over `fetcher`, bounded by `limits`.
    ///
    /// An `entries` of zero holds one entry: a cache that holds nothing
    /// would make every read a fetch, which is the traffic pattern this
    /// crate exists to avoid.
    #[must_use]
    pub fn new(fetcher: Arc<dyn Fetch>, limits: Limits) -> Self {
        let entries = NonZeroUsize::new(limits.entries).unwrap_or(NonZeroUsize::MIN);
        Self {
            fetcher,
            limits,
            cache: Mutex::new(lru::LruCache::new(entries)),
            hits: AtomicU64::new(0),
            misses: AtomicU64::new(0),
            unknowns: AtomicU64::new(0),
            refusals: AtomicU64::new(0),
        }
    }

    /// A reader over the real network, refusing what `local` already holds
    /// and refusing a private address unless `private` allows one.
    #[must_use]
    pub fn over_network(
        local: Arc<dyn LocalRepositories>,
        private: didbot_http::PrivateAddresses,
        limits: Limits,
    ) -> Self {
        Self::new(Arc::new(Network::new(local, private)), limits)
    }

    /// What this reader is bounded by.
    #[must_use]
    pub fn limits(&self) -> Limits {
        self.limits
    }

    /// What this reader has done since the process started.
    #[must_use]
    pub fn counts(&self) -> Counts {
        Counts {
            hits: self.hits.load(Ordering::Relaxed),
            misses: self.misses.load(Ordering::Relaxed),
            unknowns: self.unknowns.load(Ordering::Relaxed),
            refusals: self.refusals.load(Ordering::Relaxed),
        }
    }

    /// The answer held for `record`.
    ///
    /// Performs no I/O and never awaits, so an evaluator may call it from
    /// inside a decision. A record nothing has fetched, and one whose
    /// answer has aged out, are both [`Unknown::NotFetched`].
    pub fn read(&self, record: &RecordRef) -> Answer {
        let key = record.at_uri();
        let now = Instant::now();
        let mut cache = self.cache.lock().unwrap_or_else(|p| p.into_inner());
        match cache.get(&key) {
            Some(entry) if entry.expires > now => {
                self.hits.fetch_add(1, Ordering::Relaxed);
                entry.answer.clone()
            }
            _ => {
                self.misses.fetch_add(1, Ordering::Relaxed);
                Answer::Unknown(Unknown::NotFetched)
            }
        }
    }

    /// Fetches every record in `wanted` the cache holds no fresh answer
    /// for, and caches what comes back.
    ///
    /// Bounded three ways: at most [`Limits::per_write`] records are
    /// fetched, each fetch by [`Limits::record_deadline`], and the whole
    /// call by [`Limits::prefetch_deadline`]. A record the budget does not
    /// reach is cached as [`Unknown::Deadline`], so the write that asked
    /// gets an answer either way and the next write does not queue behind
    /// the same dead server.
    pub async fn prefetch(&self, wanted: &[RecordRef]) {
        let started = Instant::now();
        for record in wanted.iter().take(self.limits.per_write) {
            if self.fresh(&record.at_uri()) {
                continue;
            }
            let left = self
                .limits
                .prefetch_deadline
                .saturating_sub(started.elapsed());
            let budget = left.min(self.limits.record_deadline);
            let fetched = if budget.is_zero() {
                Fetched::Unknown(Unknown::Deadline)
            } else {
                let began = Instant::now();
                let fetched = match tokio::time::timeout(budget, self.fetcher.fetch(record)).await {
                    Ok(fetched) => fetched,
                    Err(_) => Fetched::Unknown(Unknown::Deadline),
                };
                tracing::debug!(
                    uri = %record,
                    outcome = outcome_of(&fetched),
                    latency_ms = began.elapsed().as_millis(),
                    "read a record from a repository this deployment does not hold"
                );
                fetched
            };
            self.store(record, fetched);
        }
    }

    /// [`prefetch`](Self::prefetch) for one record, then
    /// [`read`](Self::read) of it — for a caller that has a runtime and
    /// wants the answer now rather than at a later decision.
    pub async fn get(&self, record: &RecordRef) -> Answer {
        self.prefetch(std::slice::from_ref(record)).await;
        self.read(record)
    }

    /// Whether the cache holds an answer for `key` that has not expired.
    /// Does not count as a read: a prefetch skipping a record it already
    /// holds is not an evaluator asking for one.
    fn fresh(&self, key: &str) -> bool {
        let now = Instant::now();
        let mut cache = self.cache.lock().unwrap_or_else(|p| p.into_inner());
        cache.get(key).is_some_and(|entry| entry.expires > now)
    }

    /// Caches one fetch's outcome under `record`'s at-uri, for the TTL its
    /// kind of answer gets, and counts it.
    fn store(&self, record: &RecordRef, fetched: Fetched) {
        let (answer, ttl) = match fetched {
            Fetched::Present(value) => (Answer::Present(Arc::new(value)), self.limits.present_ttl),
            Fetched::Absent => (Answer::Absent, self.limits.absent_ttl),
            Fetched::Unknown(unknown) => {
                match unknown {
                    Unknown::Refused(_) => self.refusals.fetch_add(1, Ordering::Relaxed),
                    _ => self.unknowns.fetch_add(1, Ordering::Relaxed),
                };
                (Answer::Unknown(unknown), self.limits.unknown_ttl)
            }
        };
        let entry = Entry {
            answer,
            expires: Instant::now() + ttl,
        };
        self.cache
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .put(record.at_uri(), entry);
    }
}

/// The word a fetch's outcome is logged under.
fn outcome_of(fetched: &Fetched) -> &'static str {
    match fetched {
        Fetched::Present(_) => "present",
        Fetched::Absent => "absent",
        Fetched::Unknown(unknown) => unknown.as_str(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::collections::HashMap;

    /// A [`Fetch`] answering from a script, counting what it was asked.
    struct Stub {
        answers: Mutex<HashMap<String, Fetched>>,
        calls: AtomicU64,
    }

    impl Stub {
        fn new(answers: impl IntoIterator<Item = (String, Fetched)>) -> Arc<Self> {
            Arc::new(Self {
                answers: Mutex::new(answers.into_iter().collect()),
                calls: AtomicU64::new(0),
            })
        }

        fn calls(&self) -> u64 {
            self.calls.load(Ordering::SeqCst)
        }

        fn answer_with(&self, record: &RecordRef, fetched: Fetched) {
            self.answers
                .lock()
                .expect("answers")
                .insert(record.at_uri(), fetched);
        }
    }

    impl Fetch for Stub {
        fn fetch<'a>(&'a self, record: &'a RecordRef) -> FetchFuture<'a> {
            Box::pin(async move {
                self.calls.fetch_add(1, Ordering::SeqCst);
                self.answers
                    .lock()
                    .expect("answers")
                    .get(&record.at_uri())
                    .cloned()
                    .unwrap_or(Fetched::Absent)
            })
        }
    }

    fn gate() -> RecordRef {
        RecordRef::new(
            "did:plc:abcdefghijklmnopqrstuvwx",
            "app.bsky.feed.threadgate",
            "3kabc",
        )
    }

    /// The path every rule built on this reader takes: one fetch warms the
    /// cache, and every later read inside the TTL is answered without
    /// touching the network.
    #[tokio::test]
    async fn a_fetched_record_is_answered_from_cache_afterwards() {
        let record = gate();
        let stub = Stub::new([(record.at_uri(), Fetched::Present(json!({"allow": []})))]);
        let reader = ForeignRecords::new(stub.clone(), Limits::default());

        assert_eq!(
            reader.get(&record).await,
            Answer::Present(Arc::new(json!({"allow": []})))
        );
        reader.prefetch(std::slice::from_ref(&record)).await;
        assert_eq!(
            reader.read(&record),
            Answer::Present(Arc::new(json!({"allow": []})))
        );
        assert_eq!(stub.calls(), 1, "the second prefetch found a fresh answer");
        assert_eq!(reader.counts().hits, 2);
        assert_eq!(reader.counts().misses, 0);
    }

    /// Absence is cached too, and it is the common answer: most repositories
    /// hold no gate and no declared preference, and each of those would
    /// otherwise cost somebody else a request per write.
    #[tokio::test]
    async fn an_absence_is_cached_and_is_not_an_unknown() {
        let record = gate();
        let stub = Stub::new([(record.at_uri(), Fetched::Absent)]);
        let reader = ForeignRecords::new(stub.clone(), Limits::default());

        assert_eq!(reader.get(&record).await, Answer::Absent);
        assert_eq!(reader.read(&record), Answer::Absent);
        assert_eq!(stub.calls(), 1);
        assert_eq!(reader.counts().unknowns, 0);
    }

    /// An absence aging out is what lets somebody who writes a gate be
    /// honoured within a stated delay, and it is deliberately sooner than a
    /// value aging out.
    #[tokio::test(start_paused = true)]
    async fn an_absence_ages_out_sooner_than_a_value() {
        let absent = gate();
        let present = RecordRef::new(
            "did:plc:abcdefghijklmnopqrstuvwx",
            "app.bsky.feed.postgate",
            "3kabc",
        );
        let stub = Stub::new([
            (absent.at_uri(), Fetched::Absent),
            (present.at_uri(), Fetched::Present(json!({"x": 1}))),
        ]);
        let limits = Limits {
            absent_ttl: Duration::from_secs(60),
            present_ttl: Duration::from_secs(600),
            ..Limits::default()
        };
        let reader = ForeignRecords::new(stub, limits);
        reader.prefetch(&[absent.clone(), present.clone()]).await;

        tokio::time::advance(Duration::from_secs(61)).await;
        assert_eq!(reader.read(&absent), Answer::Unknown(Unknown::NotFetched));
        assert_eq!(
            reader.read(&present),
            Answer::Present(Arc::new(json!({"x": 1})))
        );

        tokio::time::advance(Duration::from_secs(600)).await;
        assert_eq!(reader.read(&present), Answer::Unknown(Unknown::NotFetched));
    }

    /// A server that is down is negatively cached, so one unreachable
    /// repository costs one fetch per TTL rather than one per write — and
    /// the answer stays an unknown, which is not an absence and is not a
    /// refusal of the write.
    #[tokio::test(start_paused = true)]
    async fn an_unreachable_repository_is_negatively_cached_then_retried() {
        let record = gate();
        let stub = Stub::new([(
            record.at_uri(),
            Fetched::Unknown(Unknown::Unreachable("connection refused".to_owned())),
        )]);
        let reader = ForeignRecords::new(stub.clone(), Limits::default());

        reader.prefetch(std::slice::from_ref(&record)).await;
        reader.prefetch(std::slice::from_ref(&record)).await;
        assert_eq!(stub.calls(), 1, "the unknown is held for its own TTL");
        assert!(matches!(
            reader.read(&record),
            Answer::Unknown(Unknown::Unreachable(_))
        ));

        tokio::time::advance(DEFAULT_UNKNOWN_TTL + Duration::from_secs(1)).await;
        stub.answer_with(&record, Fetched::Present(json!({"allow": []})));
        reader.prefetch(std::slice::from_ref(&record)).await;
        assert_eq!(stub.calls(), 2, "the negative entry expired and it retried");
        assert_eq!(
            reader.read(&record),
            Answer::Present(Arc::new(json!({"allow": []})))
        );
    }

    /// A refusal is counted apart from every other unknown: it is this
    /// deployment's own decision rather than somebody else's outage, and an
    /// operator reading a rising number wants to know which.
    #[tokio::test]
    async fn a_refusal_is_counted_apart_from_an_outage() {
        let refused = gate();
        let down = RecordRef::new(
            "did:plc:abcdefghijklmnopqrstuvwy",
            "app.bsky.graph.block",
            "3k",
        );
        let stub = Stub::new([
            (
                refused.at_uri(),
                Fetched::Unknown(Unknown::Refused("hosted here".to_owned())),
            ),
            (
                down.at_uri(),
                Fetched::Unknown(Unknown::Unreachable("timed out".to_owned())),
            ),
        ]);
        let reader = ForeignRecords::new(stub, Limits::default());
        reader.prefetch(&[refused, down]).await;

        let counts = reader.counts();
        assert_eq!(counts.refusals, 1);
        assert_eq!(counts.unknowns, 1);
    }

    /// The cache is bounded by entry count, not by how many records a
    /// deployment's agents happen to name.
    #[tokio::test]
    async fn the_cache_holds_no_more_than_its_declared_entries() {
        let limits = Limits {
            entries: 2,
            per_write: 8,
            ..Limits::default()
        };
        let records: Vec<RecordRef> = (0..4)
            .map(|i| {
                RecordRef::new(
                    "did:plc:abcdefghijklmnopqrstuvwx",
                    "app.bsky.feed.threadgate",
                    format!("3k{i}"),
                )
            })
            .collect();
        let stub = Stub::new(
            records
                .iter()
                .map(|r| (r.at_uri(), Fetched::Present(json!({"rkey": r.rkey})))),
        );
        let reader = ForeignRecords::new(stub, limits);
        reader.prefetch(&records).await;

        assert_eq!(
            reader.read(&records[0]),
            Answer::Unknown(Unknown::NotFetched),
            "the oldest two made room for the newest two"
        );
        assert_eq!(
            reader.read(&records[3]),
            Answer::Present(Arc::new(json!({"rkey": "3k3"})))
        );
    }

    /// One write cannot cost somebody else's server more than the declared
    /// number of requests, however many records a rule asks for.
    #[tokio::test]
    async fn one_prefetch_fetches_no_more_than_the_per_write_bound() {
        let limits = Limits {
            per_write: 2,
            ..Limits::default()
        };
        let records: Vec<RecordRef> = (0..5)
            .map(|i| {
                RecordRef::new(
                    "did:plc:abcdefghijklmnopqrstuvwx",
                    "app.bsky.feed.threadgate",
                    format!("3k{i}"),
                )
            })
            .collect();
        let stub = Stub::new([]);
        let reader = ForeignRecords::new(stub.clone(), limits);
        reader.prefetch(&records).await;
        assert_eq!(stub.calls(), 2);
    }

    /// A fetch past its deadline is an unknown rather than a wait: the
    /// write behind it finds out in bounded time that nothing was
    /// established.
    #[tokio::test(start_paused = true)]
    async fn a_fetch_past_its_deadline_is_an_unknown() {
        struct Slow;
        impl Fetch for Slow {
            fn fetch<'a>(&'a self, _record: &'a RecordRef) -> FetchFuture<'a> {
                Box::pin(async {
                    tokio::time::sleep(Duration::from_secs(3600)).await;
                    Fetched::Absent
                })
            }
        }
        let record = gate();
        let reader = ForeignRecords::new(Arc::new(Slow), Limits::default());
        reader.prefetch(std::slice::from_ref(&record)).await;
        assert_eq!(reader.read(&record), Answer::Unknown(Unknown::Deadline));
    }

    /// The whole call is bounded, not just each record in it, so a write
    /// naming several dead repositories waits once rather than once per
    /// record.
    #[tokio::test(start_paused = true)]
    async fn a_prefetch_stops_at_its_own_deadline() {
        struct Slow;
        impl Fetch for Slow {
            fn fetch<'a>(&'a self, _record: &'a RecordRef) -> FetchFuture<'a> {
                Box::pin(async {
                    tokio::time::sleep(Duration::from_secs(3600)).await;
                    Fetched::Absent
                })
            }
        }
        let limits = Limits {
            record_deadline: Duration::from_secs(2),
            prefetch_deadline: Duration::from_secs(3),
            per_write: 4,
            ..Limits::default()
        };
        let records: Vec<RecordRef> = (0..4)
            .map(|i| {
                RecordRef::new(
                    "did:plc:abcdefghijklmnopqrstuvwx",
                    "app.bsky.feed.threadgate",
                    format!("3k{i}"),
                )
            })
            .collect();
        let reader = ForeignRecords::new(Arc::new(Slow), limits);
        let started = tokio::time::Instant::now();
        reader.prefetch(&records).await;
        assert_eq!(started.elapsed(), Duration::from_secs(3));
        for record in &records {
            assert_eq!(reader.read(record), Answer::Unknown(Unknown::Deadline));
        }
    }
}