didbot_serve/

lib.rs

//! HTTP surface and the development server.
//!
//! This crate is the outside of the system: it turns provisioning into
//! [`axum`] routes, and turns what the provisioner does into a log a developer
//! can read while it happens.
//!
//! The whole surface is built around one object-safe trait,
//! [`didbot_pds::Registry`]. The provisioner is generic over its DNS provider and its
//! account store; the HTTP layer must not be, or every handler signature
//! carries type parameters that no route cares about. `Arc<dyn Registry>` is
//! where that genericity stops.
//!
//! # Running it
//!
//! The `didbot-pds` binary in this crate wires a development stack --
//! loopback DNS, an in-memory store -- and serves it, so that one command
//! produces a server for `didbot-swarm` to drive:
//!
//! ```text
//! cargo run -p didbot-serve --bin didbot-pds
//! ```
//!
//! # Routes
//!
//! | Route | Purpose |
//! | --- | --- |
//! | `GET /` | on the server's own hostname, a page naming it, its zones and its claim; elsewhere the same 404 as any unserved path |
//! | `GET /health` | liveness, and which zone this server mints under |
//! | `GET /.well-known/did.json` | the DID document for the request's Host |
//! | `GET /.well-known/atproto-did` | the DID of the account whose handle is the request's Host, 404 when nobody holds it |
//! | `POST /xrpc/bot.did.createAccount` | create an account beneath the creator the credential names |
//! | `POST /xrpc/bot.did.createSession` | a session as a hosted account, from a key-signed JWT or an OpenID Connect token |
//! | `POST /xrpc/bot.did.parkKey` | hold a public key for minutes, for the operator to admit under a name |
//! | `GET /xrpc/bot.did.describeParkedKey` | the key parked under a name |
//! | `POST /xrpc/bot.did.deleteAccount` | erase one: data gone, identity kept |
//! | `POST /xrpc/bot.did.freezeAccount`, `unfreezeAccount` | hang or lift the account's own `frozen` lock |
//! | `POST /xrpc/bot.did.deactivateAccount`, `activateAccount` | hang or lift the account's own `deactivated` lock |
//! | `GET /xrpc/bot.did.listAccounts` | list them |
//! | `POST /xrpc/bot.did.pollOperatorClaim` | ask this server to look for its operator's claim now — no credential, because the nudge confers nothing; see `auth` |
//! | `POST /xrpc/com.atproto.repo.createRecord` | write a record to a repository |
//! | `POST /xrpc/com.atproto.repo.putRecord` | write one at a key you choose |
//! | `POST /xrpc/com.atproto.repo.deleteRecord` | remove one, or make sure it is gone |
//! | `POST /xrpc/com.atproto.repo.applyWrites` | a batch of the three above |
//! | `GET /xrpc/com.atproto.repo.getRecord` | one record by collection and key |
//! | `GET /xrpc/com.atproto.repo.listRecords` | read a collection back, newest first |
//! | `GET /xrpc/com.atproto.repo.describeRepo` | the account, its document and its collections |
//! | `GET /xrpc/bot.did.stats` | accounts, records, bytes per collection, blobs by lifecycle state, the policy revision enforced |
//! | `GET /xrpc/com.atproto.server.describeServer` | this server's DID and its real zone-derived handle domains |
//! | `GET /xrpc/com.atproto.server.getServiceAuth` | a service-auth token signed as the caller's account |
//! | `GET /xrpc/com.atproto.server.getSession` | the account the caller's credential signs in as: its DID, handle and status |
//! | `GET /xrpc/com.atproto.identity.resolveHandle` | the DID a handle points at, hosted here or not |
//! | `GET /xrpc/com.atproto.identity.resolveDid` | a DID's document, hosted here or not |
//! | `GET /xrpc/app.bsky.actor.getPreferences` | the account's preferences: none, since this server stores none |
//! | `POST /xrpc/app.bsky.actor.putPreferences` | `MethodNotImplemented`, and never proxied |
//! | any other `/xrpc/` method, with `atproto-proxy` | the call, forwarded to the service that header names; see `proxy` |
//! | any other `app.bsky.*` method, without it | the same, to `[proxy] app_view` when one is set |
//!
//! Every `repo` above is an at-identifier: a DID or a handle this deployment
//! issued. Which credential each route needs, if any, is
//! [`auth::ROUTE_CREDENTIALS`] — the read surface stays `Public`, and every
//! `com.atproto.repo.*` write route takes the agent token `createAccount`
//! hands back; see that module's own doc for the full taxonomy.
//!
//! Failures are returned in atproto's error shape, `{"error":..,"message":..}`,
//! with the names the lexicons declare -- `RecordNotFound`, `RepoNotFound`,
//! `InvalidRequest`, `InvalidRecord`, `MethodNotImplemented`. See [`ApiError`]
//! for the whole mapping, and for the two conditions that get a name no lexicon
//! declares.

#![forbid(unsafe_code)]

mod analytics;
mod auth;
pub mod blob_collect;
mod blobs;
pub mod confinement_poll;
mod dashboard;
mod dashboard_accounts;
pub use dashboard_accounts::NEVER_WROTE_MIN_IDLE;
mod dashboard_apps;
mod error;
mod foreign;
pub mod handles;
mod health;
mod home;
mod landing;
pub mod lookups;
pub mod oauth;
pub mod operator;
pub mod operator_handle;
pub mod operator_poll;
pub mod policy_poll;
mod proxy;
#[cfg(test)]
mod race;
pub mod rate_limit;
mod refusal;
pub mod relay;
pub mod replay;
mod routes;
mod service_auth;
pub mod stale_sweep;
mod subscribe;
pub mod subscribers;
mod surface;
#[cfg(test)]
mod tests;
pub mod tree;
pub mod tree_reload;
mod wire;
pub mod write_meter;

pub use auth::{Credential, Disclosure, ROUTE_CREDENTIALS};
pub use blob_collect::{collect_once, BlobCollector, DEFAULT_COLLECT_INTERVAL};
pub use confinement_poll::{ConfinementPoll, DEFAULT_CONFINEMENT_INTERVAL};
pub use error::{repo_surface, ApiError, ErrorBody};
pub use health::{
    CertificateAlerts, CertificateHealth, DiskAlert, DiskHealth, DiskWatermarks, HealthSnapshot,
    HealthState, HealthTick, ZoneCertificateHealth, DEFAULT_HEALTH_INTERVAL,
};
pub use lookups::{DeclaredLookups, Reader as LookupReader};
pub use oauth::scope::{
    Action, ActionSet, MimePattern, NsidPattern, Scope, ScopeParseError, ScopeRefused, ScopeSet,
    Transition,
};
pub use operator_handle::OperatorHandle;
pub use operator_poll::{
    OperatorPoll, OperatorRecords, PollNudge, PollSchedule, MIN_POLL_INTERVAL,
};
pub use policy_poll::{list_records, PolicyPoll, PolicyPollError, PollError};
pub use proxy::{ProxyLimits, ProxySettings};
pub use routes::{
    app, app_with_auth, app_with_health, app_with_repos, cross_origin_layer, AuthState,
    StrictTransportSecurity, XrpcBody, APP_BSKY_METHODS, ATPROTO_METHODS, BOT_DID_METHODS,
    MAX_REQUEST_BODY,
};
pub use routes::{TURN_HEADER, TURN_HEADER_MAX_LENGTH};
pub use stale_sweep::{
    sweep_once, StaleSweeper, SweepSchedule, DEFAULT_SWEEP_INTERVAL, DEFAULT_SWEEP_LIMIT,
    DEFAULT_UNFINISHED_AGE,
};
pub use subscribe::{
    encode_error, encode_message, Exclusions, Frame, FrameKind, ReplayBounds, Repos, ReposStream,
    DEFAULT_REPLAY_BYTES, DEFAULT_REPLAY_FRAMES, FUTURE_CURSOR, OUTDATED_CURSOR,
};
pub use tree_reload::{TreeReloader, DEFAULT_TREE_RELOAD_INTERVAL};
pub use wire::{
    record_uri, refuse_skipped_validation, AccountProof, AccountSummary, ApplyWritesRequest,
    BlobsView, CreateAccountRequest, CreateRecordRequest, CreateRecordResponse,
    DeleteRecordRequest, DescribeRepoQuery, DescribeRepoResponse, DescribeServerResponse,
    DidRequest, EdgeState, Fragment, GetRecordQuery, GetRecordResponse, LifecycleLine,
    ListRecordsQuery, ParkKeyRequest, ParkedKeyQuery, PolicyBindingView, PolicyRecordView,
    PolicyRefusalView, PolicyView, PutRecordRequest, RecordView, RepoWrite, StatsResponse, Swap,
    TallyView, WriteResult, MAX_LIST_LIMIT,
};

use std::convert::Infallible;
use std::future::{ready, Ready};
use std::net::SocketAddr;
use std::task::{Context, Poll};

use axum::extract::connect_info::{ConnectInfo, Connected};
use axum::http::Request;
use axum::Router;
use tokio::net::TcpListener;
use tower::limit::{ConcurrencyLimit, ConcurrencyLimitLayer};
use tower::{Layer, Service};
use tracing::{info, warn};
use tracing_subscriber::fmt::time::uptime;
use tracing_subscriber::EnvFilter;

/// How many requests this server answers at once before a new one waits for
/// a slot to free up.
///
/// Backpressure rather than an outright refusal — `tower::limit::ConcurrencyLimit`
/// makes a request wait for a permit instead of failing it — because a
/// caller within its rate-limit budget (`crate::rate_limit`) asking at a bad
/// moment should be slowed, not bounced. The two are complementary: this
/// bounds how much concurrent work every route together may impose (memory,
/// open file descriptors, threads blocked on I/O), where a rate limiter
/// bounds how much *one caller* may ask for.
///
/// Applied by `limited_make_service`, which both [`serve_router`] and
/// [`serve_tls`] serve through, not as a layer on the [`Router`] itself. That
/// placement is load-bearing, not a style choice: `axum::Router::layer`
/// re-applies whatever [`tower::Layer`] it is given at more than one internal
/// dispatch point, each application constructing its own middleware
/// instance. A stateless layer does not care — a fresh `Duration` behaves
/// identically every time, which is why the router's own request timeout is
/// layered directly onto it. `ConcurrencyLimitLayer` is not
/// stateless: `Layer::layer` allocates a fresh `Semaphore` on every call, so
/// a limit applied through `Router::layer` ends up enforced against several
/// independent, unshared semaphores instead of the one number a deployment
/// configured — the bound would compile, look correct in a code review, and
/// do nothing. Wrapping the already-built `Router` once, there, and turning
/// the result into a `MakeService` the way [`tower::make::Shared`] would is
/// what keeps every per-connection clone pointing at the same semaphore. See
/// this crate's `tests::concurrency_past_the_limit_waits_instead_of_running_unbounded`
/// for the regression this would otherwise be.
pub const MAX_CONCURRENT_REQUESTS: usize = 512;

/// Filter used when `RUST_LOG` is unset.
///
/// `info` for this project's own crates, and the HTTP stack's internals turned
/// down. `hyper` and `h2` narrate every frame at DEBUG; leaving them on turns
/// the provisioning log -- the thing this server exists to show -- into a
/// minority of the output.
///
/// `tower` is deliberately not silenced here even though it is noisy in the
/// same way. `EnvFilter` matches a target by string prefix, so a `tower=warn`
/// directive also silences `tower_http::trace`, which is where this crate's
/// request log comes from.
pub const DEFAULT_LOG_FILTER: &str = "info,hyper=warn,hyper_util=warn,h2=warn,mio=warn";

/// Installs the tracing subscriber.
///
/// Honours `RUST_LOG`, falling back to [`DEFAULT_LOG_FILTER`]. Timestamps are
/// process uptime rather than wall clock: what a developer watching a
/// provisioning run wants to see is the gap between two events, and a relative
/// stamp is both shorter and easier to compare at a glance.
///
/// Safe to call more than once; the second call is a no-op rather than a
/// panic, because a test binary that installs a subscriber twice should not
/// fail for that reason.
pub fn init_tracing() {
    let filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(DEFAULT_LOG_FILTER));

    let _ = tracing_subscriber::fmt()
        .with_env_filter(filter)
        .with_timer(uptime())
        // The target says which crate spoke, which is how you tell a DNS
        // publish from a store write without reading the message.
        .with_target(true)
        .with_level(true)
        .compact()
        .try_init();
}

/// Serves an already-built router on `listener` until `SIGINT` or
/// `SIGTERM`, waiting at most [`SHUTDOWN_GRACE`] on connections that will
/// not close by themselves.
///
/// `router` is served through `limited_make_service`, the wrapping
/// [`serve_tls`] serves through too, so [`MAX_CONCURRENT_REQUESTS`] bounds
/// this listener the same way — see its own doc for why that bound is wired
/// there rather than as a layer on `router` itself.
pub async fn serve_router(listener: TcpListener, router: Router) -> std::io::Result<()> {
    // First, and before `listening` says the server is up: the accept loop
    // does not poll the future below until it is running, and a stop that
    // arrives before then is handed only to the listeners this process already
    // had. In this binary there is one — the long poll's — so the stop would be
    // logged and the accept loop would go on serving.
    let stopped = installed_shutdown_signal();
    match listener.local_addr() {
        Ok(addr) => info!(%addr, "listening"),
        Err(err) => warn!(%err, "listening on an address that could not be read back"),
    }
    // The signal is observed twice: once by `axum::serve`, to stop accepting
    // and let in-flight requests finish, and once here to start the clock on
    // [`SHUTDOWN_GRACE`]. `axum::serve` has no timeout of its own, and the
    // streams this server holds open never close by themselves, so without
    // the second observation a shutdown with one subscriber attached waits
    // forever.
    let (started, shutdown_started) = tokio::sync::oneshot::channel();
    let serve =
        axum::serve(listener, limited_make_service(router)).with_graceful_shutdown(async move {
            stopped.await;
            let _ = started.send(());
        });
    tokio::select! {
        result = serve => result,
        () = grace_expired(shutdown_started) => {
            warn!(
                after = ?SHUTDOWN_GRACE,
                "connections were still open when the shutdown grace ran out; stopping anyway"
            );
            Ok(())
        }
    }
}

/// Resolves [`SHUTDOWN_GRACE`] after `started` fires, and never if it does
/// not.
///
/// The sender is dropped without sending when serving ends for any reason
/// other than a signal — a bind that died, a listener that stopped
/// accepting — and that must not be read as a shutdown whose grace has
/// begun, so a closed channel pends here rather than starting the clock.
async fn grace_expired(started: tokio::sync::oneshot::Receiver<()>) {
    if started.await.is_err() {
        std::future::pending::<()>().await;
    }
    tokio::time::sleep(SHUTDOWN_GRACE).await;
}

/// What both listeners hand each accepted connection: `router` behind
/// [`MAX_CONCURRENT_REQUESTS`], carrying the peer's address as
/// [`ConnectInfo`].
///
/// One function rather than a wrapping per listener, because the wrapping
/// is the cap: a listener that turns `router` into a make-service any other
/// way serves without it.
fn limited_make_service(router: Router) -> LimitedMakeService {
    make_service_limited_to(router, MAX_CONCURRENT_REQUESTS)
}

/// [`limited_make_service`] at a limit a test can overrun with two requests.
fn make_service_limited_to(router: Router, limit: usize) -> LimitedMakeService {
    MakeServiceWithConnectInfo::new(ConcurrencyLimitLayer::new(limit).layer(router))
}

/// The service both listeners are driven by; see [`limited_make_service`].
type LimitedMakeService = MakeServiceWithConnectInfo<ConcurrencyLimit<Router>>;

/// Turns an already-[`Clone`] service — `limited_make_service`'s
/// [`ConcurrencyLimit`]-wrapped router — into a [`tower::Service`]
/// `axum::serve` and `axum_server` can both drive per accepted connection,
/// the same way [`tower::make::Shared`] does, but also carrying a
/// [`ConnectInfo`] extension for the connection's peer address, which
/// `crate::rate_limit::caller_key` reads by default.
///
/// `axum`'s own `Router::into_make_service_with_connect_info` cannot stand
/// in for this: it is only defined on a bare [`Router`], not on whatever
/// type wrapping one in a [`tower::Layer`] produces, and
/// [`MAX_CONCURRENT_REQUESTS`]'s own doc explains why that wrapping must
/// happen exactly once, here, rather than as a `Router::layer` — this type
/// exists so that requirement and per-connection `ConnectInfo` are not a
/// choice between one or the other.
#[derive(Clone)]
struct MakeServiceWithConnectInfo<S> {
    inner: S,
}

impl<S> MakeServiceWithConnectInfo<S> {
    fn new(inner: S) -> Self {
        Self { inner }
    }
}

impl<S, T> Service<T> for MakeServiceWithConnectInfo<S>
where
    S: Clone,
    SocketAddr: Connected<T>,
{
    type Response = ConnectInfoService<S>;
    type Error = Infallible;
    type Future = Ready<Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, target: T) -> Self::Future {
        ready(Ok(ConnectInfoService {
            inner: self.inner.clone(),
            addr: SocketAddr::connect_info(target),
        }))
    }
}

/// The per-connection service [`MakeServiceWithConnectInfo`] hands back:
/// `inner`, unchanged, with `ConnectInfo(addr)` inserted into every
/// request's extensions before the request reaches it.
#[derive(Clone)]
struct ConnectInfoService<S> {
    inner: S,
    addr: SocketAddr,
}

impl<S, ReqBody> Service<Request<ReqBody>> for ConnectInfoService<S>
where
    S: Service<Request<ReqBody>>,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: Request<ReqBody>) -> Self::Future {
        req.extensions_mut().insert(ConnectInfo(self.addr));
        self.inner.call(req)
    }
}

/// Serves an already-built router on `listener`, terminating TLS with
/// `config`, until `SIGINT` or `SIGTERM`, on the same [`SHUTDOWN_GRACE`]
/// [`serve_router`] gives the plain-HTTP path.
///
/// `config` is expected to come from `didbot_tls::manager::CertificateManager
/// ::server_config`, or anything else built the same way: **one**
/// `rustls::ServerConfig` for the whole life of this call, whose certificate
/// resolver is itself swappable. A renewal happening elsewhere replaces what
/// that resolver hands back; it does not call this function again, does not
/// rebind `listener`, and does not touch a connection already past its
/// handshake — the certificate only ever participates in the handshake that
/// created a connection. This crate does not know how to renew a
/// certificate, and does not need to: it only needs one `ServerConfig` that
/// keeps answering correctly for as long as the process runs.
///
/// This is why `serve_tls` takes a config rather than, say, a certificate
/// path: baking "reload the files and rebuild the config" in here would put
/// the rebind-avoidance property back in the caller's hands, exactly where
/// the epic that asked for this said it must not be.
///
/// Served through `limited_make_service`, the wrapping [`serve_router`]
/// serves through, so [`MAX_CONCURRENT_REQUESTS`] bounds this listener's
/// in-flight requests the same way, and with
/// [`HEADER_READ_TIMEOUT`] on the reading of a request's headers.
pub async fn serve_tls(
    listener: TcpListener,
    router: Router,
    config: std::sync::Arc<rustls::ServerConfig>,
) -> std::io::Result<()> {
    serve_tls_with_header_timeout(listener, router, config, HEADER_READ_TIMEOUT).await
}

/// [`serve_tls`], with the header-read timeout named rather than taken from
/// [`HEADER_READ_TIMEOUT`].
///
/// Public so a test can pick a timeout it can actually wait out; a
/// deployment calls [`serve_tls`].
pub async fn serve_tls_with_header_timeout(
    listener: TcpListener,
    router: Router,
    config: std::sync::Arc<rustls::ServerConfig>,
    header_read_timeout: std::time::Duration,
) -> std::io::Result<()> {
    // Registered before the listener is announced, for the reason
    // `serve_router` gives: the task below is spawned, so nothing polls it
    // until the runtime gets to it.
    let stopped = installed_shutdown_signal();
    serve_tls_inner(
        listener,
        router,
        config,
        header_read_timeout,
        stopped,
        SHUTDOWN_GRACE,
    )
    .await
}

/// [`serve_tls`], stopping when `stopped` resolves rather than on a signal
/// observed from here on, and waiting `drain` rather than
/// [`SHUTDOWN_GRACE`] for requests in flight. A process whose startup runs
/// after its handlers are installed passes [`Stopping::wait`], so a stop
/// sent during that startup is not lost.
pub async fn serve_tls_until(
    listener: TcpListener,
    router: Router,
    config: std::sync::Arc<rustls::ServerConfig>,
    stopped: impl std::future::Future<Output = ()> + Send + 'static,
    drain: std::time::Duration,
) -> std::io::Result<()> {
    serve_tls_inner(
        listener,
        router,
        config,
        HEADER_READ_TIMEOUT,
        stopped,
        drain,
    )
    .await
}

async fn serve_tls_inner(
    listener: TcpListener,
    router: Router,
    config: std::sync::Arc<rustls::ServerConfig>,
    header_read_timeout: std::time::Duration,
    stopped: impl std::future::Future<Output = ()> + Send + 'static,
    drain: std::time::Duration,
) -> std::io::Result<()> {
    match listener.local_addr() {
        Ok(addr) => info!(%addr, "listening (tls)"),
        Err(err) => warn!(%err, "listening on an address that could not be read back"),
    }
    let listener = listener.into_std()?;
    let rustls_config = axum_server::tls_rustls::RustlsConfig::from_config(config);
    let handle = axum_server::Handle::new();
    let shutdown_handle = handle.clone();
    tokio::spawn(async move {
        stopped.await;
        info!(?drain, "no longer accepting; waiting on requests in flight");
        // Graceful: let in-flight requests finish rather than cutting
        // connections mid-response, bounded by `drain` -- a subscriber
        // holding a stream open must not be able to hold the shutdown open
        // with it.
        shutdown_handle.graceful_shutdown(Some(drain));
    });
    let mut server = axum_server::from_tcp_rustls(listener, rustls_config)?;
    // A connection that has finished its TLS handshake and then sends its
    // request headers one byte at a time holds a connection, a task and a
    // buffer for free: `TimeoutLayer` on the router only starts once the
    // request has been read, so before this there was nothing bounding the
    // reading of it. `axum_server`'s own handshake timeout bounds the step
    // before; this bounds the step after.
    //
    // The timer is required: `hyper`'s HTTP/1 builder panics on a
    // `header_read_timeout` with no timer to measure it with.
    server
        .http_builder()
        .http1()
        .timer(hyper_util::rt::TokioTimer::new())
        .header_read_timeout(Some(header_read_timeout));
    server
        .handle(handle)
        .serve(limited_make_service(router))
        .await
}

/// How long a connection has to finish sending its request headers.
///
/// Generous for a real client on a slow link — a request head is a few
/// hundred bytes — and short enough that holding a connection open costs an
/// attacker a fresh handshake every time it elapses.
pub const HEADER_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);

/// How long a shutdown waits for connections that are still open before it
/// stops anyway, unless the caller names its own bound; `didbot-pds` reads
/// `[server] drain_secs`.
///
/// A bound is not optional here, for a reason specific to what this server
/// serves: `com.atproto.sync.subscribeRepos` is a stream that never ends on
/// its own. A graceful shutdown that waits for every connection to close
/// waits forever whenever one subscriber is attached — which, for a server
/// whose whole point is to be subscribed to, is the normal case rather than
/// the edge one.
///
/// What runs *after* serving ends has to fit inside the deployment's own
/// window too: the write-ahead log is synced there, and a sync that is
/// `SIGKILL`ed mid-way is the one outcome this whole path exists to avoid.
/// The unit file in `terraform-aws-didbot-pds/templates/user_data.sh.tftpl` allows 200
/// seconds from `SIGTERM` to `SIGKILL`.
pub const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10);

/// How long a shutdown waits, once serving has ended, for each
/// `subscribeRepos` connection to send what it holds and close.
///
/// A consumer that is reading takes far less to drain what it is owed, and
/// one that is not is closed by the process exiting.
pub const STREAM_CLOSE_GRACE: std::time::Duration = std::time::Duration::from_secs(2);

/// The handlers for the signals that stop this process, registered when this
/// is constructed rather than when it is awaited.
///
/// When they are registered is the whole point of the type. tokio hands a
/// signal to the listeners that existed when it arrived and to no others, so a
/// listener built after the signal has been delivered never fires, however
/// promptly it is then awaited.
struct StopHandlers {
    #[cfg(unix)]
    interrupt: Option<tokio::signal::unix::Signal>,
    #[cfg(unix)]
    terminate: Option<tokio::signal::unix::Signal>,
}

impl StopHandlers {
    #[cfg(unix)]
    fn install() -> Self {
        fn handler(
            kind: tokio::signal::unix::SignalKind,
            name: &'static str,
        ) -> Option<tokio::signal::unix::Signal> {
            match tokio::signal::unix::signal(kind) {
                Ok(stream) => Some(stream),
                Err(err) => {
                    // A handler that cannot be installed simply never fires,
                    // rather than taking down a process that is otherwise
                    // serving fine: panicking would end a healthy server, and
                    // resolving would stop it the instant it started.
                    warn!(%err, signal = name, "could not install a stop handler");
                    None
                }
            }
        }

        Self {
            interrupt: handler(tokio::signal::unix::SignalKind::interrupt(), "SIGINT"),
            terminate: handler(tokio::signal::unix::SignalKind::terminate(), "SIGTERM"),
        }
    }

    #[cfg(not(unix))]
    fn install() -> Self {
        Self {}
    }

    #[cfg(unix)]
    async fn stopped(mut self) {
        async fn fired(stream: Option<&mut tokio::signal::unix::Signal>) {
            match stream {
                Some(stream) => {
                    stream.recv().await;
                }
                None => std::future::pending().await,
            }
        }

        tokio::select! {
            () = fired(self.interrupt.as_mut()) => (),
            () = fired(self.terminate.as_mut()) => (),
        }
        info!("shutdown signal received");
    }

    #[cfg(not(unix))]
    async fn stopped(self) {
        match tokio::signal::ctrl_c().await {
            Ok(()) => info!("shutdown signal received"),
            Err(err) => {
                warn!(%err, "could not install the Ctrl-C handler");
                std::future::pending::<()>().await
            }
        }
    }
}

/// Resolves when this process is asked to stop: `SIGTERM`, or Ctrl-C. The
/// handlers are registered before this returns.
///
/// `SIGTERM` is the half a deployment actually uses: `docker stop` sends it,
/// systemd's own `ExecStop` sends it, and a process that is PID 1 in a
/// container — which this one is, see the `Dockerfile`'s `ENTRYPOINT` — gets
/// no default disposition for it, so an unhandled `SIGTERM` is not a
/// termination but a signal that is silently discarded. Waiting only on
/// Ctrl-C therefore meant every stop ran to its grace period and ended in
/// `SIGKILL`, skipping the write-ahead log's flush on the way out.
///
/// Public because more than one thing has to observe it. [`serve_router`]
/// stops accepting on it, and a caller that parks a request — the decision
/// long poll — waits on its own copy so a shutdown does not have to sit out
/// that request's whole window. Awaiting it more than once is fine; each
/// caller installs its own listener.
///
/// tokio hands a signal only to the listeners that existed when it arrived, so
/// the gap between building an observation and first polling it is a gap a
/// stop can fall into. [`serve_router`] holds that gap open across the whole of
/// its setup — it announces the listener, and only the accept loop that runs
/// afterwards polls the shutdown future — which is long enough for a real
/// `docker stop` to land in it.
pub fn installed_shutdown_signal() -> impl std::future::Future<Output = ()> + Send + 'static {
    StopHandlers::install().stopped()
}

/// One observation of the stop signal for a whole run, installed once at
/// its start and awaited by everything that stops on it.
///
/// A handler installed partway through startup misses a stop sent before
/// it. This one remembers: a stop that arrived while startup was busy is
/// still there for [`Stopping::wait`] when serving begins.
#[derive(Debug, Clone)]
pub struct Stopping(tokio::sync::watch::Receiver<bool>);

impl Stopping {
    /// Installs the handlers now. Needs a running tokio runtime.
    pub fn install() -> Self {
        let stopped = installed_shutdown_signal();
        let (sender, receiver) = tokio::sync::watch::channel(false);
        tokio::spawn(async move {
            stopped.await;
            let _ = sender.send(true);
        });
        Self(receiver)
    }

    /// A stop that arrives once `stopped` reads true, for a caller that
    /// decides for itself when a run stops.
    pub fn when(stopped: tokio::sync::watch::Receiver<bool>) -> Self {
        Self(stopped)
    }

    /// A stop that never arrives.
    pub fn never() -> Self {
        Self(tokio::sync::watch::channel(false).1)
    }

    /// Resolves once a stop has arrived, at once if one already has.
    pub async fn wait(mut self) {
        if self.0.wait_for(|stopped| *stopped).await.is_err() {
            std::future::pending::<()>().await;
        }
    }

    /// This stop, arriving only once `work` has run after it.
    ///
    /// For what has to happen while the process still serves: everything
    /// that waits on the returned one keeps going until `work` is done.
    /// `work` is not polled before the stop arrives.
    pub fn after(self, work: impl std::future::Future<Output = ()> + Send + 'static) -> Self {
        let (sender, receiver) = tokio::sync::watch::channel(false);
        tokio::spawn(async move {
            self.wait().await;
            work.await;
            let _ = sender.send(true);
        });
        Self(receiver)
    }
}