runner-manager-agent 0.4.62

//! Native Linux companion. Request bodies and credentials never enter the
//! journal. Streaming endpoints are relayed without buffering their payloads.

use std::io::{Read, Write};
use std::net::Shutdown;
use std::os::unix::fs::{FileTypeExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use super::http::{Head, MAX_JSON};
use super::*;

pub fn launch(
    cleanup: &DockerCleanup,
    attempt: AttemptId,
    repository: &str,
    socket: &Path,
    listen: Option<(
        PathBuf,
        runner_manager_platform::runner_account::RunnerAccount,
    )>,
    policy: Option<policy::Context>,
) -> io::Result<Option<std::ffi::OsString>> {
    if !socket.exists() {
        return Ok(None);
    }
    if !fs::metadata(socket)?.file_type().is_socket() {
        return Err(invalid("Docker endpoint is not a socket"));
    }
    fs::create_dir_all(&cleanup.root)?;
    fs::set_permissions(&cleanup.root, fs::Permissions::from_mode(0o700))?;
    let directory = cleanup.root.join(attempt.to_string());
    // An attempt gets exactly one proxy; never truncate a surviving journal.
    fs::create_dir(&directory)?;
    fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?;
    let path = directory.join("journal.json");
    let mut journal = Journal::new(
        cleanup.host,
        attempt,
        repository.into(),
        socket.to_path_buf(),
        PENDING_ENGINE.into(),
    );
    if let Some((listen_directory, account)) = &listen {
        // Created here, as the daemon, and handed to the runner's account with
        // the socket inside it, so nothing of the runner's decides the path.
        // A leftover at the name is accepted only as a real directory: the
        // attempt is the runner's account's, and `chmod` follows a link.
        match fs::create_dir(listen_directory) {
            Ok(()) => {}
            Err(error)
                if error.kind() == io::ErrorKind::AlreadyExists
                    && fs::symlink_metadata(listen_directory)?.is_dir() => {}
            Err(error) => return Err(error),
        }
        fs::set_permissions(listen_directory, fs::Permissions::from_mode(0o700))?;
        account.give(listen_directory)?;
        journal.listen = Some(Listen {
            socket: listen_directory.join("s"),
            uid: account.uid,
            gid: account.gid,
        });
    }
    journal.policy = policy;
    let endpoint = journal.socket(&directory);
    write_json(&path, &journal)?;
    let binary = cleanup
        .companion
        .clone()
        .map_or_else(std::env::current_exe, Ok)?;
    let mut child = std::process::Command::new(binary)
        .arg("--runner-manager-docker-proxy")
        .arg(&path)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()?;
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    while !directory.join("ready").is_file() {
        if child.try_wait()?.is_some() {
            return Err(invalid("Docker proxy failed to start"));
        }
        if std::time::Instant::now() >= deadline {
            let _ = child.kill();
            let _ = child.wait();
            return Err(invalid("Docker proxy startup timed out"));
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    // Reap the companion when it stops; its identity is durably recorded by
    // the companion before ready, so a daemon reload does not orphan ownership.
    std::thread::spawn(move || {
        let _ = child.wait();
    });
    Ok(Some(format!("unix://{}", endpoint.display()).into()))
}

pub fn sweep(
    cleanup: &DockerCleanup,
    lookup: &dyn Fn(AttemptId) -> io::Result<Option<runner_manager_domain::attempt::RunnerAttempt>>,
    dry_run: bool,
) -> io::Result<SweepReport> {
    use runner_manager_domain::attempt::AttemptState;
    use runner_manager_platform::process::Adoption;

    let configured = Config::load(&cleanup.config)?.mode;
    let mode = if dry_run { Mode::Report } else { configured };
    if !cleanup.root.exists() {
        return Ok(SweepReport::default());
    }
    let _lock = runner_manager_platform::lock::HostLock::acquire_at(
        &cleanup.root.join(".sweep.lock"),
        runner_manager_platform::lock::LockKind::DockerCleanup,
        Duration::from_secs(1),
    )
    .map_err(invalid)?;
    let entries = match fs::read_dir(&cleanup.root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(SweepReport::default()),
        Err(error) => return Err(error),
    };
    let mut total = SweepReport::default();
    for entry in entries {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }
        let name = entry.file_name();
        let Some(name) = name.to_str() else {
            continue;
        };
        if !valid_operation(name) {
            continue;
        }
        let outcome = (|| -> io::Result<()> {
            let path = entry.path().join("journal.json");
            let mut journal = Journal::load(&path)?;
            if journal.host != cleanup.host || journal.attempt.to_string() != name {
                return Err(invalid("Docker journal directory ownership mismatch"));
            }
            let Some(attempt) = lookup(journal.attempt)? else {
                return Ok(());
            };
            if attempt.state() != AttemptState::Cleaned || !attempt.execution().is_native() {
                return Ok(());
            }
            write_json(&entry.path().join("stop"), &true)?;
            if let Some(identity) = &journal.proxy {
                match identity.recheck().map_err(invalid)? {
                    Adoption::Live => {
                        total.pending += 1;
                        return Ok(());
                    }
                    Adoption::Gone | Adoption::PidRecycled { .. } => {}
                }
            }
            if mode == Mode::Off {
                return Ok(());
            }
            if journal.intents.is_empty() && journal.resources.is_empty() {
                remove_journal_files(&entry.path())?;
                return Ok(());
            }
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_err(invalid)?
                .as_secs();
            // Collect/report mode must not repeatedly inventory every historical
            // job's Docker resources. Explicit operator reports bypass this.
            if mode == Mode::Report
                && !dry_run
                && journal
                    .last_reported_secs
                    .and_then(|previous| now.checked_sub(previous))
                    .is_some_and(|elapsed| elapsed < 3600)
            {
                return Ok(());
            }
            let engine = UnixEngine(journal.upstream.clone());
            let info = engine.request("GET", "/info", None)?;
            if info.status != 200
                || info.body.get("ID").and_then(Value::as_str) != Some(journal.engine.as_str())
                || info.body.get("OSType").and_then(Value::as_str) != Some("linux")
            {
                return Err(invalid("Docker engine identity changed"));
            }
            recover_creations(&mut journal, &engine)?;
            // Persist recovery discoveries before trying any deletion.
            write_json(&path, &journal)?;
            let report = clean_resources(
                &mut journal,
                &engine,
                if dry_run { Mode::Report } else { mode },
            )?;
            if mode == Mode::Report {
                journal.last_reported_secs = Some(now);
            }
            write_json(&path, &journal)?;
            total.candidates += report.candidates;
            total.removed += report.removed;
            total.pending += report.pending;
            total.protected += report.protected;
            total.details_truncated |= report.details_truncated;
            total.append_details(report.details);
            let unresolved = journal
                .intents
                .iter()
                .filter(|intent| !intent.settled)
                .count();
            total.pending += unresolved;
            if journal.resources.is_empty() && unresolved == 0 {
                // Remove only the companion's exact files, never recursively walk
                // this directory. An unexpected entry keeps the directory present.
                remove_journal_files(&entry.path())?;
            }
            Ok(())
        })();
        if outcome.is_err() {
            total.pending += 1;
            tracing::warn!(event="docker_cleanup_journal_deferred",attempt_id=%name,"one Docker journal could not be reconciled; continuing with the others");
        }
    }
    Ok(total)
}

fn remove_journal_files(directory: &Path) -> io::Result<()> {
    let expected = ["journal.json", "runner.json", "ready", "stop", "s"];
    for entry in fs::read_dir(directory)? {
        let entry = entry?;
        if !expected
            .iter()
            .any(|name| entry.file_name() == std::ffi::OsStr::new(name))
        {
            return Err(invalid(
                "unexpected entry in Docker journal directory; metadata retained",
            ));
        }
    }
    for name in ["journal.json", "runner.json", "ready", "stop", "s"] {
        match fs::remove_file(directory.join(name)) {
            Ok(()) => {}
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
    }
    fs::remove_dir(directory)
}

#[derive(Debug, Clone)]
pub struct UnixEngine(pub PathBuf);

impl Engine for UnixEngine {
    fn request(&self, method: &str, path: &str, body: Option<&Value>) -> io::Result<Reply> {
        if !path.starts_with('/')
            || path
                .bytes()
                .any(|byte| byte.is_ascii_whitespace() || byte.is_ascii_control())
        {
            return Err(invalid("invalid local Docker API path"));
        }
        let mut stream = UnixStream::connect(&self.0)?;
        stream.set_read_timeout(Some(Duration::from_secs(30)))?;
        stream.set_write_timeout(Some(Duration::from_secs(30)))?;
        let body = body
            .map(serde_json::to_vec)
            .transpose()
            .map_err(invalid)?
            .unwrap_or_default();
        write!(
            stream,
            "{method} {path} HTTP/1.1\r\nHost: docker\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n",
            body.len()
        )?;
        stream.write_all(&body)?;
        let head = Head::read(&mut stream)?;
        let status = head
            .first
            .split_whitespace()
            .nth(1)
            .ok_or_else(|| invalid("missing Docker status"))?
            .parse()
            .map_err(invalid)?;
        let bytes =
            if head.get("Content-Length").is_some() || head.get("Transfer-Encoding").is_some() {
                head.body(&mut stream)?
            } else {
                let mut bytes = Vec::new();
                stream.take((MAX_JSON + 1) as u64).read_to_end(&mut bytes)?;
                if bytes.len() > MAX_JSON {
                    return Err(invalid("Docker response too large"));
                }
                bytes
            };
        let body = if bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&bytes).map_err(invalid)?
        };
        Ok(Reply { status, body })
    }
}

pub fn serve(path: &Path) -> io::Result<()> {
    let mut journal = Journal::load(path)?;
    let directory = path
        .parent()
        .ok_or_else(|| invalid("missing proxy directory"))?;
    let socket = journal.socket(directory);
    if socket.as_os_str().as_encoded_bytes().len() >= 108 {
        return Err(invalid("Docker proxy path too long"));
    }
    if !fs::symlink_metadata(&journal.upstream)?
        .file_type()
        .is_socket()
    {
        return Err(invalid("Docker upstream is not a Unix socket"));
    }
    let engine = UnixEngine(journal.upstream.clone());
    let info = engine.request("GET", "/info", None)?;
    let actual_engine = info
        .body
        .get("ID")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| invalid("Docker engine identity unavailable"))?;
    if info.status != 200
        || info.body.get("OSType").and_then(Value::as_str) != Some("linux")
        || (journal.engine != PENDING_ENGINE && actual_engine != journal.engine)
    {
        return Err(invalid("Docker upstream identity mismatch"));
    }
    journal.engine = actual_engine.into();
    let listener = UnixListener::bind(&socket)?;
    if let Some(listen) = &journal.listen {
        // Not `chmod`ed: it follows a link, and the runner's account owns the
        // directory and could swap one in. Only the owner can write to, and so
        // connect to, a socket `bind` made under the usual umask, and the
        // directory around it is that account's `0700` anyway.
        std::os::unix::fs::lchown(&socket, Some(listen.uid), Some(listen.gid))?;
    } else {
        fs::set_permissions(&socket, fs::Permissions::from_mode(0o600))?;
    }
    listener.set_nonblocking(true)?;
    journal.proxy = Some(ProcessIdentity::of_current_process().map_err(invalid)?);
    write_json(path, &journal)?;
    let session = Arc::new(Session {
        attempt: journal.attempt,
        socket: socket.clone(),
        repository: journal.repository.clone(),
        // The proxy's own socket is the job's to mount, wherever it lives.
        policy: journal.policy.clone().map(|mut policy| {
            policy.allowed.extend(policy::resolve(&socket));
            policy
        }),
        owned: Mutex::new(std::collections::HashSet::new()),
    });
    let shared = Arc::new(Mutex::new(journal));
    write_json(&directory.join("ready"), &true)?;
    let stop = directory.join("stop");
    let started = std::time::Instant::now();
    let mut last_probe = started
        .checked_sub(Duration::from_secs(1))
        .unwrap_or(started);
    let mut runner = None::<ProcessIdentity>;
    loop {
        if stop.exists() {
            // This is the private companion, not the daemon. Exiting also
            // stops relay threads, so a late create cannot race the sweep.
            std::process::exit(0);
        }
        if last_probe.elapsed() >= Duration::from_secs(1) {
            last_probe = std::time::Instant::now();
            if runner.is_none() {
                match fs::read(directory.join("runner.json")) {
                    Ok(bytes) => runner = Some(serde_json::from_slice(&bytes).map_err(invalid)?),
                    Err(error) if error.kind() == io::ErrorKind::NotFound => {
                        // The parent may have died between proxy readiness and
                        // starting the runner. Do not leave a helper forever.
                        if started.elapsed() >= Duration::from_secs(30) {
                            std::process::exit(0);
                        }
                    }
                    Err(error) => return Err(error),
                }
            }
            if let Some(identity) = &runner
                && !matches!(
                    identity.recheck().map_err(invalid)?,
                    runner_manager_platform::process::Adoption::Live
                )
            {
                std::process::exit(0);
            }
        }
        match listener.accept() {
            Ok((mut stream, _)) => {
                let journal = Arc::clone(&shared);
                let session = Arc::clone(&session);
                let path = path.to_path_buf();
                let engine = engine.clone();
                std::thread::spawn(move || {
                    if handle(&mut stream, &journal, &session, &path, &engine).is_err() {
                        let _ = reply(
                            &mut stream,
                            503,
                            &serde_json::json!({"message":"runner-manager Docker accounting failed; request was not acknowledged"}),
                        );
                    }
                });
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                std::thread::sleep(Duration::from_millis(50))
            }
            Err(error) => return Err(error),
        }
    }
}

/// What every request of one proxy shares, fixed when it starts.
struct Session {
    attempt: AttemptId,
    /// The proxy's own socket, which a job container gets in place of the
    /// engine's.
    socket: PathBuf,
    repository: String,
    policy: Option<policy::Context>,
    /// Container and exec ids already known to be this job's own. Only full
    /// engine ids are kept: a name can be reused by another job's container.
    owned: Mutex<std::collections::HashSet<String>>,
}

impl Session {
    fn knows(&self, id: &str) -> bool {
        self.owned.lock().is_ok_and(|owned| owned.contains(id))
    }

    fn remember(&self, id: &str) {
        if id.len() == 64
            && id.bytes().all(|byte| byte.is_ascii_hexdigit())
            && let Ok(mut owned) = self.owned.lock()
        {
            owned.insert(id.to_owned());
        }
    }

    /// Whether the container `id` (already encoded for a path) was created by
    /// this job. One the engine does not know is left for the engine to answer.
    fn owns(&self, engine: &UnixEngine, id: &str) -> io::Result<bool> {
        if self.knows(id) {
            return Ok(true);
        }
        let reply = engine.request("GET", &format!("/containers/{id}/json"), None)?;
        if reply.status == 404 {
            return Ok(true);
        }
        let owned = reply
            .body
            .pointer("/Config/Labels")
            .and_then(|labels| labels.get(policy::JOB_LABEL))
            .and_then(Value::as_str)
            == Some(self.attempt.to_string().as_str());
        if owned {
            self.remember(id);
            if let Some(full) = reply.body.get("Id").and_then(Value::as_str) {
                self.remember(full);
            }
        }
        Ok(owned)
    }
}

fn handle(
    stream: &mut UnixStream,
    shared: &Mutex<Journal>,
    session: &Session,
    path: &Path,
    engine: &UnixEngine,
) -> io::Result<()> {
    stream.set_read_timeout(Some(Duration::from_secs(30)))?;
    stream.set_write_timeout(Some(Duration::from_secs(30)))?;
    let head = Head::read(stream)?;
    let parts: Vec<_> = head.first.split_whitespace().collect();
    if parts.len() != 3 || parts[2] != "HTTP/1.1" || !parts[1].starts_with('/') {
        return Err(invalid("invalid Docker request"));
    }
    let method = parts[0];
    let (endpoint, query) = parts[1].split_once('?').unwrap_or((parts[1], ""));
    let api = policy::unversioned(endpoint);
    let create = method == "POST"
        && ["/containers/create", "/volumes/create", "/networks/create"].contains(&api);
    let exec_create = method == "POST"
        && matches!(policy::target(api), policy::Target::Container(_))
        && api.ends_with("/exec");
    let checked = session.policy.as_ref();
    let attempt = session.attempt;
    let mut violations = Vec::new();
    // With the policy off only protected paths are checked, and those only
    // ever arrive in a creation body.
    if let Some(policy) = checked
        && policy.mode != policy::Mode::Off
    {
        violations.extend(policy::check_endpoint(method, api, query));
        foreign_target(engine, session, api, &mut violations)?;
    }
    if !create && !exec_create {
        if let Some(policy) = checked
            && settle(stream, policy, session, &violations)?
        {
            return Ok(());
        }
        return relay(stream, &head, &engine.0);
    }
    if head
        .get("Expect")
        .is_some_and(|value| value.eq_ignore_ascii_case("100-continue"))
    {
        stream.write_all(b"HTTP/1.1 100 Continue\r\n\r\n")?;
    }
    let mut body: Value = serde_json::from_slice(&head.body(stream)?).map_err(invalid)?;
    if api == "/containers/create" {
        policy::redirect_engine_socket(&mut body, &session.socket);
    }
    if let Some(policy) = checked {
        let allowed = |path: &Path| policy.access(path);
        match api {
            "/containers/create" => {
                violations.extend(policy::check_container_create(&body, &allowed));
                if policy.mode != policy::Mode::Off {
                    for other in policy::referenced_containers(&body) {
                        if !session.owns(engine, &percent_encode(&other))? {
                            violations.push(policy::Violation::new(
                                "foreign_container",
                                format!("joins or borrows from container {other}"),
                            ));
                        }
                    }
                }
            }
            "/volumes/create" => violations.extend(policy::check_volume_create(&body, &allowed)),
            "/networks/create" => violations.extend(policy::check_network_create(&body)),
            _ => violations.extend(policy::check_exec_create(&body)),
        }
        if settle(stream, policy, session, &violations)? {
            return Ok(());
        }
    }
    if api == "/containers/create" {
        mark_job(&mut body, attempt)?;
    }
    if exec_create {
        let response = engine.request(method, parts[1], Some(&body))?;
        // Created in one of the job's own containers: its later start and
        // inspect need no lookup. One in another's container, let through by
        // an audit, is looked up again so each later request is recorded too.
        let foreign = violations
            .iter()
            .any(|violation| violation.rule == "foreign_container");
        if !foreign && let Some(id) = response.body.get("Id").and_then(Value::as_str) {
            session.remember(id);
        }
        return reply(stream, response.status, &response.body);
    }
    let intent = {
        let mut journal = shared
            .lock()
            .map_err(|_| invalid("Docker journal lock poisoned"))?;
        let intent = creation_intent(&journal, parts[1], &mut body)?;
        if let Some(intent) = &intent {
            journal.intents.push(intent.clone());
            write_json(path, &*journal)?;
        }
        intent
    };
    let response = engine.request(method, parts[1], Some(&body))?;
    if (200..300).contains(&response.status)
        && let Some(intent) = intent
    {
        let id = match intent.kind {
            Kind::Volume => response.body.get("Name"),
            _ => response.body.get("Id").or_else(|| response.body.get("ID")),
        }
        .and_then(Value::as_str)
        .ok_or_else(|| invalid("Docker create returned no identity"))?;
        let mut journal = shared
            .lock()
            .map_err(|_| invalid("Docker journal lock poisoned"))?;
        if intent.kind == Kind::Container {
            session.remember(id);
        }
        capture_created(&mut journal, &intent, id, engine)?;
        journal
            .intents
            .iter_mut()
            .find(|stored| stored.id == intent.id)
            .ok_or_else(|| invalid("missing Docker creation intent"))?
            .settled = true;
        write_json(path, &*journal)?;
    }
    reply(stream, response.status, &response.body)
}

/// Records `violations` and refuses the request when one of them refuses it
/// (see [`policy::Context::refuses`]). Returns whether it was refused.
fn settle(
    stream: &mut UnixStream,
    policy: &policy::Context,
    session: &Session,
    violations: &[policy::Violation],
) -> io::Result<bool> {
    let violations = violations
        .iter()
        .filter(|violation| policy.records(violation))
        .collect::<Vec<_>>();
    if violations.is_empty() {
        return Ok(false);
    }
    let refused = violations.iter().any(|violation| policy.refuses(violation));
    let at = chrono::Utc::now().to_rfc3339();
    let records = violations
        .iter()
        .map(|violation| policy::Record {
            at: at.clone(),
            attempt: session.attempt.to_string(),
            repository: session.repository.clone(),
            rule: violation.rule.clone(),
            detail: violation.detail.clone(),
            refused,
        })
        .collect::<Vec<_>>();
    // The record is evidence, not the gate: a log that cannot be written lets
    // neither an enforced request through nor stops an audited one.
    let _ = policy::record(&policy.log, &records);
    if refused {
        let rules = violations
            .iter()
            .map(|violation| format!("{} ({})", violation.rule, violation.detail))
            .collect::<Vec<_>>()
            .join("; ");
        reply(
            stream,
            403,
            &serde_json::json!({
                "message": format!("refused by the runner-manager Docker policy: {rules}")
            }),
        )?;
    }
    Ok(refused)
}

/// Adds a violation when `api` acts on a container, or an exec instance in
/// one, that this job did not create.
fn foreign_target(
    engine: &UnixEngine,
    session: &Session,
    api: &str,
    violations: &mut Vec<policy::Violation>,
) -> io::Result<()> {
    let container = match policy::target(api) {
        policy::Target::Container(id) => Some(id.to_owned()),
        policy::Target::Exec(id) if session.knows(id) => None,
        policy::Target::Exec(id) => engine
            .request("GET", &format!("/exec/{id}/json"), None)?
            .body
            .get("ContainerID")
            .and_then(Value::as_str)
            .map(percent_encode),
        policy::Target::None => None,
    };
    if let Some(container) = container
        && !session.owns(engine, &container)?
    {
        violations.push(policy::Violation::new(
            "foreign_container",
            format!("{api} acts on a container this job did not create"),
        ));
    }
    Ok(())
}

/// Labels a container this job creates as its own, replacing any value the
/// job supplied for the label itself.
fn mark_job(body: &mut Value, attempt: AttemptId) -> io::Result<()> {
    let object = body
        .as_object_mut()
        .ok_or_else(|| invalid("expected Docker creation object"))?;
    let labels = object
        .entry("Labels")
        .or_insert_with(|| Value::Object(serde_json::Map::new()));
    if labels.is_null() {
        *labels = Value::Object(serde_json::Map::new());
    }
    labels
        .as_object_mut()
        .ok_or_else(|| invalid("invalid Docker labels"))?
        .insert(policy::JOB_LABEL.into(), Value::String(attempt.to_string()));
    Ok(())
}

fn reply(stream: &mut impl Write, status: u16, body: &Value) -> io::Result<()> {
    let bytes = if body.is_null() {
        Vec::new()
    } else {
        serde_json::to_vec(body).map_err(invalid)?
    };
    write!(
        stream,
        "HTTP/1.1 {status} Docker\r\nContent-Type: application/json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n",
        bytes.len()
    )?;
    stream.write_all(&bytes)
}

fn relay(client: &mut UnixStream, head: &Head, socket: &Path) -> io::Result<()> {
    let mut upstream = UnixStream::connect(socket)?;
    upstream.write_all(&head.encode_for_relay())?;
    client.set_read_timeout(None)?;
    client.set_write_timeout(None)?;
    let mut client_read = client.try_clone()?;
    let mut upstream_write = upstream.try_clone()?;
    let send = std::thread::spawn(move || {
        let result = io::copy(&mut client_read, &mut upstream_write);
        let _ = upstream_write.shutdown(Shutdown::Write);
        result
    });
    let result = io::copy(&mut upstream, client);
    let _ = client.shutdown(Shutdown::Both);
    let _ = upstream.shutdown(Shutdown::Both);
    let _ = send.join();
    result.map(|_| ())
}

/// Recover an ambiguous create by all three ownership labels, never by age,
/// container names, or an unscoped before/after inventory difference.
pub fn recover_creations(journal: &mut Journal, engine: &dyn Engine) -> io::Result<()> {
    for intent in journal.intents.clone() {
        let labels: Vec<_> = journal
            .labels(&intent.id)
            .into_iter()
            .map(|(key, value)| format!("{key}={value}"))
            .collect();
        let filter = percent_encode(
            &serde_json::to_string(&serde_json::json!({"label":labels})).map_err(invalid)?,
        );
        let path = match intent.kind {
            Kind::Container => format!("/containers/json?all=true&filters={filter}"),
            Kind::Volume => format!("/volumes?filters={filter}"),
            Kind::Network => format!("/networks?filters={filter}"),
        };
        let response = engine.request("GET", &path, None)?;
        if response.status != 200 {
            return Err(invalid("Docker recovery inventory failed"));
        }
        let entries = if intent.kind == Kind::Volume {
            response.body.get("Volumes")
        } else {
            Some(&response.body)
        };
        if !entries.is_some_and(|entries| {
            entries.is_array() || (intent.kind == Kind::Volume && entries.is_null())
        }) {
            return Err(invalid("invalid Docker recovery inventory"));
        }
        if let Some(entries) = entries.and_then(Value::as_array) {
            for entry in entries {
                let id = if intent.kind == Kind::Volume {
                    entry.get("Name")
                } else {
                    entry.get("Id").or_else(|| entry.get("ID"))
                };
                let id = id
                    .and_then(Value::as_str)
                    .ok_or_else(|| invalid("Docker recovery inventory has no ID"))?;
                capture_created(journal, &intent, id, engine)?;
            }
        }
    }
    Ok(())
}

fn percent_encode(value: &str) -> String {
    let mut result = String::new();
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
            result.push(char::from(byte));
        } else {
            use std::fmt::Write as _;
            let _ = write!(result, "%{byte:02X}");
        }
    }
    result
}

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

    #[test]
    fn unexpected_journal_entries_keep_the_ownership_metadata() {
        let directory = tempfile::tempdir().unwrap();
        fs::write(directory.path().join("journal.json"), b"ownership").unwrap();
        fs::write(directory.path().join("unexpected"), b"preserve").unwrap();
        assert!(remove_journal_files(directory.path()).is_err());
        assert_eq!(
            fs::read(directory.path().join("journal.json")).unwrap(),
            b"ownership"
        );
        fs::remove_file(directory.path().join("unexpected")).unwrap();
        remove_journal_files(directory.path()).unwrap();
        assert!(!directory.path().exists());
    }
}