didbot_serve/

policy_poll.rs

//! The policy poll: the operator's `bot.did.policy` and
//! `bot.did.policyBinding` records, read on the operator poll's tick and
//! built into one policy set.
//!
//! # List the collection, not track individual records
//!
//! `com.atproto.repo.listRecords`, walked via its cursor every tick -- not
//! `getRecord` against a set of rkeys this server remembers from before.
//! This was an explicit design question and the owner answered it directly:
//! list the collection. Ideally this would subscribe rather than poll, but
//! there is no efficient way to subscribe to one collection in one foreign
//! repository, so polling `listRecords` on a timer is what this ships with.
//!
//! One tick lists both collections to their last page from the endpoint
//! the operator poll resolved, builds the whole set from them with
//! [`didbot_pds::policy_bindings::resolve`], and swaps the tree into the
//! gate every write and grant is judged by. A listing that fails at any page
//! builds nothing, so the gate keeps the set it has; a record the build
//! cannot honour is refused on its own, keeping the record in force at its
//! key where there is one, so neither an unreachable repository nor a
//! mid-edit one widens what an account may do.
//!
//! # Across a restart
//!
//! Every set installed is written to [`didbot_pds::PolicyStore`], and
//! [`PolicyPoll::restore`] reads it back into the gate at boot, before the
//! first tick runs. Without that, a restart is a window in which the
//! operator's denials are not in force — which would make restarting the
//! server a way around them. A file that is there and cannot be used stops
//! the boot; see that module's own doc comment for why that direction.
//!
//! What the enforced set holds, and what the newest build refused, are
//! published on `bot.did.stats` — see
//! [`didbot_pds::policy_bindings::EnforcedSet`].

use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;

use didbot_pds::policy_records::{BINDING_COLLECTION, POLICY_COLLECTION};
use didbot_pds::policy_tree::TreePolicyGate;
use didbot_pds::{
    ApplyEvent, AtUri, Boundary, Coverage, EmptiedSet, EnforcedBinding, EnforcedPolicy,
    EnforcedSet, Observed, PolicyRevision, PolicySet, PolicyStore, PolicyStoreError, Registry,
    RejectedBuild, ServerIdentity,
};
use didbot_policy_records::AcrossBuilds;
use serde::Deserialize;
use tracing::{debug, error, info, warn};

/// How many records `listRecords` is asked for per page. The maximum
/// `com.atproto.repo.listRecords` allows.
const PAGE_SIZE: u32 = 100;

/// The most pages one listing follows before giving up.
///
/// Each page is bounded in bytes; the number of them was not. A PDS that
/// answers every cursor with another full page keeps the loop running and
/// the map growing for as long as it cares to, which is an unbounded read
/// against a host this deployment does not run. At a hundred records a page
/// this is room for a hundred thousand policy records, which is far past any
/// operator's set and far short of a problem.
const MAX_PAGES: u32 = 200;

/// Why a listing could not complete. Every variant means the caller has no
/// records at all from this attempt -- see this module's doc comment on why
/// a failed listing must never apply a partial result.
#[derive(Debug, thiserror::Error, Clone)]
pub enum PollError {
    /// A network or HTTP-level failure fetching some page of the listing.
    #[error("fetching {url}: {message}")]
    Fetch {
        /// The URL that failed.
        url: String,
        /// What went wrong.
        message: String,
    },
    /// A page's body was not the shape `listRecords` promises.
    #[error("parsing response from {url}: {message}")]
    BadResponse {
        /// The URL whose response did not parse.
        url: String,
        /// What went wrong.
        message: String,
    },
    /// A listing kept answering with another full page, past the cap on how
    /// many one listing follows.
    #[error("listing {url} was still paging after {pages} pages")]
    TooManyPages {
        /// The endpoint that would not stop.
        url: String,
        /// The cap it ran into.
        pages: u32,
    },
}

#[derive(Debug, Deserialize)]
struct ListRecordsResponse {
    records: Vec<ListedRecord>,
    cursor: Option<String>,
}

#[derive(Debug, Deserialize)]
struct ListedRecord {
    uri: String,
    value: serde_json::Value,
}

/// Page through `operator_did`'s `collection` at `endpoint` to completion,
/// returning every record found keyed by at-uri.
///
/// `Err` on the first failure of any kind -- an unreachable endpoint, a
/// malformed page -- deliberately with no partial result: a caller reading
/// two collections as one revision relies on that to apply neither when
/// either fails. Takes an already resolved `endpoint`, so one resolution of
/// the operator's document serves every listing a tick makes.
pub async fn list_records(
    client: &reqwest::Client,
    endpoint: &str,
    operator_did: &str,
    collection: &str,
) -> Result<HashMap<String, serde_json::Value>, PollError> {
    let list_url = format!("{endpoint}/xrpc/com.atproto.repo.listRecords");
    let mut out = HashMap::new();
    let mut cursor: Option<String> = None;
    let mut pages: u32 = 0;
    loop {
        let mut query = vec![
            ("repo".to_string(), operator_did.to_string()),
            ("collection".to_string(), collection.to_string()),
            ("limit".to_string(), PAGE_SIZE.to_string()),
        ];
        if let Some(cursor) = &cursor {
            query.push(("cursor".to_string(), cursor.clone()));
        }
        let response = client
            .get(&list_url)
            .query(&query)
            .send()
            .await
            .map_err(|error| PollError::Fetch {
                url: list_url.clone(),
                message: error.to_string(),
            })?
            .error_for_status()
            .map_err(|error| PollError::Fetch {
                url: list_url.clone(),
                message: error.to_string(),
            })?;
        let body = didbot_http::read_bounded(response, didbot_http::MAX_JSON_BODY)
            .await
            .map_err(|error| PollError::BadResponse {
                url: list_url.clone(),
                message: error.to_string(),
            })?;
        let page: ListRecordsResponse =
            serde_json::from_slice(&body).map_err(|error| PollError::BadResponse {
                url: list_url.clone(),
                message: error.to_string(),
            })?;

        let page_len = page.records.len();
        for record in page.records {
            out.insert(record.uri, record.value);
        }

        match page.cursor {
            // atproto's convention: an empty page, or a page shorter than
            // the requested limit, means there is nothing more -- some
            // implementations still echo a cursor on the final page, so
            // relying on cursor-presence alone can loop forever against
            // one of those.
            Some(next) if page_len as u32 == PAGE_SIZE => cursor = Some(next),
            _ => break,
        }
        pages += 1;
        if pages >= MAX_PAGES {
            // A failed listing, not a short one. The all-or-nothing
            // contract is what keeps the set last built serving, and
            // answering with the pages that did arrive would install a set
            // the operator never wrote.
            return Err(PollError::TooManyPages {
                url: list_url,
                pages: MAX_PAGES,
            });
        }
    }
    Ok(out)
}

/// Why a tick installed nothing. The gate keeps the set it had.
#[derive(Debug, thiserror::Error)]
pub enum PolicyPollError {
    /// One collection could not be listed to completion.
    #[error("listing {collection}: {source}")]
    Listing {
        /// The collection whose listing failed.
        collection: &'static str,
        /// What went wrong.
        #[source]
        source: PollError,
    },
    /// The listing would have taken away every operator denial at once.
    #[error(transparent)]
    Emptied(#[from] EmptiedSet),
}

/// Why a boot could not put the kept set back in force.
#[derive(Debug, thiserror::Error)]
pub enum RestoreError {
    /// The file is there and cannot be used.
    #[error(transparent)]
    Unusable(#[from] PolicyStoreError),
    /// The file is a set this binary cannot build whole: an engine it no
    /// longer carries, an action it no longer answers.
    #[error(
        "the policy set kept in the data directory holds record(s) this build cannot honour, \
         and starting without them would widen what an agent may do: {refused}"
    )]
    Unbuildable {
        /// Every record this build could not honour, and why.
        refused: RejectedBuild,
    },
    /// The kept set was listed from a repository that is not this run's
    /// operator.
    #[error(
        "the policy set kept in the data directory was listed from {held} and this run's \
         operator is {configured}: one of the two is wrong. Point --owner back at {held}, or \
         delete the kept set to start from {configured}'s repository alone"
    )]
    AnotherOperator {
        /// The repository the kept set was listed from.
        held: String,
        /// The operator this run is configured for.
        configured: String,
    },
}

/// The operator's policy records as this deployment reads and enforces
/// them, across ticks.
pub struct PolicyPoll {
    operator_did: String,
    this_pds: String,
    gate: Arc<TreePolicyGate>,
    set: PolicySet,
    /// Where the installed set is kept so a restart comes back with it.
    /// `None` for a run with no data directory, which has nowhere durable
    /// to put one.
    store: Option<PolicyStore>,
    /// Held here, not per tick, so a policy whose new revision will not
    /// compile has the revision that did to fall back to.
    across: AcrossBuilds,
    /// The lookups of the set in force, swapped beside the tree so the
    /// write path reads what the policies it will be judged by declare.
    /// Shared with the router; see [`crate::DeclaredLookups`].
    declared_lookups: Arc<crate::DeclaredLookups>,
}

impl PolicyPoll {
    /// A poll of `operator_did`'s repository on behalf of the server whose
    /// service DID is `this_pds`, swapping each set it builds into `gate`
    /// and keeping it in `store`. `declared_lookups` is the router's, so
    /// the set that judges a write and the records read for it are the
    /// same set.
    pub fn new(
        operator_did: impl Into<String>,
        this_pds: impl Into<String>,
        gate: Arc<TreePolicyGate>,
        store: Option<PolicyStore>,
        declared_lookups: Arc<crate::DeclaredLookups>,
    ) -> Self {
        Self {
            operator_did: operator_did.into(),
            this_pds: this_pds.into(),
            gate,
            set: PolicySet::new(),
            store,
            across: AcrossBuilds::new(),
            declared_lookups,
        }
    }

    /// Puts the kept set back in force, before any tick has run.
    ///
    /// Answers the revision restored, or `None` for a deployment that has
    /// never installed one.
    ///
    /// # Errors
    ///
    /// [`RestoreError`] for a kept set this binary cannot read or cannot
    /// build whole. Both stop the boot: a server that comes online without
    /// denials it had in force is the one outcome this must not produce.
    pub fn restore(
        &self,
        registry: &dyn Registry,
    ) -> Result<Option<Arc<PolicyRevision>>, RestoreError> {
        let Some(store) = &self.store else {
            return Ok(None);
        };
        let Some(held) = store.load()? else {
            return Ok(None);
        };
        if held.observed.author != self.operator_did {
            return Err(RestoreError::AnotherOperator {
                held: held.observed.author,
                configured: self.operator_did.clone(),
            });
        }
        // Nothing is in force yet, so there is no set for a listing to
        // empty: the only refusal a restore can meet is a record.
        let bound = self
            .build(&held.observed, registry, &PolicyRevision::default())
            .expect("a build against nothing in force cannot empty a set");
        if !bound.refused.is_empty() {
            return Err(RestoreError::Unbuildable {
                refused: RejectedBuild::of(bound.refused),
            });
        }
        store.agrees(&held.digest, &bound.revision)?;
        let (revision, _) = self.set.install(bound.revision, Vec::new());
        let enforced = self.summarise(&revision, registry, Vec::new(), Vec::new());
        info!(
            digest = %revision.digest,
            policies = revision.policies.len(),
            bindings = revision.bindings.len(),
            path = %store.path().display(),
            "put the kept operator policy set back in force before the first poll"
        );
        self.declared_lookups.publish(bound.lookups);
        self.gate.swap_bound(bound.tree, bound.bindings, enforced);
        Ok(Some(revision))
    }

    /// What the newest observation refused, if it refused anything.
    pub fn rejected(&self) -> Option<Arc<RejectedBuild>> {
        self.set.rejected()
    }

    /// What the gate enforces, as an operator reads it back.
    pub fn enforced_set(&self) -> Option<Arc<EnforcedSet>> {
        self.gate.enforced_set()
    }

    /// One tick against the operator's repository at `endpoint`: list both
    /// collections, build the set, and swap it in. `registry` answers the
    /// admission tree a binding's subjects are checked against.
    pub async fn poll_once(
        &self,
        client: &reqwest::Client,
        endpoint: &str,
        registry: &dyn Registry,
    ) -> Result<Arc<PolicyRevision>, PolicyPollError> {
        let mut listed = Vec::with_capacity(2);
        for collection in [POLICY_COLLECTION, BINDING_COLLECTION] {
            let listing = list_records(client, endpoint, &self.operator_did, collection)
                .await
                .map_err(|source| PolicyPollError::Listing { collection, source })?;
            listed.push(by_record_key(listing));
        }
        let bindings = listed.pop().unwrap_or_default();
        let policies = listed.pop().unwrap_or_default();
        let observed = Observed {
            author: self.operator_did.clone(),
            policies,
            bindings,
        };

        let in_force = self.set.current();
        let bound = match self.build(&observed, registry, &in_force) {
            Ok(bound) => bound,
            Err(emptied) => {
                warn!(
                    collection = emptied.collection,
                    held = emptied.held,
                    "the operator's repository listed no records where a set is in force; \
                     nothing installed and the set in force keeps serving"
                );
                self.set
                    .reject(RejectedBuild::of(vec![didbot_pds::BuildFailure {
                        collection: emptied.collection,
                        key: String::new(),
                        reason: emptied.to_string(),
                    }]));
                return Err(PolicyPollError::Emptied(emptied));
            }
        };

        for failure in &bound.carried {
            warn!(
                collection = failure.collection,
                key = %failure.key,
                reason = %failure.reason,
                "a policy record was refused; the record in force at its key keeps serving"
            );
        }
        for failure in &bound.refused {
            warn!(
                collection = failure.collection,
                key = %failure.key,
                reason = %failure.reason,
                "a policy record was refused and nothing was in force at its key; every record \
                 beside it is installed"
            );
        }

        let digest = bound.revision.digest.clone();
        let refused = bound.refused.clone();
        let carried = bound.carried.clone();
        let (revision, events) = self.set.install(bound.revision, bound.refused);
        for event in &events {
            match event {
                ApplyEvent::Added { collection, key }
                | ApplyEvent::Updated { collection, key }
                | ApplyEvent::Removed { collection, key } => {
                    debug!(collection, %key, ?event, "operator policy poll observed a change");
                }
            }
        }
        for key in &bound.report.unbound {
            debug!(%key, "a policy no binding names reaches nobody");
        }
        for (binding, subject) in &bound.report.foreign_subjects {
            debug!(%binding, %subject, "a binding names a subject that is not an account here; ignored");
        }
        for key in &bound.report.spares_held_tokens {
            warn!(
                %key,
                "a policy denies a client before the write and not at it; tokens it already \
                 holds keep writing until they expire"
            );
        }
        for fallback in &bound.load.last_good_fallbacks {
            warn!(
                id = %fallback.id,
                enforced = ?fallback.enforced_revision,
                failed = ?fallback.failed_revision,
                reason = %fallback.reason,
                "a policy's new revision will not compile; the revision that did keeps being \
                 enforced"
            );
        }
        for warning in &bound.load.warnings {
            warn!(
                ?warning,
                "a policy compiled with something the operator should see"
            );
        }
        info!(
            %digest,
            policies = revision.policies.len(),
            bindings = revision.bindings.len(),
            bound = bound.bindings.len(),
            refused = refused.len(),
            carried = carried.len(),
            "installed the operator's policy set"
        );
        let enforced = self.summarise(&revision, registry, refused, carried);
        self.declared_lookups.publish(bound.lookups);
        self.gate.swap_bound(bound.tree, bound.bindings, enforced);

        // Kept only once it is in force, and only when it is not already
        // what is on disk: a tick that changes nothing rewrites nothing.
        //
        // A set that will not write down is still in force for this run, so
        // the tick stands rather than being thrown away. What it costs is a
        // restart coming back on the set before it, which is why this is an
        // error rather than a warning.
        if let Some(store) = &self.store {
            if in_force.digest != revision.digest {
                match store.save(&self.operator_did, &revision) {
                    Ok(()) => {
                        debug!(%digest, path = %store.path().display(), "kept the policy set in force");
                    }
                    Err(err) => error!(
                        %err,
                        %digest,
                        "the policy set in force could not be written down; a restart would come \
                         back on the set before it"
                    ),
                }
            }
        }
        Ok(revision)
    }

    /// Builds `observed` against `in_force`, with this server's identity and
    /// the registry's admission tree.
    fn build(
        &self,
        observed: &Observed,
        registry: &dyn Registry,
        in_force: &PolicyRevision,
    ) -> Result<didbot_pds::BoundTree, EmptiedSet> {
        let chain_of = |did: &str| registry.boundary(did).ok();
        didbot_pds::policy_bindings::resolve(
            observed,
            ServerIdentity {
                service_did: &self.this_pds,
                operator_did: &self.operator_did,
            },
            &chain_of,
            didbot_policy_records::builtin_policies(),
            &HashMap::new(),
            &self.across,
            in_force,
        )
    }

    /// The set as an operator reads it back, with each binding's reach
    /// resolved against the accounts this server holds right now.
    ///
    /// One pass over the accounts per installed set, which is why it is
    /// built here — on the poll's tick — and handed to the gate, rather
    /// than computed on each `bot.did.stats` call.
    fn summarise(
        &self,
        revision: &PolicyRevision,
        registry: &dyn Registry,
        refused: Vec<didbot_pds::BuildFailure>,
        carried: Vec<didbot_pds::BuildFailure>,
    ) -> EnforcedSet {
        let boundaries: Vec<(String, Boundary)> = registry
            .accounts()
            .into_iter()
            .map(|account| {
                let did = account.did.as_str().to_owned();
                let chain = registry
                    .boundary(&did)
                    .unwrap_or_else(|_| vec![did.clone()]);
                (did, Boundary::of(chain))
            })
            .collect();
        EnforcedSet {
            digest: revision.digest.clone(),
            policies: revision
                .policies
                .iter()
                .map(|(key, record)| EnforcedPolicy {
                    key: key.clone(),
                    actions: record.actions.iter().map(ToString::to_string).collect(),
                    document: record.document.type_name().to_owned(),
                })
                .collect(),
            bindings: revision
                .bindings
                .iter()
                .map(|(key, record)| {
                    let coverage = Coverage::of(record);
                    EnforcedBinding {
                        key: key.clone(),
                        policies: record
                            .policies
                            .iter()
                            .map(|policy| policy.key.clone())
                            .collect(),
                        accounts: boundaries
                            .iter()
                            .filter(|(_, boundary)| coverage.covers(boundary, &self.this_pds))
                            .map(|(did, _)| did.clone())
                            .collect(),
                    }
                })
                .collect(),
            refused,
            carried,
        }
    }
}

/// A listing keyed by at-uri, re-keyed by record key: the key a binding
/// names a policy by, and the one a refusal names a record by.
fn by_record_key(
    listing: HashMap<String, serde_json::Value>,
) -> BTreeMap<String, serde_json::Value> {
    listing
        .into_iter()
        .map(|(uri, value)| {
            let key = AtUri::parse(&uri)
                .ok()
                .and_then(AtUri::rkey)
                .unwrap_or_else(|| uri.rsplit('/').next().unwrap_or(&uri))
                .to_owned();
            (key, value)
        })
        .collect()
}

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

    const COLLECTION: &str = "bot.did.policy";

    fn value(revision: u64) -> serde_json::Value {
        json!({
            "language": "regex",
            "revision": revision,
            "applicability": { "write": { "collections": ["c1"] } },
            "predicate": {}
        })
    }

    /// An operator PDS that answers one prepared page per request, echoing
    /// a cursor until the last one.
    async fn fake_operator_pds(
        pages: Vec<Vec<(&'static str, serde_json::Value)>>,
    ) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
        use axum::extract::Query;
        use axum::response::IntoResponse;
        use axum::routing::get;
        use axum::{Json, Router};
        use std::sync::atomic::{AtomicUsize, Ordering};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");

        let pages = Arc::new(pages);
        let page_index = Arc::new(AtomicUsize::new(0));

        let router = Router::new().route(
            "/xrpc/com.atproto.repo.listRecords",
            get(move |Query(_params): Query<HashMap<String, String>>| {
                let pages = pages.clone();
                let page_index = page_index.clone();
                async move {
                    let i = page_index.fetch_add(1, Ordering::SeqCst);
                    let page = pages.get(i).cloned().unwrap_or_default();
                    let records: Vec<_> = page
                        .iter()
                        .map(
                            |(uri, value)| json!({ "uri": uri, "cid": "bafyfake", "value": value }),
                        )
                        .collect();
                    let has_more = i + 1 < pages.len();
                    let body = if has_more {
                        json!({ "records": records, "cursor": format!("page-{}", i + 1) })
                    } else {
                        json!({ "records": records })
                    };
                    Json(body).into_response()
                }
            }),
        );

        let handle = tokio::spawn(async move {
            axum::serve(listener, router).await.ok();
        });
        (addr, handle)
    }

    /// An operator whose host accepts the connection and never answers
    /// costs the listing exactly `didbot_http::DEFAULT_TIMEOUT`, then a
    /// fetch error.
    #[tokio::test(start_paused = true)]
    async fn an_operator_host_that_never_answers_is_a_fetch_error_after_the_client_timeout() {
        let addr = didbot_http::test_support::hung_server();
        let endpoint = format!("http://{addr}");
        let client = didbot_http::client();
        let started = tokio::time::Instant::now();
        let listing = list_records(&client, &endpoint, "did:web:op.example", COLLECTION);
        let err = tokio::time::timeout(didbot_http::DEFAULT_TIMEOUT * 2, listing)
            .await
            .expect("the client's own timeout fires first")
            .expect_err("nothing ever answers");
        assert!(matches!(err, PollError::Fetch { .. }), "{err:?}");
        assert_eq!(started.elapsed(), didbot_http::DEFAULT_TIMEOUT);
    }

    #[tokio::test]
    async fn a_single_page_is_fetched_in_full() {
        let (addr, server) =
            fake_operator_pds(vec![vec![("at://op/bot.did.policy/a", value(1))]]).await;
        let endpoint = format!("http://{addr}");
        let client = didbot_http::client();

        let listing = list_records(&client, &endpoint, "did:web:op.example", COLLECTION)
            .await
            .expect("should succeed");
        assert_eq!(listing.len(), 1);
        server.abort();
    }

    #[tokio::test]
    async fn a_full_page_followed_by_a_short_page_is_paginated_through() {
        let full_page: Vec<_> = (0..PAGE_SIZE)
            .map(|i| {
                let uri: &'static str =
                    Box::leak(format!("at://op/bot.did.policy/{i}").into_boxed_str());
                (uri, value(1))
            })
            .collect();
        let second_page = vec![("at://op/bot.did.policy/extra", value(1))];
        let (addr, server) = fake_operator_pds(vec![full_page, second_page]).await;
        let endpoint = format!("http://{addr}");
        let client = didbot_http::client();

        let listing = list_records(&client, &endpoint, "did:web:op.example", COLLECTION)
            .await
            .expect("should succeed");
        assert_eq!(listing.len() as u32, PAGE_SIZE + 1);
        server.abort();
    }

    /// A PDS that answers every cursor with another full page. Each page is
    /// bounded in bytes and the number of them was not, so this loop ran and
    /// this map grew for as long as the host cared to keep answering.
    async fn endless_operator_pds() -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
        use axum::extract::Query;
        use axum::routing::get;
        use axum::{Json, Router};
        use std::sync::atomic::{AtomicU32, Ordering};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let served = Arc::new(AtomicU32::new(0));
        let router = Router::new().route(
            "/xrpc/com.atproto.repo.listRecords",
            get(move |Query(_): Query<HashMap<String, String>>| {
                let counted = served.clone();
                async move {
                    let page = counted.fetch_add(1, Ordering::SeqCst);
                    let records: Vec<_> = (0..PAGE_SIZE)
                        .map(|n| {
                            json!({
                                "uri": format!("at://op/bot.did.policy/p{page}-{n}"),
                                "cid": "bafyfake",
                                "value": value(1),
                            })
                        })
                        .collect();
                    Json(json!({ "records": records, "cursor": format!("page-{}", page + 1) }))
                }
            }),
        );
        let handle = tokio::spawn(async move {
            axum::serve(listener, router).await.ok();
        });
        (addr, handle)
    }

    /// The listing stops at the cap and reports a failure rather than an
    /// answer. A short listing would install a set the operator never wrote;
    /// a failure leaves the set last built serving.
    #[tokio::test]
    async fn a_listing_that_never_ends_stops_at_the_cap_and_fails() {
        let (addr, server) = endless_operator_pds().await;
        let endpoint = format!("http://{addr}");
        let client = didbot_http::client();
        let error = list_records(&client, &endpoint, "did:web:op.example", COLLECTION)
            .await
            .expect_err("a listing that never ends is not an answer");
        match error {
            PollError::TooManyPages { pages, .. } => assert_eq!(pages, MAX_PAGES),
            other => panic!("{other:?}"),
        }
        server.abort();
    }
}