hyperhive/swarm-builder/src/main.rs

//! The swarm builder's build service.
//!
//! Connects to the swarm queue as its own OIDC client, takes build requests
//! from swarm-controller on [`swarm_wire::builder::SUBJECT`], builds
//! each agent config at its commit, answers on the request's reply subject,
//! and logs one line per build. It writes nothing to the forge: the
//! controller turns the reply into a commit status. It also answers
//! [`swarm_wire::builder::ARCHETYPES_SUBJECT`] with the agent
//! archetypes its hyperhive source declares, which the controller stores.
//!
//! Lock requests on [`swarm_wire::builder::LOCK_SUBJECT`] carry the swarm meta
//! flake's `flake.nix`, and the answer is the `flake.lock` `nix` wrote for it in
//! a scratch directory. The controller commits it.
//!
//! Meta build requests on [`swarm_wire::builder::META_SUBJECT`] name an agent
//! and an `internal/meta` commit, and the answer is the store path of that
//! agent's toplevel, built at the flake's own lock. They take the build slot
//! PR checks take. The controller records the path; the builder writes nothing.
//! When a cache is set (see [`push`]), a green meta build is pushed to it before
//! the reply, and the reply says whether that worked.
//!
//! The store identity comes from the `BAO_*` environment. The queue client
//! secret is read at start and again when authelia refuses it, the forge token
//! once per build, both from the store under that identity.

use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use futures_util::StreamExt as _;
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::forge::Credential;
use swarm_secret_client::stored_secret::StoredSecret;
use swarm_wire::builder::{
    ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, BuildFromMeta,
    FromMetaReply, LOCK_DEADLINE, LOCK_SUBJECT, LockMeta, LockReply, META_SUBJECT, QUEUE_GROUP,
    REPLY_DEADLINE, Reply, SUBJECT,
};

mod archetypes;
mod nix_build;
mod push;

/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
/// writes it, matching the leaf's subject.
const CERT_ROLE: &str = "swarm-builder";

/// The environment prefix `swarm-builder.nix` sets.
const PREFIX: &str = "SWARM_BUILDER";

/// The longest the store login and forge token read may take before a build.
/// Each store request alone may take 30 s, so without this bound the two could
/// spend the minute `REPLY_DEADLINE` leaves after `BUILD_DEADLINE`.
const TOKEN_DEADLINE: Duration = Duration::from_secs(20);

/// What is held back from the minute `REPLY_DEADLINE` leaves after
/// `BUILD_DEADLINE`: [`TOKEN_DEADLINE`], then 10 s to publish the reply and for
/// it to reach the controller. Both come after the slot wait.
const SLOT_MARGIN: Duration = Duration::from_secs(30);

/// The longest a request may wait for the build slot. A build that gets the
/// slot later could still be running when the controller stops waiting for the
/// reply.
const MAX_SLOT_WAIT: Duration = REPLY_DEADLINE
    .saturating_sub(BUILD_DEADLINE)
    .saturating_sub(SLOT_MARGIN);

/// What is held back from a meta build's reply deadline for publishing the reply
/// after its push.
const REPLY_MARGIN: Duration = Duration::from_secs(10);

/// How long the push of a meta build that ends at now may run: at most
/// [`push::PUSH_DEADLINE`], and never past `reply_by`, so a push cannot make the
/// reply later than the controller waits.
fn push_budget(reply_by: tokio::time::Instant) -> Duration {
    push::PUSH_DEADLINE.min(reply_by.saturating_duration_since(tokio::time::Instant::now()))
}

#[tokio::main]
async fn main() -> Result<()> {
    hive_log::init();
    swarm_secret_client::install_crypto_provider();
    let var = |name: &str| {
        let full = format!("{PREFIX}_{name}");
        std::env::var(&full).with_context(|| format!("{full} is unset"))
    };
    let client_id = var("OIDC_CLIENT_ID")?;
    let builder = Arc::new(nix_build::Builder {
        nix: PathBuf::from(var("NIX")?),
        check_flake: var("CHECK_FLAKE")?,
        hyperhive: var("HYPERHIVE_FLAKE")?,
        nixpkgs: var("NIXPKGS_FLAKE")?,
        forge_url: forge_url()?,
        config_org: var("CONFIG_ORG")?,
        deadline: BUILD_DEADLINE,
        lock_deadline: LOCK_DEADLINE,
        system: var("SYSTEM")?,
        cache: push_target()?,
    });
    let secret =
        swarm_secret_client::read_with_startup_retry("reading the queue client secret", || {
            client_secret(&client_id)
        })
        .await?;
    let evaluator = archetypes::Evaluator {
        nix: builder.nix.clone(),
        hyperhive: builder.hyperhive.clone(),
        nixpkgs: builder.nixpkgs.clone(),
        deadline: ARCHETYPES_DEADLINE,
    };
    let reread_id = client_id.clone();
    let secret = StoredSecret::new(client_id.clone(), secret, move || {
        let client_id = reread_id.clone();
        async move { client_secret(&client_id).await.map_err(Into::into) }
    });
    let queue = swarm_queue_client::connect(QueueConfig {
        url: var("NATS_URL")?,
        token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
        client_id,
        client_secret: ClientSecret::Stored(secret),
        ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE"))
            .ok()
            .map(PathBuf::from),
    })
    .await?;
    let mut requests = queue
        .queue_subscribe(SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {SUBJECT}"))?;
    tracing::info!(subject = SUBJECT, "builder: awaiting builds");
    let evaluations = queue
        .queue_subscribe(ARCHETYPES_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {ARCHETYPES_SUBJECT}"))?;
    // Outside the build slot below: a list refresh does not wait on a build.
    tokio::spawn(answer_evaluations(evaluations, queue.clone(), evaluator));
    let locks = queue
        .queue_subscribe(LOCK_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {LOCK_SUBJECT}"))?;
    // Outside the build slot too: a lock builds nothing, so it does not wait
    // behind an hour-long build.
    tokio::spawn(answer_locks(locks, queue.clone(), Arc::clone(&builder)));

    // One build at a time. swarm-controller already sends one at a time; this
    // holds the bound for any request it did not serialise, and `in_slot`
    // refuses one that waited too long behind another. Meta builds take the
    // same slot as PR checks.
    let slot = Arc::new(tokio::sync::Semaphore::new(1));
    serve_meta_builds(&queue, &builder, &slot).await?;

    while let Some(msg) = requests.next().await {
        let received = tokio::time::Instant::now();
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a build arrived with no reply subject; dropping it");
            continue;
        };
        let queue = queue.clone();
        let slot = Arc::clone(&slot);
        let builder = Arc::clone(&builder);
        tokio::spawn(async move {
            let reply = match serde_json::from_slice::<BuildAgent>(&msg.payload) {
                Ok(job) => {
                    let Some(reply) =
                        in_slot(&slot, received, || busy(&job), handle(&job, &builder)).await
                    else {
                        return;
                    };
                    reply
                }
                Err(e) => {
                    tracing::warn!(error = %e, "an undecodable build; answering with the error");
                    Reply::Failed {
                        error_tail: format!("the builder could not decode the request: {e}"),
                    }
                }
            };
            match serde_json::to_vec(&reply) {
                Ok(body) => {
                    if let Err(e) = queue.publish(reply_to, body.into()).await {
                        tracing::warn!(error = %e, "publishing a build's reply failed");
                    }
                }
                Err(e) => tracing::warn!(error = %e, "encoding a build's reply failed"),
            }
        });
    }
    bail!("the subscription to {SUBJECT} ended")
}

/// Answer every archetype evaluation request, one at a time, with the
/// evaluation's result.
async fn answer_evaluations(
    mut requests: async_nats::Subscriber,
    queue: async_nats::Client,
    evaluator: archetypes::Evaluator,
) {
    while let Some(msg) = requests.next().await {
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("an archetype evaluation arrived with no reply subject; dropping it");
            continue;
        };
        let reply = evaluator.eval().await;
        match &reply {
            swarm_wire::builder::ArchetypesReply::Evaluated { archetypes, .. } => {
                tracing::info!(count = archetypes.len(), "archetypes evaluated");
            }
            swarm_wire::builder::ArchetypesReply::Failed { error_tail } => {
                tracing::warn!(%error_tail, "archetype evaluation failed");
            }
        }
        match serde_json::to_vec(&reply) {
            Ok(body) => {
                if let Err(e) = queue.publish(reply_to, body.into()).await {
                    tracing::warn!(error = %e, "publishing an evaluation's reply failed");
                }
            }
            Err(e) => tracing::warn!(error = %e, "encoding an evaluation's reply failed"),
        }
    }
    tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended");
}

/// Answer every meta-flake lock request, one at a time, with the lock `nix`
/// wrote or why it did not.
async fn answer_locks(
    mut requests: async_nats::Subscriber,
    queue: async_nats::Client,
    builder: Arc<nix_build::Builder>,
) {
    while let Some(msg) = requests.next().await {
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a meta-flake lock arrived with no reply subject; dropping it");
            continue;
        };
        let reply = handle_lock(&msg.payload, &builder).await;
        match serde_json::to_vec(&reply) {
            Ok(body) => {
                if let Err(e) = queue.publish(reply_to, body.into()).await {
                    tracing::warn!(error = %e, "publishing a lock's reply failed");
                }
            }
            Err(e) => tracing::warn!(error = %e, "encoding a lock's reply failed"),
        }
    }
    tracing::error!("the subscription to {LOCK_SUBJECT} ended");
}

/// Decode one lock request, lock it and log the lock.
async fn handle_lock(payload: &[u8], builder: &nix_build::Builder) -> LockReply {
    let job: LockMeta = match serde_json::from_slice(payload) {
        Ok(job) => job,
        Err(e) => {
            tracing::warn!(error = %e, "an undecodable lock; answering with the error");
            return LockReply::Failed {
                error_tail: format!("the builder could not decode the request: {e}"),
            };
        }
    };
    let reply = match within(TOKEN_DEADLINE, forge_token()).await {
        Ok(token) => builder.lock(&job, &token).await,
        Err(e) => LockReply::Failed {
            error_tail: format!("{e:#}"),
        },
    };
    match &reply {
        LockReply::Locked { .. } => tracing::info!("swarm meta flake locked"),
        LockReply::Failed { error_tail } => {
            tracing::warn!(%error_tail, "swarm meta flake lock failed");
        }
    }
    reply
}

/// Subscribe to meta build requests and answer them in the background.
async fn serve_meta_builds(
    queue: &async_nats::Client,
    builder: &Arc<nix_build::Builder>,
    slot: &Arc<tokio::sync::Semaphore>,
) -> Result<()> {
    let requests = queue
        .queue_subscribe(META_SUBJECT, QUEUE_GROUP.to_owned())
        .await
        .with_context(|| format!("subscribing to {META_SUBJECT}"))?;
    tokio::spawn(answer_meta_builds(
        requests,
        queue.clone(),
        Arc::clone(builder),
        Arc::clone(slot),
    ));
    Ok(())
}

/// Answer every agent build from the meta flake, each in its own task, with the
/// build's result.
async fn answer_meta_builds(
    mut requests: async_nats::Subscriber,
    queue: async_nats::Client,
    builder: Arc<nix_build::Builder>,
    slot: Arc<tokio::sync::Semaphore>,
) {
    while let Some(msg) = requests.next().await {
        let received = tokio::time::Instant::now();
        let Some(reply_to) = msg.reply.clone() else {
            tracing::warn!("a meta build arrived with no reply subject; dropping it");
            continue;
        };
        let (queue, builder, slot) = (queue.clone(), Arc::clone(&builder), Arc::clone(&slot));
        tokio::spawn(async move {
            let reply = match serde_json::from_slice::<BuildFromMeta>(&msg.payload) {
                Ok(job) => {
                    let Some(reply) = in_slot(
                        &slot,
                        received,
                        || busy_meta(&job),
                        handle_meta(&job, &builder, received + REPLY_DEADLINE - REPLY_MARGIN),
                    )
                    .await
                    else {
                        return;
                    };
                    reply
                }
                Err(e) => {
                    tracing::warn!(error = %e, "an undecodable meta build; answering with the error");
                    FromMetaReply::Failed {
                        error_tail: format!("the builder could not decode the request: {e}"),
                    }
                }
            };
            match serde_json::to_vec(&reply) {
                Ok(body) => {
                    if let Err(e) = queue.publish(reply_to, body.into()).await {
                        tracing::warn!(error = %e, "publishing a meta build's reply failed");
                    }
                }
                Err(e) => tracing::warn!(error = %e, "encoding a meta build's reply failed"),
            }
        });
    }
    tracing::error!("the subscription to {META_SUBJECT} ended");
}

/// `build`'s reply once a request holds the build slot, or `refuse`'s, without
/// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`.
/// `None` only when the slot is closed.
async fn in_slot<T>(
    slot: &tokio::sync::Semaphore,
    received: tokio::time::Instant,
    refuse: impl FnOnce() -> T,
    build: impl Future<Output = T>,
) -> Option<T> {
    let Ok(permit) = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await else {
        return Some(refuse());
    };
    let _permit = permit.ok()?;
    Some(build.await)
}

/// The refusal text of a request that waited [`MAX_SLOT_WAIT`] for the slot.
fn busy_tail() -> String {
    format!(
        "the swarm builder was busy with another build for {} s, too long for this one \
         to finish before swarm-controller stops waiting for it",
        MAX_SLOT_WAIT.as_secs()
    )
}

/// Log and answer a PR check that waited too long for the slot.
fn busy(job: &BuildAgent) -> Reply {
    tracing::warn!(
        agent = %job.agent,
        rev = %job.rev,
        pr = job.pr,
        waited_secs = MAX_SLOT_WAIT.as_secs(),
        "build refused: another build held the slot too long for this one to finish \
         before swarm-controller stops waiting"
    );
    Reply::Failed {
        error_tail: busy_tail(),
    }
}

/// Log and answer a meta build that waited too long for the slot.
fn busy_meta(job: &BuildFromMeta) -> FromMetaReply {
    tracing::warn!(
        agent = %job.agent,
        meta_rev = %job.meta_rev,
        waited_secs = MAX_SLOT_WAIT.as_secs(),
        "meta build refused: another build held the slot too long for this one to finish \
         before swarm-controller stops waiting"
    );
    FromMetaReply::Unavailable {
        reason: busy_tail(),
    }
}

/// Build one meta request and log the build.
async fn handle_meta(
    job: &BuildFromMeta,
    builder: &nix_build::Builder,
    reply_by: tokio::time::Instant,
) -> FromMetaReply {
    let reply = match within(TOKEN_DEADLINE, forge_token()).await {
        Ok(token) => builder.build_from_meta(job, &token).await,
        Err(e) => FromMetaReply::Unavailable {
            reason: format!("{e:#}"),
        },
    };
    let reply = push::cache_built(
        reply,
        builder.cache.as_ref(),
        attic_push_token(),
        push_budget(reply_by),
        &job.agent,
    )
    .await;
    match &reply {
        FromMetaReply::Built {
            out_path,
            system,
            cached,
        } => tracing::info!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %out_path,
            %system,
            cached,
            "agent meta build built"
        ),
        FromMetaReply::Failed { error_tail } => tracing::warn!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %error_tail,
            "agent meta build failed"
        ),
        FromMetaReply::Unavailable { reason } => tracing::warn!(
            agent = %job.agent,
            meta_rev = %job.meta_rev,
            %reason,
            "agent meta build not run"
        ),
    }
    reply
}

/// Build one request and log the build.
async fn handle(job: &BuildAgent, builder: &nix_build::Builder) -> Reply {
    let reply = match within(TOKEN_DEADLINE, forge_token()).await {
        Ok(token) => builder.build(job, &token).await,
        Err(e) => Reply::Failed {
            error_tail: format!("{e:#}"),
        },
    };
    log_build(job, &reply);
    reply
}

/// `read`'s result, or an error once it has run for `deadline`.
async fn within<T>(deadline: Duration, read: impl Future<Output = Result<T>>) -> Result<T> {
    tokio::time::timeout(deadline, read)
        .await
        .unwrap_or_else(|_| {
            Err(anyhow!(
                "reading the forge token took more than {} s",
                deadline.as_secs()
            ))
        })
}

/// The one journal line a build leaves.
fn log_build(job: &BuildAgent, reply: &Reply) {
    match reply {
        Reply::Built { out_path } => tracing::info!(
            agent = %job.agent,
            rev = %job.rev,
            pr = job.pr,
            %out_path,
            "agent config built"
        ),
        Reply::Failed { error_tail } => tracing::warn!(
            agent = %job.agent,
            rev = %job.rev,
            pr = job.pr,
            %error_tail,
            "agent config build failed"
        ),
    }
}

/// The token the builder pushes closures to the cache with, read on every push so
/// a rotation takes effect on the next one.
async fn attic_push_token() -> Result<String> {
    let path = swarm_secret_client::builder::attic_push_token_path()?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<swarm_secret_client::builder::AtticPushToken> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(t) if !t.value.trim().is_empty() => Ok(t.value.trim().to_owned()),
        _ => bail!("nothing at {path}: the cache push token has not been written"),
    }
}

/// The forge token the builder fetches config repos with, read on every
/// build so a rotation takes effect on the next one.
async fn forge_token() -> Result<String> {
    let path = swarm_secret_client::builder::forge_token_path()?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<Credential> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
        _ => bail!("nothing at {path}: swarm-controller has not minted the builder's forge token"),
    }
}

async fn client_secret(client_id: &str) -> Result<String> {
    use swarm_secret_client::builder::{ClientSecret, client_secret_path};

    let path = client_secret_path(client_id)?;
    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")?;
    let stored: Option<ClientSecret> = store
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
        _ => bail!(
            "nothing at {path}: authelia's mint, in its container, has not \
             stored this service's client secret"
        ),
    }
}

/// The cache the unit's environment names, if any. systemd sets
/// `RUNTIME_DIRECTORY` for the unit's runtime directory.
fn push_target() -> Result<Option<push::Remote>> {
    let var = |name: &str| std::env::var(name).ok();
    push::Remote::from_settings(
        var(&format!("{PREFIX}_ATTIC_DOMAIN")),
        var(&format!("{PREFIX}_ATTIC_CACHE")),
        var("RUNTIME_DIRECTORY"),
    )
}

/// The forge's base URL, which must be `https`: the credential helper offers
/// the token to `https` requests for its host alone.
fn forge_url() -> Result<String> {
    let name = format!("{PREFIX}_FORGE_URL");
    let url = std::env::var(&name).with_context(|| format!("{name} is unset"))?;
    let url = url.trim_end_matches('/');
    if !url.starts_with("https://") || url["https://".len()..].contains('/') {
        bail!("{name} is {url:?}, not https://<host>[:port]");
    }
    Ok(url.to_owned())
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, Ordering};

    use tokio::sync::Semaphore;
    use tokio::time::Instant;

    use super::*;

    fn job() -> BuildAgent {
        BuildAgent {
            agent: "atlas".to_owned(),
            rev: "abc".to_owned(),
            pr: 3,
        }
    }

    /// A request behind a build that holds the slot for `held`: its reply,
    /// whether its own build ran, and how long after arriving it was answered.
    async fn behind_a_build_of(held: Duration) -> (Option<Reply>, bool, Duration) {
        let slot = Arc::new(Semaphore::new(1));
        let running = Arc::clone(&slot).acquire_owned().await.expect("open");
        tokio::spawn(async move {
            tokio::time::sleep(held).await;
            drop(running);
        });
        let received = Instant::now();
        let ran = AtomicBool::new(false);
        let build = async {
            ran.store(true, Ordering::SeqCst);
            Reply::Built {
                out_path: "/nix/store/x".to_owned(),
            }
        };
        let reply = in_slot(&slot, received, || busy(&job()), build).await;
        (reply, ran.load(Ordering::SeqCst), received.elapsed())
    }

    #[tokio::test(start_paused = true)]
    async fn a_request_that_waited_past_the_slot_limit_is_refused_without_building() {
        let (reply, built, answered_after) =
            behind_a_build_of(MAX_SLOT_WAIT + Duration::from_mins(1)).await;
        assert!(
            matches!(&reply, Some(Reply::Failed { error_tail }) if error_tail.contains("busy")),
            "{reply:?}"
        );
        assert!(!built, "a refused request must not build");
        assert_eq!(answered_after, MAX_SLOT_WAIT);
    }

    #[tokio::test(start_paused = true)]
    async fn a_request_that_gets_the_slot_in_time_builds() {
        let (reply, built, _) =
            behind_a_build_of(MAX_SLOT_WAIT.saturating_sub(Duration::from_secs(1))).await;
        assert!(matches!(reply, Some(Reply::Built { .. })), "{reply:?}");
        assert!(built);
    }

    /// A meta build waits on the slot a PR check holds, and is refused at the
    /// same limit.
    #[tokio::test(start_paused = true)]
    async fn a_meta_build_shares_the_slot_with_pr_checks() {
        let meta = BuildFromMeta {
            agent: "atlas".to_owned(),
            meta_rev: "abc".to_owned(),
        };
        let slot = Arc::new(Semaphore::new(1));
        let check = Arc::clone(&slot).acquire_owned().await.expect("open");
        let received = Instant::now();
        let refused = in_slot(&slot, received, || busy_meta(&meta), async {
            FromMetaReply::Built {
                out_path: "/nix/store/x".to_owned(),
                system: "s".to_owned(),
                cached: false,
            }
        })
        .await;
        assert!(
            matches!(&refused, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")),
            "{refused:?}"
        );
        assert_eq!(received.elapsed(), MAX_SLOT_WAIT);
        // Control: once the check lets go, the same request builds.
        drop(check);
        let built = in_slot(&slot, Instant::now(), || busy_meta(&meta), async {
            FromMetaReply::Built {
                out_path: "/nix/store/x".to_owned(),
                system: "s".to_owned(),
                cached: false,
            }
        })
        .await;
        assert!(
            matches!(built, Some(FromMetaReply::Built { .. })),
            "{built:?}"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_token_read_past_its_deadline_is_an_error_and_a_prompt_one_is_kept() {
        let started = Instant::now();
        let stuck = within(TOKEN_DEADLINE, std::future::pending::<Result<String>>()).await;
        assert_eq!(started.elapsed(), TOKEN_DEADLINE);
        assert!(
            matches!(&stuck, Err(e) if e.to_string().contains("took more than 20 s")),
            "{stuck:?}"
        );
        let prompt = within(TOKEN_DEADLINE, async { Ok("t".to_owned()) }).await;
        assert_eq!(prompt.expect("read"), "t");
    }

    #[tokio::test(start_paused = true)]
    async fn a_push_gets_the_deadline_unless_the_reply_is_due_sooner() {
        let now = Instant::now();
        assert_eq!(
            push_budget(now + Duration::from_hours(1)),
            push::PUSH_DEADLINE
        );
        assert_eq!(
            push_budget(now + Duration::from_secs(90)),
            Duration::from_secs(90)
        );
        assert_eq!(push_budget(now), Duration::ZERO);
    }

    #[test]
    fn a_build_started_at_the_slot_limit_ends_inside_the_reply_deadline() {
        assert!(MAX_SLOT_WAIT > Duration::ZERO);
        assert!(MAX_SLOT_WAIT + BUILD_DEADLINE < REPLY_DEADLINE);
        assert!(SLOT_MARGIN >= Duration::from_secs(30));
        assert!(SLOT_MARGIN >= TOKEN_DEADLINE + Duration::from_secs(10));
        assert!(MAX_SLOT_WAIT + SLOT_MARGIN + BUILD_DEADLINE <= REPLY_DEADLINE);
    }
}