didbot_serve/routes/

sync.rs

use super::*;

/// The slot a never-ending stream holds for as long as it is open, or `None`
/// when this deployment is already holding as many as it will.
///
/// Keyed on the caller the per-address rate limiters are keyed on, so one
/// caller cannot spend the whole total by itself; see
/// [`crate::subscribers::SubscriberLimit`].
pub(crate) fn subscriber_slot(
    state: &AppState,
    headers: &HeaderMap,
    peer: Option<std::net::IpAddr>,
    route: &str,
) -> Option<crate::subscribers::SubscriberSlot> {
    let key = caller_key(headers, peer, state.trust_forwarded_headers);
    let slot = state.subscriber_limit.take(&key);
    if slot.is_none() {
        info!(
            route,
            held = state.subscriber_limit.held(),
            "refusing a subscription: this server is already holding as many streams as it will"
        );
    }
    slot
}

/// What a caller past the subscriber bound is told.
///
/// A retryable refusal rather than a silent hold: a consumer that cannot
/// tell a full server from a quiet one backs off on neither.
pub(crate) fn too_many_subscribers(route: &str) -> Response {
    ApiError::rate_limited(format!(
        "this server is already holding as many streams as it will; retry {route} shortly"
    ))
    .into_response()
}

/// `GET /xrpc/com.atproto.sync.getRepo`
///
/// The whole repository as a CARv1 file: a signed commit over a Merkle search
/// tree over every record. This is the only answer this server gives that a
/// reader does not have to take on trust — resolve the DID, take the key out
/// of the document, check the signature.
///
/// `since` is a revision the caller last synced at, and the answer is then the
/// blocks it does not already hold — the same commit at the root either way.
/// A revision this deployment cannot place, because it never minted it or has
/// since trimmed it out of that repository's trail, is answered with the whole
/// repository rather than with an error: the caller asked what it is missing,
/// and all of it is true. See `didbot_pds::history`.
///
/// It is answered with the whole repository *and said so*, in [`SINCE`]. A
/// full export and a delta are both a CAR rooted at the current commit, so a
/// caller comparing bytes cannot tell one from the other, and one that cannot
/// tell cannot notice it is paying for a full export on every reconnect. The
/// lexicon declares no error for this and its output encoding is a CAR, so a
/// header is the only place the fact fits without this server answering
/// something `com.atproto.sync.getRepo` does not define.
pub(crate) async fn get_repo(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<GetRepoQuery>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    let export = match sync_read(&state, "com.atproto.sync.getRepo", move |registry| {
        registry.export_repo(&query.did, query.since.as_deref())
    })
    .await
    {
        Ok(export) => export,
        Err(response) => return response,
    };
    match since_header(&export.delta) {
        Some(value) => ([CAR, (SINCE, value)], export.car).into_response(),
        None => ([CAR], export.car).into_response(),
    }
}

/// What a `getRepo` answer says about the `since` it was asked for.
///
/// Absent when nothing was asked, `Diff` when the delta was answered, and
/// `OutdatedCursor` when it could not be and the whole repository was sent
/// instead. The two values are the firehose's own vocabulary — see
/// `didbot_serve::subscribe::OUTDATED_CURSOR` — because a cursor this server
/// can no longer place is the same condition on either endpoint, and a
/// consumer that already switches on one name should not have to learn a
/// second.
pub(crate) const SINCE: header::HeaderName = header::HeaderName::from_static("x-didbot-since");

/// The [`SINCE`] value for one answer, or `None` when no `since` was asked.
pub(crate) fn since_header(delta: &Delta) -> Option<&'static str> {
    match delta {
        Delta::Whole => None,
        Delta::Since(_) => Some("Diff"),
        Delta::OutdatedCursor(_) => Some(crate::subscribe::OUTDATED_CURSOR),
    }
}

/// The content type a repository, a proof or a bag of blocks goes out as.
///
/// A reader that dispatches on it — `goat repo ls` does not, but a browser
/// does — must not be told this is JSON.
pub(crate) const CAR: (header::HeaderName, &str) =
    (header::CONTENT_TYPE, "application/vnd.ipld.car");

/// Maps a failure onto the error name `com.atproto.sync.*` declares for it.
///
/// The names in those lexicons are the ones a generated client switches on,
/// and they are not this server's own: a repository nobody minted is
/// `RepoNotFound` here even though every other route calls the same condition
/// `UnknownAccount`. The status codes stay this server's — 404 for a
/// repository that is not there, 400 for a CID that names nothing — because
/// the name is the normative half and a status that says the truth is better
/// than one chosen to match a framework's default.
pub(crate) fn sync_error(state: &AppState, err: &ProvisionError) -> ApiError {
    match err {
        ProvisionError::UnknownAccount { did } => repo_not_served(state, did),
        ProvisionError::BlockNotFound { .. } => {
            ApiError::new(StatusCode::BAD_REQUEST, "BlockNotFound", err.to_string())
        }
        ProvisionError::Blob(_) => crate::blobs::blob_error(err),
        other => ApiError::from(other),
    }
}

/// Why a `com.atproto.sync.*` route serves no repository for `did`.
///
/// Read through [`didbot_pds::HostedAccount::sync_status`], so a route names
/// only a status `subscribeRepos` already announced. `deleted` has no error
/// name in these lexicons, and a reserved or provisioning account has no
/// announced status, so both are `RepoNotFound`. A suspended or deactivated
/// repository is a 403: it is here, and withheld.
pub(crate) fn repo_not_served(state: &AppState, did: &str) -> ApiError {
    let status = state
        .registry
        .account(did)
        .and_then(|account| account.sync_status())
        .and_then(|wire| wire.status);
    match status {
        Some(didbot_pds::AccountStatus::Suspended) => ApiError::new(
            StatusCode::FORBIDDEN,
            "RepoSuspended",
            format!("the repository for `{did}` is suspended"),
        ),
        Some(didbot_pds::AccountStatus::Deactivated) => ApiError::new(
            StatusCode::FORBIDDEN,
            "RepoDeactivated",
            format!("the repository for `{did}` is deactivated"),
        ),
        Some(didbot_pds::AccountStatus::Deleted) | None => ApiError::new(
            StatusCode::NOT_FOUND,
            "RepoNotFound",
            format!("no repository here for `{did}`"),
        ),
    }
}

/// `GET /xrpc/com.atproto.sync.getLatestCommit`
///
/// The commit CID and revision the repository is currently at, which is what
/// a consumer compares its own position against. It costs what an export
/// costs, because a derived repository has to be rebuilt to be named;
/// what it saves the caller is the blocks.
pub(crate) async fn get_latest_commit(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<DidRequest>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    match sync_read(
        &state,
        "com.atproto.sync.getLatestCommit",
        move |registry| registry.latest_commit(&query.did),
    )
    .await
    {
        Ok(head) => Json(json!({ "cid": head.head, "rev": head.rev })).into_response(),
        Err(response) => response,
    }
}

/// `GET /xrpc/com.atproto.sync.getRepoStatus`
///
/// What `com.atproto.sync.subscribeRepos`'s `#account` says about one
/// account, on request, with the `rev` a caller compares a relay's copy
/// against. The `active` and `status` pair is read through
/// [`didbot_pds::HostedAccount::sync_status`], the mapping the stream
/// announces from, so this route and the next frame say the same thing. A
/// relay holding an older frame is brought level by the announcement a boot
/// makes of every position that changed — see
/// [`didbot_pds::Provisioner::announce_accounts`]. An
/// account that mapping
/// announces nothing for — reserved, or still provisioning — is
/// `RepoNotFound` here for the same reason it is silent there: a relay was
/// never told it exists, and this route must not be the first to say so.
pub(crate) async fn get_repo_status(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<DidRequest>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    let unknown = || ProvisionError::UnknownAccount {
        did: query.did.clone(),
    };
    let Some(account) = state.registry.account(&query.did) else {
        return sync_error(&state, &unknown()).into_response();
    };
    let Some(status) = account.sync_status() else {
        return sync_error(&state, &unknown()).into_response();
    };
    let mut body = json!({ "did": account.did.as_str(), "active": status.active });
    if let Some(reason) = status.status {
        body["status"] = json!(reason.as_str());
    }
    if status.active {
        let did = account.did.as_str().to_owned();
        match sync_read(&state, "com.atproto.sync.getRepoStatus", move |registry| {
            registry.latest_commit(&did)
        })
        .await
        {
            Ok(head) => body["rev"] = json!(head.rev),
            Err(response) => return response,
        }
    }
    debug!(did = %account.did, active = status.active, status = ?status.status, "reported a repository's status");
    Json(body).into_response()
}

/// `GET /xrpc/com.atproto.sync.getBlocks`
///
/// Named blocks and nothing else, in a CAR file that declares no root:
/// the caller asked for blocks by name rather than for something to walk, so
/// there is nothing to start walking from. A CID this repository does not
/// hold refuses the whole request rather than returning a short file, because
/// a caller cannot tell a block that was left out from one it never asked for.
pub(crate) async fn get_blocks(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    RawQuery(query): RawQuery,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let query = query.unwrap_or_default();
    let Some(did) = repeated(&query, "did").into_iter().next() else {
        return ApiError::bad_request("getBlocks needs a `did`").into_response();
    };
    let cids = repeated(&query, "cids");
    if cids.is_empty() {
        return ApiError::bad_request("getBlocks needs at least one `cids`").into_response();
    }
    match sync_read(&state, "com.atproto.sync.getBlocks", move |registry| {
        registry.get_blocks(&did, &cids)
    })
    .await
    {
        Ok(car) => ([CAR], car).into_response(),
        Err(response) => response,
    }
}

/// `GET /xrpc/com.atproto.sync.getRecord`
///
/// The blocks that prove what this repository says about one record: the
/// signed commit, the Merkle search tree nodes from its root down to where
/// the record's key sits, and the record itself. A verifier checks the
/// signature with the key in the account's DID document, then walks the nodes
/// by hashing each one and following the link the previous one named.
///
/// # A record that is not there is a 200
///
/// The lexicon declares `RecordNotFound` and this never returns it. The
/// method's own description is that it proves "the existence or non-existence
/// of a record", and an error body cannot prove anything: the walk that ends
/// at the node the key would have to be in, and does not find it, is the
/// proof that it is not in the repository. So both answers are the same CAR
/// file with the same root, and the difference between them is whether the
/// last node names the key. That is what the reference implementation does
/// too.
pub(crate) async fn sync_get_record(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<SyncGetRecordQuery>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    match sync_read(&state, "com.atproto.sync.getRecord", move |registry| {
        registry.record_proof(&query.did, &query.collection, &query.rkey)
    })
    .await
    {
        Ok(car) => ([CAR], car).into_response(),
        Err(response) => response,
    }
}

/// `GET /xrpc/com.atproto.sync.listRepos`
///
/// Every repository this deployment hosts, in DID order, with the commit each
/// one is at. The cursor is the last DID of the page: a position in a total
/// order rather than an offset into a list that moves under the caller.
pub(crate) async fn list_repos(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<ListReposQuery>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    // This route names every account this deployment holds, so it is the
    // same decision `bot.did.listAccounts` is; see `auth::Disclosure`.
    if let Err(err) =
        auth::require_disclosure(state.disclosure.list_accounts, "com.atproto.sync.listRepos")
    {
        return err.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    let (repos, cursor) = match sync_read(&state, "com.atproto.sync.listRepos", move |registry| {
        registry.list_repos(query.effective_limit(), query.cursor.as_deref())
    })
    .await
    {
        Ok(page) => page,
        Err(response) => return response,
    };
    match cursor {
        Some(cursor) => Json(json!({ "repos": repos, "cursor": cursor })).into_response(),
        None => Json(json!({ "repos": repos })).into_response(),
    }
}

/// `GET /xrpc/com.atproto.sync.subscribeRepos`
///
/// `Credential::Open`: the lexicon says "public and does not require auth",
/// and a relay that had to authenticate could not crawl this server.
///
/// A WebSocket of framed DAG-CBOR, one `#commit` per commit this server makes,
/// resumable from a `cursor`. This is the method a relay crawls — see
/// [`crate::subscribe`].
///
/// The cursor is read here and the connection is opened either way: an error
/// the lexicon declares is a frame on the socket followed by a close, not an
/// HTTP status, because a client that asked for an upgrade is not reading for
/// one.
pub(crate) async fn subscribe_repos(
    State(state): State<AppState>,
    peer: Peer,
    headers: HeaderMap,
    upgrade: WebSocketUpgrade,
    Query(params): Query<SubscribeParams>,
) -> Response {
    // Refused *before* the upgrade, deliberately, and this is the choice
    // `plan/onboarding.md` asks to be made explicitly rather than fallen
    // into. The alternative -- accept the upgrade and hold the connection
    // open with no frames on it -- is indistinguishable, from a relay's
    // side, from a healthy server with nothing to say: relays back off on a
    // failed connect and do not back off on silence, so holding would
    // produce a subscription that looks live, indexes nothing, and gives
    // its operator no signal at all. Refusing pre-upgrade keeps the answer
    // on the HTTP exchange, where a status code and a `Retry-After` can
    // carry it.
    if let Err(err) = require_lifecycle(
        &state,
        |p| p.serves_reads,
        "com.atproto.sync.subscribeRepos",
    ) {
        return err.into_response();
    }
    // Refused pre-upgrade for the same reason, and under the same decision
    // `bot.did.listAccounts` is: this stream announces each account as it
    // appears and as it writes. See `auth::Disclosure`.
    if let Err(err) = auth::require_disclosure(
        state.disclosure.list_accounts,
        "com.atproto.sync.subscribeRepos",
    ) {
        return err.into_response();
    }
    let Some(slot) = subscriber_slot(&state, &headers, peer.0, "com.atproto.sync.subscribeRepos")
    else {
        return too_many_subscribers("com.atproto.sync.subscribeRepos");
    };
    let cursor = params.cursor();
    let exclude = params.exclusions();
    let upgrade = upgrade
        .max_message_size(MAX_CONSUMER_MESSAGE)
        .max_frame_size(MAX_CONSUMER_MESSAGE);
    // The slot rides into the pump, so it is given back when the socket
    // closes rather than when this handler returns.
    upgrade.on_upgrade(move |socket| pump_repos(socket, state.repos, cursor, exclude, slot))
}

/// The largest message, and the largest frame, a `subscribeRepos` consumer
/// may send.
///
/// A consumer sends nothing this server reads: the incoming half is polled
/// only to see the close. WebSocket control frames carry at most 125 bytes
/// (RFC 6455 §5.5), and four kibibytes is thirty times that. axum's own
/// defaults, 64 MiB a message and 16 MiB a frame, are what one consumer could
/// otherwise make this server buffer on each slot [`subscriber_slot`] gives
/// it. A consumer that sends more is disconnected.
pub(crate) const MAX_CONSUMER_MESSAGE: usize = 4 * 1024;

/// What a `subscribeRepos` connection asked for.
#[derive(Debug, Default, serde::Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub(crate) struct SubscribeParams {
    /// The last sequence number the consumer applied.
    ///
    /// A string rather than an integer so that a cursor which is not one is
    /// read as no cursor at all rather than rejected before the upgrade. It
    /// reaches this server as a query parameter a human may well have typed,
    /// and answering a WebSocket handshake with a `400` would leave them with
    /// an HTTP status where they wanted a stream.
    cursor: Option<String>,
    /// Drop `#commit` frames on this connection.
    exclude_commit: Option<String>,
    /// Drop `#identity` frames on this connection.
    exclude_identity: Option<String>,
    /// Drop `#account` frames on this connection.
    exclude_account: Option<String>,
}

impl SubscribeParams {
    /// The cursor as the lexicon's integer, or `None` if there is not one.
    fn cursor(&self) -> Option<i64> {
        let cursor = self
            .cursor
            .as_deref()
            .map(str::trim)
            .filter(|c| !c.is_empty())?;
        match cursor.parse::<i64>() {
            Ok(seq) => Some(seq),
            Err(_) => {
                info!(
                    cursor,
                    "a subscribeRepos cursor was not an integer; streaming from now"
                );
                None
            }
        }
    }

    /// Which message types this connection asked not to be sent.
    fn exclusions(&self) -> crate::subscribe::Exclusions {
        crate::subscribe::Exclusions {
            commit: flag(self.exclude_commit.as_deref()),
            identity: flag(self.exclude_identity.as_deref()),
            account: flag(self.exclude_account.as_deref()),
        }
    }
}

/// Reads a boolean query parameter.
///
/// Present-and-not-false is true, so `?excludeIdentity` on its own works the
/// way somebody typing it expects. Anything unrecognisable reads as false,
/// which is the safe direction for a delivery filter: the consumer is sent
/// more than it asked for rather than less.
pub(crate) fn flag(value: Option<&str>) -> bool {
    match value.map(str::trim) {
        None => false,
        Some("false" | "0") => false,
        Some(_) => true,
    }
}

/// Writes one consumer's frames onto its socket until either end stops.
///
/// Reads are drained and discarded: this stream carries nothing from the
/// consumer, but a socket whose incoming half is never polled never sees the
/// close frame that says the consumer has gone, so the task would outlive it.
///
/// The end of a connection that began is written at `info`, with the cursor
/// it asked for, why it ended, how long it lasted and how many frames it was
/// sent. A consumer closed for falling behind is written at `warn`.
pub(crate) async fn pump_repos(
    socket: WebSocket,
    repos: Repos,
    cursor: Option<i64>,
    exclude: crate::subscribe::Exclusions,
    // Held, not read: dropped when this function returns, which is what
    // gives the subscription's slot back. See `subscriber_slot`.
    _slot: crate::subscribers::SubscriberSlot,
) {
    let mut stream = match repos.subscribe(cursor, exclude) {
        Ok(stream) => stream,
        Err(frame) => {
            let mut socket = socket;
            let _ = socket.send(Message::Binary(frame.into())).await;
            // `Message::Close` rather than `SinkExt::close`, so this crate
            // does not take a dependency on `futures-util` for one method.
            let _ = socket.send(Message::Close(None)).await;
            return;
        }
    };
    let mut socket = socket;
    let began = std::time::Instant::now();
    let mut sent: u64 = 0;
    let mut stopping = false;
    let reason = loop {
        tokio::select! {
            // Nothing is expected on this half, and it is polled anyway: a
            // socket whose incoming frames are never read never sees the
            // close that says the consumer has gone, so this task would
            // outlive it.
            incoming = socket.recv() => match incoming {
                None => break "the consumer hung up".to_owned(),
                // A message past `MAX_CONSUMER_MESSAGE` lands here, as does a
                // broken socket.
                Some(Err(error)) => break format!("the consumer's socket failed: {error}"),
                Some(Ok(_)) => {}
            },
            frame = stream.next() => {
                let Some(bytes) = frame else {
                    break if stream.lagged() > 0 {
                        "the consumer fell further behind than the server buffers".to_owned()
                    } else {
                        stopping = true;
                        "the server is closing the stream".to_owned()
                    };
                };
                if let Err(error) = socket.send(Message::Binary(bytes)).await {
                    break format!("a frame could not be sent: {error}");
                }
                sent += 1;
            }
        }
    };
    let lasted_secs = began.elapsed().as_secs();
    if stream.lagged() > 0 {
        warn!(
            ?cursor,
            reason,
            lasted_secs,
            sent,
            lagged = stream.lagged(),
            "subscribeRepos consumer disconnected"
        );
    } else {
        info!(
            ?cursor,
            reason, lasted_secs, sent, "subscribeRepos consumer disconnected"
        );
    }
    if stream.lagged() > 0 {
        // The one place this server closes a connection on purpose. The
        // lexicon names the case and the consumer has to know it has a gap;
        // see `ReposStream::next`.
        let _ = socket
            .send(Message::Binary(
                crate::subscribe::encode_error(
                    "ConsumerTooSlow",
                    "this connection fell further behind than the server buffers; \
                     reconnect with the last cursor you applied",
                )
                .into(),
            ))
            .await;
    }
    // A stop says so with 1001, "going away", rather than a bare close, so a
    // relay reads it as the server leaving and not as an error in its own
    // connection. It reconnects with the cursor it holds.
    let close = stopping.then(|| CloseFrame {
        code: close_code::AWAY,
        reason: "the server is stopping".into(),
    });
    let _ = socket.send(Message::Close(close)).await;
}