gridwork 0.0.2

//! The built binary, run as a caller runs it.
//!
//! These drive `gw` as a subprocess rather than calling `run` in-process, and
//! that is the point: the socket path comes from the environment, and a
//! subprocess can be given its own without mutating a variable every other test
//! in the binary shares. It also means the exit code being asserted is the one
//! the shell would see.
//!
//! Most cases need nothing but the binary. The last one needs a database,
//! because the only way to prove the client half of the wire is to put a real
//! daemon on the other end of it.

use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};

/// A directory the daemon will accept as a socket's parent.
fn private_dir(tag: &str) -> PathBuf {
    use std::os::unix::fs::PermissionsExt;
    let dir = std::env::temp_dir().join(format!("gw-cli-{}-{tag}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create dir");
    std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).expect("chmod");
    dir
}

/// Run `gw` with a socket path of the caller's choosing.
fn gw(socket: &Path, line: &str) -> Output {
    gw_env(line, &[("GWK_SOCKET_PATH", &socket.to_string_lossy())])
}

/// Run `gw` with exactly the environment given — nothing inherited that matters.
///
/// A subprocess is what makes this safe: the credentials and the socket path a
/// case needs go to the child, rather than into a variable every other case in
/// this binary would see.
fn gw_env(line: &str, env: &[(&str, &str)]) -> Output {
    let mut command = Command::new(env!("CARGO_BIN_EXE_gw"));
    command.args(line.split_whitespace());
    for (name, value) in env {
        command.env(name, value);
    }
    command.output().expect("run gw")
}

fn code(output: &Output) -> i32 {
    output.status.code().expect("gw exited on a signal")
}

fn json(output: &Output) -> serde_json::Value {
    let text = String::from_utf8_lossy(&output.stdout);
    serde_json::from_str(text.trim()).unwrap_or_else(|e| panic!("stdout is not JSON ({e}): {text}"))
}

#[test]
fn the_command_tree_is_printed_as_prose_and_everything_else_as_json() {
    let dir = private_dir("help");
    let socket = dir.join("absent.sock");

    let help = gw(&socket, "--help");
    assert_eq!(code(&help), 0);
    let text = String::from_utf8_lossy(&help.stdout);
    // The one output a human reads. Every other answer is machine JSON, which is
    // why this case exists: to pin the exception rather than let it spread.
    assert!(text.starts_with("gw —"), "{text}");
    assert!(text.contains("gw kernel health"), "{text}");

    let info = gw(&socket, "build-info");
    assert_eq!(code(&info), 0);
    let info = json(&info);
    assert_eq!(info["type"], "build_info");
    assert!(info["contract_version"].is_number(), "{info}");
    // Present either way: a build stamped from a clean checkout reports its
    // revision and an unstamped one reports null. What must not happen is the
    // key being absent, because a caller comparing this against the revision
    // genesis recorded would read "absent" as "same".
    assert!(info.get("public_revision").is_some(), "{info}");
    if let Some(revision) = info["public_revision"].as_str() {
        assert_eq!(revision.len(), 40, "{revision}");
    }
}

#[test]
fn a_mistake_exits_two_and_says_what_it_was() {
    let dir = private_dir("usage");
    let socket = dir.join("absent.sock");

    for line in ["nonsense", "projection list tsak", "kernel activate"] {
        let out = gw(&socket, line);
        assert_eq!(
            code(&out),
            2,
            "{line}: {:?}",
            String::from_utf8_lossy(&out.stdout)
        );
        let answer = json(&out);
        // The protocol's own error shape, so a caller parses one thing whether
        // the refusal came from the kernel or from the command line.
        assert_eq!(answer["type"], "error");
        assert_eq!(answer["code"], "validation");
    }

    // The projection refusal lists what would have worked. A closed set that
    // refuses without naming its members makes the caller go and read source.
    let answer = json(&gw(&socket, "projection list tsak"));
    let message = answer["message"].as_str().expect("message");
    assert!(message.contains("attention_item"), "{message}");
}

#[test]
fn a_daemon_that_is_not_there_is_unavailable_rather_than_a_crash() {
    let dir = private_dir("absent");
    let socket = dir.join("absent.sock");

    let out = gw(&socket, "kernel health");
    // 5, not 10: nothing is wrong with `gw`, and nothing is wrong with the
    // request. The kernel is not running, and trying again later is the fix.
    assert_eq!(code(&out), 5);
    let answer = json(&out);
    assert_eq!(answer["code"], "storage");
    assert!(
        answer["message"]
            .as_str()
            .expect("message")
            .contains("absent.sock"),
        "{answer}"
    );
}

/// The runtime role `admin::init` grants to. Created by the maintenance pool.
const RUNTIME_ROLE: &str = "gwk_cli_runtime";
const TEST_KEK: [u8; 32] = [0x5a; 32];

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn the_binary_talks_to_a_real_daemon() {
    use gwk_kernel::blob::store::PgBlobStore;
    use gwk_kernel::config::{ADMIN_DATABASE_URL_ENV, AdminConfig, BlobConfig, RUNTIME_ROLE_ENV};
    use gwk_kernel::wire::listen::Listener;
    use gwk_kernel::wire::serve::Daemon;
    use gwk_kernel::{PgEventStore, admin, connect_pool};
    use secrecy::SecretString;
    use sqlx::PgPool;

    let admin_url = std::env::var("GWK_TEST_ADMIN_DATABASE_URL")
        .expect("GWK_TEST_ADMIN_DATABASE_URL must point at a PostgreSQL superuser DSN");
    let maintenance = PgPool::connect(&admin_url).await.expect("maintenance");
    // No CREATE ROLE IF NOT EXISTS; the failure case is "already exists", which
    // is the state this wanted. A role genuinely absent fails at the GRANT.
    let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "CREATE ROLE {RUNTIME_ROLE} NOLOGIN;"
    )))
    .execute(&maintenance)
    .await;

    let name = format!("gwk_cli_{}", std::process::id());
    let url = {
        let (prefix, _) = admin_url.rsplit_once('/').expect("a /database suffix");
        format!("{prefix}/{name}")
    };
    let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "DROP DATABASE IF EXISTS {name} WITH (FORCE);"
    )))
    .execute(&maintenance)
    .await;
    sqlx::raw_sql(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name};")))
        .execute(&maintenance)
        .await
        .expect("create database");

    let pool = connect_pool(&SecretString::from(url.clone()), 8)
        .await
        .expect("connect");
    let config = AdminConfig::from_lookup(move |key| match key {
        ADMIN_DATABASE_URL_ENV => Some(url.clone()),
        RUNTIME_ROLE_ENV => Some(RUNTIME_ROLE.to_owned()),
        _ => None,
    })
    .expect("admin config");
    admin::init(&pool, &config).await.expect("init");

    let dir = private_dir("live");
    let blob_root = dir.join("blobs");
    let blobs = PgBlobStore::open(
        pool.clone(),
        BlobConfig::new(blob_root.clone(), TEST_KEK, "kek-test".to_owned()).expect("blob config"),
    )
    .await
    .expect("blob store");
    let store = PgEventStore::open(pool).await.expect("store");
    store
        .ensure_genesis(&"a1b2c3d4e5".repeat(4))
        .await
        .expect("genesis");

    let socket = dir.join("gwk.sock");
    let listener = Listener::bind(&socket).await.expect("bind");
    let daemon = std::sync::Arc::new(
        Daemon::new(store.with_blobs(blobs), "a1b2c3d4e5".repeat(4)).expect("daemon"),
    );
    let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
    let serving = tokio::spawn(async move {
        let _ = gwk_kernel::wire::serve::run(listener, daemon, async move {
            let _ = stopped.await;
        })
        .await;
    });

    // Run the binary on a blocking thread: it is a subprocess doing synchronous
    // I/O against a daemon living in this runtime, so waiting for it here would
    // park the very task that has to answer it.
    let socket_for_blocking = socket.clone();
    let answers = tokio::task::spawn_blocking(move || {
        let socket = socket_for_blocking.as_path();
        let health = gw(socket, "kernel health");
        let status = gw(socket, "kernel status");
        let events = gw(socket, "event read --limit 1");
        let tasks = gw(socket, "projection list task");
        let missing = gw(socket, "projection get task t-nope");

        // A blob, all the way there and back through the built binary.
        let payload = socket.with_file_name("payload.bin");
        let mut file = std::fs::File::create(&payload).expect("create");
        let plaintext: Vec<u8> = (0..5_000u32).map(|i| (i % 251) as u8).collect();
        file.write_all(&plaintext).expect("write");
        drop(file);
        let put = gw(
            socket,
            &format!(
                "blob put --file {} --media-type text/plain",
                payload.display()
            ),
        );
        (
            health, status, events, tasks, missing, put, plaintext, payload,
        )
    })
    .await
    .expect("join");
    let (health, status, events, tasks, missing, put, plaintext, payload) = answers;

    assert_eq!(
        code(&health),
        0,
        "{:?}",
        String::from_utf8_lossy(&health.stderr)
    );
    // The contract's own value, not a shape this program invented: a sealed
    // kernel is READY and says so.
    assert_eq!(
        json(&health),
        serde_json::json!({"type": "health", "ready": true, "sealed": true})
    );

    assert_eq!(code(&status), 0);
    let status = json(&status);
    assert_eq!(status["type"], "status");
    assert_eq!(status["public_revision"], "a1b2c3d4e5".repeat(4));

    assert_eq!(code(&events), 0);
    let events = json(&events);
    // Genesis, and nothing else — the log of a kernel that has only been
    // initialized.
    assert_eq!(
        events["events"].as_array().expect("events").len(),
        1,
        "{events}"
    );

    assert_eq!(code(&tasks), 0);
    // An empty page is an answer, and a cursor that says the walk is over.
    assert_eq!(json(&tasks)["records"], serde_json::json!([]));

    // Absent exits 4, distinctly from a refusal and from unavailability.
    assert_eq!(code(&missing), 4);
    assert_eq!(json(&missing)["code"], "not_found");

    assert_eq!(code(&put), 0, "{:?}", String::from_utf8_lossy(&put.stderr));
    let put = json(&put);
    assert_eq!(put["type"], "blob_committed");
    assert_eq!(put["deduplicated"], false);
    let address = put["descriptor"]["address"]
        .as_str()
        .expect("an address")
        .to_owned();

    let back = payload.with_file_name("back.bin");
    let (stat, got) = tokio::task::spawn_blocking({
        let socket = socket.clone();
        let back = back.clone();
        let address = address.clone();
        move || {
            let stat = gw(&socket, &format!("blob stat {address}"));
            let got = gw(
                &socket,
                &format!("blob get {address} --output {}", back.display()),
            );
            (stat, got)
        }
    })
    .await
    .expect("join");

    assert_eq!(code(&stat), 0);
    assert_eq!(json(&stat)["descriptor"]["media_type"], "text/plain");
    assert_eq!(code(&got), 0, "{:?}", String::from_utf8_lossy(&got.stderr));
    assert_eq!(
        std::fs::read(&back).expect("read back"),
        plaintext,
        "the blob did not survive the round trip"
    );

    let _ = stop.send(());
    serving.await.expect("join");
    let _ = std::fs::remove_dir_all(&dir);
    let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "DROP DATABASE IF EXISTS {name} WITH (FORCE);"
    )))
    .execute(&maintenance)
    .await;
}

/// A login-capable, NON-superuser role — the only kind a daemon may run as.
///
/// Separate from the role the other live case grants to, which is `NOLOGIN`:
/// this one has to be able to connect, because the whole point is to prove the
/// privilege check passes for a credential `admin init` actually granted.
const DAEMON_ROLE: &str = "gwk_cli_daemon";
/// The throwaway password for that role on an ephemeral test database. Not a
/// secret: the superuser DSN this case is handed already carries CI's own.
const DAEMON_PASSWORD: &str = "ci";

#[tokio::test]
#[ignore = "requires PostgreSQL"]
async fn the_admin_door_initializes_a_database_and_the_daemon_serves_it() {
    use base64::prelude::{BASE64_STANDARD, Engine as _};
    use sqlx::PgPool;

    let admin_url = std::env::var("GWK_TEST_ADMIN_DATABASE_URL")
        .expect("GWK_TEST_ADMIN_DATABASE_URL must point at a PostgreSQL superuser DSN");
    let maintenance = PgPool::connect(&admin_url).await.expect("maintenance");
    // Created if absent, then forced into the shape this case needs. Both are
    // allowed to fail: the second is what makes a leftover role from an earlier
    // run usable rather than a reason to stop.
    for statement in [
        format!("CREATE ROLE {DAEMON_ROLE} LOGIN PASSWORD '{DAEMON_PASSWORD}';"),
        format!(
            "ALTER ROLE {DAEMON_ROLE} LOGIN PASSWORD '{DAEMON_PASSWORD}' NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS;"
        ),
    ] {
        let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(statement))
            .execute(&maintenance)
            .await;
    }

    let live = format!("gwk_daemon_{}", std::process::id());
    let scratch = format!("gwk_scratch_{}", std::process::id());
    for name in [&live, &scratch] {
        let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
            "DROP DATABASE IF EXISTS {name} WITH (FORCE);"
        )))
        .execute(&maintenance)
        .await;
        sqlx::raw_sql(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name};")))
            .execute(&maintenance)
            .await
            .expect("create database");
    }

    let (prefix, _) = admin_url.rsplit_once('/').expect("a /database suffix");
    let admin_dsn = format!("{prefix}/{live}");
    // The runtime credential: the same server and database, as the granted role
    // rather than as a superuser. A daemon handed a superuser refuses to start,
    // which is a rule this case would otherwise never reach.
    let runtime_dsn = {
        let (scheme, rest) = prefix.split_once("://").expect("a scheme");
        let host = rest.rsplit_once('@').map_or(rest, |(_, host)| host);
        format!("{scheme}://{DAEMON_ROLE}:{DAEMON_PASSWORD}@{host}/{live}")
    };

    let dir = private_dir("daemon");
    let socket = dir.join("gwk.sock");
    let blob_root = dir.join("blobs");
    let revision = "a1b2c3d4e5".repeat(4);
    let kek = BASE64_STANDARD.encode([0x5a_u8; 32]);
    let blob_root_text = blob_root.to_string_lossy().into_owned();
    let admin_env: Vec<(&str, &str)> = vec![
        ("GWK_ADMIN_DATABASE_URL", &admin_dsn),
        ("GWK_RUNTIME_ROLE", DAEMON_ROLE),
        ("GWK_PUBLIC_REVISION", &revision),
        ("GWK_BLOB_ROOT", &blob_root_text),
        ("GWK_BLOB_KEK", &kek),
        ("GWK_BLOB_KEK_ID", "kek-test"),
    ];

    // The override rule, pinned in both directions rather than assumed to be in
    // one of them: a STAMPED build states a fact about the bytes compiled and
    // refuses the environment, an unstamped one has no fact to state and takes
    // it. Which case this is depends on whether the tree was clean when
    // `build.rs` last ran — so a case that hardcoded `revision` here passed on a
    // developer's tree and failed on CI, where the checkout is clean and the
    // stamp is the merge commit. `build-info` reports the STAMP ONLY, by design;
    // it is the oracle for which case we are in, not for the resolved value.
    let stamp = json(&gw_env("build-info", &admin_env))["public_revision"]
        .as_str()
        .map(str::to_owned);
    let reported = stamp.clone().unwrap_or_else(|| revision.clone());
    assert!(
        reported.len() == 40 && reported.bytes().all(|b| b.is_ascii_hexdigit()),
        "a reported revision is a full hex revision, got {reported:?}"
    );
    if let Some(stamped) = &stamp {
        assert_ne!(
            stamped, &revision,
            "this case can only prove the override rule if the two differ"
        );
    }

    let init = gw_env("admin init", &admin_env);
    assert_eq!(code(&init), 0, "{}", String::from_utf8_lossy(&init.stdout));
    let answer = json(&init);
    assert_eq!(answer["type"], "admin_initialized");
    assert_eq!(answer["outcome"], "initialized");
    assert_eq!(answer["public_revision"], reported);

    // Again, unchanged: the contract is already installed and genesis is
    // idempotent under its own key, so a second run is a no-op and not a second
    // epoch.
    let again = gw_env("admin init", &admin_env);
    assert_eq!(
        code(&again),
        0,
        "{}",
        String::from_utf8_lossy(&again.stdout)
    );
    assert_eq!(json(&again)["outcome"], "already_initialized");

    let verified = gw_env("admin verify", &admin_env);
    assert_eq!(
        code(&verified),
        0,
        "{}",
        String::from_utf8_lossy(&verified.stdout)
    );
    let answer = json(&verified);
    assert_eq!(answer["target"], "initialized");
    assert_eq!(answer["runtime_role_exists"], true);
    // The load-bearing assertion of the whole case: the role `admin init`
    // granted to is one the daemon will accept. If the grants and the refusal
    // list ever disagree, `gw daemon` can never start, and nothing short of
    // running both halves finds that.
    assert_eq!(answer["violations"], serde_json::json!([]));
    assert_eq!(
        answer["detail"]["contract_sha256"],
        answer["expected_contract_sha256"]
    );

    let rebuilt = gw_env(
        &format!("admin rebuild-projections --scratch-database {scratch}"),
        &admin_env,
    );
    assert_eq!(
        code(&rebuilt),
        0,
        "{}",
        String::from_utf8_lossy(&rebuilt.stdout)
    );
    let answer = json(&rebuilt);
    // A replay of the log into an empty scratch agrees with the live
    // projections. It also proves the reader store did not fence anybody: the
    // daemon below starts afterwards and appends nothing, but `admin init`
    // already claimed an epoch, and a rebuild that had claimed another would
    // have made this database unwritable.
    assert_eq!(answer["agrees"], true, "{answer}");
    assert_eq!(answer["live_hash"], answer["rebuilt_hash"]);

    // The service itself, as a service: its own process, its own credential.
    let mut serving = Command::new(env!("CARGO_BIN_EXE_gw"))
        .arg("daemon")
        .env("GWK_DATABASE_URL", &runtime_dsn)
        .env("GWK_SOCKET_PATH", &socket)
        .env("GWK_BLOB_ROOT", &blob_root)
        .env("GWK_BLOB_KEK", &kek)
        .env("GWK_BLOB_KEK_ID", "kek-test")
        .env("GWK_PUBLIC_REVISION", &revision)
        .spawn()
        .expect("spawn the daemon");

    // Poll rather than sleep once: a fixed wait is either flaky or slow, and
    // "answers health" is the only definition of started that matters.
    let mut health = None;
    for _ in 0..100 {
        let out = gw(&socket, "kernel health");
        if code(&out) == 0 {
            health = Some(out);
            break;
        }
        assert!(
            serving.try_wait().expect("wait").is_none(),
            "the daemon exited before it served: {}",
            String::from_utf8_lossy(&out.stdout)
        );
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    let health = health.expect("the daemon never answered health");
    assert_eq!(
        json(&health),
        serde_json::json!({"type": "health", "ready": true, "sealed": true})
    );

    let status = gw(&socket, "kernel status");
    assert_eq!(code(&status), 0);
    let answer = json(&status);
    // The revision the daemon resolved, reported back — the comparison
    // `build-info` and genesis exist for. `reported`, not `revision`: a stamped
    // build ignores the environment on purpose (see above).
    assert_eq!(answer["public_revision"], reported);
    assert_eq!(answer["sealed"], true);

    // Retention, through the admin door rather than the socket — which is the
    // whole reason these four verbs live there. Uploading goes over the wire; pin,
    // unpin, sweep, and shred need a credential, and none of them has a wire
    // request to reach through.
    let payload = dir.join("hold.bin");
    std::fs::write(&payload, b"a blob nothing in the log points at").expect("write a payload");
    let put = gw(
        &socket,
        &format!(
            "blob put --file {} --media-type text/plain",
            payload.display()
        ),
    );
    assert_eq!(code(&put), 0, "{}", String::from_utf8_lossy(&put.stdout));
    let held = json(&put)["descriptor"]["address"]
        .as_str()
        .expect("an address")
        .to_owned();

    let pinned = gw_env(&format!("admin blob pin {held} evidence-1"), &admin_env);
    assert_eq!(
        code(&pinned),
        0,
        "{}",
        String::from_utf8_lossy(&pinned.stdout)
    );
    assert_eq!(json(&pinned)["type"], "blob_pinned");

    // Nothing in this log references the blob, so the only thing standing between
    // it and the sweep is the pin.
    let swept = gw_env("admin blob sweep", &admin_env);
    assert_eq!(
        code(&swept),
        0,
        "{}",
        String::from_utf8_lossy(&swept.stdout)
    );
    let removed = json(&swept);
    assert!(
        !removed["removed"]
            .as_array()
            .expect("an array")
            .iter()
            .any(|address| address == &serde_json::Value::String(held.clone())),
        "a pinned blob was swept: {removed}"
    );
    assert_eq!(code(&gw(&socket, &format!("blob stat {held}"))), 0);

    // Released, and the same sweep reclaims it. Absent afterwards, not
    // tombstoned: a sweep is reclamation of something nothing pointed at, and it
    // leaves no claim behind.
    let unpinned = gw_env(&format!("admin blob unpin {held} evidence-1"), &admin_env);
    assert_eq!(
        code(&unpinned),
        0,
        "{}",
        String::from_utf8_lossy(&unpinned.stdout)
    );
    let swept = gw_env("admin blob sweep", &admin_env);
    assert_eq!(code(&swept), 0);
    assert!(
        json(&swept)["removed"]
            .as_array()
            .expect("an array")
            .iter()
            .any(|address| address == &serde_json::Value::String(held.clone())),
        "the unpinned blob survived a sweep: {}",
        String::from_utf8_lossy(&swept.stdout)
    );
    let gone = gw(&socket, &format!("blob stat {held}"));
    assert_eq!(code(&gone), 4, "{}", String::from_utf8_lossy(&gone.stdout));
    assert_eq!(json(&gone)["code"], "not_found");

    // Shred is the other half and it is the opposite answer: permanent, and
    // REFUSED rather than absent, because a shredded address is a claim the
    // kernel keeps making. The container may still be on disk; the key is gone.
    let doomed = dir.join("shred.bin");
    std::fs::write(&doomed, b"a blob that will be cryptographically erased").expect("write");
    let put = gw(
        &socket,
        &format!(
            "blob put --file {} --media-type text/plain",
            doomed.display()
        ),
    );
    assert_eq!(code(&put), 0);
    let erased = json(&put)["descriptor"]["address"]
        .as_str()
        .expect("an address")
        .to_owned();
    let shredded = gw_env(&format!("admin blob shred {erased}"), &admin_env);
    assert_eq!(
        code(&shredded),
        0,
        "{}",
        String::from_utf8_lossy(&shredded.stdout)
    );
    assert_eq!(json(&shredded)["type"], "blob_shredded");
    let refused = gw(
        &socket,
        &format!(
            "blob get {erased} --output {}",
            dir.join("never.bin").display()
        ),
    );
    assert_eq!(
        code(&refused),
        6,
        "{}",
        String::from_utf8_lossy(&refused.stdout)
    );
    assert_eq!(json(&refused)["code"], "blob_tombstoned");
    assert!(
        !dir.join("never.bin").exists(),
        "a refused read still wrote a file"
    );

    // A second daemon on the same database must refuse rather than race: one
    // writer per store, enforced by an advisory lock that is never waited on.
    let second = gw_env(
        "daemon",
        &[
            ("GWK_DATABASE_URL", &runtime_dsn),
            (
                "GWK_SOCKET_PATH",
                &dir.join("second.sock").to_string_lossy(),
            ),
            ("GWK_BLOB_ROOT", &blob_root.to_string_lossy()),
            ("GWK_BLOB_KEK", &kek),
            ("GWK_BLOB_KEK_ID", "kek-test"),
            ("GWK_PUBLIC_REVISION", &revision),
        ],
    );
    assert_eq!(
        code(&second),
        3,
        "{}",
        String::from_utf8_lossy(&second.stdout)
    );
    assert_eq!(json(&second)["code"], "fenced");

    // SIGKILL: the crash the daemon gets no say in. Not SIGTERM — the point is
    // that no shutdown code runs at all, so nothing releases the writer lock and
    // nothing removes the socket. Everything after this is what the NEXT daemon
    // has to cope with, and it is the only path in this suite that produces a
    // genuinely stale socket beside a genuinely abandoned lock.
    Command::new("kill")
        .args(["-KILL", &serving.id().to_string()])
        .status()
        .expect("signal the daemon");
    let died = serving.wait().expect("wait");
    assert!(!died.success(), "SIGKILL is not a clean exit: {died:?}");
    assert!(
        socket.exists(),
        "a killed daemon cannot have cleaned up after itself"
    );

    // Its connections die with it, and the advisory lock goes when the backend
    // does — which the server learns from a closed socket rather than from the
    // signal, so it is prompt but not synchronous. Waiting for the backends to
    // disappear is the precondition, not a retry loop dressed up as one.
    let mut released = false;
    for _ in 0..100 {
        let live_backends: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM pg_stat_activity WHERE datname = $1 AND usename = $2",
        )
        .bind(&live)
        .bind(DAEMON_ROLE)
        .fetch_one(&maintenance)
        .await
        .expect("count the daemon's backends");
        if live_backends == 0 {
            released = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    assert!(released, "the killed daemon's backends outlived it");

    // A replacement starts on the same socket path and the same database. Three
    // things have to be true at once for this to work: the abandoned advisory
    // lock is takeable (`acquire` never waits, so a lock a corpse still held
    // would exit 3), the stale socket is probed and replaced rather than refused,
    // and recovery reaches a verdict on a log nobody checkpointed on the way out.
    let mut restarted = Command::new(env!("CARGO_BIN_EXE_gw"))
        .arg("daemon")
        .env("GWK_DATABASE_URL", &runtime_dsn)
        .env("GWK_SOCKET_PATH", &socket)
        .env("GWK_BLOB_ROOT", &blob_root)
        .env("GWK_BLOB_KEK", &kek)
        .env("GWK_BLOB_KEK_ID", "kek-test")
        .env("GWK_PUBLIC_REVISION", &revision)
        .spawn()
        .expect("spawn the replacement");

    let mut recovered = None;
    for _ in 0..100 {
        let out = gw(&socket, "kernel health");
        if code(&out) == 0 {
            recovered = Some(out);
            break;
        }
        if let Some(exited) = restarted.try_wait().expect("wait") {
            panic!("the replacement refused to start after a crash: {exited:?}");
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    let recovered = recovered.expect("the replacement never answered health");
    assert_eq!(json(&recovered)["ready"], true);
    // And it is serving the same log, not a fresh one: the epoch it reports is
    // past the crashed daemon's, because claiming write authority bumps it.
    let status = gw(&socket, "kernel status");
    assert_eq!(code(&status), 0);
    assert_eq!(json(&status)["public_revision"], reported);

    // SIGTERM is how a service manager asks. The socket goes with it, or the
    // next start would take the stale-takeover path for no reason.
    Command::new("kill")
        .args(["-TERM", &restarted.id().to_string()])
        .status()
        .expect("signal the daemon");
    let stopped = restarted.wait().expect("wait");
    assert!(stopped.success(), "the daemon exited {stopped:?}");
    assert!(!socket.exists(), "shutdown left the socket behind");

    let _ = std::fs::remove_dir_all(&dir);
    for name in [&live, &scratch] {
        let _ = sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
            "DROP DATABASE IF EXISTS {name} WITH (FORCE);"
        )))
        .execute(&maintenance)
        .await;
    }
}