kache 1.2.2

//! Publication of compiler outputs handed off by a wrapper.
//!
//! A C compile through a build script runs nearly serially: cc-rs gets the
//! jobserver tokens rustc leaves over, and it cannot start the next file
//! until the wrapper for the previous one exits. Everything the wrapper does
//! after the object is written (staging, the store put, the durability
//! hand-off, the event log) is therefore wall-clock time on the build's
//! critical path. This module lets the wrapper hand that work to the daemon
//! and exit as soon as the daemon has accepted it.
//!
//! The contract:
//! - The wrapper snapshots its outputs into files it owns under the store's
//!   staging directory, drops its own key lock, and sends
//!   [`PublishCcRequest`]. The daemon takes the key lock itself before it
//!   answers, so a peer computing the same key sees the build as contended
//!   and waits for the commit, as it would for a wrapper that was still
//!   storing.
//! - The daemon answers `ok` only once the job is queued behind that lock.
//!   An atomic receipt resolves lost replies: the daemon renames pending
//!   to accepted before enqueueing; the wrapper can cancel by removing pending.
//!   A refusal before the daemon holds the key falls back to the wrapper.
//!   A cancellation after the daemon has snapshotted and holds the key is
//!   published by the daemon, so a slow snapshot cannot drop the compile.
//! - The daemon owns the handoff files from acceptance on and removes them
//!   once the put has copied them. A daemon that dies with jobs queued loses
//!   those stores, never a build: the compile has already produced its
//!   outputs and returned its exit code.
//! - The compiler never runs in the daemon.

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};

use crate::config::Config;
use crate::daemon::{Daemon, Response, UploadJob};
use crate::events::{BuildEvent, EventResult};
use crate::store::{BuildClaim, KeyLock, Store};

/// Jobs waiting for the worker. A small object takes a few milliseconds, but
/// a large C++ unit (Firefox's `Unified_*.cpp`) takes much longer, so a big
/// build can end with the queue full. When it is full the wrapper publishes
/// itself; whatever is queued at shutdown is drained (see
/// [`drain_publications`]).
pub(crate) const PUBLISH_QUEUE_CAPACITY: usize = 256;

/// How long shutdown waits for the publish worker to finish another job
/// before it gives up on the rest of the queue.
pub(crate) const PUBLISH_DRAIN_STALL: Duration = Duration::from_secs(10);

/// How long a wrapper waits for the daemon to accept a hand-off. The daemon's
/// side of it is a file lock and a channel send; anything slower means the
/// daemon is busy and the wrapper is better off doing the work.
pub(crate) const PUBLISH_HANDOFF_TIMEOUT: Duration = Duration::from_millis(25);

/// Name of the directory under the store's staging area that holds files a
/// wrapper handed to the daemon.
pub(crate) const HANDOFF_DIR: &str = "handoff";

/// One output the wrapper snapshotted for the daemon.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub(crate) struct HandoffFile {
    /// Absolute path of the snapshot, under the store's handoff directory.
    pub path: String,
    /// Name the file takes in the entry.
    pub store_name: String,
}

/// A wrapper's request that the daemon store a cc compile it has finished.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub(crate) struct PublishCcRequest {
    /// Client executable mtime. Daemons from 1.0.x and earlier drain when it
    /// is newer than their own; later daemons ignore it.
    #[serde(default)]
    pub client_epoch: u64,
    /// Release requesting an upgrade. Absent from legacy and read-only requests.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub client_version: Option<String>,
    pub cache_key: String,
    pub crate_name: String,
    pub target: String,
    pub files: Vec<HandoffFile>,
    pub stdout: String,
    pub stderr: String,
    pub compile_time_ms: u64,
    /// Whether the entry may be uploaded to a writable remote.
    pub publishes_to_remote: bool,
    /// The build event the wrapper would have logged, with the store fields
    /// still empty. The daemon fills them in and writes it.
    pub event: BuildEvent,
    /// The read-set memo of a deferred compile, recorded here instead of in
    /// the wrapper: a hundred-row write transaction the next compile of the
    /// build script would otherwise wait for.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub memo: Option<CcMemoHandoff>,
}

/// A preprocess memo as the wrapper captured it: the fingerprints and
/// content hashes of everything the compile read, under the key of its
/// arguments. Each input carries the time it was observed, and the store
/// records its stamp only if the stamp had settled by then.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub(crate) struct CcMemoHandoff {
    pub memo_key: String,
    pub preprocessed_hash: String,
    pub inputs: Vec<crate::cache_key::CcPreprocessMemoInput>,
}

/// A job the handler accepted: the request plus the key lock it holds.
pub(crate) struct PublishJob {
    request: PublishCcRequest,
    _lock: KeyLock,
}

/// The daemon's publish queue: a bounded channel drained by one blocking
/// worker with its own store connection.
pub(crate) struct PublishQueue {
    tx: std::sync::Mutex<Option<tokio::sync::mpsc::Sender<PublishJob>>>,
    progress: Arc<PublishProgress>,
}

/// Accepted jobs not yet published, and jobs published so far. Shutdown
/// reads them after the daemon itself is gone, so they live apart from it.
#[derive(Default)]
pub(crate) struct PublishProgress {
    pending: std::sync::atomic::AtomicUsize,
    done: std::sync::atomic::AtomicU64,
}

impl PublishProgress {
    pub(crate) fn pending(&self) -> usize {
        self.pending.load(std::sync::atomic::Ordering::Acquire)
    }

    pub(crate) fn done(&self) -> u64 {
        self.done.load(std::sync::atomic::Ordering::Acquire)
    }

    fn accepted(&self) {
        self.pending
            .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
    }

    fn finished(&self) {
        self.pending
            .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
        self.done.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
    }
}

impl PublishQueue {
    pub(crate) fn new() -> Self {
        Self {
            tx: std::sync::Mutex::new(None),
            progress: Arc::new(PublishProgress::default()),
        }
    }

    pub(crate) fn progress(&self) -> Arc<PublishProgress> {
        Arc::clone(&self.progress)
    }

    pub(crate) fn set_sender(&self, tx: tokio::sync::mpsc::Sender<PublishJob>) {
        if let Ok(mut slot) = self.tx.lock() {
            *slot = Some(tx);
        }
    }

    fn sender(&self) -> Option<tokio::sync::mpsc::Sender<PublishJob>> {
        self.tx.lock().ok().and_then(|slot| slot.clone())
    }

    /// Drop the sender so the worker drains what is queued and exits.
    pub(crate) fn close(&self) {
        if let Ok(mut slot) = self.tx.lock() {
            slot.take();
        }
    }
}

/// Where a wrapper snapshots outputs it hands to the daemon.
pub(crate) fn handoff_dir(config: &Config) -> PathBuf {
    config.store_dir().join("staging").join(HANDOFF_DIR)
}

/// Whether `path` is one the daemon may ingest: an absolute path under this
/// store's handoff directory, with no `..` to walk out of it.
fn handoff_path_is_owned(config: &Config, path: &Path) -> bool {
    path.is_absolute()
        && path.starts_with(handoff_dir(config))
        && !path
            .components()
            .any(|component| matches!(component, std::path::Component::ParentDir))
}

/// Why a hand-off was refused, in the response's error text. The wrapper only
/// needs to know it was refused; the text is for `kache daemon` logs.
fn refused(reason: &str) -> Response {
    Response::err(format!("{REFUSED}{reason}"))
}

const REFUSED: &str = "publish refused: ";

/// The wrapper's budget ran out before the daemon copied its files, so it
/// cancelled, removed them and stores the compile itself. Under load this is
/// routine and only logged at debug level (#1362).
const WITHDRAWN: &str = "the wrapper withdrew the hand-off first";

/// Whether a failed request's `error` is a refusal the wrapper recovers from
/// on its own, so the daemon need not warn about it.
pub(crate) fn refusal_is_routine(error: &str) -> bool {
    error.strip_prefix(REFUSED) == Some(WITHDRAWN)
}

/// The wrapper removes its receipt before the files it sent, so a snapshot
/// that found a file missing while the receipt is also gone lost that race.
fn handoff_withdrawn(request: &PublishCcRequest, error: &anyhow::Error) -> bool {
    receipt_cancelled(error)
        && receipt_paths(request)
            .is_ok_and(|(pending, _)| std::fs::symlink_metadata(pending).is_err())
}

impl Daemon {
    /// Accept a wrapper's hand-off if the key can be claimed and the queue has
    /// room. Claiming happens before the answer: from the wrapper's return
    /// onwards a peer must see the key as contended, not absent.
    pub(crate) async fn handle_publish_cc(self: &Arc<Self>, request: PublishCcRequest) -> Response {
        if !crate::cache_key::is_valid_cache_key(&request.cache_key) {
            return refused("invalid cache key");
        }
        if request.files.is_empty() {
            return refused("no files");
        }
        let config = self.config().clone();
        if let Some(file) = request
            .files
            .iter()
            .find(|file| !handoff_path_is_owned(&config, Path::new(&file.path)))
        {
            return refused(&format!("{} is outside the handoff directory", file.path));
        }
        let Some(tx) = self.publish_queue().sender() else {
            return refused("queue closed");
        };
        // Reserve before claiming ownership: once the receipt is accepted,
        // queue insertion must not fail because another handler took its slot.
        let permit = match tx.try_reserve_owned() {
            Ok(permit) => permit,
            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => return refused("queue full"),
            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                return refused("queue closed");
            }
        };
        let key = request.cache_key.clone();
        let daemon = Arc::clone(self);
        let claim =
            tokio::task::spawn_blocking(move || daemon.with_store(|store| store.claim_build(&key)))
                .await;
        let lock = match claim {
            Ok(Ok(BuildClaim::Acquired(lock))) => lock,
            Ok(Ok(BuildClaim::Committed(_))) => return refused("already committed"),
            Ok(Ok(BuildClaim::Contended)) => return refused("key contended"),
            Ok(Err(error)) => return refused(&format!("claim failed: {error:#}")),
            Err(error) => return refused(&format!("claim task failed: {error}")),
        };
        // Own separate links before acknowledging. A client may time out
        // after acceptance and remove every path it sent us.
        let claimed = tokio::task::spawn_blocking(move || {
            let files = request
                .files
                .iter()
                .map(|file| (PathBuf::from(&file.path), file.store_name.clone()))
                .collect::<Vec<_>>();
            let owned = match snapshot_for_handoff(&config, &files) {
                Ok(owned) => owned,
                Err(error) if handoff_withdrawn(&request, &error) => return Ok(None),
                Err(error) => return Err(error),
            };
            // Rename and client cancellation race on the same pending file.
            // Exactly one wins, even if the socket reply is lost. A missing
            // receipt means the wrapper already gave up. It publishes only if
            // it can take the key back, so keep these snapshots and the claim.
            if let Err(error) = accept_receipt(&request)
                && !receipt_cancelled(&error)
            {
                remove_handoff_files(&owned);
                return Err(error);
            }
            remove_handoff_files(&request.files);
            let mut request = request;
            request.files = owned;
            Ok::<_, anyhow::Error>(Some(PublishJob {
                request,
                _lock: lock,
            }))
        })
        .await;
        let job = match claimed {
            Ok(Ok(Some(job))) => job,
            Ok(Ok(None)) => return refused(WITHDRAWN),
            Ok(Err(error)) => return refused(&format!("snapshot failed: {error:#}")),
            Err(error) => return refused(&format!("snapshot task failed: {error}")),
        };
        self.publish_queue().progress.accepted();
        permit.send(job);
        Response::ok()
    }
}

/// Wait for the publish worker to drain the queue after [`PublishQueue::close`].
///
/// There is no fixed budget: a large build can end with a full queue of big
/// objects that take minutes to publish, and each job left behind is an
/// entry the next build compiles again. The wait ends when the worker exits,
/// or when it has finished nothing for `stall`, which means it is wedged.
/// Returns how many accepted jobs were left unpublished.
pub(crate) async fn drain_publications(
    mut done: tokio::sync::oneshot::Receiver<()>,
    progress: &PublishProgress,
    stall: Duration,
) -> usize {
    let tick = stall.min(Duration::from_secs(1));
    let mut last_done = progress.done();
    let mut last_change = tokio::time::Instant::now();
    loop {
        // Each pass ends within a tick of the last check, and a check past
        // `stall` returns, so the wait never runs far past the window. A
        // hang would otherwise read as a slow drain.
        debug_assert!(
            last_change.elapsed() < stall * 10 + tick,
            "drain kept waiting past its stall window"
        );
        match tokio::time::timeout(tick, &mut done).await {
            Ok(_) => return progress.pending(),
            Err(_) => {
                let now_done = progress.done();
                if now_done != last_done {
                    // The count only grows; a change that is not growth
                    // would keep the stall clock from ever running out.
                    debug_assert!(now_done > last_done, "publish progress went backwards");
                    last_done = now_done;
                    last_change = tokio::time::Instant::now();
                } else if last_change.elapsed() >= stall {
                    return progress.pending();
                }
            }
        }
    }
}

/// Run [`run_publish_worker`] on a thread of its own. The receiver resolves
/// once the worker has drained the queue and exited, and the caller bounds
/// that wait. A plain thread rather than `spawn_blocking`: a worker wedged in
/// a put must not hold up the runtime's teardown.
pub(crate) fn spawn_publish_worker(
    daemon: Arc<Daemon>,
    rx: tokio::sync::mpsc::Receiver<PublishJob>,
) -> std::io::Result<tokio::sync::oneshot::Receiver<()>> {
    let runtime = tokio::runtime::Handle::current();
    let (done_tx, done) = tokio::sync::oneshot::channel();
    std::thread::Builder::new()
        .name("kache-publish".to_string())
        .spawn(move || {
            // Uploads are queued through the runtime (see `enqueue_upload`).
            let _runtime = runtime.enter();
            run_publish_worker(daemon, rx);
            let _ = done_tx.send(());
        })?;
    Ok(done)
}

/// Run the publish worker until the queue closes and drains. Blocking, with
/// its own store connection so the daemon's shared store mutex never waits
/// on a put.
pub(crate) fn run_publish_worker(
    daemon: Arc<Daemon>,
    mut rx: tokio::sync::mpsc::Receiver<PublishJob>,
) {
    let config = daemon.config().clone();
    let store = match Store::open(&config) {
        Ok(store) => store,
        Err(error) => {
            tracing::warn!("publish worker could not open the store: {error:#}; hand-offs refused");
            daemon.publish_queue().close();
            // Drain so accepted jobs release their locks and files.
            while let Some(job) = rx.blocking_recv() {
                discard_handoff_files(&job.request);
                daemon.publish_queue().progress.finished();
            }
            return;
        }
    };
    while let Some(job) = rx.blocking_recv() {
        publish_one(&daemon, &config, &store, job);
    }
}

/// Store one handed-off compile, log its event, and release what the job
/// held. Every failure is recorded on the event; none of them can reach a
/// build.
fn publish_one(daemon: &Arc<Daemon>, config: &Config, store: &Store, job: PublishJob) {
    let PublishJob { request, _lock } = job;
    let store_started = std::time::Instant::now();
    let files: Vec<(PathBuf, String)> = request
        .files
        .iter()
        .map(|file| (PathBuf::from(&file.path), file.store_name.clone()))
        .collect();
    let bytes_before = StoreBytes::snapshot();
    let mut event = request.event.clone();
    let put = store.put_with_compile_time_independent(
        &request.cache_key,
        &request.crate_name,
        &[],
        &[],
        &request.target,
        "",
        &files,
        &request.stdout,
        &request.stderr,
        request.compile_time_ms,
    );
    // As after a wrapper's put: a size check after a stored entry, recovery
    // after one refused for lack of space.
    crate::wrapper::after_store(config, store, put.as_ref().err());
    match put {
        Ok(put) => {
            event.result = if put.is_full_dup() {
                EventResult::Dup
            } else {
                EventResult::Miss
            };
            event.store_output_blobs = put.output_blobs;
            event.store_duplicate_blobs = put.duplicate_blobs;
            event.store_new_blobs = put.new_blobs;
            bytes_before.charge(&mut event);
            if let Some(memo) = &request.memo
                && let Err(error) = store.file_hash_cache().put_cc_preprocess_memo_inputs(
                    &memo.memo_key,
                    &memo.preprocessed_hash,
                    &memo.inputs,
                )
            {
                tracing::debug!("daemon could not record the cc memo: {error}");
            }
            if request.publishes_to_remote && config.remote.is_some() {
                enqueue_upload(daemon, config, &request);
            }
        }
        Err(error) => {
            event.store_error = crate::wrapper::store_error_for_event(&error);
            tracing::warn!(
                "daemon could not store the cc entry for {}: {}",
                request.crate_name,
                event.store_error
            );
        }
    }
    // This work overlaps later compiles. Keep it out of the wrapper's
    // synchronous store phase and report it separately.
    event.daemon_store_ms = store_started.elapsed().as_millis() as u64;
    event.store_handed_off = true;
    // The put has its own copy (staged then published); the snapshots are
    // done. Drop the lock only after the files are gone: a peer that takes
    // the key next must find a committed entry, and nothing of ours left.
    discard_handoff_files(&request);
    crate::wrapper::write_event(config, &event);
    drop(_lock);
    daemon.publish_queue().progress.finished();
}

fn discard_handoff_files(request: &PublishCcRequest) {
    remove_handoff_files(&request.files);
}

/// Queue the entry for upload the way the wrapper would have: a durable
/// intent first, then the daemon's own upload pipeline.
fn enqueue_upload(daemon: &Arc<Daemon>, config: &Config, request: &PublishCcRequest) {
    if config.remote_readonly {
        daemon.warn_readonly_upload_skipped();
        return;
    }
    let job = UploadJob {
        key: request.cache_key.clone(),
        entry_dir: config
            .store_dir()
            .join(&request.cache_key)
            .to_string_lossy()
            .into_owned(),
        crate_name: request.crate_name.clone(),
        client_epoch: 0,
        client_version: Some(crate::VERSION.to_owned()),
    };
    let daemon = Arc::clone(daemon);
    if let Ok(handle) = tokio::runtime::Handle::try_current() {
        let response = handle.block_on(daemon.handle_upload(&job));
        if let Some(error) = response.error {
            tracing::debug!("upload of {} not queued: {error}", request.crate_name);
        }
    }
}

/// Bytes recorded on the publication thread. Remote imports may run on
/// other threads at the same time and must not be charged to this job.
struct StoreBytes {
    reflinked: u64,
    hardlinked: u64,
    copied: u64,
    cross_device: u64,
    permission: u64,
    ineligible: u64,
    other: u64,
}

impl StoreBytes {
    fn snapshot() -> Self {
        let bytes = crate::opcounts::store_thread_bytes();
        Self {
            reflinked: bytes.reflinked,
            hardlinked: bytes.hardlinked,
            copied: bytes.copied,
            cross_device: bytes.copy_cross_device,
            permission: bytes.copy_permission,
            ineligible: bytes.copy_ineligible,
            other: bytes.copy_other,
        }
    }

    fn charge(&self, event: &mut BuildEvent) {
        let now = Self::snapshot();
        event.store_reflinked_bytes = now.reflinked.saturating_sub(self.reflinked);
        event.store_hardlinked_bytes = now.hardlinked.saturating_sub(self.hardlinked);
        event.store_copied_bytes = now.copied.saturating_sub(self.copied);
        event.store_copy_cross_device_bytes = now.cross_device.saturating_sub(self.cross_device);
        event.store_copy_permission_bytes = now.permission.saturating_sub(self.permission);
        event.store_copy_ineligible_bytes = now.ineligible.saturating_sub(self.ineligible);
        event.store_copy_other_bytes = now.other.saturating_sub(self.other);
    }
}

// ── Wrapper side ─────────────────────────────────────────────────

/// Outcome of offering a compile to the daemon.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Handoff {
    /// The daemon holds the key and will publish; the wrapper is done.
    Accepted,
    /// The wrapper publishes itself; `reason` is for the debug log.
    Declined(String),
}

/// Offer `request` to the daemon. Bounded by [`PUBLISH_HANDOFF_TIMEOUT`]:
/// a daemon that is slow to answer is one the wrapper should not wait for.
pub(crate) fn hand_off_cc_publish(config: &Config, request: &PublishCcRequest) -> Handoff {
    hand_off_cc_publish_within(config, request, PUBLISH_HANDOFF_TIMEOUT)
}

/// [`hand_off_cc_publish`] with the wait for the daemon supplied.
pub(crate) fn hand_off_cc_publish_within(
    config: &Config,
    request: &PublishCcRequest,
    budget: Duration,
) -> Handoff {
    let _trace = crate::phase_trace::phase("store_handoff");
    let socket = config.socket_path();
    #[cfg(unix)]
    if !socket.exists() {
        return Handoff::Declined("no daemon socket".to_string());
    }
    let receipt = match HandoffReceipt::new(request) {
        Ok(receipt) => receipt,
        Err(error) => return Handoff::Declined(format!("creating handoff receipt: {error:#}")),
    };
    let reply = send_handoff(&socket, request, budget);
    finish_handoff(&receipt, reply)
}

/// The wrapper owns the receipt directory until it has resolved the socket
/// result. The daemon atomically renames pending to accepted before enqueueing.
/// Removing pending cancels a late handler, so it cannot publish after fallback.
struct HandoffReceipt {
    pending: PathBuf,
    accepted: PathBuf,
}

fn receipt_paths(request: &PublishCcRequest) -> Result<(PathBuf, PathBuf)> {
    let parent = request
        .files
        .first()
        .and_then(|file| Path::new(&file.path).parent())
        .context("handoff has no snapshot directory")?;
    Ok((parent.join("pending"), parent.join("accepted")))
}

impl HandoffReceipt {
    fn new(request: &PublishCcRequest) -> Result<Self> {
        let (pending, accepted) = receipt_paths(request)?;
        std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&pending)?;
        Ok(Self { pending, accepted })
    }
}

impl Drop for HandoffReceipt {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.pending);
        let _ = std::fs::remove_file(&self.accepted);
        if let Some(parent) = self.pending.parent() {
            let _ = std::fs::remove_dir(parent);
        }
    }
}

fn accept_receipt(request: &PublishCcRequest) -> Result<()> {
    let (pending, accepted) = receipt_paths(request)?;
    std::fs::rename(pending, accepted).context("handoff cancelled before acceptance")
}

/// The wrapper removes `pending` to cancel. `rename` then fails with
/// [`std::io::ErrorKind::NotFound`]. Every other failure still has a receipt
/// the wrapper may publish from, so the daemon must release the key.
fn receipt_cancelled(error: &anyhow::Error) -> bool {
    error.chain().any(|cause| {
        cause
            .downcast_ref::<std::io::Error>()
            .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
    })
}

fn finish_handoff(receipt: &HandoffReceipt, reply: Result<String>) -> Handoff {
    match std::fs::remove_file(&receipt.pending) {
        Ok(()) => {
            let reason = match reply {
                Ok(line) => match serde_json::from_str::<Response>(&line) {
                    Ok(response) => response.error.unwrap_or_else(|| {
                        "daemon replied without accepting the receipt".to_string()
                    }),
                    Err(error) => format!("unreadable daemon reply: {error}"),
                },
                Err(error) => format!("{error:#}"),
            };
            Handoff::Declined(reason)
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            // The daemon won the rename, and owns both publication and event.
            // The socket reply is no longer needed to establish ownership.
            if receipt.accepted.exists() {
                Handoff::Accepted
            } else {
                Handoff::Declined("handoff receipt disappeared".to_string())
            }
        }
        Err(error) => {
            // Cancellation could not be established. Leave this optional put
            // to the daemon instead of racing it with a second publication.
            tracing::debug!("could not cancel cc handoff receipt: {error}");
            Handoff::Accepted
        }
    }
}

/// Bound connection, writes, and reads by the same deadline. The general
/// daemon client only bounds reads and can spend seconds sending a memo to
/// a stalled daemon.
#[cfg(unix)]
fn send_handoff(socket: &Path, request: &PublishCcRequest, budget: Duration) -> Result<String> {
    use interprocess::local_socket::{ConnectOptions, traits::Stream as _};
    use std::io::{Read, Write};
    let deadline = std::time::Instant::now() + budget;
    let remaining = || -> std::io::Result<Duration> {
        deadline
            .checked_duration_since(std::time::Instant::now())
            .filter(|duration| !duration.is_zero())
            .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::TimedOut, "cc handoff deadline"))
    };
    let mut line = serde_json::to_vec(&crate::daemon::Request::PublishCc(Box::new(
        request.clone(),
    )))?;
    line.push(b'\n');
    let mut stream = ConnectOptions::new()
        .name(crate::transport::socket_name(socket)?)
        .wait_mode(interprocess::ConnectWaitMode::Timeout(remaining()?))
        .connect_sync()?;
    let mut pending = line.as_slice();
    while !pending.is_empty() {
        peer_may_have_closed(stream.set_send_timeout(Some(remaining()?)))?;
        let written = stream.write(pending)?;
        anyhow::ensure!(written != 0, "daemon closed during handoff write");
        pending = &pending[written..];
    }
    let mut response = Vec::new();
    loop {
        peer_may_have_closed(stream.set_recv_timeout(Some(remaining()?)))?;
        let mut buffer = [0; 1024];
        let read = stream.read(&mut buffer)?;
        anyhow::ensure!(read != 0, "daemon closed before handoff reply");
        response.extend_from_slice(&buffer[..read]);
        anyhow::ensure!(response.len() <= 8192, "oversized handoff reply");
        if response.contains(&b'\n') {
            return Ok(String::from_utf8(response)?);
        }
    }
}

/// A socket timeout that could not be set because the daemon already closed
/// its end. macOS reports that as EINVAL from `setsockopt`; the write or
/// read that follows then reports the close as what it is, instead of the
/// handoff failing with "Invalid argument". The timeouts passed here are
/// never zero, so EINVAL has no other cause.
#[cfg(unix)]
fn peer_may_have_closed(result: std::io::Result<()>) -> std::io::Result<()> {
    match result {
        Err(error) if error.raw_os_error() == Some(libc::EINVAL) => Ok(()),
        other => other,
    }
}

#[cfg(windows)]
use send_handoff_windows as send_handoff;

#[cfg(windows)]
fn send_handoff_windows(
    socket: &Path,
    request: &PublishCcRequest,
    budget: Duration,
) -> Result<String> {
    crate::daemon::send_request_with_timeout(
        socket,
        &crate::daemon::Request::PublishCc(Box::new(request.clone())),
        budget,
    )
}

/// Snapshot `files` (each `(source, store_name)`) into the handoff directory
/// so they outlive the wrapper. Returns the daemon-visible list; on any
/// failure the snapshots made so far are removed and the error is returned,
/// leaving the wrapper to publish from its own staging as before.
pub(crate) fn snapshot_for_handoff(
    config: &Config,
    files: &[(PathBuf, String)],
) -> Result<Vec<HandoffFile>> {
    let _trace = crate::phase_trace::phase("handoff_snapshot");
    let dir = handoff_dir(config);
    std::fs::create_dir_all(&dir)
        .with_context(|| format!("creating handoff directory {}", dir.display()))?;
    // Each request has its own directory. PID-based names can collide with
    // a previous request still queued in the daemon, or with a reused PID.
    let batch = tempfile::Builder::new().prefix("cc-").tempdir_in(&dir)?;
    let mut snapshots = Vec::with_capacity(files.len());
    for (index, (source, store_name)) in files.iter().enumerate() {
        let target = batch
            .path()
            .join(format!("{index}-{}", handoff_file_name(store_name)));
        // Inputs are private staging files, never compiler-owned outputs.
        if std::fs::hard_link(source, &target).is_err() {
            std::fs::copy(source, &target)
                .with_context(|| format!("snapshotting {} for the daemon", source.display()))?;
        }
        snapshots.push(HandoffFile {
            path: target.to_string_lossy().into_owned(),
            store_name: store_name.clone(),
        });
    }
    let _ = batch.keep();
    Ok(snapshots)
}

/// A store name flattened to one path component.
fn handoff_file_name(store_name: &str) -> String {
    store_name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// What [`sweep_orphaned_handoffs`] reclaimed.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HandoffSweep {
    pub(crate) removed: usize,
    pub(crate) bytes_reclaimed: u64,
}

/// Whether a request directory whose newest entry changed at `newest` has
/// outlived every legitimate request.
fn handoff_is_abandoned(
    newest: std::time::SystemTime,
    now: std::time::SystemTime,
    grace: Duration,
) -> bool {
    now.duration_since(newest).is_ok_and(|age| age >= grace)
}

/// The latest mtime of `dir` and its entries, without following links. `None`
/// when anything cannot be read, so an unreadable request is left alone.
fn newest_mtime(dir: &Path) -> Option<(std::time::SystemTime, u64)> {
    let mut newest = std::fs::symlink_metadata(dir).ok()?.modified().ok()?;
    let mut bytes = 0;
    for entry in std::fs::read_dir(dir).ok()? {
        let meta = std::fs::symlink_metadata(entry.ok()?.path()).ok()?;
        newest = newest.max(meta.modified().ok()?);
        bytes += meta.len();
    }
    Some((newest, bytes))
}

/// Remove request directories a crash left under the handoff directory.
///
/// A request directory holds a wrapper's snapshots and receipt until the
/// daemon has published them, which takes milliseconds from a bounded queue.
/// One whose newest entry is older than `grace` belongs to no live request:
/// the daemon that accepted it died, or the wrapper died before handing it
/// over. The store's own staging sweep only looks at files and so never
/// reached these. Only `cc-` directories directly under the handoff directory
/// are considered, a link is never followed, and nothing else in the store is
/// walked.
pub(crate) fn sweep_orphaned_handoffs(config: &Config, grace: Duration) -> HandoffSweep {
    let mut sweep = HandoffSweep::default();
    let Ok(entries) = std::fs::read_dir(handoff_dir(config)) else {
        return sweep;
    };
    let now = std::time::SystemTime::now();
    for entry in entries.flatten() {
        let path = entry.path();
        let is_request = entry.file_name().to_string_lossy().starts_with("cc-")
            && std::fs::symlink_metadata(&path).is_ok_and(|meta| meta.is_dir());
        if !is_request {
            continue;
        }
        let Some((newest, bytes)) = newest_mtime(&path) else {
            continue;
        };
        if handoff_is_abandoned(newest, now, grace) && std::fs::remove_dir_all(&path).is_ok() {
            sweep.removed += 1;
            sweep.bytes_reclaimed += bytes;
        }
    }
    sweep
}

pub(crate) fn remove_handoff_files(files: &[HandoffFile]) {
    for file in files {
        let _ = std::fs::remove_file(&file.path);
        if let Some(parent) = Path::new(&file.path).parent() {
            // Only empty request directories are removed; other snapshots
            // in the batch keep the directory alive until their turn.
            let _ = std::fs::remove_dir(parent);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_request_is_abandoned_exactly_one_grace_after_its_last_change() {
        let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
        let grace = Duration::from_secs(3600);
        assert!(!handoff_is_abandoned(now, now, grace));
        // One millisecond, not one nanosecond: Windows keeps SystemTime in
        // 100 ns steps and would round a nanosecond away.
        assert!(!handoff_is_abandoned(
            now - grace + Duration::from_millis(1),
            now,
            grace
        ));
        assert!(handoff_is_abandoned(now - grace, now, grace));
        // An entry dated after the clock (a skewed filesystem) is kept.
        assert!(!handoff_is_abandoned(
            now + Duration::from_secs(1),
            now,
            grace
        ));
    }

    #[test]
    fn the_handoff_sweep_reclaims_only_old_request_directories() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let root = handoff_dir(&config);
        let request = |name: &str| {
            let path = root.join(name);
            std::fs::create_dir_all(&path).unwrap();
            std::fs::write(path.join("0-foo.o"), b"object").unwrap();
            std::fs::write(path.join("accepted"), b"").unwrap();
            path
        };
        let hours_ago = filetime::FileTime::from_system_time(
            std::time::SystemTime::now() - Duration::from_secs(2 * 3600),
        );
        let age = |path: &Path| {
            for entry in std::fs::read_dir(path).unwrap() {
                filetime::set_file_mtime(entry.unwrap().path(), hours_ago).unwrap();
            }
            filetime::set_file_mtime(path, hours_ago).unwrap();
        };

        let abandoned = request("cc-dead");
        age(&abandoned);
        let live = request("cc-live");
        // Old directory, but one file is new: a request still being written.
        let mixed = request("cc-mixed");
        age(&mixed);
        std::fs::write(mixed.join("1-bar.o"), b"new").unwrap();
        // Not a request directory, however old.
        let other = request("keep-me");
        age(&other);
        let stray = root.join("cc-file");
        std::fs::write(&stray, b"x").unwrap();
        filetime::set_file_mtime(&stray, hours_ago).unwrap();

        let sweep = sweep_orphaned_handoffs(&config, Duration::from_secs(3600));

        assert_eq!(sweep.removed, 1, "{sweep:?}");
        assert_eq!(sweep.bytes_reclaimed, b"object".len() as u64);
        assert!(!abandoned.exists());
        assert!(live.exists());
        assert!(mixed.exists());
        assert!(other.exists());
        assert!(stray.exists());
    }

    #[cfg(unix)]
    #[test]
    fn the_handoff_sweep_never_follows_a_link_out_of_the_store() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let root = handoff_dir(&config);
        std::fs::create_dir_all(&root).unwrap();
        let outside = dir.path().join("outside");
        std::fs::create_dir(&outside).unwrap();
        std::fs::write(outside.join("precious"), b"keep").unwrap();
        std::os::unix::fs::symlink(&outside, root.join("cc-link")).unwrap();
        let old = filetime::FileTime::from_unix_time(1_000_000_000, 0);
        filetime::set_symlink_file_times(root.join("cc-link"), old, old).unwrap();

        let sweep = sweep_orphaned_handoffs(&config, Duration::from_secs(3600));

        assert_eq!(sweep.removed, 0);
        assert!(outside.join("precious").exists());
    }

    #[test]
    fn handoff_paths_must_sit_under_the_store_handoff_dir() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let inside = handoff_dir(&config).join("1-0-foo.o");
        assert!(handoff_path_is_owned(&config, &inside));
        assert!(!handoff_path_is_owned(&config, Path::new("foo.o")));
        assert!(!handoff_path_is_owned(
            &config,
            &dir.path().join("elsewhere.o")
        ));
        let escaping = handoff_dir(&config).join("..").join("..").join("index.db");
        assert!(!handoff_path_is_owned(&config, &escaping));
    }

    #[test]
    fn snapshots_land_in_the_handoff_dir_and_are_removed_together_on_failure() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let good = dir.path().join("a.o");
        std::fs::write(&good, b"object").unwrap();
        let snapshots =
            snapshot_for_handoff(&config, &[(good.clone(), "a.o".to_string())]).unwrap();
        assert_eq!(snapshots.len(), 1);
        let path = Path::new(&snapshots[0].path);
        assert!(handoff_path_is_owned(&config, path));
        assert_eq!(std::fs::read(path).unwrap(), b"object");
        assert!(
            path.file_name()
                .unwrap()
                .to_str()
                .unwrap()
                .ends_with("0-a.o")
        );

        let missing = dir.path().join("missing.o");
        let error = snapshot_for_handoff(
            &config,
            &[
                (good.clone(), "a.o".to_string()),
                (missing, "sub/dir/b.o".to_string()),
            ],
        )
        .unwrap_err();
        assert!(error.to_string().contains("missing.o"), "{error:#}");
        let leftovers: Vec<_> = std::fs::read_dir(handoff_dir(&config))
            .unwrap()
            .map(|entry| entry.unwrap().file_name())
            .filter(|name| name != path.parent().unwrap().file_name().unwrap())
            .collect();
        assert!(leftovers.is_empty(), "{leftovers:?}");
        assert_eq!(std::fs::read(path).unwrap(), b"object");
        remove_handoff_files(&snapshots);
        assert!(!path.exists());
    }

    #[test]
    fn store_names_flatten_to_one_component() {
        assert_eq!(handoff_file_name("foo.o"), "foo.o");
        assert_eq!(handoff_file_name("sub/dir/foo.d"), "sub_dir_foo.d");
        assert_eq!(handoff_file_name("../x"), ".._x");
        assert_eq!(handoff_file_name("lib-a_b.o"), "lib-a_b.o");
    }

    fn handoff_request(config: &Config, key: &str, dir: &Path) -> PublishCcRequest {
        let object = dir.join("a.o");
        std::fs::write(&object, b"object bytes").unwrap();
        let files = snapshot_for_handoff(config, &[(object, "a.o".to_string())]).unwrap();
        PublishCcRequest {
            client_epoch: 0,
            client_version: None,
            cache_key: key.to_string(),
            crate_name: "a.c".to_string(),
            target: "x86_64".to_string(),
            files,
            stdout: String::new(),
            stderr: "warning: w\n".to_string(),
            compile_time_ms: 7,
            publishes_to_remote: false,
            event: crate::events::BuildEvent::new_for_test("a.c", EventResult::Miss),
            memo: None,
        }
    }

    fn key(label: &str) -> String {
        blake3::hash(label.as_bytes()).to_hex().to_string()
    }

    /// A read-only daemon stores every C/C++ compile handed to it and queues
    /// none for upload. Only a writable client marks its compile for the
    /// remote, so only that skip warns.
    #[test]
    fn a_read_only_daemon_warns_only_about_a_writable_clients_compile() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = crate::test_support::test_config(dir.path().join("cache"));
        config.remote = Some(crate::config::RemoteConfig::test_s3("bucket", "prefix"));
        config.remote_readonly = true;
        config.auto_gc = false;
        let store = Store::open(&config).unwrap();
        for (label, writable_client) in [("read-only client", false), ("writable client", true)] {
            let daemon = Arc::new(Daemon::new(config.clone()));
            let mut request = handoff_request(&config, &key(label), dir.path());
            request.publishes_to_remote = writable_client;
            let cache_key = request.cache_key.clone();
            let BuildClaim::Acquired(lock) = store.claim_build(&cache_key).unwrap() else {
                panic!("fresh key");
            };
            publish_one(
                &daemon,
                &config,
                &store,
                PublishJob {
                    request,
                    _lock: lock,
                },
            );
            assert!(store.get(&cache_key).unwrap().is_some(), "{label}: stored");
            assert_eq!(
                daemon.readonly_skip_warned_at().is_some(),
                writable_client,
                "{label}"
            );
        }
        let queued = std::fs::read_dir(config.upload_spool_dir()).map_or(0, Iterator::count);
        assert_eq!(queued, 0, "nothing is queued for upload");
    }

    /// The worker stores what the handler accepted: the entry is committed
    /// under the key, the event carries the daemon's store fields, and the
    /// snapshots are gone.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn an_accepted_handoff_is_stored_logged_and_cleaned_up() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, rx) = tokio::sync::mpsc::channel(4);
        daemon.publish_queue().set_sender(tx);
        let worker = spawn_publish_worker(Arc::clone(&daemon), rx).unwrap();
        let mut request = handoff_request(&config, &key("accepted"), dir.path());
        let _request_receipt = HandoffReceipt::new(&request).unwrap();
        let header = dir.path().join("a.h");
        std::fs::write(&header, b"#define A 1\n").unwrap();
        let fingerprint = crate::cache_key::FileFingerprint::from_path(&header).unwrap();
        let settled = crate::cache_key::CcPreprocessMemoInput {
            name: "a.h".to_string(),
            observed_ns: fingerprint.mtime_ns.max(fingerprint.ctime_ns)
                + crate::cache_key::HASH_SETTLE_NS,
            fingerprint,
            content: "c".repeat(64),
            mapped: "d".repeat(64),
        };
        // From a wrapper that sends no observation time.
        let unobserved = crate::cache_key::CcPreprocessMemoInput {
            name: "b.h".to_string(),
            observed_ns: 0,
            ..settled.clone()
        };
        let memo = CcMemoHandoff {
            memo_key: "m".repeat(64),
            preprocessed_hash: "p".repeat(64),
            inputs: vec![settled.clone(), unobserved.clone()],
        };
        request.memo = Some(memo.clone());
        request.event.schema = 26;
        request.event.rebuilt_package = Some("native-package".into());
        request.event.rebuilt_fingerprint = Some(request.cache_key.clone());
        request.event.rebuilt_path = Some("/actual/compiler/a.o".into());
        request.event.rebuilt_paths =
            vec!["/actual/compiler/a.o".into(), "/actual/compiler/a.d".into()];
        let snapshot = PathBuf::from(&request.files[0].path);
        let response = daemon.handle_publish_cc(request.clone()).await;
        assert!(response.ok, "{:?}", response.error);
        daemon.publish_queue().close();
        tokio::time::timeout(Duration::from_secs(10), worker)
            .await
            .expect("closing the queue lets the worker drain and exit")
            .unwrap();

        let store = Store::open(&config).unwrap();
        let meta = store
            .get(&request.cache_key)
            .unwrap()
            .expect("entry committed");
        assert_eq!(meta.files.len(), 1);
        assert_eq!(meta.files[0].name, "a.o");
        assert_eq!(meta.stderr, "warning: w\n");
        let recorded = store
            .file_hash_cache()
            .get_cc_preprocess_memo(&memo.memo_key)
            .unwrap()
            .expect("the memo is recorded with the entry");
        assert_eq!(recorded.preprocessed_hash, memo.preprocessed_hash);
        assert_eq!(recorded.inputs.len(), 2);
        assert_eq!(
            recorded.inputs[0],
            crate::cache_key::CcPreprocessMemoInput {
                observed_ns: 0,
                ..settled
            }
        );
        assert_eq!(recorded.inputs[1].content, unobserved.content);
        assert_ne!(
            recorded.inputs[1].fingerprint, unobserved.fingerprint,
            "a stamp with no observation time is not recorded"
        );
        assert!(
            !snapshot.exists(),
            "the daemon removes the snapshot after the put"
        );
        let events = std::fs::read_to_string(config.event_log_path()).unwrap();
        let event: BuildEvent = serde_json::from_str(events.lines().last().unwrap()).unwrap();
        assert!(event.store_handed_off);
        assert_eq!(event.schema, 26);
        assert_eq!(event.rebuilt_package, request.event.rebuilt_package);
        assert_eq!(event.rebuilt_fingerprint, request.event.rebuilt_fingerprint);
        assert_eq!(event.rebuilt_path, request.event.rebuilt_path);
        assert_eq!(event.rebuilt_paths, request.event.rebuilt_paths);
        assert_eq!(event.result, EventResult::Miss);
        assert_eq!(event.store_new_blobs, 1);
        assert!(event.store_error.is_empty());
        // The put's bytes are charged to the event, however they landed.
        assert!(
            event.store_reflinked_bytes + event.store_hardlinked_bytes + event.store_copied_bytes
                > 0,
            "{event:?}"
        );
    }

    /// Only an entry the wrapper marked for the remote is queued for upload,
    /// even when a remote is configured.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn only_an_entry_marked_for_the_remote_is_queued_for_upload() {
        const CHILD_ROOT: &str = "KACHE_TEST_REMOTE_PUBLISH_ROOT";
        let Some(root) = std::env::var_os(CHILD_ROOT) else {
            // A stuck publish_one owns a blocking task that runtime teardown
            // cannot cancel. Keep the worker in a process we can kill and reap.
            let dir = tempfile::tempdir().unwrap();
            let log_path = dir.path().join("publisher.log");
            let log = std::fs::File::create(&log_path).unwrap();
            struct FixtureChild(std::process::Child);
            impl Drop for FixtureChild {
                fn drop(&mut self) {
                    if !matches!(self.0.try_wait(), Ok(Some(_))) {
                        let _ = self.0.kill();
                    }
                    let _ = self.0.wait();
                }
            }
            let mut child = FixtureChild(
                std::process::Command::new(std::env::current_exe().unwrap())
                    .args([
                        "--exact",
                        "daemon_publish::tests::only_an_entry_marked_for_the_remote_is_queued_for_upload",
                        "--nocapture",
                        "--test-threads=1",
                    ])
                    .env(CHILD_ROOT, dir.path())
                    .current_dir(dir.path())
                    .stdout(log.try_clone().unwrap())
                    .stderr(log)
                    .spawn()
                    .unwrap(),
            );
            let deadline = std::time::Instant::now() + Duration::from_secs(30);
            loop {
                match child.0.try_wait() {
                    Ok(Some(status)) => {
                        child.0.wait().unwrap();
                        let log = std::fs::read(&log_path).unwrap();
                        let log = String::from_utf8_lossy(&log);
                        assert!(
                            status.success(),
                            "remote publish fixture failed: {status}\n{log}"
                        );
                        assert_eq!(
                            std::fs::read(dir.path().join("publisher-complete")).unwrap(),
                            b"ok",
                            "the child must finish the upload assertions"
                        );
                        return;
                    }
                    Ok(None) if std::time::Instant::now() < deadline => {
                        std::thread::sleep(Duration::from_millis(10));
                    }
                    result => {
                        let killed = child.0.kill();
                        let waited = child.0.wait();
                        let log = std::fs::read(&log_path).unwrap_or_default();
                        let log = String::from_utf8_lossy(&log);
                        panic!(
                            "remote publish fixture did not finish: {result:?}; \
                             kill: {killed:?}; wait: {waited:?}\n{log}"
                        );
                    }
                }
            }
        };
        let dir = PathBuf::from(root);
        let mut config = crate::test_support::test_config(dir.join("cache"));
        config.remote = Some(crate::config::RemoteConfig::test_s3("bucket", "artifacts"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (upload_tx, mut uploads) = tokio::sync::mpsc::unbounded_channel();
        daemon.set_upload_tx(upload_tx);
        for (label, publishes_to_remote) in [("local-only", false), ("shared", true)] {
            let mut request = handoff_request(&config, &key(label), &dir);
            request.publishes_to_remote = publishes_to_remote;
            let (daemon, config) = (Arc::clone(&daemon), config.clone());
            tokio::task::spawn_blocking(move || {
                let store = Store::open(&config).unwrap();
                let BuildClaim::Acquired(lock) = store.claim_build(&request.cache_key).unwrap()
                else {
                    panic!("fresh key");
                };
                publish_one(
                    &daemon,
                    &config,
                    &store,
                    PublishJob {
                        request,
                        _lock: lock,
                    },
                );
            })
            .await
            .unwrap();
        }
        let queued = uploads.try_recv().expect("the shared entry is queued");
        assert_eq!(queued.key, key("shared"));
        assert!(uploads.try_recv().is_err(), "the local-only entry is not");
        std::fs::write(dir.join("publisher-complete"), b"ok").unwrap();
    }

    /// The wrapper's side and the daemon's connection handler, over a real
    /// socket: the request reaches the publish queue, the wrapper sees its
    /// receipt accepted, and a wrapper built after the daemon starts its
    /// restart like any other request would.
    #[cfg(unix)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_handoff_over_the_socket_reaches_the_publish_queue() {
        use interprocess::local_socket::traits::tokio::Listener as _;
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        daemon.publish_queue().set_sender(tx);
        let socket = config.socket_path();
        std::fs::create_dir_all(socket.parent().unwrap()).unwrap();
        let listener = crate::transport::ListenerOptions::new()
            .name(crate::transport::socket_name(&socket).unwrap())
            .create_tokio()
            .unwrap();
        let lifecycle = Arc::new(kunobi_daemon::Lifecycle::default());
        let server = {
            let (daemon, lifecycle) = (Arc::clone(&daemon), Arc::clone(&lifecycle));
            tokio::spawn(async move {
                let stream = listener.accept().await.unwrap();
                crate::daemon::handle_connection(stream, &daemon, &lifecycle).await
            })
        };
        let mut request = handoff_request(&config, &key("socket"), dir.path());
        request.client_epoch = 0;
        request.client_version = Some("999.0.0".into());
        let outcome = {
            let (config, request) = (config.clone(), request.clone());
            tokio::task::spawn_blocking(move || {
                hand_off_cc_publish_within(&config, &request, Duration::from_secs(10))
            })
            .await
            .unwrap()
        };
        assert_eq!(outcome, Handoff::Accepted);
        let job = tokio::time::timeout(Duration::from_secs(10), rx.recv())
            .await
            .expect("the accepted job is queued")
            .unwrap();
        assert_eq!(job.request.cache_key, request.cache_key);
        drop(job);
        tokio::time::timeout(Duration::from_secs(10), server)
            .await
            .expect("the handler ends with the connection")
            .unwrap()
            .unwrap();
        assert!(
            !lifecycle.accepting_calls(),
            "a newer wrapper drains the daemon"
        );
    }

    /// Refusals never take the key: a full queue, a foreign path, a key a
    /// peer holds, and an invalid key all leave the wrapper to publish.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn refusals_leave_the_key_free_and_the_snapshots_alone() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let store = Store::open(&config).unwrap();

        // No sender yet: closed.
        let request = handoff_request(&config, &key("closed"), dir.path());
        let _request_receipt = HandoffReceipt::new(&request).unwrap();
        let response = daemon.handle_publish_cc(request.clone()).await;
        assert!(!response.ok);
        assert!(response.error.unwrap().contains("queue closed"));
        assert!(Path::new(&request.files[0].path).exists());
        assert!(matches!(
            store.claim_build(&request.cache_key).unwrap(),
            BuildClaim::Acquired(_)
        ));

        // A queue of one: the second hand-off finds it full.
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        daemon.publish_queue().set_sender(tx);
        let first = handoff_request(&config, &key("first"), dir.path());
        let _first_receipt = HandoffReceipt::new(&first).unwrap();
        assert!(daemon.handle_publish_cc(first.clone()).await.ok);
        let second = handoff_request(&config, &key("second"), dir.path());
        let _second_receipt = HandoffReceipt::new(&second).unwrap();
        let response = daemon.handle_publish_cc(second.clone()).await;
        assert!(!response.ok);
        assert!(response.error.unwrap().contains("queue full"));
        assert!(matches!(
            store.claim_build(&second.cache_key).unwrap(),
            BuildClaim::Acquired(_)
        ));
        // The accepted one holds its key until the worker is done with it.
        assert!(matches!(
            store.claim_build(&first.cache_key).unwrap(),
            BuildClaim::Contended
        ));
        let job = rx.recv().await.unwrap();
        drop(job);
        assert!(matches!(
            store.claim_build(&first.cache_key).unwrap(),
            BuildClaim::Acquired(_)
        ));

        // A key a peer holds.
        let held = handoff_request(&config, &key("held"), dir.path());
        let _held_receipt = HandoffReceipt::new(&held).unwrap();
        let _peer = match store.claim_build(&held.cache_key).unwrap() {
            BuildClaim::Acquired(lock) => lock,
            _ => panic!("fresh key"),
        };
        let response = daemon.handle_publish_cc(held).await;
        assert!(!response.ok);
        assert!(response.error.unwrap().contains("contended"));

        // A file outside the handoff directory.
        let mut foreign = handoff_request(&config, &key("foreign"), dir.path());
        let _foreign_receipt = HandoffReceipt::new(&foreign).unwrap();
        foreign.files[0].path = dir.path().join("a.o").to_string_lossy().into_owned();
        let response = daemon.handle_publish_cc(foreign).await;
        assert!(!response.ok);
        assert!(response.error.unwrap().contains("outside"));

        let mut invalid = handoff_request(&config, &key("invalid"), dir.path());
        let _invalid_receipt = HandoffReceipt::new(&invalid).unwrap();
        invalid.cache_key = "nope".to_string();
        assert!(!daemon.handle_publish_cc(invalid).await.ok);
    }

    /// The progress a queue hands out is the one its jobs update, so the
    /// shutdown drain sees them.
    #[test]
    fn a_queue_hands_out_the_progress_its_jobs_update() {
        let queue = PublishQueue::new();
        let seen = queue.progress();
        queue.progress.accepted();
        assert_eq!(seen.pending(), 1);
        queue.progress.finished();
        assert_eq!((seen.pending(), seen.done()), (0, 1));
    }

    /// A queue that takes longer than any fixed budget to drain is still
    /// drained, as long as the worker keeps finishing jobs.
    #[tokio::test]
    async fn shutdown_waits_for_a_slow_but_moving_worker() {
        let progress = Arc::new(PublishProgress::default());
        for _ in 0..10 {
            progress.accepted();
        }
        let (done_tx, done) = tokio::sync::oneshot::channel();
        let worker = Arc::clone(&progress);
        tokio::spawn(async move {
            // Ten jobs, each slower than half the stall window: the whole
            // queue takes several stall windows, but no single gap does.
            for _ in 0..10 {
                tokio::time::sleep(Duration::from_millis(60)).await;
                worker.finished();
            }
            let _ = done_tx.send(());
        });
        let left = drain_publications(done, &progress, Duration::from_millis(150)).await;
        assert_eq!(left, 0);
        assert_eq!(progress.done(), 10);
    }

    /// A worker that finishes nothing for the stall window is given up on,
    /// and the jobs it never reached are counted.
    #[tokio::test]
    async fn shutdown_gives_up_on_a_wedged_worker_and_counts_what_is_left() {
        let progress = Arc::new(PublishProgress::default());
        for _ in 0..3 {
            progress.accepted();
        }
        progress.finished();
        let (_done_tx, done) = tokio::sync::oneshot::channel::<()>();
        let started = std::time::Instant::now();
        let left = drain_publications(done, &progress, Duration::from_millis(100)).await;
        assert_eq!(left, 2);
        assert!(started.elapsed() < Duration::from_secs(5));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn client_cleanup_after_a_lost_reply_cannot_remove_queued_outputs() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, mut rx) = tokio::sync::mpsc::channel(2);
        daemon.publish_queue().set_sender(tx);
        let request = handoff_request(&config, &key("lost-reply"), dir.path());
        let receipt = HandoffReceipt::new(&request).unwrap();
        assert!(daemon.handle_publish_cc(request.clone()).await.ok);
        assert_eq!(
            finish_handoff(&receipt, Err(anyhow::anyhow!("lost socket reply"))),
            Handoff::Accepted,
        );
        drop(receipt);
        // The reply is lost. The client times out, removes its snapshots,
        // and a subsequent invocation stages different bytes under its PID.
        remove_handoff_files(&request.files);
        let replacement = dir.path().join("replacement.o");
        std::fs::write(&replacement, b"different object").unwrap();
        let later = snapshot_for_handoff(&config, &[(replacement, "a.o".into())]).unwrap();
        let job = rx.recv().await.unwrap();
        assert_ne!(job.request.files, request.files);
        assert_eq!(
            std::fs::read(&job.request.files[0].path).unwrap(),
            b"object bytes"
        );
        let store = Store::open(&config).unwrap();
        publish_one(&daemon, &config, &store, job);
        assert!(store.get(&request.cache_key).unwrap().is_some());
        let events = crate::events::read_events(&config.event_log_path()).unwrap();
        assert_eq!(events.len(), 1);
        assert!(events[0].store_handed_off);
        assert!(events[0].store_error.is_empty());
        assert_eq!(std::fs::read(&later[0].path).unwrap(), b"different object");
        remove_handoff_files(&later);
        assert_eq!(std::fs::read_dir(handoff_dir(&config)).unwrap().count(), 0);
    }

    /// The wrapper's 25 ms budget ran out and it removed the receipt. The
    /// daemon already holds the key, so it stores the snapshot itself. While
    /// that job is queued the wrapper's reclaim sees the key held; after the
    /// put it sees a commit and does not store a second copy.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_cancelled_receipt_is_published_when_the_daemon_holds_the_key() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        daemon.publish_queue().set_sender(tx);
        let request = handoff_request(&config, &key("cancelled"), dir.path());
        let receipt = HandoffReceipt::new(&request).unwrap();
        assert_eq!(
            finish_handoff(&receipt, Err(anyhow::anyhow!("timed out"))),
            Handoff::Declined("timed out".to_string()),
        );
        let response = daemon.handle_publish_cc(request.clone()).await;
        assert!(response.ok, "{response:?}");
        let job = rx.try_recv().expect("the cancelled hand-off is queued");
        assert_eq!(job.request.cache_key, request.cache_key);
        assert_ne!(job.request.files, request.files);
        assert_eq!(
            std::fs::read(&job.request.files[0].path).unwrap(),
            b"object bytes"
        );
        let store = Store::open(&config).unwrap();
        assert!(matches!(
            store.claim_build(&request.cache_key).unwrap(),
            BuildClaim::Contended
        ));
        publish_one(&daemon, &config, &store, job);
        assert!(store.get(&request.cache_key).unwrap().is_some());
        assert!(matches!(
            store.claim_build(&request.cache_key).unwrap(),
            BuildClaim::Committed(_)
        ));
        let events = crate::events::read_events(&config.event_log_path()).unwrap();
        assert_eq!(events.len(), 1);
        assert!(
            events[0].store_error.is_empty(),
            "{:?}",
            events[0].store_error
        );
        drop(receipt);
        assert_eq!(std::fs::read_dir(handoff_dir(&config)).unwrap().count(), 0);
    }

    #[cfg(unix)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_receipt_rename_that_is_not_cancellation_still_frees_the_key() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        daemon.publish_queue().set_sender(tx);
        let request = handoff_request(&config, &key("rename-denied"), dir.path());
        let receipt = HandoffReceipt::new(&request).unwrap();
        let parent = receipt.pending.parent().unwrap();
        let original = std::fs::metadata(parent).unwrap().permissions();
        let mut readonly = original.clone();
        readonly.set_readonly(true);
        std::fs::set_permissions(parent, readonly).unwrap();
        let response = daemon.handle_publish_cc(request.clone()).await;
        std::fs::set_permissions(parent, original).unwrap();
        assert!(!response.ok, "{response:?}");
        assert!(response.error.unwrap().contains("snapshot failed"));
        assert!(matches!(
            rx.try_recv(),
            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
        ));
        let store = Store::open(&config).unwrap();
        assert!(matches!(
            store.claim_build(&request.cache_key).unwrap(),
            BuildClaim::Acquired(_)
        ));
        drop(receipt);
        remove_handoff_files(&request.files);
    }

    /// A wrapper that withdrew removed its receipt, then its files. Only
    /// that pair is routine; files missing under a live receipt still warn.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn files_gone_after_the_receipt_is_a_routine_withdrawal() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let daemon = Arc::new(Daemon::new(config.clone()));
        let (tx, mut rx) = tokio::sync::mpsc::channel(2);
        daemon.publish_queue().set_sender(tx);
        for withdrawn in [true, false] {
            let request = handoff_request(&config, &key(&format!("gone-{withdrawn}")), dir.path());
            let receipt = HandoffReceipt::new(&request).unwrap();
            if withdrawn {
                std::fs::remove_file(&receipt.pending).unwrap();
            }
            remove_handoff_files(&request.files);
            let error = daemon
                .handle_publish_cc(request.clone())
                .await
                .error
                .unwrap();
            assert_eq!(refusal_is_routine(&error), withdrawn, "{error}");
            assert_eq!(error.contains("snapshot failed"), !withdrawn, "{error}");
            let store = Store::open(&config).unwrap();
            assert!(matches!(
                store.claim_build(&request.cache_key).unwrap(),
                BuildClaim::Acquired(_)
            ));
        }
        assert!(rx.try_recv().is_err());
        assert!(!refusal_is_routine(WITHDRAWN));
        assert!(!refusal_is_routine("publish refused: queue full"));
    }

    #[test]
    fn only_a_missing_receipt_counts_as_cancellation() {
        let cancelled =
            anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::NotFound, "rename"))
                .context("handoff cancelled before acceptance");
        let denied = anyhow::Error::from(std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            "rename",
        ))
        .context("handoff cancelled before acceptance");
        assert!(receipt_cancelled(&cancelled));
        assert!(!receipt_cancelled(&denied));
        assert!(!receipt_cancelled(&anyhow::Error::msg(
            "handoff cancelled before acceptance"
        )));
    }

    #[test]
    fn cancellation_resolves_refused_invalid_and_unclaimed_replies() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        for (reply, reason) in [
            ("{\"ok\":false,\"error\":\"queue closed\"}", "queue closed"),
            ("not json", "unreadable daemon reply"),
            ("{\"ok\":true}", "without accepting the receipt"),
        ] {
            let request = handoff_request(&config, &key(reply), dir.path());
            let receipt = HandoffReceipt::new(&request).unwrap();
            assert!(
                matches!(finish_handoff(&receipt, Ok(reply.to_string())), Handoff::Declined(message) if message.contains(reason))
            );
            assert!(!receipt.pending.exists());
            assert!(!receipt.accepted.exists());
            assert!(accept_receipt(&request).is_err());
        }
    }

    #[test]
    fn receipt_creation_never_cleans_up_an_existing_owner() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let mut request = handoff_request(&config, &key("receipt-owner"), dir.path());
        let receipt = HandoffReceipt::new(&request).unwrap();
        assert!(HandoffReceipt::new(&request).is_err());
        assert!(receipt.pending.exists());
        accept_receipt(&request).unwrap();
        assert_eq!(
            finish_handoff(&receipt, Ok("{\"ok\":true}".into())),
            Handoff::Accepted
        );
        let accepted = receipt.accepted.clone();
        drop(receipt);
        assert!(!accepted.exists());
        request.files.clear();
        assert!(HandoffReceipt::new(&request).is_err());
    }

    #[test]
    fn vanished_receipt_declines_but_failed_cancellation_does_not_race_the_daemon() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let request = handoff_request(&config, &key("receipt-errors"), dir.path());
        let receipt = HandoffReceipt::new(&request).unwrap();
        std::fs::remove_file(&receipt.pending).unwrap();
        assert_eq!(
            finish_handoff(&receipt, Err(anyhow::anyhow!("socket error"))),
            Handoff::Declined("handoff receipt disappeared".into())
        );
        std::fs::create_dir(&receipt.pending).unwrap();
        assert_eq!(
            finish_handoff(&receipt, Err(anyhow::anyhow!("socket error"))),
            Handoff::Accepted
        );
    }

    #[test]
    fn receipt_protocol_does_not_reach_legacy_publish_handlers() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let request = handoff_request(&config, &key("protocol"), dir.path());
        let value =
            serde_json::to_value(crate::daemon::Request::PublishCc(Box::new(request))).unwrap();
        assert!(value.get("publish_cc_v2").is_some());
        assert!(serde_json::from_value::<crate::daemon::Request>(value.clone()).is_ok());
        let legacy = serde_json::json!({"publish_cc": value["publish_cc_v2"]});
        assert!(serde_json::from_value::<crate::daemon::Request>(legacy).is_err());
    }

    #[cfg(unix)]
    #[test]
    fn handoff_budget_covers_a_daemon_that_never_reads_the_request() {
        use std::os::unix::net::UnixListener;
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let socket = dir.path().join("stalled.sock");
        let listener = UnixListener::bind(&socket).unwrap();
        let (done_tx, done_rx) = std::sync::mpsc::channel();
        let server = std::thread::spawn(move || {
            let (_stream, _) = listener.accept().unwrap();
            done_rx.recv_timeout(Duration::from_secs(3)).unwrap();
        });
        let mut request = handoff_request(&config, &key("stalled"), dir.path());
        request.stderr = "x".repeat(1024 * 1024);
        let start = std::time::Instant::now();
        assert!(send_handoff(&socket, &request, Duration::from_millis(500)).is_err());
        let elapsed = start.elapsed();
        // A client that gave up before connecting would leave the server
        // in `accept` for good; a connection of our own releases it.
        let _unblock = std::os::unix::net::UnixStream::connect(&socket);
        done_tx.send(()).unwrap();
        server.join().unwrap();
        assert!(elapsed < Duration::from_secs(2), "handoff took {elapsed:?}");
    }

    #[cfg(unix)]
    #[test]
    fn handoff_waits_for_a_complete_reply_and_rejects_eof() {
        use std::io::{BufRead, Write};
        use std::os::unix::net::UnixListener;
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        for reply in [b"{\"ok\":true}\n".as_slice(), b""] {
            let socket = dir.path().join("reply.sock");
            let listener = UnixListener::bind(&socket).unwrap();
            let server = std::thread::spawn(move || {
                let (mut stream, _) = listener.accept().unwrap();
                let mut request = String::new();
                std::io::BufReader::new(&stream)
                    .read_line(&mut request)
                    .unwrap();
                assert!(request.contains("publish_cc"));
                stream.write_all(reply).unwrap();
            });
            let request = handoff_request(&config, &key("reply"), dir.path());
            let result = send_handoff(&socket, &request, Duration::from_secs(1));
            if reply.is_empty() {
                assert!(
                    result
                        .unwrap_err()
                        .to_string()
                        .contains("before handoff reply")
                );
            } else {
                assert_eq!(result.unwrap(), "{\"ok\":true}\n");
            }
            server.join().unwrap();
            std::fs::remove_file(socket).unwrap();
        }
    }

    #[cfg(unix)]
    #[test]
    fn a_timeout_refused_by_a_closed_peer_leaves_the_close_to_the_next_read() {
        let einval = std::io::Error::from_raw_os_error(libc::EINVAL);
        assert!(peer_may_have_closed(Err(einval)).is_ok());
        let other = std::io::Error::from_raw_os_error(libc::EBADF);
        assert_eq!(
            peer_may_have_closed(Err(other)).unwrap_err().raw_os_error(),
            Some(libc::EBADF)
        );
        assert!(peer_may_have_closed(Ok(())).is_ok());
    }

    #[cfg(unix)]
    #[test]
    fn handing_off_without_a_socket_declines_at_once() {
        let dir = tempfile::tempdir().unwrap();
        let config = crate::test_support::test_config(dir.path().join("cache"));
        let request = PublishCcRequest {
            client_epoch: 0,
            client_version: None,
            cache_key: "k".repeat(64),
            crate_name: "a.c".to_string(),
            target: String::new(),
            files: Vec::new(),
            stdout: String::new(),
            stderr: String::new(),
            compile_time_ms: 1,
            publishes_to_remote: false,
            event: crate::events::BuildEvent::new_for_test("a.c", crate::events::EventResult::Miss),
            memo: None,
        };
        let started = std::time::Instant::now();
        let outcome = hand_off_cc_publish(&config, &request);
        assert!(matches!(outcome, Handoff::Declined(reason) if reason.contains("socket")));
        assert!(started.elapsed() < Duration::from_millis(100));
    }
}