didbot_serve/

operator_poll.rs

//! Drives `didbot_pds::OperatorRelation`'s poll against a real operator repository,
//! and reacts to what it finds. See `plan/handshake.md`.
//!
//! `didbot-pds` holds no HTTP client and no async runtime, so
//! [`didbot_pds::ClaimSource`] is a plain synchronous trait; the fetch that
//! actually reaches the operator's repository lives here instead, where both
//! are already dependencies. The bridge is `Prefetched`, a private
//! `ClaimSource` below: this module does the one HTTP round trip a poll needs
//! *before* calling
//! [`didbot_pds::OperatorRelation::poll`], then hands the already-fetched result to
//! it through a [`didbot_pds::ClaimSource`] that does no I/O of its own. That
//! keeps the decision logic in `didbot-pds`, testable with no network at all,
//! while the network reaches this crate exactly once per tick.
//!
//! # What each poll actually does
//!
//! 1. Resolve the operator's DID document through
//!    [`didbot_identity::document_url`], which dispatches by method, to find
//!    its `#atproto_pds` service endpoint.
//! 2. `com.atproto.repo.getRecord` at that endpoint, for
//!    `bot.did.operator` at the record key `didbot_pds::OperatorRelation::hostname`
//!    — a single direct lookup, not a list-and-filter, because that is what
//!    keying the record by hostname buys.
//! 3. Hand the result to [`didbot_pds::OperatorRelation::poll`], and act on the
//!    [`didbot_pds::OperatorTransition`] it reports: throw or release the
//!    self-triggered pause, and republish the server's own registration
//!    record once claimed.
//! 4. Look for the same record once per root account — one the operator
//!    created from their own repository — keyed by *that* hostname, and
//!    report to the registry whether it still stands. A record missing for
//!    longer than the grace window quarantines the account and everything
//!    beneath it; see [`refresh_root_standing`].
//!
//! # What is unverified
//!
//! This has never been run against a real operator PDS — there is no such
//! server reachable from where this was built. The `getRecord` shape, the
//! DID resolution and the error handling for "no such record" versus "the
//! server is actually down" are implemented against atproto's documented
//! XRPC conventions, not exercised against a live implementation. The
//! `did:plc` path reaches `https://plc.directory` in production and is
//! exercised here against a loopback stand-in for it.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration as StdDuration;

use didbot_http::PrivateAddresses;
use didbot_identity::{DidDocument, ResolveError};
use didbot_pds::allowance::Allowances;
use didbot_pds::{
    AccountState, ClaimFetchError, ClaimSource, Estop, EstopCause, EstopMode, HostedAccount, Lock,
    OperatorClaim, OperatorRelation, OperatorTransition, Party, Registry, ServerLifecycle, Tag,
};
use serde::Deserialize;
use time::Duration;
use time::OffsetDateTime;
use tokio::sync::{watch, Notify};
use tokio::task::JoinHandle;
use tokio::time::MissedTickBehavior;
use tracing::{info, warn};

use crate::operator_handle::OperatorHandle;
use crate::policy_poll::PolicyPoll;

/// The collection this poll reads. `didbot-pds` holds no schema for a
/// collection defined by someone else's repository convention here, so this
/// is a plain string rather than a `didbot_lexicon::nsid` constant — the
/// constant lives with the lexicon, in `didbot_lexicon::nsid::OPERATOR`.
const COLLECTION: &str = didbot_lexicon::nsid::OPERATOR;

/// How often this deployment ships polling for an operator record, absent
/// configuration. Independent of [`didbot_pds::DEFAULT_GRACE_WINDOW`]: the
/// interval is how often the network is asked, the grace window is how long
/// a missed answer is tolerated before it means something. Five minutes
/// against a six-hour grace window is 72 chances to succeed before a
/// transient outage could ever matter.
///
/// The two compose into the bound an operator actually cares about: a
/// revocation is *acted on* — the self-triggered pause thrown — no later
/// than the grace window plus one interval after the record stops being
/// readable, because the poll that notices can be a full interval away from
/// the instant the window elapsed. Six hours and five minutes, as this
/// deployment ships, and
/// `a_revocation_is_acted_on_within_the_grace_window_plus_one_interval`
/// below holds the two constants to it.
///
/// That is the bound for *this server's own* behaviour. It says nothing
/// about a third party, who is not polling anything: a reader learns of a
/// revocation when they next read the operator's record.
pub const DEFAULT_POLL_INTERVAL: StdDuration = StdDuration::from_secs(5 * 60);

/// A [`ClaimSource`] holding a result already fetched, so
/// [`didbot_pds::OperatorRelation::poll`] — synchronous by design — can be called
/// with no I/O of its own. Not `pub`: the whole point is that nothing
/// outside this module's own poll tick constructs one.
struct Prefetched(Result<Option<OperatorClaim>, ClaimFetchError>);

impl ClaimSource for Prefetched {
    fn get_claim(
        &self,
        _operator_did: &str,
        _hostname: &str,
    ) -> Result<Option<OperatorClaim>, ClaimFetchError> {
        self.0.clone()
    }
}

/// What `com.atproto.repo.getRecord` answers with on success.
#[derive(Debug, Deserialize)]
struct GetRecordResponse {
    value: serde_json::Value,
}

/// The atproto error name a repository uses for "there is no record at that
/// key" -- the one 400 this poll reads as absence rather than as a failure.
const RECORD_NOT_FOUND: &str = "RecordNotFound";

/// Resolves `operator_did`'s `#atproto_pds` service endpoint, then fetches
/// the `bot.did.operator` record at `hostname` from it.
///
/// The document URL comes from [`didbot_identity::document_url`], so every
/// method that resolver speaks — `did:web` against the hostname the DID
/// names, `did:plc` against the PLC directory — is a method an operator can
/// be configured with here, and a method it does not speak is refused by
/// name rather than by failing to parse as something else.
///
/// `Ok(None)` for "reached the operator's PDS and it has no record at that
/// key" -- a 404, or a 400 whose body names [`RECORD_NOT_FOUND`].
/// [`ClaimFetchError::Unresolvable`] for a DID with no document URL at all.
/// [`ClaimFetchError::Unreachable`] for everything this module cannot tell
/// apart from an outage: connection failure, a document that does not parse
/// or claims a different DID, a 400 that means something other than absence,
/// or any other status this module does not recognise.
#[cfg(test)]
async fn fetch_claim(
    client: &reqwest::Client,
    operator_did: &str,
    hostname: &str,
    private: PrivateAddresses,
) -> Result<Option<OperatorClaim>, ClaimFetchError> {
    let (endpoint, pinned) = resolve_endpoint(client, operator_did, private).await?;
    fetch_claim_from(&pinned, &endpoint, operator_did, hostname).await
}

/// Resolves `operator_did`'s document and reads its `#atproto_pds`
/// endpoint out of it, on its own so one tick resolves the operator once
/// and reads every record it needs from the same endpoint.
///
/// Answers with the endpoint and the client to read it through: one
/// pinned to what the endpoint's host resolved to and following no
/// redirect. Reading through anything else would let a name that answers
/// differently the second time, or a 3xx from the operator's PDS, land the
/// request at an address the check never saw.
pub(crate) async fn resolve_endpoint(
    client: &reqwest::Client,
    operator_did: &str,
    private: PrivateAddresses,
) -> Result<(String, reqwest::Client), ClaimFetchError> {
    let (_, endpoint, pinned) = resolve_operator(client, operator_did, private).await?;
    Ok((endpoint, pinned))
}

/// [`resolve_endpoint`], with the document it read the endpoint from.
async fn resolve_operator(
    client: &reqwest::Client,
    operator_did: &str,
    private: PrivateAddresses,
) -> Result<(DidDocument, String, reqwest::Client), ClaimFetchError> {
    let document_url =
        didbot_identity::document_url(operator_did).map_err(|error| match error {
            ResolveError::UnsupportedMethod { .. } | ResolveError::InvalidDid { .. } => {
                ClaimFetchError::Unresolvable(format!("operator did {operator_did}: {error}"))
            }
            other => ClaimFetchError::Unreachable(format!("operator did {operator_did}: {other}")),
        })?;
    resolve_operator_at(client, &document_url, operator_did, private).await
}

/// [`fetch_claim`] from `document_url` onward: fetch the document there,
/// verify it belongs to `operator_did`, and read the claim out of the PDS it
/// names.
///
/// Split from [`fetch_claim`] so this crate's tests can point the whole
/// exchange at a loopback stand-in for a directory that is otherwise only
/// reachable over the public internet — `https://plc.directory`, which
/// [`didbot_identity::document_url`] is the sole source of. Which URL a DID
/// maps to is `didbot-identity`'s to decide and to test; what this module
/// does with the document there is tested here.
#[cfg(test)]
async fn fetch_claim_at(
    client: &reqwest::Client,
    document_url: &str,
    operator_did: &str,
    hostname: &str,
    private: PrivateAddresses,
) -> Result<Option<OperatorClaim>, ClaimFetchError> {
    let (_, endpoint, pinned) =
        resolve_operator_at(client, document_url, operator_did, private).await?;
    fetch_claim_from(&pinned, &endpoint, operator_did, hostname).await
}

/// [`resolve_operator`] from `document_url` onward.
///
/// The endpoint the document names is checked against
/// [`didbot_http::REFUSED_BLOCKS`] before anything fetches from it. The
/// operator's DID is configuration, but the `#atproto_pds` endpoint inside
/// their document is not: it is a URL this server is told to fetch from by
/// a document it just read, which is the same kind of URL a `client_id` is.
async fn resolve_operator_at(
    client: &reqwest::Client,
    document_url: &str,
    operator_did: &str,
    private: PrivateAddresses,
) -> Result<(DidDocument, String, reqwest::Client), ClaimFetchError> {
    let response = client
        .get(document_url)
        .send()
        .await
        .map_err(|error| ClaimFetchError::Unreachable(format!("fetching {document_url}: {error}")))?
        .error_for_status()
        .map_err(|error| {
            ClaimFetchError::Unreachable(format!("fetching {document_url}: {error}"))
        })?;
    let body = didbot_http::read_bounded(response, didbot_http::MAX_JSON_BODY)
        .await
        .map_err(|error| {
            ClaimFetchError::Unreachable(format!("reading {document_url}: {error}"))
        })?;

    let document: DidDocument = serde_json::from_slice(&body).map_err(|error| {
        ClaimFetchError::Unreachable(format!("parsing {document_url}: {error}"))
    })?;
    // The same check `didbot_identity::resolve` makes, for the same reason
    // the DID spec gives: "the DID declared in the document (in the `id`
    // field) should always be verified against what was expected." A
    // document served at the right URL claiming a different subject is a
    // document about somebody else, and reading a claim out of the PDS it
    // names would be reading somebody else's repository.
    if document.id != operator_did {
        return Err(ClaimFetchError::Unreachable(format!(
            "the did document at {document_url} claims {:?}, expected {operator_did:?}",
            document.id
        )));
    }

    let endpoint = document.pds_endpoint().ok_or_else(|| {
        ClaimFetchError::Unreachable(format!(
            "{operator_did}'s did document names no #atproto_pds service"
        ))
    })?;
    let url = url::Url::parse(endpoint).map_err(|error| {
        ClaimFetchError::Unreachable(format!(
            "the did document at {document_url} names {endpoint} as its #atproto_pds \
             service, which is not a URL: {error}"
        ))
    })?;
    let pinned = didbot_http::pinned_client(&url, private)
        .await
        .map_err(|error| {
            ClaimFetchError::Unreachable(format!(
                "the did document at {document_url} names {endpoint} as its #atproto_pds \
                 service: {error}"
            ))
        })?;
    let endpoint = endpoint.to_owned();
    Ok((document, endpoint, pinned))
}

/// Reads the `bot.did.operator` record at `hostname` from the operator's
/// repository at `endpoint`.
pub(crate) async fn fetch_claim_from(
    client: &reqwest::Client,
    endpoint: &str,
    operator_did: &str,
    hostname: &str,
) -> Result<Option<OperatorClaim>, ClaimFetchError> {
    let record_url = format!("{endpoint}/xrpc/com.atproto.repo.getRecord");
    let response = client
        .get(&record_url)
        .query(&[
            ("repo", operator_did),
            ("collection", COLLECTION),
            ("rkey", hostname),
        ])
        .send()
        .await
        .map_err(|error| ClaimFetchError::Unreachable(format!("fetching {record_url}: {error}")))?;

    // atproto answers "no such record" as a 400 naming `RecordNotFound` in
    // the error body, not a 404 -- but a well-behaved implementation may
    // reasonably choose either, so both can mean "no claim."
    //
    // The status alone does not say which. A 400 is also what a PDS answers
    // for a collection it does not know, an rkey it will not accept, or a
    // proxy in front of it rewriting the query, and reading those as "there
    // is no operator claim" pauses provisioning for a reason nothing names.
    // So a 400 only means absence when the body says `RecordNotFound`; every
    // other 400 is a repository this poll could not get an answer out of.
    let status = response.status().as_u16();
    if status == 404 {
        return Ok(None);
    }
    if status == 400 {
        // Under the bound the record itself is read under: a body past it
        // names no error, and the poll says it could not read the answer.
        let body = didbot_http::read_bounded(response, didbot_http::MAX_JSON_BODY)
            .await
            .map(didbot_http::body_text)
            .unwrap_or_default();
        let named = serde_json::from_str::<serde_json::Value>(&body)
            .ok()
            .and_then(|value| {
                value
                    .get("error")
                    .and_then(serde_json::Value::as_str)
                    .map(str::to_owned)
            });
        return match named.as_deref() {
            Some(RECORD_NOT_FOUND) => Ok(None),
            Some(other) => Err(ClaimFetchError::Unreachable(format!(
                "{record_url} answered 400 {other}, which is not {RECORD_NOT_FOUND}: \
                 the repository was reached and refused the request"
            ))),
            None => Err(ClaimFetchError::Unreachable(format!(
                "{record_url} answered 400 with no atproto error name: {body}"
            ))),
        };
    }
    let response = response
        .error_for_status()
        .map_err(|error| ClaimFetchError::Unreachable(format!("{record_url} answered: {error}")))?;
    let body = didbot_http::read_bounded(response, didbot_http::MAX_JSON_BODY)
        .await
        .map_err(|error| ClaimFetchError::Unreachable(format!("reading {record_url}: {error}")))?;
    let body: GetRecordResponse = serde_json::from_slice(&body)
        .map_err(|error| ClaimFetchError::Unreachable(format!("parsing {record_url}: {error}")))?;

    Ok(OperatorClaim::from_json(&body.value))
}

/// Builds the [`OperatorRelation`] a poll needs.
///
/// The operator is configuration — a server is launched knowing who runs
/// it, and never discovers that from the network. The claim is read at the
/// server's own authority as its DID spells it — a development server's
/// port included — which is the key `didbot operate` writes and the key
/// [`refresh_root_standing`] reads every root's record at.
pub fn configure(
    registry: &dyn Registry,
    operator_did: &str,
    grace: time::Duration,
) -> Arc<OperatorRelation> {
    let server_did = registry.service_did();
    let hostname = didbot_identity::did::AccountDid::parse(&server_did)
        .expect("the server's own DID is a did:web")
        .authority();
    Arc::new(OperatorRelation::new(
        operator_did.to_owned(),
        hostname,
        server_did,
        grace,
    ))
}

/// The shortest interval between two polls, however many nudges arrive.
///
/// This is the whole anti-amplification property, and it is a floor on
/// *outbound* work rather than a limit on inbound requests. A token bucket
/// on the route would bound how many nudges are accepted while still letting
/// an accepted burst turn into a burst of polls at somebody else's PDS; a
/// floor bounds the thing that must not amplify directly, so the worst an
/// abusive caller achieves is one poll per this interval — which is what the
/// timer was going to do anyway.
///
/// Ten seconds. Short enough that an operator who has just run
/// `didbot operate` sees the server flip within one breath rather than waiting
/// out [`DEFAULT_POLL_INTERVAL`], and long enough that the operator's PDS is
/// never asked more than six times a minute by this server no matter what
/// arrives at the socket.
pub const MIN_POLL_INTERVAL: StdDuration = StdDuration::from_secs(10);

/// The most one poll tick -- the claim's up to two sequential HTTP
/// round trips, then the local decision -- may take before the tick is
/// abandoned.
///
/// A backstop on top of the client's own budget, not a substitute for it:
/// [`didbot_http::client`] falls back to a client with no timeout at all if
/// its builder ever fails to construct one, and this is what still bounds a
/// tick if that happens. Thirty seconds is triple
/// [`didbot_http::DEFAULT_TIMEOUT`], which is the most either HTTP call
/// normally takes.
const POLL_TICK_TIMEOUT: StdDuration = StdDuration::from_secs(30);

/// The most root accounts one tick re-reads the operator's record for.
///
/// Sized the way [`crate::stale_sweep::DEFAULT_SWEEP_LIMIT`] is, against the
/// round trips behind it rather than anything in this process: a hundred
/// `getRecord`s at the operator's PDS finish well inside
/// `POLL_TICK_TIMEOUT` against a healthy one. A deployment with more roots
/// than this reads the rest on the ticks that follow, each resuming where
/// the last stopped.
pub const VOUCH_REREAD_LIMIT: usize = 100;

/// What [`refresh_root_standing`] carries from one tick to the next.
#[derive(Debug, Default)]
pub struct RootStanding {
    /// The last root read, so the next tick resumes past it.
    resume_after: Option<String>,
    /// When each root's record was last read present — or first read
    /// missing, for one this poll has never seen present — so a lapse is
    /// measured against the same grace window the server's own claim gets.
    last_seen: BTreeMap<String, OffsetDateTime>,
}

/// A request that this server look for its operator's claim sooner than its
/// timer would.
///
/// **Urgency, never authority.** Nothing here asserts a claim, carries a
/// credential, or names an operator; the only thing a caller can influence
/// is *when* this server next reads a repository it was going to read
/// anyway, and the answer still comes from that read. A server that took
/// anybody's word for a claim would be a server that could be claimed by
/// whoever asked, which is the whole reason the flip is observation-only.
///
/// Nudges **coalesce**: many arriving inside one [`MIN_POLL_INTERVAL`]
/// window set one flag and produce one poll, not one poll each. See
/// [`OperatorPoll::spawn`].
#[derive(Debug)]
pub struct PollNudge {
    notify: Notify,
    /// Whether an [`OperatorPoll`] is listening, so a wait on a channel
    /// nothing drains answers at once.
    served: AtomicBool,
    /// How many polls have begun.
    begun: AtomicU64,
    /// The last poll to finish: its number, and whether it read the
    /// operator's records.
    finished: watch::Sender<(u64, bool)>,
}

impl Default for PollNudge {
    fn default() -> Self {
        Self {
            notify: Notify::new(),
            served: AtomicBool::new(false),
            begun: AtomicU64::new(0),
            finished: watch::Sender::new((0, false)),
        }
    }
}

impl PollNudge {
    /// A fresh nudge channel.
    pub fn new() -> Self {
        Self::default()
    }

    /// Nudges, then waits up to `within` for a poll that began after the
    /// nudge to finish. Answers whether that poll read the operator's
    /// records, so a caller refused for want of an allowance can check
    /// again against what the operator's repository holds now.
    ///
    /// The nudge is still bound by the floor, so a poll that ran moments
    /// ago makes this wait out the rest of [`MIN_POLL_INTERVAL`].
    pub async fn nudge_and_wait(&self, within: StdDuration) -> bool {
        if !self.served.load(Ordering::Acquire) {
            return false;
        }
        let mut finished = self.finished.subscribe();
        let before = self.begun.load(Ordering::Acquire);
        self.nudge();
        let waited =
            tokio::time::timeout(within, finished.wait_for(|(poll, _)| *poll > before)).await;
        matches!(waited, Ok(Ok(done)) if done.1)
    }

    /// Asks the poll task to look now, if it is not already about to.
    ///
    /// Cheap and infallible on purpose: the cost of an inbound nudge has to
    /// stay a wake-up on an existing task, or the route in front of it
    /// becomes a way to make this server do work.
    pub fn nudge(&self) {
        self.notify.notify_one();
    }
}

/// When an [`OperatorPoll`] polls: its timer, and the floor a nudge
/// cannot get under.
///
/// The two together, rather than two parameters, because they are one
/// decision: how often this server reads somebody else's repository. Reading
/// them apart is how a deployment ends up with a tight floor and a loose
/// timer, or the reverse, and no idea which one it meant.
#[derive(Debug, Clone, Copy)]
pub struct PollSchedule {
    /// How often the poll fires on its own.
    pub interval: StdDuration,
    /// The shortest gap between two polls, however many nudges arrive. See
    /// [`MIN_POLL_INTERVAL`], which is what every deployment passes; a field
    /// only so this module's own tests can exercise the bound without
    /// sleeping through it.
    pub floor: StdDuration,
}

impl Default for PollSchedule {
    /// [`DEFAULT_POLL_INTERVAL`] and [`MIN_POLL_INTERVAL`].
    fn default() -> Self {
        Self {
            interval: DEFAULT_POLL_INTERVAL,
            floor: MIN_POLL_INTERVAL,
        }
    }
}

/// A running operator poll, stopped by [`OperatorPoll::stop`] or by
/// dropping it.
pub struct OperatorPoll {
    handle: JoinHandle<()>,
    stop: Arc<Notify>,
}

impl OperatorPoll {
    /// Starts polling `relation` against the real network on `schedule`,
    /// driving `estop`'s self latch from what it finds, admitting the
    /// reservations `registry` holds that the operator has vouched for, and
    /// reporting to `lifecycle` whether a claim currently stands. Each tick
    /// that reads the operator's DID document also starts a check of the
    /// handle it claims, into `operator_handle`, which the tick does not
    /// wait for.
    ///
    /// The first tick fires immediately, not after one interval: a server
    /// that waited a full interval before its first chance to find an
    /// already-existing claim would refuse writes for no reason a fresh
    /// restart should ever cause.
    ///
    /// `nudge` lets a caller ask for a poll sooner than the schedule's
    /// interval, bounded by its floor. A nudge arriving less than the floor
    /// since the last poll waits out the remainder rather than polling, and
    /// every further nudge inside that window coalesces into the same single
    /// poll. So an unbounded burst of nudges costs this server one extra
    /// outbound request, not one per nudge — see [`PollNudge`].
    #[allow(clippy::too_many_arguments)]
    pub fn spawn(
        relation: Arc<OperatorRelation>,
        estop: Arc<Estop>,
        lifecycle: Arc<ServerLifecycle>,
        registry: Arc<dyn Registry>,
        client: reqwest::Client,
        private: PrivateAddresses,
        schedule: PollSchedule,
        nudge: Arc<PollNudge>,
        policies: Option<Arc<PolicyPoll>>,
        records_read: Arc<OperatorRecords>,
        operator_handle: Option<Arc<OperatorHandle>>,
    ) -> Self {
        let stop = Arc::new(Notify::new());
        let notified = stop.clone();
        nudge.served.store(true, Ordering::Release);
        let handle = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(schedule.interval);
            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
            // The floor is kept as an instant rather than a counter so a
            // nudge that arrives during a slow poll cannot stack: whatever
            // happened while the poll ran, the next one is still no earlier
            // than `MIN_POLL_INTERVAL` after this one started.
            let mut earliest_next = tokio::time::Instant::now();
            // Where the last tick's re-read of root records stopped, and
            // when each was last seen; see `RootStanding`.
            let mut standing = RootStanding::default();
            loop {
                let nudged = tokio::select! {
                    biased;
                    () = notified.notified() => break,
                    _ = ticker.tick() => false,
                    () = nudge.notify.notified() => true,
                };
                if nudged {
                    let now = tokio::time::Instant::now();
                    if now < earliest_next {
                        // Wait out the floor instead of polling. Any further
                        // nudges arriving meanwhile re-arm the same `Notify`
                        // and are absorbed by the one poll below -- which is
                        // exactly the coalescing this is for.
                        tokio::select! {
                            biased;
                            () = notified.notified() => break,
                            () = tokio::time::sleep_until(earliest_next) => {}
                        }
                    }
                }
                // Absorb a nudge that arrived while the floor was being
                // waited out: it is asking for exactly the poll about to
                // happen, and `Notify` holds one permit, so leaving it armed
                // would buy the burst a second poll it has already been
                // given. A nudge arriving *during* the poll below is
                // deliberately not absorbed -- that one may have been sent
                // after the read it would need to see.
                {
                    let pending = nudge.notify.notified();
                    tokio::pin!(pending);
                    tokio::select! {
                        biased;
                        () = &mut pending => {}
                        () = std::future::ready(()) => {}
                    }
                }
                earliest_next = tokio::time::Instant::now() + schedule.floor;
                let poll = nudge.begun.fetch_add(1, Ordering::AcqRel) + 1;
                let ticked = tokio::time::timeout(
                    POLL_TICK_TIMEOUT,
                    tick(
                        &relation,
                        &estop,
                        &lifecycle,
                        registry.as_ref(),
                        &client,
                        OffsetDateTime::now_utc(),
                        private,
                        &mut standing,
                        policies.as_deref(),
                        &records_read,
                        operator_handle.as_ref(),
                    ),
                )
                .await;
                nudge
                    .finished
                    .send_replace((poll, matches!(ticked, Ok((_, true)))));
                if ticked.is_err() {
                    // No outcome was recorded for this tick -- not even the
                    // "could not reach the operator's repository" one
                    // `poll_once` itself would have logged, because it never
                    // returned. The next tick is unaffected: `earliest_next`
                    // is already set, and the ticker fires again on its own
                    // schedule.
                    warn!(
                        operator = relation.operator_did(),
                        hostname = relation.hostname(),
                        timeout = ?POLL_TICK_TIMEOUT,
                        "operator poll tick did not finish within its timeout; abandoning it"
                    );
                }
            }
            nudge.served.store(false, Ordering::Release);
        });
        Self { handle, stop }
    }

    /// Stops the poll and waits for the task to finish.
    pub async fn stop(self) {
        self.stop.notify_one();
        let _ = self.handle.await;
    }
}

/// One poll: fetch, decide, act. Split out from [`OperatorPoll::spawn`]'s
/// closure, and public, so a caller — an integration test walking the
/// bootstrap, in particular — can drive exactly one tick against a real
/// operator repository without spawning a task or waiting on a timer. The
/// spawned poll runs this and, on the same resolved endpoint, the
/// [`PolicyPoll`] it was given. Every call here re-reads standing vouches
/// from the first host; the spawned poll is what carries on between ticks.
pub async fn poll_once(
    relation: &OperatorRelation,
    estop: &Estop,
    lifecycle: &ServerLifecycle,
    registry: &dyn Registry,
    client: &reqwest::Client,
    now: OffsetDateTime,
    private: PrivateAddresses,
) -> OperatorTransition {
    tick(
        relation,
        estop,
        lifecycle,
        registry,
        client,
        now,
        private,
        &mut RootStanding::default(),
        None,
        &OperatorRecords::new(),
        None,
    )
    .await
    .0
}

/// [`poll_once`] with `standing` carried for [`refresh_root_standing`],
/// followed by `policies`' own tick against the endpoint the claim was
/// read from. A tick that could not resolve the operator reads no policies
/// either, and the gate keeps the set it had. Answers the transition, and
/// whether the operator's records were read and their allowances installed.
#[allow(clippy::too_many_arguments)]
async fn tick(
    relation: &OperatorRelation,
    estop: &Estop,
    lifecycle: &ServerLifecycle,
    registry: &dyn Registry,
    client: &reqwest::Client,
    now: OffsetDateTime,
    private: PrivateAddresses,
    standing: &mut RootStanding,
    policies: Option<&PolicyPoll>,
    records_read: &OperatorRecords,
    handle: Option<&Arc<OperatorHandle>>,
) -> (OperatorTransition, bool) {
    let resolved = match resolve_operator(client, relation.operator_did(), private).await {
        Ok((document, endpoint, pinned)) => {
            // Not awaited: the claim below never waits on a name for a page.
            if let Some(handle) = handle {
                handle.refresh_in_background(document, private);
            }
            Ok((endpoint, pinned))
        }
        Err(error) => Err(error),
    };
    let fetched = match &resolved {
        Ok((endpoint, pinned)) => {
            fetch_claim_from(
                pinned,
                endpoint,
                relation.operator_did(),
                relation.hostname(),
            )
            .await
        }
        Err(error) => Err(error.clone()),
    };
    if let Err(error) = &fetched {
        warn!(
            operator = relation.operator_did(),
            hostname = relation.hostname(),
            %error,
            "operator poll could not reach the operator's repository"
        );
    }
    let source = Prefetched(fetched);
    let transition = relation.poll(&source, now);
    // Reported to the lifecycle on every tick, before anything is thrown or
    // released: `didbot_pds::ServerState` is what the write and provisioning
    // surfaces consult, and it has to have seen this poll before a caller
    // can observe the e-stop's reaction to it.
    //
    // One boolean. "Never claimed" and "claim lapsed" are the same lifecycle
    // state with the same recovery, so the distinction stops here, where the
    // grace window that produced it lives. `GracePending` counts as standing
    // -- the window exists so a blip at the operator's PDS does not take a
    // server's writes down.
    lifecycle.observe_operator_claim(matches!(
        transition,
        OperatorTransition::NewlyClaimed
            | OperatorTransition::StillClaimed
            | OperatorTransition::GracePending
    ));
    match transition {
        OperatorTransition::NewlyClaimed => {
            info!(
                operator = relation.operator_did(),
                "found the operator's record; the claim stands"
            );
            estop.release_self();
        }
        OperatorTransition::NewlyLapsed => {
            warn!(
                operator = relation.operator_did(),
                hostname = relation.hostname(),
                "no matching operator record found within the grace window; pausing the server \
                 until the operator's claim reappears \
                 (see plan/alerts.md -- this deployment does not yet write an alert record for it)"
            );
            // A legitimate stop: by the time a claim can lapse the server
            // has been operational, so there is something running to halt.
            // `ServerPolicy::may_halt` says so as data, and this asserts
            // the two agree rather than trusting that they do -- a lapse
            // derived from anywhere the stop is unavailable would mean the
            // lifecycle and this module disagree about what has happened.
            debug_assert!(
                lifecycle.policy().may_halt,
                "a lapse threw a stop from a state whose policy forbids one"
            );
            estop.throw_self(EstopMode::Pause);
        }
        OperatorTransition::StillClaimed
        | OperatorTransition::StillLapsed
        | OperatorTransition::GracePending
        | OperatorTransition::StillUnclaimed => {}
    }
    // A cheap sanity check that costs nothing on the ordinary path: a poll
    // reporting anything other than a Cause::OperatorMissing self-pause after
    // a lapse would mean this module and `Estop` have disagreed about who
    // engaged the latch.
    if matches!(
        transition,
        OperatorTransition::NewlyLapsed | OperatorTransition::StillLapsed
    ) {
        debug_assert_eq!(estop.status().cause, Some(EstopCause::OperatorMissing));
    }
    // Claimed or not: a root's standing only ever says what the operator's
    // repository says, and having reached that repository is all this
    // needs. The same operator, the same repository, one record key per
    // root.
    if let Ok((endpoint, pinned)) = &resolved {
        refresh_root_standing(
            registry,
            pinned,
            endpoint,
            relation.operator_did(),
            standing,
            now,
            relation.grace_window(),
        )
        .await;
    }
    let allowances_read = match &resolved {
        Ok((endpoint, pinned)) => {
            refresh_allowances(
                registry,
                pinned,
                endpoint,
                relation.operator_did(),
                records_read,
            )
            .await
        }
        Err(_) => false,
    };
    if let (Some(policies), Ok((endpoint, pinned))) = (policies, &resolved) {
        if let Err(error) = policies.poll_once(pinned, endpoint, registry).await {
            warn!(
                operator = relation.operator_did(),
                %error,
                "operator policy poll applied nothing; the set last built keeps serving"
            );
        }
    }
    (transition, allowances_read)
}

/// One `bot.did.operator` record in the operator's repository, as the last
/// listing read it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct OperatorRecordLine {
    /// The record's own address in the operator's repository.
    pub uri: String,
    /// The account the record names.
    pub subject: String,
    /// When the writer says they wrote it, by the writer's own clock.
    pub created_at: Option<String>,
}

/// Every `bot.did.operator` record the last successful listing of the
/// operator's repository read.
///
/// An app holding the operator's session writes as the operator, and their
/// PDS cannot tell the two apart. So the operator recognising a record they
/// did not write is the defence, and that needs the records in front of
/// them: where each one is, who it names, and when its writer says it was
/// written.
/// Kept here because the poll already lists them on every tick for
/// [`refresh_allowances`], and a page that refetched them would put a
/// remote read behind a panel.
///
/// `None` until a listing has succeeded, which is a different fact from a
/// repository that holds none.
#[derive(Debug, Default)]
pub struct OperatorRecords {
    last_read: Mutex<Option<Vec<OperatorRecordLine>>>,
}

impl OperatorRecords {
    /// A store nothing has written to yet.
    pub fn new() -> Self {
        Self::default()
    }

    /// What the last successful listing read.
    pub fn read(&self) -> Option<Vec<OperatorRecordLine>> {
        self.last_read
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    /// Replaces what the last listing read.
    pub(crate) fn store(&self, lines: Vec<OperatorRecordLine>) {
        *self
            .last_read
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(lines);
    }
}

/// Lists every `bot.did.operator` record in the operator's repository and
/// installs the `creates` allowances they carry; see
/// [`didbot_pds::allowance`]. A listing that fails leaves the allowances
/// last installed in force, the same way a failed policy listing leaves
/// the last policy set serving: a blip at the operator's PDS must not
/// widen or narrow what hosted accounts may create.
///
/// Every record in the collection is read, not only the ones naming
/// accounts hosted here; an allowance for an account this server does not
/// host is indexed and never matched.
///
/// A record naming an account some hosted account created carries no
/// allowance, and is logged instead. The operator admits an account by
/// creating it from their own repository, so a record that names a name
/// already taken beneath a hosted creator is one the creator raced the
/// operator to: honouring it would hand the creator whatever the operator
/// meant to grant the account they were admitting. Records naming a root,
/// or a name no account holds yet, are indexed as before.
///
/// Answers whether the listing succeeded and its allowances were installed.
pub async fn refresh_allowances(
    registry: &dyn Registry,
    client: &reqwest::Client,
    endpoint: &str,
    operator_did: &str,
    records_read: &OperatorRecords,
) -> bool {
    match crate::policy_poll::list_records(client, endpoint, operator_did, COLLECTION).await {
        Ok(records) => {
            let hosted_parents: BTreeMap<String, String> = registry
                .accounts()
                .into_iter()
                .filter_map(|account| {
                    let parent = account.parent?;
                    Some((account.did.as_str().to_owned(), parent.as_str().to_owned()))
                })
                .collect();
            let allowances = Allowances::from_records(records.iter().filter_map(|(uri, value)| {
                let claim = OperatorClaim::from_json(value)?;
                if let Some(parent) = hosted_parents.get(&claim.subject) {
                    warn!(
                        record = %uri,
                        subject = %claim.subject,
                        parent = %parent,
                        "the operator's record names an account a hosted account created, \
                         so it grants nothing; the operator admits an account by creating it"
                    );
                    return None;
                }
                Some((claim.subject, claim.creates))
            }));
            registry.set_allowances(allowances);
            records_read.store(
                records
                    .iter()
                    .filter_map(|(uri, value)| {
                        Some(OperatorRecordLine {
                            uri: uri.clone(),
                            subject: value["subject"].as_str()?.to_owned(),
                            created_at: value["createdAt"].as_str().map(str::to_owned),
                        })
                    })
                    .collect(),
            );
            true
        }
        Err(error) => {
            warn!(
                operator = operator_did,
                %error,
                "could not list the operator's records; the allowances last installed stay in force"
            );
            false
        }
    }
}

/// Re-reads the operator's record for every root account and reports each
/// one's standing to the registry. Answers with the DIDs whose standing
/// changed.
///
/// A root account is one the operator created from their own repository:
/// active, hosted here, with no hosted parent, and not the server itself.
/// The same `getRecord` the server's own claim is read with, at the record
/// key the root's own hostname is. A record counts when it names the
/// root's DID exactly.
///
/// Absence is the only thing read as revocation, and only once it has
/// lasted `grace`, the window the server's own claim gets and for the
/// same reason: a server cannot tell "the operator revoked" from "the
/// operator's PDS is down". `fetch_claim_from` answers `Ok(None)` for a
/// repository that was reached and holds no record at the key, and
/// [`ClaimFetchError`] for everything it could not get an answer out of;
/// an error leaves every standing as it was and ends the pass, since every
/// root reads from the same endpoint. [`Registry::set_operator_standing`]
/// quarantines a lapsed root and everything beneath it, and releases them
/// when the record returns.
///
/// At most [`VOUCH_REREAD_LIMIT`] roots per call, in DID order from just
/// past where the last call stopped and wrapping to the first.
pub async fn refresh_root_standing(
    registry: &dyn Registry,
    client: &reqwest::Client,
    endpoint: &str,
    operator_did: &str,
    standing: &mut RootStanding,
    now: OffsetDateTime,
    grace: Duration,
) -> Vec<String> {
    let mut roots: Vec<HostedAccount> = registry
        .accounts()
        .into_iter()
        .filter(|account| account.state == AccountState::Active && account.parent.is_none())
        .collect();
    roots.sort_by(|a, b| a.did.as_str().cmp(b.did.as_str()));
    if roots.is_empty() {
        standing.resume_after = None;
        return Vec::new();
    }
    let start = standing.resume_after.as_deref().map_or(0, |last| {
        roots.partition_point(|root| root.did.as_str() <= last)
    });
    let window = roots.len().min(VOUCH_REREAD_LIMIT);
    let mut changed = Vec::new();
    for root in roots.iter().cycle().skip(start).take(window) {
        let did = root.did.as_str();
        let rkey = root.did.did().authority();
        let found = match fetch_claim_from(client, endpoint, operator_did, &rkey).await {
            Ok(found) => found,
            Err(error) => {
                warn!(
                    did,
                    %error,
                    "could not re-read the operator's record for a root account; \
                     leaving its standing as it is until the repository answers"
                );
                break;
            }
        };
        let present = match found {
            Some(claim) if claim.subject == did => true,
            Some(claim) => {
                warn!(
                    did,
                    subject = %claim.subject,
                    "the record at this root's key names a different did; not its record"
                );
                false
            }
            None => false,
        };
        let stands = if present {
            standing.last_seen.insert(did.to_owned(), now);
            true
        } else {
            // A quarantine already hung on the root is this server's own
            // record that the window ran out before, and it outlives the
            // process that hung it. Starting the clock over on a first read
            // would give a revoked subtree another whole window, every
            // restart.
            let already = root.locks.holds(Tag::new(Lock::Quarantined, Party::Parent));
            let since = *standing.last_seen.entry(did.to_owned()).or_insert_with(|| {
                if already {
                    now - grace
                } else {
                    now
                }
            });
            now - since < grace
        };
        match registry.set_operator_standing(did, stands) {
            Ok(true) => {
                info!(
                    did,
                    present, stands, "the operator's record for a root account changed standing"
                );
                changed.push(did.to_owned());
            }
            Ok(false) => {}
            Err(error) => {
                warn!(did, %error, "could not update a root account's standing");
            }
        }
        standing.resume_after = Some(did.to_owned());
    }
    changed
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::extract::Query;
    use axum::response::IntoResponse;
    use axum::routing::get;
    use axum::{Json, Router};
    use serde_json::json;
    use std::collections::HashMap;
    use std::net::SocketAddr;

    // ------------------------------------------------------------------
    // The revocation bound: what the two constants compose into
    // ------------------------------------------------------------------

    /// An operator deletes their record. How long until this server has
    /// actually stopped?
    ///
    /// Driven the way the spawned task drives it — a poll every
    /// [`DEFAULT_POLL_INTERVAL`], each one finding nothing — so the answer
    /// crosses both constants rather than restating either. The pause lands
    /// on the first tick at or after the grace window elapses, which is
    /// never later than one interval past it.
    #[test]
    fn a_revocation_is_acted_on_within_the_grace_window_plus_one_interval() {
        let grace = didbot_pds::DEFAULT_GRACE_WINDOW;
        let interval = time::Duration::seconds(
            DEFAULT_POLL_INTERVAL
                .as_secs()
                .try_into()
                .unwrap_or(i64::MAX),
        );
        let relation = OperatorRelation::new(
            "did:web:operator.example",
            "agents.example",
            "did:web:agents.example",
            grace,
        );

        let start = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid timestamp");
        let claim = OperatorClaim::new("did:web:agents.example", start);
        assert_eq!(
            relation.poll(&Prefetched(Ok(Some(claim))), start),
            OperatorTransition::NewlyClaimed
        );

        // From here the record is gone. Every subsequent tick finds nothing.
        let mut now = start;
        let lapsed_at = loop {
            now += interval;
            assert!(
                now - start <= grace + interval,
                "still running {} past the revocation",
                now - start
            );
            if relation.poll(&Prefetched(Ok(None)), now) == OperatorTransition::NewlyLapsed {
                break now;
            }
        };
        assert!(
            lapsed_at - start >= grace,
            "paused before the grace window it promised: an operator outage would trip it"
        );
        assert!(!relation.is_claimed());
    }

    // ------------------------------------------------------------------
    // fetch_claim: a real round trip against a loopback server over TLS
    // ------------------------------------------------------------------

    /// A client that trusts `ca_pem` on top of the roots it is built with.
    ///
    /// The fake below serves TLS, because every server this project talks to
    /// does and no client here can be told otherwise. This is the same seam
    /// a development stack uses through `didbot_http::EXTRA_CA_CERTS`,
    /// reached directly so two tests running at once do not share one
    /// environment variable.
    fn client_trusting(ca_pem: &str) -> reqwest::Client {
        didbot_http::builder()
            .add_root_certificate(
                reqwest::Certificate::from_pem(ca_pem.as_bytes()).expect("the authority parses"),
            )
            .build()
            .expect("a client with one extra root builds")
    }

    /// The path the fake serves an operator's document at in addition to
    /// `/.well-known/did.json`, standing in for the one directory a
    /// `did:plc` document is fetched from. `https://plc.directory` is not
    /// reachable from a test, and `didbot_identity::document_url` is the
    /// only thing that names it, so the fake offers a second path and the
    /// `did:plc` tests hand that URL to [`fetch_claim_at`] directly.
    const DIRECTORY_PATH: &str = "/plc-directory-stand-in";

    /// A minimal stand-in for an operator's PDS: serves a did:web document
    /// naming itself as the `#atproto_pds`, and answers getRecord for
    /// whatever `bot.did.operator` records `records` holds, keyed by rkey.
    async fn fake_operator_pds(
        records: HashMap<String, serde_json::Value>,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>, String) {
        fake_operator_pds_counting(records, Arc::new(std::sync::atomic::AtomicUsize::new(0))).await
    }

    /// [`fake_operator_pds`], publishing a document that claims `did` rather
    /// than the `did:web` its own address implies — an operator whose
    /// identity is a `did:plc`, whose PDS is still an ordinary host.
    async fn fake_operator_pds_for(
        did: &str,
        records: HashMap<String, serde_json::Value>,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>, String) {
        fake_operator_pds_inner(
            records,
            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            "RecordNotFound",
            Some(did.to_owned()),
        )
        .await
    }

    /// [`fake_operator_pds`], counting every `getRecord` it answers — which
    /// is one per poll, and therefore the outbound rate the nudge must not
    /// amplify.
    async fn fake_operator_pds_counting(
        records: HashMap<String, serde_json::Value>,
        hits: Arc<std::sync::atomic::AtomicUsize>,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>, String) {
        fake_operator_pds_inner(records, hits, "RecordNotFound", None).await
    }

    /// [`fake_operator_pds`], answering a miss with a caller-chosen atproto
    /// error name rather than `RecordNotFound` -- the only way to tell a 400
    /// that means "no such record" from a 400 that means something else.
    async fn fake_operator_pds_answering(
        records: HashMap<String, serde_json::Value>,
        miss: &'static str,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>, String) {
        fake_operator_pds_inner(
            records,
            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            miss,
            None,
        )
        .await
    }

    /// The one fake both of the above are views of.
    async fn fake_operator_pds_inner(
        records: HashMap<String, serde_json::Value>,
        hits: Arc<std::sync::atomic::AtomicUsize>,
        miss: &'static str,
        did_override: Option<String>,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>, String) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let base = format!("https://{addr}");
        let (tls, ca_pem) = didbot_tls::test_support::server_config(&["127.0.0.1"]);
        // The clients this fake is reached through are built inside the
        // code under test, so they are told once, for the process.
        didbot_http::test_support::trust(&ca_pem);
        let did =
            did_override.unwrap_or_else(|| format!("did:web:{}%3A{}", addr.ip(), addr.port()));

        let document = json!({
            "@context": ["https://www.w3.org/ns/did/v1"],
            "id": did,
            "alsoKnownAs": ["at://operator.example"],
            "service": [{
                "id": format!("{did}#atproto_pds"),
                "type": "AtprotoPersonalDataServer",
                "serviceEndpoint": base,
            }],
        });

        let well_known = document.clone();
        let router = Router::new()
            .route(
                "/.well-known/did.json",
                get(move || {
                    let document = well_known.clone();
                    async move { Json(document) }
                }),
            )
            .route(
                DIRECTORY_PATH,
                get(move || {
                    let document = document.clone();
                    async move { Json(document) }
                }),
            )
            .route(
                "/xrpc/com.atproto.repo.getRecord",
                get(move |Query(params): Query<HashMap<String, String>>| {
                    let records = records.clone();
                    let hits = hits.clone();
                    async move {
                        hits.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                        let rkey = params.get("rkey").cloned().unwrap_or_default();
                        match records.get(&rkey) {
                            Some(value) => Json(json!({
                                "uri": format!("at://operator/bot.did.operator/{rkey}"),
                                "cid": "bafyfake",
                                "value": value,
                            }))
                            .into_response(),
                            None => (
                                axum::http::StatusCode::BAD_REQUEST,
                                Json(json!({"error": miss})),
                            )
                                .into_response(),
                        }
                    }
                }),
            );

        let handle = tokio::spawn(async move {
            crate::serve_tls(listener, router, tls).await.ok();
        });
        (addr, handle, ca_pem)
    }

    /// A stand-in operator PDS whose `getRecord` answers a `302` pointing
    /// at another host rather than a record.
    async fn fake_operator_pds_redirecting(
        to: String,
    ) -> (SocketAddr, tokio::task::JoinHandle<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let base = format!("http://{addr}");
        let did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let document = json!({
            "@context": ["https://www.w3.org/ns/did/v1"],
            "id": did,
            "service": [{
                "id": format!("{did}#atproto_pds"),
                "type": "AtprotoPersonalDataServer",
                "serviceEndpoint": base,
            }],
        });
        let well_known = document.clone();
        let router = Router::new()
            .route(
                "/.well-known/did.json",
                get(move || {
                    let document = well_known.clone();
                    async move { Json(document) }
                }),
            )
            .route(
                DIRECTORY_PATH,
                get(move || {
                    let document = document.clone();
                    async move { Json(document) }
                }),
            )
            .route(
                "/xrpc/com.atproto.repo.getRecord",
                get(move |Query(params): Query<HashMap<String, String>>| {
                    let to = to.clone();
                    async move {
                        let rkey = params.get("rkey").cloned().unwrap_or_default();
                        axum::response::Redirect::temporary(&format!(
                            "{to}/xrpc/com.atproto.repo.getRecord\
                             ?repo=operator&collection=bot.did.operator&rkey={rkey}"
                        ))
                    }
                }),
            );

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

    /// The endpoint is checked once, so the read of it follows no redirect:
    /// a 3xx from the operator's PDS would land the request at an address
    /// that check never saw.
    #[tokio::test]
    async fn a_redirect_from_the_operators_pds_is_not_followed() {
        let hostname = "pds.example.test";
        let mut records = HashMap::new();
        records.insert(
            hostname.to_owned(),
            json!({
                "subject": "did:web:pds.example.test",
                "createdAt": "2026-01-01T00:00:00Z",
            }),
        );
        let hits = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        // Nothing ever connects to the server the redirect names, so the
        // authority it signs from is not needed here.
        let (elsewhere, elsewhere_handle, _ca) =
            fake_operator_pds_counting(records, hits.clone()).await;
        let (addr, handle) = fake_operator_pds_redirecting(format!("https://{elsewhere}")).await;
        let document_url = format!("http://{addr}{DIRECTORY_PATH}");

        let result = fetch_claim_at(
            &didbot_http::client(),
            &document_url,
            &format!("did:web:{}%3A{}", addr.ip(), addr.port()),
            hostname,
            PrivateAddresses::Allowed,
        )
        .await;
        assert!(
            matches!(result, Err(ClaimFetchError::Unreachable(_))),
            "a redirect is not a record, got {result:?}"
        );
        assert_eq!(
            hits.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "the redirect was followed"
        );

        handle.abort();
        elsewhere_handle.abort();
    }

    /// The `#atproto_pds` endpoint inside the operator's document is a URL
    /// a fetched document told this server to fetch from. A deployment on
    /// the public internet refuses one on its own network, and says which
    /// endpoint it refused.
    #[tokio::test]
    async fn an_endpoint_the_document_names_on_this_network_is_refused() {
        let (addr, handle, ca) = fake_operator_pds(HashMap::new()).await;
        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = client_trusting(&ca);

        let error = fetch_claim(
            &client,
            &operator_did,
            "pds.example.test",
            PrivateAddresses::Refused,
        )
        .await
        .expect_err("the endpoint is on this deployment's own network");
        let message = error.to_string();
        assert!(message.contains(&format!("https://{addr}")), "{message}");
        assert!(
            message.contains("will not fetch from"),
            "the refusal names the address list: {message}"
        );

        handle.abort();
    }

    #[tokio::test]
    async fn a_matching_record_is_fetched_and_parsed() {
        let hostname = "pds.example.test";
        let mut records = HashMap::new();
        records.insert(
            hostname.to_owned(),
            json!({
                "subject": "did:web:pds.example.test",
                "createdAt": "2026-01-01T00:00:00Z",
            }),
        );
        let (addr, handle, ca) = fake_operator_pds(records).await;
        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = client_trusting(&ca);

        let claim = fetch_claim(&client, &operator_did, hostname, PrivateAddresses::Allowed)
            .await
            .expect("should succeed")
            .expect("should find a claim");
        assert_eq!(claim.subject, "did:web:pds.example.test");

        handle.abort();
    }

    /// An operator whose `did:web` host accepts the connection and never
    /// speaks costs a tick's DID resolution exactly
    /// `didbot_http::DEFAULT_CONNECT_TIMEOUT` — the handshake is part of
    /// connecting — and then `Unreachable`.
    #[tokio::test(start_paused = true)]
    async fn an_operator_host_that_never_answers_is_unreachable_after_the_client_timeout() {
        let addr = didbot_http::test_support::hung_server();
        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = didbot_http::client();
        let started = tokio::time::Instant::now();
        let fetching = fetch_claim(
            &client,
            &operator_did,
            "pds.example.test",
            PrivateAddresses::Allowed,
        );
        let err = tokio::time::timeout(didbot_http::DEFAULT_TIMEOUT * 2, fetching)
            .await
            .expect("the client's own timeout fires first")
            .expect_err("nothing ever answers");
        assert!(matches!(err, ClaimFetchError::Unreachable(_)), "{err:?}");
        assert_eq!(started.elapsed(), didbot_http::DEFAULT_CONNECT_TIMEOUT);
    }

    #[tokio::test]
    async fn a_missing_record_is_ok_none_not_an_error() {
        let (addr, handle, ca) = fake_operator_pds(HashMap::new()).await;
        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = client_trusting(&ca);

        let claim = fetch_claim(
            &client,
            &operator_did,
            "pds.example.test",
            PrivateAddresses::Allowed,
        )
        .await
        .expect("a 400 RecordNotFound is not an unreachable repository");
        assert!(claim.is_none());

        handle.abort();
    }

    #[tokio::test]
    async fn a_400_that_is_not_record_not_found_is_unreachable_not_an_absent_claim() {
        // The operator's PDS answered, and what it said was that the request
        // was wrong -- a collection it does not know, a malformed rkey, a
        // proxy in front of it rewriting the query. Reading that as "there is
        // no operator claim" pauses provisioning for a reason nothing names.
        let (addr, handle, ca) =
            fake_operator_pds_answering(HashMap::new(), "InvalidRequest").await;
        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = client_trusting(&ca);

        let result = fetch_claim(
            &client,
            &operator_did,
            "pds.example.test",
            PrivateAddresses::Allowed,
        )
        .await;
        assert!(
            matches!(result, Err(ClaimFetchError::Unreachable(_))),
            "a 400 the poll cannot read as RecordNotFound must be loud, got {result:?}"
        );

        handle.abort();
    }

    /// A 400's body is read under the bound the record is read under, so one
    /// naming `RecordNotFound` past it is an answer the poll could not read,
    /// not an absent claim.
    #[tokio::test]
    async fn a_400_body_past_the_bound_is_unreachable_not_an_absent_claim() {
        let padding = "x".repeat(didbot_http::MAX_JSON_BODY);
        let router = Router::new().route(
            "/xrpc/com.atproto.repo.getRecord",
            get(move || {
                let padding = padding.clone();
                async move {
                    (
                        axum::http::StatusCode::BAD_REQUEST,
                        Json(json!({ "error": RECORD_NOT_FOUND, "message": padding })),
                    )
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        let handle = tokio::spawn(async move {
            axum::serve(listener, router).await.ok();
        });

        let result = fetch_claim_from(
            &didbot_http::client(),
            &format!("http://{addr}"),
            "did:web:operator.example",
            "pds.example.test",
        )
        .await;
        assert!(
            matches!(result, Err(ClaimFetchError::Unreachable(_))),
            "an oversized 400 was read as an absent claim: {result:?}"
        );

        handle.abort();
    }

    /// The operator whose identity is a `did:plc` — the ordinary case for a
    /// person with a Bluesky account — reaches a confirmed claim.
    ///
    /// The whole path runs: the document is fetched, checked to be the one
    /// belonging to that DID, and its `#atproto_pds` service is where the
    /// record is read from; the record is then handed to the same
    /// `OperatorRelation::poll` a live tick uses, which is where "confirmed"
    /// actually means something. The one step held out is the URL the
    /// directory lives at, which `didbot-identity` owns and tests, and which
    /// `a_did_plc_resolves_against_the_plc_directory` below pins from here.
    #[tokio::test]
    async fn a_did_plc_operator_reaches_a_confirmed_claim() {
        let operator_did = "did:plc:nlzmjyfv6loqtxyzvdcznwgf";
        let hostname = "pds.example.test";
        let server_did = "did:web:pds.example.test";

        let mut records = HashMap::new();
        records.insert(
            hostname.to_owned(),
            json!({
                "subject": server_did,
                "createdAt": "2026-01-01T00:00:00Z",
            }),
        );
        let (addr, handle, ca) = fake_operator_pds_for(operator_did, records).await;
        let document_url = format!("https://{addr}{DIRECTORY_PATH}");
        let client = client_trusting(&ca);

        let claim = fetch_claim_at(
            &client,
            &document_url,
            operator_did,
            hostname,
            PrivateAddresses::Allowed,
        )
        .await
        .expect("the directory answered and the operator's pds was reached")
        .expect("the operator's repository holds a claim at this hostname");
        assert_eq!(claim.subject, server_did);

        let relation =
            OperatorRelation::new(operator_did, hostname, server_did, time::Duration::hours(6));
        assert_eq!(
            relation.poll(
                &Prefetched(Ok(Some(claim))),
                OffsetDateTime::from_unix_timestamp(1_800_000_000).expect("valid timestamp"),
            ),
            OperatorTransition::NewlyClaimed,
            "a did:plc operator's claim confirms exactly as a did:web one does"
        );
        assert!(relation.is_claimed());

        handle.abort();
    }

    /// A directory that answers with a document belonging to somebody else
    /// is not an answer about this operator, and the claim in the repository
    /// that document names is not this operator's claim.
    #[tokio::test]
    async fn a_document_claiming_another_did_is_not_read_as_the_operators() {
        let hostname = "pds.example.test";
        let mut records = HashMap::new();
        records.insert(
            hostname.to_owned(),
            json!({
                "subject": "did:web:pds.example.test",
                "createdAt": "2026-01-01T00:00:00Z",
            }),
        );
        let (addr, handle, ca) =
            fake_operator_pds_for("did:plc:nlzmjyfv6loqtxyzvdcznwgf", records).await;
        let document_url = format!("https://{addr}{DIRECTORY_PATH}");
        let client = client_trusting(&ca);

        let result = fetch_claim_at(
            &client,
            &document_url,
            "did:plc:vwmrkybzt2ncwyipgnu6cocx",
            hostname,
            PrivateAddresses::Allowed,
        )
        .await;
        assert!(
            matches!(result, Err(ClaimFetchError::Unreachable(_))),
            "the document at this url claims a different did, got {result:?}"
        );

        handle.abort();
    }

    /// The half held out of the test above: which URL each method's document
    /// is fetched from. `did:web` is the hostname the DID names, `did:plc`
    /// the PLC directory — and this poll takes both from the one dispatch in
    /// `didbot-identity` rather than deciding either itself.
    #[test]
    fn a_did_plc_resolves_against_the_plc_directory() {
        let plc = "did:plc:nlzmjyfv6loqtxyzvdcznwgf";
        assert_eq!(
            didbot_identity::document_url(plc).expect("did:plc is a method this poll resolves"),
            format!("{}/{plc}", didbot_identity::DEFAULT_PLC_DIRECTORY)
        );
        assert_eq!(
            didbot_identity::document_url("did:web:operator.example")
                .expect("did:web is a method this poll resolves"),
            "https://operator.example/.well-known/did.json"
        );
    }

    /// A DID this deployment has no way to resolve is reported as its own
    /// error, and still lands in the grace window: nothing about being
    /// unable to look confirms a claim.
    #[tokio::test]
    async fn an_operator_did_this_poll_cannot_resolve_never_confirms_a_claim() {
        let client = didbot_http::client();
        let result = fetch_claim(
            &client,
            "did:key:zQ3shrubbery",
            "pds.example.test",
            PrivateAddresses::Allowed,
        )
        .await;
        assert!(
            matches!(result, Err(ClaimFetchError::Unresolvable(_))),
            "a method with no document url is this deployment's configuration, \
             not somebody else's outage, got {result:?}"
        );

        let relation = OperatorRelation::new(
            "did:key:zQ3shrubbery",
            "pds.example.test",
            "did:web:pds.example.test",
            time::Duration::hours(6),
        );
        let now = OffsetDateTime::from_unix_timestamp(1_800_000_000).expect("valid timestamp");
        assert_eq!(
            relation.poll(&Prefetched(result), now),
            OperatorTransition::StillUnclaimed,
            "an unresolvable operator did leaves the server unclaimed"
        );
        assert!(!relation.is_claimed());
    }

    /// The other side of the same taxonomy as
    /// `an_operator_did_this_poll_cannot_resolve_never_confirms_a_claim`: a
    /// DID that resolves fine to a host that is not answering is somebody
    /// else's outage, and stays distinct from one this deployment cannot
    /// resolve at all.
    #[tokio::test]
    async fn a_closed_port_is_unreachable_rather_than_unresolvable() {
        // Bind and immediately drop, freeing the port with nothing listening.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("local addr");
        drop(listener);

        let operator_did = format!("did:web:{}%3A{}", addr.ip(), addr.port());
        let client = didbot_http::client();
        let result = fetch_claim(
            &client,
            &operator_did,
            "pds.example.test",
            PrivateAddresses::Allowed,
        )
        .await;
        assert!(matches!(result, Err(ClaimFetchError::Unreachable(_))));
    }
    // ------------------------------------------------------------------
    // The nudge, and the bound that stops it amplifying
    // ------------------------------------------------------------------

    /// The property the whole nudge design exists to hold: a burst of
    /// "please look now" costs this server *one* extra outbound request, not
    /// one per nudge. The bound is on the outbound rate, so it does not
    /// matter how many arrive or how fast.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_burst_of_nudges_produces_at_most_one_extra_poll() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let hits = Arc::new(AtomicUsize::new(0));
        let (addr, server, ca) = fake_operator_pds_counting(HashMap::new(), hits.clone()).await;
        let operator = format!("did:web:{}%3A{}", addr.ip(), addr.port());

        let relation = Arc::new(OperatorRelation::new(
            operator,
            "pds.example.test",
            "did:web:pds.example.test",
            time::Duration::hours(6),
        ));
        let nudge = Arc::new(PollNudge::new());
        let poll = OperatorPoll::spawn(
            relation,
            Arc::new(Estop::default()),
            Arc::new(ServerLifecycle::unclaimed()),
            Arc::new(crate::tests::FakeRegistry::new()) as Arc<dyn Registry>,
            client_trusting(&ca),
            PrivateAddresses::Allowed,
            PollSchedule {
                // Long enough that the timer never fires a second time
                // inside this test, so every poll after the immediate first
                // one is attributable to a nudge.
                interval: StdDuration::from_secs(3600),
                // Wide enough that a poll -- a TLS handshake and two
                // requests -- finishes inside the window each assertion
                // below is about, on a machine doing other things.
                floor: StdDuration::from_millis(360),
            },
            nudge.clone(),
            None,
            Arc::new(OperatorRecords::new()),
            None,
        );

        // Let the immediate first tick land.
        tokio::time::sleep(StdDuration::from_millis(180)).await;
        let after_first = hits.load(Ordering::SeqCst);
        assert_eq!(after_first, 1, "the first tick should fire immediately");

        // Now shout. Every one of these lands inside the floor window the
        // first poll opened.
        for _ in 0..200 {
            nudge.nudge();
        }
        tokio::time::sleep(StdDuration::from_millis(120)).await;
        assert_eq!(
            hits.load(Ordering::SeqCst),
            1,
            "a nudge inside the floor window must wait, not poll"
        );

        // And when the floor expires, the whole burst collapses into one.
        tokio::time::sleep(StdDuration::from_millis(600)).await;
        assert_eq!(
            hits.load(Ordering::SeqCst),
            2,
            "200 nudges must cost one extra outbound request, not 200"
        );

        poll.stop().await;
        server.abort();
    }

    /// And the nudge is not decorative: it does bring a poll forward, which
    /// is the whole reason to accept one.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_nudge_brings_the_next_poll_forward() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let hits = Arc::new(AtomicUsize::new(0));
        let (addr, server, ca) = fake_operator_pds_counting(HashMap::new(), hits.clone()).await;
        let operator = format!("did:web:{}%3A{}", addr.ip(), addr.port());

        let relation = Arc::new(OperatorRelation::new(
            operator,
            "pds.example.test",
            "did:web:pds.example.test",
            time::Duration::hours(6),
        ));
        let nudge = Arc::new(PollNudge::new());
        let poll = OperatorPoll::spawn(
            relation,
            Arc::new(Estop::default()),
            Arc::new(ServerLifecycle::unclaimed()),
            Arc::new(crate::tests::FakeRegistry::new()) as Arc<dyn Registry>,
            client_trusting(&ca),
            PrivateAddresses::Allowed,
            PollSchedule {
                interval: StdDuration::from_secs(3600),
                // See the floor in the test above: a poll now opens a TLS
                // connection, and each window has to cover one.
                floor: StdDuration::from_millis(150),
            },
            nudge.clone(),
            None,
            Arc::new(OperatorRecords::new()),
            None,
        );

        tokio::time::sleep(StdDuration::from_millis(300)).await;
        assert_eq!(hits.load(Ordering::SeqCst), 1);
        // Past the floor now, so this one is taken immediately rather than
        // waited out.
        nudge.nudge();
        tokio::time::sleep(StdDuration::from_millis(300)).await;
        assert_eq!(
            hits.load(Ordering::SeqCst),
            2,
            "an hour-long timer plus a nudge should still have polled twice"
        );

        poll.stop().await;
        server.abort();
    }

    /// A handle lookup that never answers holds up neither the claim nor
    /// the next poll: the check runs beside the tick, not inside it.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_handle_lookup_that_never_answers_does_not_hold_the_claim() {
        let mut records = HashMap::new();
        records.insert(
            "pds.example.test".to_owned(),
            json!({
                "subject": "did:web:pds.example.test",
                "createdAt": "2026-01-01T00:00:00Z",
            }),
        );
        let (addr, server, ca) = fake_operator_pds(records).await;
        let relation = Arc::new(OperatorRelation::new(
            format!("did:web:{}%3A{}", addr.ip(), addr.port()),
            "pds.example.test",
            "did:web:pds.example.test",
            time::Duration::hours(6),
        ));
        let lifecycle = Arc::new(ServerLifecycle::unclaimed());
        let nudge = Arc::new(PollNudge::new());
        let poll = OperatorPoll::spawn(
            relation.clone(),
            Arc::new(Estop::default()),
            lifecycle.clone(),
            Arc::new(crate::tests::FakeRegistry::new()) as Arc<dyn Registry>,
            client_trusting(&ca),
            PrivateAddresses::Allowed,
            PollSchedule {
                interval: StdDuration::from_secs(3600),
                floor: StdDuration::from_millis(150),
            },
            nudge.clone(),
            None,
            Arc::new(OperatorRecords::new()),
            Some(Arc::new(OperatorHandle::new(
                crate::operator_handle::tests::Stalls,
            ))),
        );
        tokio::time::sleep(StdDuration::from_millis(300)).await;
        assert!(relation.is_claimed(), "the first tick found the claim");
        assert_eq!(lifecycle.state(), didbot_pds::ServerState::Claimed);

        // Well inside `LOOKUP_TIMEOUT`, with the first lookup still stalled.
        let started = tokio::time::Instant::now();
        nudge.nudge_and_wait(StdDuration::from_secs(3)).await;
        assert!(
            started.elapsed() < StdDuration::from_secs(2),
            "the next poll waited on the lookup: {:?}",
            started.elapsed()
        );
        assert!(relation.is_claimed());

        poll.stop().await;
        server.abort();
    }

    /// A caller waiting on a nudge hears back when the poll it asked for
    /// finishes, not when its own bound runs out; and with no poll running
    /// it hears back at once.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_wait_on_a_nudge_ends_with_the_poll_it_asked_for() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let idle = PollNudge::new();
        let started = tokio::time::Instant::now();
        assert!(!idle.nudge_and_wait(StdDuration::from_secs(5)).await);
        assert!(started.elapsed() < StdDuration::from_secs(1));

        let hits = Arc::new(AtomicUsize::new(0));
        let (addr, server, ca) = fake_operator_pds_counting(HashMap::new(), hits.clone()).await;
        let relation = Arc::new(OperatorRelation::new(
            format!("did:web:{}%3A{}", addr.ip(), addr.port()),
            "pds.example.test",
            "did:web:pds.example.test",
            time::Duration::hours(6),
        ));
        let nudge = Arc::new(PollNudge::new());
        let poll = OperatorPoll::spawn(
            relation,
            Arc::new(Estop::default()),
            Arc::new(ServerLifecycle::unclaimed()),
            Arc::new(crate::tests::FakeRegistry::new()) as Arc<dyn Registry>,
            client_trusting(&ca),
            PrivateAddresses::Allowed,
            PollSchedule {
                interval: StdDuration::from_secs(3600),
                floor: StdDuration::from_millis(150),
            },
            nudge.clone(),
            None,
            Arc::new(OperatorRecords::new()),
            None,
        );
        tokio::time::sleep(StdDuration::from_millis(300)).await;
        assert_eq!(hits.load(Ordering::SeqCst), 1);

        let started = tokio::time::Instant::now();
        // This fake lists no records, so the poll reads none.
        assert!(!nudge.nudge_and_wait(StdDuration::from_secs(5)).await);
        assert!(
            started.elapsed() < StdDuration::from_secs(4),
            "the wait ends with the poll, not the bound: {:?}",
            started.elapsed()
        );
        assert_eq!(hits.load(Ordering::SeqCst), 2);

        poll.stop().await;
        server.abort();
    }
}