didbot_serve/

relay.rs

//! `com.atproto.sync.requestCrawl` and `notifyOfUpdate`, called outward.
//!
//! These are the two `com.atproto.*` methods this crate is a *client* of
//! rather than a server for. A relay discovers a new personal data server by
//! subscribing to its `subscribeRepos` — but nothing makes a relay dial an
//! address it has never heard of. `requestCrawl` is the announcement: "I
//! exist, my hostname is this, please open a subscription." Without ever
//! sending it, `subscribeRepos` is a stream nobody is listening to, and the
//! symptom is silence — no error, no refused write, just an account that
//! never appears anywhere a relay indexes. `notifyOfUpdate` is the same
//! request against a different, deprecated NSID — upstream's own schema
//! says to use `requestCrawl` instead — kept here because a relay that
//! dropped a subscription mid-stream is the one case its description names
//! specifically, and an operator recovering from that gets to call the
//! method upstream still names for it.
//!
//! # When either fires
//!
//! Neither fires on its own. Both are operator-triggered: naming a relay
//! does not announce anything by itself, so an operator asks explicitly
//! rather than a restart asking for them. There is no startup announce and
//! no timer. This module itself has no opinion on when it is called; it
//! only answers what happened when it was.
//!
//! # What a failure does
//!
//! Nothing to serving, because nothing on the request-serving path calls
//! into this module at all. [`RelayClient::request_crawl`] and
//! [`RelayClient::notify_of_update`] return a typed [`RelayCrawlError`] to
//! their caller, distinguishing a relay that refused (`Refused`, including
//! `HostBanned`) from one that could not be reached at all
//! (`Unreachable`). There is no log-only failure path here: the operator
//! who asked is the channel.
//!
//! # Which relays, and localhost
//!
//! Configurable — [`RelaySection`](didbot_config::sections::RelaySection) in
//! `didbot-config`, or `--relay-hostname` on `didbot-pds` directly, once per
//! relay — never hardcoded, because this project vendors nobody's relay address as a
//! default. A development run against `*.localhost` has no relay reachable
//! from it and no hostname a relay could ever resolve back to this machine,
//! so `didbot-pds.rs` never wires `ANNOUNCE` up for a loopback zone even when
//! a relay is configured; see that binary's own call site.

use std::time::Duration;

use serde::Serialize;
use serde_json::Value;

/// `com.atproto.sync.requestCrawl`'s NSID, for the URL this client builds
/// and for anything that wants to name the method in a log line.
pub const REQUEST_CRAWL_NSID: &str = "com.atproto.sync.requestCrawl";

/// `com.atproto.sync.notifyOfUpdate`'s NSID. Deprecated upstream in favor of
/// `requestCrawl` — same request shape, same effect, different name — kept
/// here for the one case its own description names: recovering from a gap
/// after a relay's subscription dropped.
pub const NOTIFY_OF_UPDATE_NSID: &str = "com.atproto.sync.notifyOfUpdate";

/// How long [`RelayClient::request_crawl`] and
/// [`RelayClient::notify_of_update`] wait before giving up.
///
/// A relay this server cannot reach must not hang the operator command that
/// calls either forever; a generous bound rather than a tight one, since an
/// operator invoking this waits on it interactively and nothing here is
/// latency-sensitive.
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);

/// How much of a non-success response body [`RelayCrawlError::Refused`]
/// keeps: a relay answering with a large page must not flood the log this
/// failure is reported to.
///
/// How much is *read* to get it is
/// [`didbot_http::MAX_JSON_BODY`], the bound this
/// workspace holds any body from a server it does not operate to. Past it the
/// message says the body went unread, which a relay answering an endless
/// error page earns.
const MAX_ERROR_BODY: usize = 512;

/// Body of both `requestCrawl`'s and `notifyOfUpdate`'s request — the two
/// lexicons declare the identical `{"hostname": ..}` shape.
#[derive(Debug, Clone, Serialize)]
struct RequestCrawlBody {
    hostname: String,
}

/// The JSON body a `requestCrawl` or `notifyOfUpdate` call sends, built in
/// isolation from any network concern so a test can check it against either
/// vendored lexicon's `input` schema without standing up a server.
pub fn request_crawl_body(hostname: &str) -> Value {
    serde_json::to_value(RequestCrawlBody {
        hostname: hostname.to_owned(),
    })
    .expect("a struct of one string always serializes")
}

/// Why a relay did not end up crawling this server.
///
/// Both variants are returned to the caller rather than swallowed — see the
/// module documentation for who that caller is and what it does with one.
#[derive(Debug, thiserror::Error)]
pub enum RelayCrawlError {
    /// The request never completed: nothing listening, DNS, TLS, or the
    /// timeout.
    #[error("cannot reach relay {relay} at {url}: {source}")]
    Unreachable {
        /// The relay's configured base URL.
        relay: String,
        /// The full URL that was tried.
        url: String,
        /// The transport failure `reqwest` reported.
        #[source]
        source: reqwest::Error,
    },
    /// The relay answered, but not with success — including `HostBanned`,
    /// the one error `requestCrawl`'s own document declares.
    #[error("relay {relay} refused the crawl request: {status} {body}")]
    Refused {
        /// The relay's configured base URL.
        relay: String,
        /// The HTTP status the relay answered with.
        status: u16,
        /// The response body, truncated to `MAX_ERROR_BODY`.
        body: String,
    },
}

/// Calls `requestCrawl` or `notifyOfUpdate` against one relay.
///
/// A thin wrapper around [`reqwest::Client`] rather than a trait: the seam a
/// test needs is which URL this client talks to, not a second implementation
/// of what it does with the response, so [`RelayClient::new`] takes the base
/// URL directly and a test points it at a throwaway local server instead of
/// a real relay. See this module's own tests.
#[derive(Debug, Clone)]
pub struct RelayClient {
    http: reqwest::Client,
    /// The relay's base URL, without a trailing slash — `https://bsky.network`,
    /// or `http://127.0.0.1:PORT` in a test.
    base_url: String,
}

/// The relays an operator's announce reaches, and the hostname it announces.
///
/// Built once at startup from `--relay-hostname`/`[relay]` and held rather
/// than acted on: naming a relay does not announce anything by itself.
#[derive(Debug, Clone)]
pub struct RelayTarget {
    /// One client per relay, in the order they were named.
    pub relays: Vec<RelayClient>,
    /// This server's own hostname — what gets announced, not the relay's.
    pub hostname: String,
}

impl RelayTarget {
    /// Announces [`Self::hostname`] to every relay at once, over
    /// `requestCrawl`, or `notifyOfUpdate` when `resume` is set.
    ///
    /// Answers one outcome per relay, each beside that relay's base URL, in
    /// the order the relays were named. One relay failing does not stop the
    /// others being asked.
    pub async fn announce(&self, resume: bool) -> Vec<(String, Result<(), RelayCrawlError>)> {
        let mut calls = tokio::task::JoinSet::new();
        for (index, client) in self.relays.iter().cloned().enumerate() {
            let hostname = self.hostname.clone();
            calls.spawn(async move {
                let outcome = if resume {
                    client.notify_of_update(&hostname).await
                } else {
                    client.request_crawl(&hostname).await
                };
                (index, client.base_url, outcome)
            });
        }
        let mut outcomes = calls.join_all().await;
        outcomes.sort_by_key(|(index, _, _)| *index);
        outcomes
            .into_iter()
            .map(|(_, relay, outcome)| (relay, outcome))
            .collect()
    }
}

impl RelayClient {
    /// Builds a client pointed at `base_url`.
    ///
    /// Falls back to a default-configured [`reqwest::Client`] if one cannot
    /// be built with the timeout applied: a one-shot announcement degrading
    /// is preferable to a startup path that refuses to construct.
    pub fn new(base_url: impl Into<String>) -> Self {
        let http = didbot_http::builder()
            .timeout(REQUEST_TIMEOUT)
            .build()
            .unwrap_or_default();
        Self {
            http,
            base_url: base_url.into(),
        }
    }

    /// Asks this client's relay to crawl `hostname`, over `requestCrawl`.
    pub async fn request_crawl(&self, hostname: &str) -> Result<(), RelayCrawlError> {
        self.call(REQUEST_CRAWL_NSID, hostname).await
    }

    /// Asks this client's relay to resume crawling `hostname`, over the
    /// deprecated `notifyOfUpdate` — the method upstream's own schema names
    /// for recovering from a gap after a dropped subscription; see this
    /// module's doc.
    pub async fn notify_of_update(&self, hostname: &str) -> Result<(), RelayCrawlError> {
        self.call(NOTIFY_OF_UPDATE_NSID, hostname).await
    }

    /// Shared body of [`Self::request_crawl`] and [`Self::notify_of_update`]:
    /// both lexicons take the identical `{"hostname": ..}` input and report
    /// success or failure identically, so only the NSID in the URL differs.
    async fn call(&self, nsid: &str, hostname: &str) -> Result<(), RelayCrawlError> {
        let url = format!("{}/xrpc/{nsid}", self.base_url);
        let body = request_crawl_body(hostname);
        let response = self
            .http
            .post(&url)
            .json(&body)
            .send()
            .await
            .map_err(|source| RelayCrawlError::Unreachable {
                relay: self.base_url.clone(),
                url: url.clone(),
                source,
            })?;
        let status = response.status();
        if !status.is_success() {
            // Bounded on the way in, then truncated. The truncation is what
            // keeps the log readable; the bound is what decides how much of a
            // relay's answer this process ever holds, and a relay is a server
            // nobody here operates.
            let body = match didbot_http::read_bounded(response, didbot_http::MAX_JSON_BODY).await {
                Ok(bytes) => String::from_utf8_lossy(&bytes)
                    .chars()
                    .take(MAX_ERROR_BODY)
                    .collect(),
                Err(err) => format!("<unread: {err}>"),
            };
            return Err(RelayCrawlError::Refused {
                relay: self.base_url.clone(),
                status: status.as_u16(),
                body,
            });
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use axum::extract::State;
    use axum::http::StatusCode;
    use axum::routing::post;
    use axum::{Json, Router};
    use tokio::net::TcpListener;

    use super::*;

    /// Every request body the fake relay below received, shared with the
    /// test that started it.
    type Received = std::sync::Arc<std::sync::Mutex<Vec<Value>>>;

    /// A fake relay's fixed state: the response it always answers, and
    /// where it records what it was sent.
    type FakeRelayState = ((StatusCode, Value), Received);

    /// Starts a fake relay on loopback that always answers `outcome`,
    /// recording every request body it received.
    async fn fake_relay(outcome: (StatusCode, Value)) -> (String, Received) {
        let received: Received = Default::default();
        let state: FakeRelayState = (outcome, received.clone());
        let handler = move |State((outcome, received)): State<FakeRelayState>,
                            Json(body): Json<Value>| async move {
            received.lock().expect("poisoned").push(body);
            (outcome.0, Json(outcome.1))
        };
        // Both NSIDs answered by the same handler: `request_crawl` and
        // `notify_of_update` are exercised against one fixture rather than
        // two, since the two lexicons differ only in which path is called.
        let app = Router::new()
            .route(&format!("/xrpc/{REQUEST_CRAWL_NSID}"), post(handler))
            .route(&format!("/xrpc/{NOTIFY_OF_UPDATE_NSID}"), post(handler))
            .with_state(state);
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("loopback binds");
        let addr = listener.local_addr().expect("bound listener has an addr");
        tokio::spawn(async move {
            axum::serve(listener, app).await.expect("fake relay serves");
        });
        (format!("http://{addr}"), received)
    }

    #[test]
    fn the_request_body_is_exactly_what_the_lexicon_requires() {
        let body = request_crawl_body("pds.example.com");
        assert_eq!(body, serde_json::json!({"hostname": "pds.example.com"}));
    }

    #[tokio::test]
    async fn a_relay_that_accepts_is_reported_ok() {
        let (base_url, received) = fake_relay((StatusCode::OK, serde_json::json!({}))).await;
        let client = RelayClient::new(base_url);
        client
            .request_crawl("pds.example.com")
            .await
            .expect("the fake relay answers success");
        assert_eq!(
            received.lock().expect("poisoned").as_slice(),
            [serde_json::json!({"hostname": "pds.example.com"})]
        );
    }

    /// Every named relay is asked, one refusing or unreachable does not stop
    /// the rest, and the outcomes come back in the order the relays were
    /// named.
    #[tokio::test]
    async fn an_announce_asks_every_relay_and_reports_each() {
        let (accepting, accepted) = fake_relay((StatusCode::OK, serde_json::json!({}))).await;
        let (refusing, refused) = fake_relay((
            StatusCode::FORBIDDEN,
            serde_json::json!({"error": "HostBanned"}),
        ))
        .await;
        // Bound and dropped, so nothing listens there.
        let gone = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("loopback binds")
            .local_addr()
            .expect("bound listener has an addr");
        let unreachable = format!("http://{gone}");
        let target = RelayTarget {
            relays: [&refusing, &unreachable, &accepting]
                .into_iter()
                .map(|base_url| RelayClient::new(base_url.clone()))
                .collect(),
            hostname: "pds.example.com".to_owned(),
        };

        let outcomes = target.announce(false).await;

        let relays: Vec<&str> = outcomes.iter().map(|(relay, _)| relay.as_str()).collect();
        assert_eq!(relays, [&refusing, &unreachable, &accepting]);
        assert!(matches!(
            outcomes[0].1,
            Err(RelayCrawlError::Refused { status: 403, .. })
        ));
        assert!(matches!(
            outcomes[1].1,
            Err(RelayCrawlError::Unreachable { .. })
        ));
        assert!(outcomes[2].1.is_ok());
        let body = serde_json::json!({"hostname": "pds.example.com"});
        assert_eq!(
            accepted.lock().expect("poisoned").as_slice(),
            std::slice::from_ref(&body)
        );
        assert_eq!(refused.lock().expect("poisoned").as_slice(), [body]);
    }

    #[tokio::test]
    async fn a_relay_that_refuses_is_reported_and_named() {
        let (base_url, _received) = fake_relay((
            StatusCode::FORBIDDEN,
            serde_json::json!({"error": "HostBanned", "message": "no"}),
        ))
        .await;
        let client = RelayClient::new(base_url);
        let err = client
            .request_crawl("pds.example.com")
            .await
            .expect_err("a 403 is a refusal");
        match err {
            RelayCrawlError::Refused { status, body, .. } => {
                assert_eq!(status, 403);
                assert!(body.contains("HostBanned"), "body was {body}");
            }
            other => panic!("expected Refused, got {other:?}"),
        }
    }

    /// A response body that declares no length and runs well past the bound
    /// a relay's answer is read against.
    struct Oversized {
        remaining: usize,
    }

    impl futures_core::Stream for Oversized {
        type Item = Result<bytes::Bytes, std::convert::Infallible>;

        fn poll_next(
            mut self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Option<Self::Item>> {
            const CHUNK: usize = 64 * 1024;
            if self.remaining == 0 {
                return std::task::Poll::Ready(None);
            }
            let take = self.remaining.min(CHUNK);
            self.remaining -= take;
            std::task::Poll::Ready(Some(Ok(bytes::Bytes::from(vec![b'x'; take]))))
        }
    }

    /// The refusal is still reported, and the page it came with is not held
    /// in memory to report it. A relay is a server nobody here operates, and
    /// the only bound on a `text()` of its error body is how long the request
    /// may take.
    #[tokio::test]
    async fn a_refusal_with_an_outsized_body_is_reported_without_reading_it() {
        let app = Router::new().route(
            &format!("/xrpc/{REQUEST_CRAWL_NSID}"),
            post(|| async {
                (
                    StatusCode::SERVICE_UNAVAILABLE,
                    axum::body::Body::from_stream(Oversized {
                        remaining: didbot_http::MAX_JSON_BODY * 2,
                    }),
                )
            }),
        );
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("loopback binds");
        let addr = listener.local_addr().expect("bound listener has an addr");
        tokio::spawn(async move {
            axum::serve(listener, app).await.expect("fake relay serves");
        });

        let client = RelayClient::new(format!("http://{addr}"));
        let err = client
            .request_crawl("pds.example.com")
            .await
            .expect_err("a 503 is a refusal");
        match err {
            RelayCrawlError::Refused { status, body, .. } => {
                assert_eq!(status, 503);
                assert!(
                    !body.contains("xxxx"),
                    "a body past the bound was read anyway: {body}"
                );
                assert!(body.contains("exceeds"), "body was {body}");
            }
            other => panic!("expected Refused, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn an_unreachable_relay_is_reported_rather_than_hanging_or_panicking() {
        // A bound-then-dropped listener's port has nothing behind it, so the
        // connection is refused immediately rather than this test waiting
        // out the real timeout.
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("loopback binds");
        let addr = listener.local_addr().expect("bound listener has an addr");
        drop(listener);

        let client = RelayClient::new(format!("http://{addr}"));
        let err = client
            .request_crawl("pds.example.com")
            .await
            .expect_err("nothing is listening");
        assert!(matches!(err, RelayCrawlError::Unreachable { .. }));
    }

    /// A relay that accepts the connection and never answers costs a call
    /// exactly [`REQUEST_TIMEOUT`], then `Unreachable`.
    #[tokio::test(start_paused = true)]
    async fn a_relay_that_never_answers_is_unreachable_after_the_request_timeout() {
        let addr = didbot_http::test_support::hung_server();
        let client = RelayClient::new(format!("http://{addr}"));
        let started = tokio::time::Instant::now();
        let err =
            tokio::time::timeout(REQUEST_TIMEOUT * 2, client.request_crawl("pds.example.com"))
                .await
                .expect("the client's own timeout fires first")
                .expect_err("nothing ever answers");
        assert!(
            matches!(&err, RelayCrawlError::Unreachable { source, .. } if source.is_timeout()),
            "{err:?}"
        );
        assert_eq!(started.elapsed(), REQUEST_TIMEOUT);
    }

    /// `notify_of_update` is a thin wrapper differing from `request_crawl`
    /// only in which NSID it posts to — this checks it actually reaches the
    /// deprecated path and not `requestCrawl`'s, and still reports success.
    #[tokio::test]
    async fn notify_of_update_reaches_its_own_nsid_and_is_reported_ok() {
        let (base_url, received) = fake_relay((StatusCode::OK, serde_json::json!({}))).await;
        let client = RelayClient::new(base_url);
        client
            .notify_of_update("pds.example.com")
            .await
            .expect("the fake relay answers success");
        assert_eq!(
            received.lock().expect("poisoned").as_slice(),
            [serde_json::json!({"hostname": "pds.example.com"})]
        );
    }

    /// A relay refusing `notifyOfUpdate` reports the same `Refused` shape
    /// `request_crawl` does — both go through [`RelayClient::call`].
    #[tokio::test]
    async fn notify_of_update_refusal_is_reported_and_named() {
        let (base_url, _received) = fake_relay((
            StatusCode::FORBIDDEN,
            serde_json::json!({"error": "HostBanned", "message": "no"}),
        ))
        .await;
        let client = RelayClient::new(base_url);
        let err = client
            .notify_of_update("pds.example.com")
            .await
            .expect_err("a 403 is a refusal");
        assert!(matches!(err, RelayCrawlError::Refused { status: 403, .. }));
    }
}