didbot_serve/oauth/

permission_set.rs

//! What an `include:` scope grants: the permission set it names, resolved
//! from the set's lexicon authority and cached.
//!
//! The lexicon spec's "Lexicon Publication and Resolution" is the path. The
//! set's NSID less its name, reversed, is a domain: `app.bsky.authFullApp`
//! is `bsky.app`. The `_lexicon` TXT record under it names a DID. That
//! DID's document names the server holding its repository and the key its
//! commits are signed with. The set is the `com.atproto.lexicon.schema`
//! record keyed by its NSID in that repository. It arrives as a
//! `com.atproto.sync.getRecord` proof, and nothing in it is read until the
//! proof verifies against that key.
//!
//! [`PermissionSets::expand`] replaces each `include:` in a request with the
//! atoms [`didbot_scope::Include::grants`] reads out of its set. A set that
//! does not resolve, for any reason, grants nothing, and the rest of the
//! request is judged without it. It never fails the request.
//!
//! # What one request can cost
//!
//! Each set costs at most three requests: the DNS query, the DID document,
//! and the proof. Each is a GET through [`didbot_http::pinned_client`] with
//! a bounded body. One request resolves at most [`MAX_SETS`] sets, each
//! within [`SET_DEADLINE`] and all of them within [`EXPANSION_DEADLINE`].
//! The cache is shared by every account and holds at most [`CACHE_ENTRIES`]
//! sets, each read from a proof of at most [`MAX_PROOF_BYTES`].

use std::future::Future;
use std::num::NonZeroUsize;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use didbot_http::PrivateAddresses;
use didbot_identity::DidDocument;
use didbot_lexicon::lexicon::LexiconDoc;
use didbot_repo::CommitKey;
use serde_json::Value;
// `tokio`'s clock, so a test can advance the cache's lifetimes without
// waiting.
use tokio::time::Instant;

use crate::oauth::scope::{Include, Scope, ScopeSet, MAX_GRANT_ATOMS, MAX_GRANT_BYTES};

/// The collection a lexicon is published in.
const SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema";

/// The atproto error a repository names when it holds no record at a key.
const RECORD_NOT_FOUND: &str = "RecordNotFound";

/// How long a resolved set is used before it is resolved again.
///
/// **Reasonable range: 15 minutes to 24 hours.** The permission spec puts
/// the floor at an access token's lifetime and calls 24 hours a firm
/// ceiling. A set its authority changes reaches new sign-ins within this.
pub const FRESH_FOR: Duration = Duration::from_secs(60 * 60);

/// How long the last resolved copy of a set stands in while its authority
/// cannot be reached.
///
/// **Reasonable range: 1 to 90 days.** Ninety days is the permission spec's
/// recommendation for a set referenced by a new sign-in. An authority that
/// answers, and not with the set, ends the copy at once.
pub const KEEP_FOR: Duration = Duration::from_secs(90 * 24 * 60 * 60);

/// How long a failed resolution is remembered before a sign-in tries again.
///
/// **Reasonable range: 10 seconds to 5 minutes.** Long enough that a set
/// whose authority is down costs one lookup per interval rather than one
/// per sign-in; short enough that an outage ends by itself.
pub const RETRY_AFTER: Duration = Duration::from_secs(60);

/// How long resolving one set may take: three requests, each usually well
/// under a second.
pub const SET_DEADLINE: Duration = Duration::from_secs(4);

/// How long resolving every set in one request may take. With the client's
/// metadata and key set fetched first, a push stays inside
/// `crate::routes::REQUEST_TIMEOUT`.
pub const EXPANSION_DEADLINE: Duration = Duration::from_secs(8);

/// The most sets one request resolves. Past it, an `include:` grants
/// nothing.
pub const MAX_SETS: usize = 16;

/// How many sets the cache holds.
pub const CACHE_ENTRIES: usize = 256;

/// The largest `com.atproto.sync.getRecord` proof read. A set with a
/// hundred methods is about ten kilobytes of proof.
pub const MAX_PROOF_BYTES: usize = 64 * 1024;

/// The most of a refusal's body read, for the atproto error it names.
const MAX_ERROR_BYTES: usize = 64 * 1024;

/// Why a set did not resolve.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum Unresolved {
    /// The set's authority names no DID, not exactly one, or one this server
    /// cannot resolve.
    #[error("no lexicon authority: {0}")]
    NoAuthority(String),
    /// A server on the way did not answer, or answered with an error.
    #[error("unreachable: {0}")]
    Unreachable(String),
    /// A server answered with something that is not what the spec says it
    /// publishes: a document for another DID, a proof that does not verify,
    /// no record at the set's NSID, or a record that is not a
    /// `permission-set`.
    #[error("not a permission set: {0}")]
    Invalid(String),
    /// The resolution did not finish in time.
    #[error("resolving it took longer than allowed")]
    Deadline,
}

impl Unresolved {
    /// Whether the authority answered, rather than failing to: the last
    /// copy of the set no longer stands in for it.
    fn answered(&self) -> bool {
        matches!(self, Self::NoAuthority(_) | Self::Invalid(_))
    }
}

/// A future a [`Resolve`] implementation hands back, boxed so the trait can
/// be held as `Arc<dyn Resolve>`.
pub type ResolveFuture<'a> =
    Pin<Box<dyn Future<Output = Result<Vec<Value>, Unresolved>> + Send + 'a>>;

/// One resolution of a set: its published `permissions`.
///
/// Apart from the cache, so a test can drive the cache with no network.
pub trait Resolve: Send + Sync {
    /// Resolves the set `nsid` names.
    fn resolve<'a>(&'a self, nsid: &'a str) -> ResolveFuture<'a>;
}

/// The resolution the lexicon spec describes, over the network.
pub struct Network {
    doh: String,
    private: PrivateAddresses,
}

impl Network {
    /// Asks `doh`, a DNS-over-HTTPS resolver answering the JSON form, for
    /// the authority's TXT record. The resolver is this deployment's own
    /// configuration, so it may be on a private address. Every server an
    /// authority names is refused one unless `private` allows it.
    pub fn new(doh: impl Into<String>, private: PrivateAddresses) -> Self {
        Self {
            doh: doh.into(),
            private,
        }
    }

    /// The DID the `_lexicon` TXT record for `nsid`'s authority names.
    ///
    /// Exactly one `did=` value, at exactly that name. No other name is
    /// tried.
    async fn authority(&self, nsid: &str) -> Result<String, Unresolved> {
        let name = lexicon_name(nsid);
        let records = didbot_http::doh::lookup(
            &self.doh,
            &name,
            didbot_http::doh::TXT,
            PrivateAddresses::Allowed,
        )
        .await
        .map_err(|error| Unresolved::Unreachable(format!("{name}: {error}")))?;
        let dids: Vec<&str> = records
            .iter()
            .filter_map(|record| record.strip_prefix("did="))
            .collect();
        let [did] = dids.as_slice() else {
            return Err(Unresolved::NoAuthority(format!(
                "{name} names {} DIDs, not one",
                dids.len()
            )));
        };
        didbot_identity::validate_did(did)
            .map_err(|error| Unresolved::NoAuthority(format!("{name} names {did}: {error}")))?;
        Ok((*did).to_owned())
    }

    /// The server holding `did`'s repository, and the key its commits are
    /// signed with.
    async fn identity(&self, did: &str) -> Result<(String, CommitKey), Unresolved> {
        let url = didbot_identity::document_url(did)
            .map_err(|error| Unresolved::NoAuthority(error.to_string()))?;
        let body = self.get(&url, didbot_http::MAX_JSON_BODY).await?;
        let document: DidDocument = serde_json::from_slice(&body)
            .map_err(|error| Unresolved::Invalid(format!("parsing {url}: {error}")))?;
        if document.id != did {
            return Err(Unresolved::Invalid(format!(
                "the document at {url} is {:?}'s, not {did:?}'s",
                document.id
            )));
        }
        let endpoint = document
            .pds_endpoint()
            .ok_or_else(|| Unresolved::Invalid(format!("{did} names no #atproto_pds service")))?;
        let key = document
            .signing_key_multibase()
            .ok_or_else(|| Unresolved::Invalid(format!("{did} names no #atproto key")))?;
        let key = CommitKey::from_multibase(key)
            .map_err(|error| Unresolved::Invalid(format!("{did}'s #atproto key: {error}")))?;
        Ok((endpoint.to_owned(), key))
    }

    /// The set's `permissions`, out of the proof `endpoint` answers with.
    async fn record(
        &self,
        endpoint: &str,
        did: &str,
        key: &CommitKey,
        nsid: &str,
    ) -> Result<Vec<Value>, Unresolved> {
        let mut url = url::Url::parse(&format!(
            "{}/xrpc/com.atproto.sync.getRecord",
            endpoint.trim_end_matches('/')
        ))
        .map_err(|error| Unresolved::Invalid(format!("{did}'s server {endpoint}: {error}")))?;
        url.query_pairs_mut()
            .append_pair("did", did)
            .append_pair("collection", SCHEMA_COLLECTION)
            .append_pair("rkey", nsid);
        let absent =
            || Unresolved::Invalid(format!("{did}'s repository holds no lexicon at {nsid}"));
        let response = self.send(url.as_str()).await?;
        if response.status() == reqwest::StatusCode::BAD_REQUEST {
            let body = didbot_http::read_bounded(response, MAX_ERROR_BYTES)
                .await
                .unwrap_or_default();
            let error = serde_json::from_slice::<Value>(&body)
                .ok()
                .and_then(|body| body.get("error")?.as_str().map(str::to_owned));
            return Err(match error.as_deref() {
                Some(RECORD_NOT_FOUND) => absent(),
                _ => Unresolved::Unreachable(format!("{url} answered 400 {error:?}")),
            });
        }
        let proof = Self::body(url.as_str(), response, MAX_PROOF_BYTES).await?;
        let record = didbot_repo::read_record(&proof, did, key, SCHEMA_COLLECTION, nsid)
            .map_err(|error| Unresolved::Invalid(format!("{did}'s proof of {nsid}: {error}")))?
            .ok_or_else(absent)?;
        permissions_of(record, nsid)
    }

    /// A GET through a client pinned to `url`'s own addresses, following no
    /// redirect, with its body read against `max_bytes`.
    async fn get(&self, url: &str, max_bytes: usize) -> Result<Vec<u8>, Unresolved> {
        let response = self.send(url).await?;
        Self::body(url, response, max_bytes).await
    }

    /// The same GET, with the status left to the caller.
    async fn send(&self, url: &str) -> Result<reqwest::Response, Unresolved> {
        let unreachable = Unresolved::Unreachable;
        let parsed =
            url::Url::parse(url).map_err(|error| unreachable(format!("{url}: {error}")))?;
        let client = didbot_http::pinned_client(&parsed, self.private)
            .await
            .map_err(|error| unreachable(error.to_string()))?;
        client
            .get(parsed)
            .send()
            .await
            .map_err(|error| unreachable(format!("fetching {url}: {error}")))
    }

    /// A success's body, read against `max_bytes`.
    async fn body(
        url: &str,
        response: reqwest::Response,
        max_bytes: usize,
    ) -> Result<Vec<u8>, Unresolved> {
        let unreachable = Unresolved::Unreachable;
        let response = response
            .error_for_status()
            .map_err(|error| unreachable(format!("{url} answered: {error}")))?;
        didbot_http::read_bounded(response, max_bytes)
            .await
            .map_err(|error| unreachable(format!("reading {url}: {error}")))
    }
}

impl Resolve for Network {
    fn resolve<'a>(&'a self, nsid: &'a str) -> ResolveFuture<'a> {
        Box::pin(async move {
            let did = self.authority(nsid).await?;
            let (endpoint, key) = self.identity(&did).await?;
            self.record(&endpoint, &did, &key, nsid).await
        })
    }
}

/// Where `nsid`'s authority publishes its DID: `_lexicon.`, then the NSID
/// less its name, reversed.
fn lexicon_name(nsid: &str) -> String {
    let (authority, _) = nsid.rsplit_once('.').unwrap_or((nsid, ""));
    let reversed: Vec<&str> = authority.rsplit('.').collect();
    format!("_lexicon.{}", reversed.join("."))
}

/// The `permissions` of the `permission-set` a lexicon record defines as
/// `main`, or why it defines none.
fn permissions_of(record: Value, nsid: &str) -> Result<Vec<Value>, Unresolved> {
    let lexicon: LexiconDoc = serde_json::from_value(record)
        .map_err(|error| Unresolved::Invalid(format!("{nsid} is not a lexicon: {error}")))?;
    if lexicon.lexicon != 1 || lexicon.id != nsid {
        return Err(Unresolved::Invalid(format!(
            "the record at {nsid} is lexicon {}, version {}",
            lexicon.id, lexicon.lexicon
        )));
    }
    if lexicon.main_type() != Some("permission-set") {
        return Err(Unresolved::Invalid(format!(
            "{nsid} defines no permission-set as main"
        )));
    }
    lexicon
        .main()
        .and_then(|main| main.get("permissions"))
        .and_then(Value::as_array)
        .cloned()
        .ok_or_else(|| Unresolved::Invalid(format!("{nsid} lists no permissions")))
}

/// What the cache knows about one set.
struct Entry {
    /// The last copy that resolved, and when.
    copy: Option<(Arc<Vec<Value>>, Instant)>,
    /// When resolving the set was last tried.
    tried: Instant,
    /// Why that try failed, if it did.
    failed: Option<Unresolved>,
}

impl Entry {
    /// The last copy, while it is young enough to stand in for the set.
    fn kept(&self, now: Instant) -> Option<Arc<Vec<Value>>> {
        self.copy
            .as_ref()
            .filter(|(_, resolved)| now.duration_since(*resolved) < KEEP_FOR)
            .map(|(copy, _)| copy.clone())
    }
}

/// Every set this deployment has resolved, shared by every account.
pub struct PermissionSets {
    resolver: Arc<dyn Resolve>,
    cache: Mutex<lru::LruCache<String, Entry>>,
}

impl PermissionSets {
    /// Sets resolved by `resolver`.
    pub fn new(resolver: Arc<dyn Resolve>) -> Self {
        Self {
            resolver,
            cache: Mutex::new(lru::LruCache::new(
                NonZeroUsize::new(CACHE_ENTRIES).expect("a constant above zero"),
            )),
        }
    }

    /// Sets resolved over the network; see [`Network::new`].
    pub fn over_network(doh: impl Into<String>, private: PrivateAddresses) -> Self {
        Self::new(Arc::new(Network::new(doh, private)))
    }

    /// `requested` with each `include:` replaced by the atoms its set
    /// grants.
    ///
    /// An `include:` grants nothing when it names no set, when its set does
    /// not resolve, and when its atoms would take the request past what one
    /// grant may hold ([`MAX_GRANT_ATOMS`], [`MAX_GRANT_BYTES`]). Every
    /// other atom is kept as it was asked for.
    pub async fn expand(&self, requested: &ScopeSet) -> ScopeSet {
        let started = Instant::now();
        let mut atoms: Vec<Scope> = requested
            .0
            .iter()
            .filter(|atom| !matches!(atom, Scope::Include { .. }))
            .cloned()
            .collect();
        let mut bytes: usize = atoms.iter().map(printed_len).sum();
        let mut sets = 0;
        for atom in &requested.0 {
            let Scope::Include { nsid, aud } = atom else {
                continue;
            };
            let granted = match self.include(nsid, aud.as_deref(), started, &mut sets).await {
                Ok(granted) => granted,
                Err(why) => {
                    tracing::debug!(include = %atom, reason = %why, "an included permission set grants nothing");
                    continue;
                }
            };
            let added: usize = granted.iter().map(printed_len).sum();
            if atoms.len() + granted.len() > MAX_GRANT_ATOMS || bytes + added > MAX_GRANT_BYTES {
                tracing::debug!(
                    include = %atom,
                    atoms = granted.len(),
                    "an included permission set grants nothing: its atoms would pass what one grant may hold"
                );
                continue;
            }
            bytes += added;
            atoms.extend(granted);
        }
        ScopeSet::new(atoms)
    }

    /// What one `include:` grants, or why it grants nothing.
    async fn include(
        &self,
        nsid: &str,
        aud: Option<&str>,
        started: Instant,
        sets: &mut usize,
    ) -> Result<Vec<Scope>, String> {
        let include = Include::new(nsid, aud).map_err(|error| error.to_string())?;
        *sets += 1;
        if *sets > MAX_SETS {
            return Err(format!("the request names more than {MAX_SETS} sets"));
        }
        let left = EXPANSION_DEADLINE.saturating_sub(started.elapsed());
        if left.is_zero() {
            return Err("the request's other sets used its time".to_owned());
        }
        let permissions = self
            .permissions(include.nsid(), left.min(SET_DEADLINE))
            .await
            .map_err(|error| error.to_string())?;
        Ok(include.grants(&permissions))
    }

    /// The published `permissions` of the set `nsid` names, from the cache
    /// or resolved within `budget`.
    async fn permissions(
        &self,
        nsid: &str,
        budget: Duration,
    ) -> Result<Arc<Vec<Value>>, Unresolved> {
        if let Some(answer) = self.cached(nsid, Instant::now()) {
            return answer;
        }
        let began = Instant::now();
        let outcome = tokio::time::timeout(budget, self.resolver.resolve(nsid)).await;
        let resolved = match outcome {
            Ok(resolved) => resolved,
            // This request ran out of time, not the set: nothing was learned
            // about the set, so nothing is remembered.
            Err(_) if budget < SET_DEADLINE => {
                return self
                    .lock()
                    .peek(nsid)
                    .and_then(|entry| entry.kept(Instant::now()))
                    .ok_or(Unresolved::Deadline);
            }
            Err(_) => Err(Unresolved::Deadline),
        };
        tracing::debug!(
            nsid,
            outcome = resolved
                .as_ref()
                .map_or_else(ToString::to_string, |_| "resolved".to_owned()),
            latency_ms = began.elapsed().as_millis(),
            "resolved a permission set"
        );
        self.remember(nsid, resolved)
    }

    /// The cache's answer for `nsid` at `now`, if it has one that needs no
    /// resolution: a copy still fresh, or a failure too recent to retry.
    fn cached(&self, nsid: &str, now: Instant) -> Option<Result<Arc<Vec<Value>>, Unresolved>> {
        let mut cache = self.lock();
        let entry = cache.get(nsid)?;
        let since = now.duration_since(entry.tried);
        match &entry.failed {
            None if since < FRESH_FOR => entry.kept(now).map(Ok),
            Some(why) if since < RETRY_AFTER => Some(entry.kept(now).ok_or_else(|| why.clone())),
            _ => None,
        }
    }

    /// Stores what resolving `nsid` just established, and answers with what
    /// a sign-in should use: the new copy, or the last one while the
    /// authority cannot be reached and the copy is kept.
    fn remember(
        &self,
        nsid: &str,
        resolved: Result<Vec<Value>, Unresolved>,
    ) -> Result<Arc<Vec<Value>>, Unresolved> {
        let now = Instant::now();
        let mut cache = self.lock();
        let previous = cache.pop(nsid);
        let (entry, answer) = match resolved {
            Ok(permissions) => {
                let permissions = Arc::new(permissions);
                let entry = Entry {
                    copy: Some((permissions.clone(), now)),
                    tried: now,
                    failed: None,
                };
                (entry, Ok(permissions))
            }
            Err(why) => {
                let copy = previous
                    .and_then(|entry| entry.copy)
                    .filter(|(_, resolved)| {
                        !why.answered() && now.duration_since(*resolved) < KEEP_FOR
                    });
                let answer = copy
                    .as_ref()
                    .map(|(copy, _)| copy.clone())
                    .ok_or_else(|| why.clone());
                let entry = Entry {
                    copy,
                    tried: now,
                    failed: Some(why),
                };
                (entry, answer)
            }
        };
        cache.put(nsid.to_owned(), entry);
        answer
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, lru::LruCache<String, Entry>> {
        self.cache
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }
}

impl Default for PermissionSets {
    /// Sets resolved over the public internet, through Cloudflare's
    /// resolver, `didbot_http::doh::CLOUDFLARE`.
    fn default() -> Self {
        Self::over_network(didbot_http::doh::CLOUDFLARE, PrivateAddresses::Refused)
    }
}

/// How many bytes `atom` adds to a grant's scope string, its separator
/// included.
fn printed_len(atom: &Scope) -> usize {
    atom.to_string().len() + 1
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use serde_json::json;

    use super::*;

    /// A [`Resolve`] answering from a script and counting what it was asked.
    struct Script {
        answers: Mutex<HashMap<String, Result<Vec<Value>, Unresolved>>>,
        calls: AtomicUsize,
    }

    impl Script {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                answers: Mutex::new(HashMap::new()),
                calls: AtomicUsize::new(0),
            })
        }

        fn answer(&self, nsid: &str, answer: Result<Vec<Value>, Unresolved>) {
            self.answers
                .lock()
                .expect("answers")
                .insert(nsid.to_owned(), answer);
        }

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

    impl Resolve for Script {
        fn resolve<'a>(&'a self, nsid: &'a str) -> ResolveFuture<'a> {
            Box::pin(async move {
                self.calls.fetch_add(1, Ordering::SeqCst);
                self.answers
                    .lock()
                    .expect("answers")
                    .get(nsid)
                    .cloned()
                    .unwrap_or_else(|| Err(Unresolved::NoAuthority(nsid.to_owned())))
            })
        }
    }

    fn repo(collection: &str) -> Vec<Value> {
        vec![json!({
            "type": "permission",
            "resource": "repo",
            "collection": [collection],
        })]
    }

    fn scope(s: &str) -> ScopeSet {
        ScopeSet::parse(s).expect("a scope")
    }

    #[test]
    fn the_authority_is_the_nsid_less_its_name_reversed() {
        assert_eq!(lexicon_name("app.bsky.authFullApp"), "_lexicon.bsky.app");
        assert_eq!(
            lexicon_name("edu.university.dept.lab.blogging.getBlogPost"),
            "_lexicon.blogging.lab.dept.university.edu"
        );
    }

    /// The rest of the request is judged as asked, whatever happens to the
    /// sets beside it.
    #[tokio::test]
    async fn a_set_that_does_not_resolve_grants_nothing_and_fails_nothing() {
        let script = Script::new();
        script.answer("com.example.authFull", Ok(repo("com.example.post")));
        let sets = PermissionSets::new(script.clone());
        let expanded = sets
            .expand(&scope(
                "atproto include:com.example.authFull include:org.example.authGone \
                 include:not-an-nsid rpc:com.example.get?aud=*",
            ))
            .await;
        assert_eq!(
            expanded.to_string(),
            "atproto repo:com.example.post rpc:com.example.get?aud=*"
        );
        assert_eq!(
            script.calls(),
            2,
            "an include naming no NSID is not resolved"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_resolved_set_is_reused_until_it_is_stale() {
        let script = Script::new();
        script.answer("com.example.authFull", Ok(repo("com.example.post")));
        let sets = PermissionSets::new(script.clone());
        let request = scope("include:com.example.authFull");

        sets.expand(&request).await;
        sets.expand(&request).await;
        assert_eq!(script.calls(), 1);

        script.answer("com.example.authFull", Ok(repo("com.example.like")));
        tokio::time::advance(FRESH_FOR).await;
        assert_eq!(
            sets.expand(&request).await.to_string(),
            "repo:com.example.like"
        );
        assert_eq!(script.calls(), 2, "a stale set is resolved again");
    }

    /// The spec: "if resolution fails, the previously existing stale
    /// values can be used". A failure is remembered, so a set whose
    /// authority is down costs one lookup per interval.
    #[tokio::test(start_paused = true)]
    async fn a_failed_refresh_keeps_the_last_copy_and_is_retried_later() {
        let script = Script::new();
        script.answer("com.example.authFull", Ok(repo("com.example.post")));
        let sets = PermissionSets::new(script.clone());
        let request = scope("include:com.example.authFull");
        sets.expand(&request).await;

        script.answer(
            "com.example.authFull",
            Err(Unresolved::Unreachable("down".to_owned())),
        );
        tokio::time::advance(FRESH_FOR).await;
        assert_eq!(
            sets.expand(&request).await.to_string(),
            "repo:com.example.post"
        );
        assert_eq!(
            sets.expand(&request).await.to_string(),
            "repo:com.example.post"
        );
        assert_eq!(script.calls(), 2, "the failure is not retried at once");

        tokio::time::advance(RETRY_AFTER).await;
        sets.expand(&request).await;
        assert_eq!(script.calls(), 3, "and is retried after a while");

        tokio::time::advance(KEEP_FOR).await;
        assert_eq!(
            sets.expand(&request).await.to_string(),
            "",
            "a copy past its keep grants nothing"
        );
    }

    /// The last copy stands in only while the authority cannot be reached.
    /// An authority that answers, and not with the set, ends it at once.
    #[tokio::test(start_paused = true)]
    async fn an_authority_that_withdraws_a_set_ends_its_last_copy() {
        let script = Script::new();
        script.answer("com.example.authFull", Ok(repo("com.example.post")));
        let sets = PermissionSets::new(script.clone());
        let request = scope("include:com.example.authFull");
        sets.expand(&request).await;

        script.answer(
            "com.example.authFull",
            Err(Unresolved::Invalid("no record there".to_owned())),
        );
        tokio::time::advance(FRESH_FOR).await;
        assert_eq!(sets.expand(&request).await.to_string(), "");
        assert_eq!(
            sets.expand(&request).await.to_string(),
            "",
            "nor while the answer is remembered"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_set_that_never_resolved_is_not_asked_for_again_at_once() {
        let script = Script::new();
        let sets = PermissionSets::new(script.clone());
        let request = scope("include:org.example.authGone");
        sets.expand(&request).await;
        sets.expand(&request).await;
        assert_eq!(script.calls(), 1);
        tokio::time::advance(RETRY_AFTER).await;
        sets.expand(&request).await;
        assert_eq!(script.calls(), 2);
    }

    /// A slow authority is remembered as failed, and costs a request no
    /// more than its deadline.
    #[tokio::test(start_paused = true)]
    async fn a_slow_set_is_cut_off_at_its_deadline() {
        struct Slow;
        impl Resolve for Slow {
            fn resolve<'a>(&'a self, _nsid: &'a str) -> ResolveFuture<'a> {
                Box::pin(async {
                    tokio::time::sleep(Duration::from_secs(3600)).await;
                    Ok(Vec::new())
                })
            }
        }
        let sets = PermissionSets::new(Arc::new(Slow));
        let started = Instant::now();
        let expanded = sets
            .expand(&scope(
                "atproto include:com.example.authOne include:com.example.authTwo \
                 include:com.example.authThree",
            ))
            .await;
        assert_eq!(expanded.to_string(), "atproto");
        assert_eq!(started.elapsed(), EXPANSION_DEADLINE);
    }

    #[tokio::test]
    async fn one_request_resolves_at_most_its_bound_of_sets() {
        let script = Script::new();
        let names: Vec<String> = (0..MAX_SETS + 4)
            .map(|i| format!("include:com.example.auth{i}"))
            .collect();
        PermissionSets::new(script.clone())
            .expand(&scope(&names.join(" ")))
            .await;
        assert_eq!(script.calls(), MAX_SETS);
    }

    /// A set that would take the request past what a grant may hold grants
    /// nothing, and the sets that fit are still granted.
    #[tokio::test]
    async fn a_set_past_what_a_grant_holds_grants_nothing() {
        let script = Script::new();
        let methods: Vec<String> = (0..MAX_GRANT_ATOMS)
            .map(|i| format!("com.example.get{i}"))
            .collect();
        script.answer(
            "com.example.authHuge",
            Ok(vec![json!({
                "type": "permission",
                "resource": "rpc",
                "lxm": methods,
                "aud": "*",
            })]),
        );
        script.answer("org.example.authSmall", Ok(repo("org.example.post")));
        let expanded = PermissionSets::new(script)
            .expand(&scope(
                "atproto include:com.example.authHuge include:org.example.authSmall",
            ))
            .await;
        assert_eq!(expanded.to_string(), "atproto repo:org.example.post");
    }
}