//! The relay's open window kept on disk and read back (`src/checkpoint.rs`): a relay killed
//! outright, with SIGKILL, loses none of the counts its checkpoint held, states the span it
//! cannot count as unmetered, and never counts anything twice.
//!
//! - **A hard kill.** `wai-relay` itself, started from a configuration, bound to port 0, its
//!   port read from its log. A publisher and a viewer; window 0 sealed; two more groups; the
//!   checkpoint written; the process killed with SIGKILL, its sessions never ended. Started
//!   again from the same configuration it still answers window 0, the last it sealed, as it
//!   stated it before; and its next window is window 1, the same index, holding the killed
//!   sessions' counts beside the new sessions', the statement marked as resumed with the
//!   checkpoint's instant and the span from it to the restart, unmetered. The room's record
//!   recomputes from the parts, and over both windows every group is counted once.
//! - **Planted: a checkpoint older than the last sealed cut.** The checkpoint that held window
//!   1's counts is put back after window 1 was sealed, and the relay killed and started again:
//!   refused `checkpoint_stale`, none of its counts in window 2, the span since the seal
//!   unmetered. Left unrefused, the two groups would be billed twice.
//! - **Planted: a truncated checkpoint, and one of another relay's key.** Each refused with its
//!   reason, in the statement of the next window, with no count read from it.
//! - **A clean stop loses nothing and leaves nothing uncounted.** Stopped as SIGTERM stops it,
//!   every session ended, the relay writes the window's counts once more; started again, the
//!   window holds them and its mark states no unmetered span.
//! - A relay refuses to start with a checkpoint and no key of its own to sign it with.

#![cfg(unix)]

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::Parameters;
use joule_moq_relay::checkpoint::{self, Checkpoint, SignedCheckpoint};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::metering::HopRecord;
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::window::SignedWindow;
use joule_moq_relay::RelayConfig;
use serde_json::{json, Value};
use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;

fn now_us() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
}

fn hexkey(k: &SigningKey) -> String {
    hex::encode(k.verifying_key().to_bytes())
}

/// A directory of the test's own: in `/dev/shm` where the host has one (memory), else the
/// temporary directory. What these tests hold is the relay's logic, and a checkpoint read
/// back after SIGKILL needs no sync (the killed process's writes are in the kernel's cache);
/// each write still syncs, and on a runner's disk under other jobs one synced write took up to
/// 23.5 s (the forge's linux-arm64 host, 11 November 2026: median 18 ms, p90 3.4 s), past
/// the bounds here, while a host whose temporary directory is memory passed. A seal that
/// waits on its disk is held by its own test below.
fn dir(what: &str) -> PathBuf {
    let name = format!("wai-relay-ckpt-{}-{what}", std::process::id());
    let shm = Path::new("/dev/shm").join(&name);
    let _ = std::fs::remove_dir_all(&shm);
    if std::fs::create_dir_all(&shm).is_ok() {
        return shm;
    }
    let d = std::env::temp_dir().join(name);
    let _ = std::fs::remove_dir_all(&d);
    std::fs::create_dir_all(&d).unwrap();
    d
}

async fn ask(path: &Path, req: Value) -> Value {
    timeout(STEP * 2, async {
        let s = tokio::net::UnixStream::connect(path).await.expect("the control socket");
        let (r, mut w) = s.into_split();
        w.write_all(format!("{req}\n").as_bytes()).await.unwrap();
        let mut line = String::new();
        BufReader::new(r).read_line(&mut line).await.unwrap();
        serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("an answer: {e}: {line:?}"))
    })
    .await
    .expect("an answer in time")
}

async fn ok(path: &Path, req: Value) -> Value {
    let a = ask(path, req.clone()).await;
    assert_eq!(a["ok"], true, "{req} answered {a}");
    a
}

async fn statement(path: &Path, window: u64) -> SignedWindow {
    let a = ok(path, json!({"op": "window", "window": window})).await;
    let w: SignedWindow = serde_json::from_value(a["statement"].clone()).expect("a signed window");
    w.verify().expect("the relay signed it");
    w
}

async fn record(path: &Path, room: &str, t0: u64, window: u64) -> HopRecord {
    let a = ok(path, json!({"op": "record", "room": room, "t0_unix_us": t0, "window": window})).await;
    let h = |k: &str| hex::decode(a[k].as_str().unwrap()).unwrap();
    HopRecord::decode(&h("record"), &h("ledger")).expect("the record reads")
}

/// A `wai-relay` process: killed and reaped when this is dropped, however the test ends.
struct Process {
    child: std::process::Child,
    websocket: std::net::SocketAddr,
}

impl Process {
    /// Start `wai-relay --config <config>` and read where its WebSocket listener bound from its
    /// log: the relay binds port 0 itself, and no port is picked for it.
    fn start(config: &Path) -> Process {
        let mut child = std::process::Command::new(env!("CARGO_BIN_EXE_wai-relay"))
            .arg("--config")
            .arg(config)
            .env("NO_COLOR", "1")
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::inherit())
            .spawn()
            .expect("wai-relay starts");
        let out = child.stdout.take().expect("its log");
        let (tx, rx) = std::sync::mpsc::channel();
        // The log is read to its end, so the relay never blocks on a full pipe.
        std::thread::spawn(move || {
            for line in std::io::BufReader::new(out).lines().map_while(Result::ok) {
                if let Some(at) = line.split("on tcp ").nth(1).and_then(|r| r.split(',').next()).and_then(|a| a.trim().parse().ok()) {
                    let _ = tx.send(at);
                }
            }
        });
        let mut p = Process { child, websocket: "127.0.0.1:0".parse().unwrap() };
        p.websocket = rx.recv_timeout(STEP).expect("the relay logs where its WebSocket listener bound");
        p
    }

    /// SIGKILL: no handler runs, no session ends, nothing is written.
    fn kill(mut self) {
        self.child.kill().expect("SIGKILL");
        let _ = self.child.wait();
    }

    async fn client(&self) -> Client {
        let url = format!("ws://{}/moq", self.websocket);
        let conn = timeout(STEP, websocket::connect(&url, MuxConfig::default())).await.expect("in time").expect("a WebSocket session");
        Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP")
    }
}

impl Drop for Process {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

/// The checkpoint at `path`, once it is there and `holds` is true of it.
async fn checkpoint_when(path: &Path, holds: impl Fn(&Checkpoint) -> bool) -> (Checkpoint, Vec<u8>) {
    let mut last = None;
    for _ in 0..(STEP.as_millis() / 20) {
        if let Ok(bytes) = std::fs::read(path)
            && let Ok(c) = serde_json::from_slice::<SignedCheckpoint>(&bytes)
        {
            if holds(&c.checkpoint) {
                return (c.checkpoint, bytes);
            }
            last = Some(c.checkpoint);
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    panic!("no checkpoint that holds the counts in time; the last: {last:?}");
}

fn media_in(parts: &[joule_moq_relay::SessionSummary], media: &str) -> u64 {
    parts.iter().filter_map(|p| p.tally.publisher.tracks.get(media)).map(|t| t.objects_received).sum()
}

fn media_out(parts: &[joule_moq_relay::SessionSummary], media: &str) -> u64 {
    parts.iter().filter_map(|p| p.tally.subscriber.tracks.get(media)).map(|t| t.objects_forwarded).sum()
}

/// A publisher and a viewer of `t` on `relay`, the viewer subscribed.
async fn pair(relay: &Process, t: &Tracks) -> (Client, Client, Subscribed, Vec<(joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)>) {
    let p = relay.client().await;
    let held = publish_tracks(&p, t, None).await;
    let v = relay.client().await;
    let s = subscribe_both(&v, t, Parameters::new()).await;
    (p, v, s, held)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_killed_outright_reads_its_open_window_back() {
    timeout(Duration::from_secs(300), kill_run()).await.expect("the test finishes in time");
}

async fn kill_run() {
    let t0 = now_us() - 1_000_000;
    let room = hex::encode([0xa1; 32]);
    let pk = SigningKey::from_bytes(&[0xa2; 32]);
    let d = dir("kill");
    let (sock, ckpt, cfg) = (d.join("c.sock"), d.join("open.json"), d.join("relay.toml"));
    std::fs::write(d.join("relay.key"), hex::encode([0xa3u8; 32])).unwrap();
    std::fs::write(
        &cfg,
        format!(
            "websocket_listen = \"127.0.0.1:0\"\n[summary]\nkey_path = {:?}\n[metering]\nenergy = \"off\"\n[control]\nsocket = {:?}\ncheckpoint_path = {:?}\ncheckpoint_interval_ms = 50\nflush_grace_ms = 200\n",
            d.join("relay.key"),
            sock,
            ckpt
        ),
    )
    .unwrap();
    let t = Tracks::new(ns(&[&room, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));

    // ── The first run: window 0 sealed, two groups after it, then SIGKILL ──
    let first = Process::start(&cfg);
    let (p, v, s, _held) = pair(&first, &t).await;
    for g in 0..3 {
        publish_group(&p, &pk, &t, g, &payload("k", g, 300)).await;
    }
    let got = collect(&v, STEP, |x| x.len() >= 6).await;
    assert_eq!(verify(&got, &s, &pk).0, 3);
    checkpoint_when(&ckpt, |c| media_out(&c.parts, &media) == 3).await;
    let f = ok(&sock, json!({"op": "flush"})).await;
    assert_eq!(f["window"], 0);
    let w0 = statement(&sock, 0).await;
    assert_eq!((media_in(&w0.statement.parts, &media), media_out(&w0.statement.parts, &media)), (3, 3));
    assert!(w0.statement.resumed.is_empty(), "a window one run held from end to end carries no mark");
    for g in 3..5 {
        publish_group(&p, &pk, &t, g, &payload("k", g, 300)).await;
    }
    let got = collect(&v, STEP, |x| x.len() >= 4).await;
    assert_eq!(got.len(), 4, "groups 3 and 4 delivered");
    // The checkpoint holds them: of window 1, both sessions' counts since the seal.
    let (c1, c1_bytes) = checkpoint_when(&ckpt, |c| c.window == 1 && media_in(&c.parts, &media) == 2 && media_out(&c.parts, &media) == 2).await;
    assert!(!c1.clean && c1.parts.len() == 2, "{c1:?}");
    let old: Vec<u64> = c1.parts.iter().map(|p| p.session).collect();
    let killed_at = now_us();
    first.kill();
    drop((p, v));

    // ── The second run reads the window back ──
    let second = Process::start(&cfg);
    let restarted_by = now_us();
    let again = statement(&sock, 0).await;
    assert_eq!(again, w0, "the last window it sealed, as it stated it before the kill");
    let (p, v, s, _held) = pair(&second, &t).await;
    for g in 5..7 {
        publish_group(&p, &pk, &t, g, &payload("k", g, 300)).await;
    }
    let got = collect(&v, STEP, |x| x.len() >= 4).await;
    assert_eq!(verify(&got, &s, &pk).0, 2);
    checkpoint_when(&ckpt, |c| media_out(&c.parts, &media) == 4).await;
    let f = ok(&sock, json!({"op": "flush"})).await;
    assert_eq!(f["window"], 1, "the window the checkpoint was of, under the same index");
    let w1 = statement(&sock, 1).await;
    assert_eq!(w1.statement.from_unix_us, w0.statement.to_unix_us, "it begins where window 0 ended");
    // The mark: resumed from the checkpoint, and the span no count holds, unmetered.
    assert_eq!(w1.statement.resumed.len(), 1, "{:?}", w1.statement.resumed);
    let m = &w1.statement.resumed[0];
    assert_eq!(m.checkpoint_refused, None);
    let at = m.checkpoint_unix_us.expect("the checkpoint it read");
    assert!(at >= c1.at_unix_us && at <= killed_at, "a checkpoint taken before the kill: {at}");
    let span = m.unmetered.clone().expect("the span from the checkpoint to the restart is unmetered, never zero");
    assert_eq!(span.from_unix_us, Some(at));
    assert!(span.to_unix_us == m.restarted_unix_us && span.to_unix_us >= killed_at && span.to_unix_us <= restarted_by, "{span:?}");
    // The killed sessions' counts beside the new sessions', under ids of their own.
    let ids: Vec<u64> = w1.statement.parts.iter().map(|p| p.session).collect();
    assert!(old.iter().all(|o| ids.contains(o)), "the killed sessions' parts: {ids:?}");
    assert_eq!(ids.len(), 4, "two killed, two new, no id given twice: {ids:?}");
    let of = |keep: &dyn Fn(u64) -> bool| w1.statement.parts.iter().filter(|p| keep(p.session)).cloned().collect::<Vec<_>>();
    let (killed, new) = (of(&|s| old.contains(&s)), of(&|s| !old.contains(&s)));
    assert_eq!((media_in(&killed, &media), media_out(&killed, &media)), (2, 2), "groups 3 and 4, from the checkpoint");
    assert_eq!((media_in(&new, &media), media_out(&new, &media)), (2, 2), "groups 5 and 6, from this run");
    assert!(killed.iter().all(|p| p.closed_unix_ms * 1000 <= killed_at && p.close_code.is_none()), "their time ends at the checkpoint");
    let rec1 = record(&sock, &room, t0, 1).await;
    w1.check_record(&rec1, t0).expect("the room's record over the resumed window recomputes from its parts");
    assert_eq!(rec1.joules, 0, "its energy is the unmetered marker");
    // Every group once, over both windows.
    assert_eq!(media_in(&w0.statement.parts, &media) + media_in(&w1.statement.parts, &media), 7, "seven groups in, each counted once");
    assert_eq!(media_out(&w0.statement.parts, &media) + media_out(&w1.statement.parts, &media), 7);

    // ── Planted: the checkpoint from before window 1's seal, put back ──
    second.kill();
    drop((p, v));
    std::fs::write(&ckpt, &c1_bytes).unwrap();
    let third = Process::start(&cfg);
    let (p, v, s, _held) = pair(&third, &t).await;
    publish_group(&p, &pk, &t, 7, &payload("k", 7, 300)).await;
    let got = collect(&v, STEP, |x| x.len() >= 2).await;
    assert_eq!(verify(&got, &s, &pk).0, 1);
    checkpoint_when(&ckpt, |c| c.window == 2 && media_out(&c.parts, &media) == 1).await;
    let f = ok(&sock, json!({"op": "flush"})).await;
    assert_eq!(f["window"], 2);
    let w2 = statement(&sock, 2).await;
    let m = &w2.statement.resumed[0];
    let why = m.checkpoint_refused.clone().expect("the stale checkpoint is refused, with its reason");
    assert!(why.starts_with(checkpoint::STALE), "{why}");
    assert_eq!(m.checkpoint_unix_us, None);
    assert_eq!(m.unmetered.as_ref().and_then(|u| u.from_unix_us), Some(w1.statement.to_unix_us), "unmetered since the last seal");
    assert_eq!((media_in(&w2.statement.parts, &media), media_out(&w2.statement.parts, &media)), (1, 1), "group 7 alone: groups 3 and 4 are not billed twice");
    assert!(w2.statement.parts.iter().all(|p| !old.contains(&p.session)), "a refused checkpoint's sessions");
    assert_eq!(statement(&sock, 1).await, w1, "window 1 as it was sealed");
    eprintln!("planted faults caught (kill): 1 of 1");
    p.close(0, "done");
    v.close(0, "done");
    drop(third);
    let _ = std::fs::remove_dir_all(&d);
}

// ── A truncated checkpoint, another relay's, and a clean stop ───────────────

fn configured(d: &Path, seed: [u8; 32]) -> impl Fn(&mut RelayConfig) + '_ {
    std::fs::write(d.join("relay.key"), hex::encode(seed)).unwrap();
    move |c: &mut RelayConfig| {
        c.control.socket = Some(d.join("c.sock"));
        c.control.checkpoint_path = Some(d.join("open.json"));
        c.control.checkpoint_interval_ms = 50;
        c.control.flush_grace_ms = 100;
        c.summary.key_path = Some(d.join("relay.key"));
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_checkpoint_that_cannot_be_trusted_is_refused_and_a_clean_stop_leaves_nothing_uncounted() {
    timeout(Duration::from_secs(240), faults_run()).await.expect("the test finishes in time");
}

async fn faults_run() {
    let room = hex::encode([0xb1; 32]);
    let pk = SigningKey::from_bytes(&[0xb2; 32]);
    let d = dir("faults");
    let seed = [0xb3u8; 32];
    let (sock, ckpt) = (d.join("c.sock"), d.join("open.json"));
    let t = Tracks::new(ns(&[&room, &hexkey(&pk), "video"]), "0", 10, 11, 64);
    let media = joule_moq::draft22::names::render(&t.ns, Some(&t.media));

    // A run that counts three groups and stops cleanly, as SIGTERM stops it.
    let server = relay(configured(&d, seed)).await;
    let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (v, _ev) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let s = subscribe_both(&v, &t, Parameters::new()).await;
    for g in 0..3 {
        publish_group(&p, &pk, &t, g, &payload("f", g, 300)).await;
    }
    let got = collect(&v, STEP, |x| x.len() >= 6).await;
    assert_eq!(verify(&got, &s, &pk).0, 3);
    let (open, _) = checkpoint_when(&ckpt, |c| media_out(&c.parts, &media) == 3).await;
    assert!(!open.clean, "a checkpoint of a running relay");
    p.close(0, "done");
    v.close(0, "done");
    server.shutdown().await;
    let (stopped, whole) = checkpoint_when(&ckpt, |c| c.clean).await;
    assert_eq!((media_in(&stopped.parts, &media), media_out(&stopped.parts, &media)), (3, 3), "the stop wrote every count");

    // Started again: the window holds them, and nothing is uncounted.
    let server = relay(configured(&d, seed)).await;
    let marks = server.relay().resumed();
    assert_eq!(marks.len(), 1);
    assert!(marks[0].checkpoint_refused.is_none() && marks[0].unmetered.is_none() && marks[0].checkpoint_unix_us == Some(stopped.at_unix_us), "{marks:?}");
    ok(&sock, json!({"op": "flush"})).await;
    let w0 = statement(&sock, 0).await;
    assert_eq!((media_in(&w0.statement.parts, &media), media_out(&w0.statement.parts, &media)), (3, 3), "a clean stop loses nothing");
    assert_eq!(w0.statement.resumed, marks, "and the window says it was resumed, with no unmetered span");
    assert!(server.relay().resumed().is_empty(), "the seal took the mark");
    server.shutdown().await;

    // Planted: each refused with its reason, in the next window's statement, and counted nothing.
    let mut caught = 0;
    let other = SigningKey::from_bytes(&[0xb4; 32]);
    let theirs = {
        let mut c = stopped.clone();
        c.relay_key = hexkey(&other);
        c.window = 1;
        serde_json::to_vec(&SignedCheckpoint::sign(c, &other)).unwrap()
    };
    let raised = {
        let mut c: SignedCheckpoint = serde_json::from_slice(&whole).unwrap();
        c.checkpoint.window = 1;
        c.checkpoint.parts[0].tally.subscriber.objects_forwarded += 100;
        serde_json::to_vec(&c).unwrap()
    };
    let planted = [(whole[..whole.len() * 2 / 3].to_vec(), checkpoint::TRUNCATED), (theirs, checkpoint::OTHER_RELAY), (raised, checkpoint::BAD_SIGNATURE)];
    for (window, (planted, reason)) in (1u64..).zip(planted) {
        std::fs::write(&ckpt, &planted).unwrap();
        let server = relay(configured(&d, seed)).await;
        let f = ok(&sock, json!({"op": "flush"})).await;
        assert_eq!(f["window"], window, "{reason}: the window after the last sealed");
        let w = statement(&sock, window).await;
        let m = &w.statement.resumed[0];
        assert!(m.checkpoint_refused.as_deref().is_some_and(|r| r.starts_with(reason)), "{reason}: {m:?}");
        assert!(w.statement.parts.is_empty(), "{reason}: a refused checkpoint's counts were read: {:?}", w.statement.parts);
        assert!(m.unmetered.is_some() && m.checkpoint_unix_us.is_none(), "{reason}: the span since the last seal is unmetered: {m:?}");
        caught += 1;
        server.shutdown().await;
    }
    eprintln!("planted faults caught (checkpoint): {caught} of 3");
    assert_eq!(caught, 3);

    // No key of its own: no checkpoint could be read back, so the relay does not start.
    let mut c = RelayConfig::loopback();
    c.control.checkpoint_path = Some(ckpt.clone());
    let e = joule_moq_relay::Relay::new(c).err().expect("a checkpoint with no summary key");
    assert!(e.contains("summary.key_path"), "{e}");
    let _ = std::fs::remove_dir_all(&d);
}

// ── A seal that waits on its disk ───────────────────────────────────────────

/// A seal writes the window it sealed beside the checkpoint, synced; a disk under other work
/// can hold that write for seconds (above). Here the write waits until something reads it: the
/// file it is written through is a FIFO. On a runtime of one worker a timer still fires while
/// the seal waits, so the seal waits off the workers that carry sessions; once the FIFO is
/// read the seal finishes, its write refused (a FIFO cannot be synced) and counted, and the
/// checkpoint after it written. A watchdog thread reads the FIFO after 5 s whatever happens, so
/// a seal that holds the worker fails the test rather than hanging it.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn a_seal_that_waits_on_its_disk_leaves_the_workers_free() {
    let d = dir("disk");
    let ckpt = d.join("open.json");
    let fifo = PathBuf::from(format!("{}.writing", checkpoint::sealed_path(&ckpt).display()));
    assert!(std::process::Command::new("mkfifo").arg(&fifo).status().expect("mkfifo").success());
    let conf = configured(&d, [0xc3; 32]);
    // No checkpoint on the interval while the test runs: the seal's own writes alone.
    let server = relay(|c| {
        conf(c);
        c.control.checkpoint_interval_ms = 3_600_000;
    })
    .await;
    let released = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let watchdog = {
        let (fifo, released) = (fifo.clone(), released.clone());
        std::thread::spawn(move || {
            for _ in 0..100 {
                if released.load(std::sync::atomic::Ordering::SeqCst) {
                    return;
                }
                std::thread::sleep(Duration::from_millis(50));
            }
            let _ = std::fs::read(&fifo);
        })
    };
    let r = server.relay().clone();
    let sealing = tokio::spawn(async move { r.flush_and_seal(Duration::ZERO).await });
    let t0 = std::time::Instant::now();
    tokio::time::sleep(Duration::from_millis(300)).await;
    let waited = t0.elapsed();
    assert!(waited < Duration::from_secs(3), "a 300 ms timer fired after {waited:?}: the seal's write held the runtime's worker");
    released.store(true, std::sync::atomic::Ordering::SeqCst);
    let body = {
        let fifo = fifo.clone();
        tokio::task::spawn_blocking(move || std::fs::read(&fifo)).await.unwrap().expect("the seal's write, read through the FIFO")
    };
    let w = timeout(STEP, sealing).await.expect("the seal finishes once its write is read").unwrap();
    assert_eq!(w.index, 0);
    let sealed: checkpoint::SignedSealed = serde_json::from_slice(&body).expect("the sealed window, whole");
    assert_eq!(sealed.sealed.window.window, 0);
    let (written, failed) = server.relay().checkpoints();
    assert_eq!(written, 1, "the checkpoint after the seal written");
    #[cfg(target_os = "linux")]
    assert_eq!(failed, 1, "the sealed window's write refused: a FIFO is not synced (EINVAL)");
    let _ = failed;
    let back: SignedCheckpoint = serde_json::from_slice(&std::fs::read(&ckpt).unwrap()).unwrap();
    assert_eq!(back.checkpoint.window, 1, "the checkpoint after the seal is of the window open now");
    watchdog.join().unwrap();
    server.shutdown().await;
    let _ = std::fs::remove_dir_all(&d);
}