//! The edge between two circuits: a producer and a consumer joined by stamped, coalescing
//! deltas equal one circuit running both. Evidence: `context/test-evidence/zset.md`.

mod support;

use proptest::prelude::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use support::Text;
use zset::{
    ChannelPipe, Circuit, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, Flow, Frame,
    Input, Output, Pipe, Recv, Refused, Stamp, ZSet, channel_pipes,
};

type Z = ZSet<u8>;
type Counts = ZSet<(u8, i64)>;

fn z(entries: &[(u8, i64)]) -> Z {
    entries.iter().copied().collect()
}

fn mapped(d: &Z) -> Z {
    d.map(|k| k % 7)
}

/// The producer's graph: one map.
fn producer() -> (Circuit, Input<u8>, Output<u8>) {
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let m = c.map(&s, |k| k % 7);
    let o = c.output(&m);
    (c, i, o)
}

/// The consumer's graph: distinct, then a count per group.
fn consumer_tail(c: &mut Circuit, s: &zset::Stream<Z>) -> Output<(u8, i64)> {
    let d = c.distinct(s);
    let g = c.index_by(&d, |k| k % 3);
    let a = c.aggregate(&g, |_, vals| Some(vals.len() as i64));
    c.output(&a)
}

fn consumer() -> (Circuit, Input<u8>, Output<(u8, i64)>) {
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = consumer_tail(&mut c, &s);
    (c, i, o)
}

/// Both graphs in one circuit: what the edge must be equal to.
fn single() -> (Circuit, Input<u8>, Output<(u8, i64)>) {
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let m = c.map(&s, |k| k % 7);
    let o = consumer_tail(&mut c, &m);
    (c, i, o)
}

/// A pipe end that can lose what is sent through it.
struct FaultPipe<W> {
    inner: ChannelPipe<W>,
    lose: Arc<AtomicUsize>,
}

impl<W: Send> Pipe<W> for FaultPipe<W> {
    fn try_send(&mut self, wire: W) -> Result<(), Refused<W>> {
        if self.lose.load(Ordering::SeqCst) > 0 {
            self.lose.fetch_sub(1, Ordering::SeqCst);
            return Ok(());
        }
        self.inner.try_send(wire)
    }

    fn try_recv(&mut self) -> Result<W, Recv> {
        self.inner.try_recv()
    }
}

fn fault<W: Send>(
    cap: usize,
) -> (
    FaultPipe<W>,
    FaultPipe<W>,
    Arc<AtomicUsize>,
    Arc<AtomicUsize>,
) {
    let (a, b) = channel_pipes(cap);
    let (la, lb) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
    (
        FaultPipe {
            inner: a,
            lose: la.clone(),
        },
        FaultPipe {
            inner: b,
            lose: lb.clone(),
        },
        la,
        lb,
    )
}

type Sink<C> = EdgeSink<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
type Source<C> = EdgeSource<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;

/// Two circuits joined by an edge, and the combined circuit they must equal.
struct World<C: EdgeCodec<u8> + Clone> {
    codec: C,
    prod: Circuit,
    p_in: Input<u8>,
    p_out: Output<u8>,
    sink: Sink<C>,
    lose_fwd: Arc<AtomicUsize>,
    cons: Circuit,
    c_out: Output<(u8, i64)>,
    src: Source<C>,
    lose_back: Arc<AtomicUsize>,
    got: Counts,
    p_total: Z,
    epoch: u64,
    one: Circuit,
    one_in: Input<u8>,
    one_out: Output<(u8, i64)>,
    want: Counts,
}

impl<C: EdgeCodec<u8> + Clone> World<C> {
    fn new(codec: C, cap: usize) -> Self {
        let (prod, p_in, p_out) = producer();
        let (cons, c_in, c_out) = consumer();
        let (one, one_in, one_out) = single();
        let (a, b, lose_fwd, lose_back) = fault(cap);
        World {
            sink: EdgeSink::new(codec.clone(), a, 1),
            src: EdgeSource::new(codec.clone(), b, c_in),
            codec,
            prod,
            p_in,
            p_out,
            lose_fwd,
            cons,
            c_out,
            lose_back,
            got: Counts::new(),
            p_total: Z::new(),
            epoch: 1,
            one,
            one_in,
            one_out,
            want: Counts::new(),
        }
    }

    fn commit(&mut self, d: &Z) {
        self.prod.begin();
        self.prod.push(&self.p_in, d.clone());
        self.prod.step();
        self.prod.commit();
        self.sink.committed_from(&mut self.prod, &self.p_out);
        self.p_total += &mapped(d);
        self.one.push(&self.one_in, d.clone());
        self.one.step();
        self.want += self.one.take(&self.one_out);
    }

    fn abort(&mut self, d: &Z) {
        let before = self.sink.stamp();
        self.prod.begin();
        self.prod.push(&self.p_in, d.clone());
        self.prod.step();
        self.prod.rollback();
        assert_eq!(
            self.sink.stamp(),
            before,
            "[abort] a rolled-back command was stamped"
        );
        assert_eq!(
            self.prod.take(&self.p_out),
            Z::new(),
            "[abort] its output survived"
        );
    }

    fn step_consumer(&mut self) {
        self.cons.step();
        self.got += self.cons.take(&self.c_out);
    }

    fn restart_sink(&mut self) {
        self.epoch += 1;
        let old = std::mem::replace(
            &mut self.sink,
            EdgeSink::new(self.codec.clone(), fault_placeholder(), 0),
        );
        self.sink = old.restart(self.epoch, 0, self.p_total.clone());
    }

    fn restart_source(&mut self) {
        let (cons, c_in, c_out) = consumer();
        let old = std::mem::replace(
            &mut self.src,
            EdgeSource::new(self.codec.clone(), fault_placeholder(), c_in),
        );
        self.src = old.restart(c_in);
        self.cons = cons;
        self.c_out = c_out;
        self.got = Counts::new();
    }

    /// Runs the edge until it is idle, with nothing lost any more.
    fn settle(&mut self) {
        self.lose_fwd.store(0, Ordering::SeqCst);
        self.lose_back.store(0, Ordering::SeqCst);
        for _ in 0..200 {
            self.sink.pump();
            let read = self.src.poll(&mut self.cons, 8);
            self.step_consumer();
            if !self.sink.backlog()
                && !self.src.stale()
                && self.src.stamp() == Some(self.sink.stamp())
            {
                break;
            }
            if read == 0 {
                // Nothing arrived: the host's timeout fires and the receiver asks again.
                self.src.connection_lost();
            }
        }
        assert_eq!(
            self.src.stamp(),
            Some(self.sink.stamp()),
            "[settle] never caught up"
        );
        self.step_consumer();
    }
}

/// A pipe end that is never used: only to fill a slot while the real one is moved.
fn fault_placeholder<W: Send>() -> FaultPipe<W> {
    fault(1).0
}

#[derive(Clone, Debug)]
enum Act {
    Commit(Vec<(u8, i64)>),
    Abort(Vec<(u8, i64)>),
    Pump,
    Poll(usize),
    StepConsumer,
    LoseForward(usize),
    LoseBack(usize),
    RestartSink,
    RestartSource,
    ConnectionLost,
}

fn delta() -> impl Strategy<Value = Vec<(u8, i64)>> {
    prop::collection::vec((0u8..14, -2i64..=2), 0..4)
}

fn act() -> impl Strategy<Value = Act> {
    prop_oneof![
        6 => delta().prop_map(Act::Commit),
        1 => delta().prop_map(Act::Abort),
        4 => Just(Act::Pump),
        4 => (1usize..4).prop_map(Act::Poll),
        3 => Just(Act::StepConsumer),
        1 => (1usize..3).prop_map(Act::LoseForward),
        1 => (1usize..3).prop_map(Act::LoseBack),
        1 => Just(Act::RestartSink),
        1 => Just(Act::RestartSource),
        1 => Just(Act::ConnectionLost),
    ]
}

fn run<C: EdgeCodec<u8> + Clone>(codec: C, cap: usize, acts: &[Act]) {
    let mut w = World::new(codec, cap);
    for a in acts {
        match a {
            Act::Commit(d) => w.commit(&d.iter().copied().collect()),
            Act::Abort(d) => w.abort(&d.iter().copied().collect()),
            Act::Pump => {
                w.sink.pump();
            }
            Act::Poll(n) => {
                w.src.poll(&mut w.cons, *n);
            }
            Act::StepConsumer => w.step_consumer(),
            Act::LoseForward(n) => w.lose_fwd.store(*n, Ordering::SeqCst),
            Act::LoseBack(n) => w.lose_back.store(*n, Ordering::SeqCst),
            Act::RestartSink => w.restart_sink(),
            Act::RestartSource => w.restart_source(),
            Act::ConnectionLost => w.src.connection_lost(),
        }
    }
    w.settle();
    assert_eq!(
        w.got, w.want,
        "[edge] the consumer's output after the edge settles"
    );
}

proptest! {
    /// The differential: any history of commits, aborts, pumps, polls, losses and restarts,
    /// once settled, leaves the consumer where one combined circuit would be.
    #[test]
    fn two_circuits_on_an_edge_equal_one_circuit(
        acts in prop::collection::vec(act(), 0..40), cap in 1usize..4,
    ) {
        run(Direct, cap, &acts);
    }

    /// The same histories through a text wire form.
    #[test]
    fn the_same_histories_through_a_text_codec(
        acts in prop::collection::vec(act(), 0..40), cap in 1usize..4,
    ) {
        run(Text, cap, &acts);
    }
}

/// Runs the edge until quiet, stepping the consumer between polls.
fn drive<C: EdgeCodec<u8>>(
    sink: &mut EdgeSink<u8, C, impl Pipe<C::Wire>>,
    src: &mut EdgeSource<u8, C, impl Pipe<C::Wire>>,
    c: &mut Circuit,
) {
    for _ in 0..20 {
        sink.pump();
        src.poll(c, 100);
        c.step();
    }
}

/// A consumer circuit that only mirrors its input, and a sink and source on a fault pipe.
fn mirror(
    cap: usize,
) -> (
    Sink<Direct>,
    Source<Direct>,
    Circuit,
    Output<u8>,
    Arc<AtomicUsize>,
) {
    let (a, b, lose_fwd, _) = fault(cap);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = c.output(&s);
    (
        EdgeSink::new(Direct, a, 1),
        EdgeSource::new(Direct, b, i),
        c,
        o,
        lose_fwd,
    )
}

#[test]
fn deltas_arrive_in_order_and_are_applied() {
    let (mut sink, mut src, mut c, out, _) = mirror(8);
    sink.committed(z(&[(1, 1)]));
    sink.pump();
    sink.committed(z(&[(2, 1)]));
    sink.pump();
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(c.take(&out), z(&[(1, 1), (2, 1)]));
    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
}

#[test]
fn a_full_pipe_coalesces_exactly_and_keeps_the_range() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(1);
    let mut sink = EdgeSink::new(Direct, a, 1);
    sink.committed(z(&[(1, 1)]));
    assert_eq!(sink.pump(), Flow::Idle);
    for k in 2..6 {
        sink.committed(z(&[(k, 1), (1, -1)]));
        assert_eq!(sink.pump(), Flow::Backlogged, "[full] command {k}");
    }
    assert!(matches!(b.try_recv(), Ok(Frame::Delta(d)) if (d.from, d.to) == (0, 1)));
    assert_eq!(sink.pump(), Flow::Idle);
    let Ok(Frame::Delta(d)) = b.try_recv() else {
        panic!("a merged delta")
    };
    assert_eq!(
        (d.from, d.to),
        (1, 5),
        "[range] the merged delta covers (1, 5]"
    );
    assert_eq!(
        d.delta,
        z(&[(2, 1), (3, 1), (4, 1), (5, 1), (1, -4)]),
        "[sum] coalescing sums exactly"
    );
}

#[test]
fn an_empty_command_is_still_a_stamp() {
    let (mut sink, mut src, mut c, _, _) = mirror(8);
    sink.committed(Z::new());
    sink.committed(Z::new());
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
}

#[test]
fn a_lost_delta_is_seen_in_the_stamps_and_corrected() {
    let (mut sink, mut src, mut c, out, lose) = mirror(8);
    drive(&mut sink, &mut src, &mut c);
    sink.committed(z(&[(1, 1)]));
    sink.pump();
    src.poll(&mut c, 10);
    lose.store(1, Ordering::SeqCst);
    sink.committed(z(&[(2, 1)]));
    sink.pump();
    sink.committed(z(&[(3, 1)]));
    sink.pump();
    src.poll(&mut c, 10);
    assert!(
        src.stale(),
        "[gap] the third delta does not follow the first"
    );
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(c.take(&out), z(&[(1, 1), (2, 1), (3, 1)]), "[correction]");
    assert!(!src.stale());
}

#[test]
fn a_duplicate_delta_is_applied_once() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 10);
    let join = answer(&mut b, 1, 0, Z::new());
    let d = Frame::Delta(Delta {
        epoch: 1,
        from: 0,
        to: 1,
        delta: z(&[(1, 1)]),
    });
    b.try_send(join).unwrap();
    b.try_send(d.clone()).unwrap();
    b.try_send(d).unwrap();
    src.poll(&mut c, 10);
    assert!(!src.stale(), "[dup] a duplicate is not a gap");
    c.step();
    assert_eq!(c.take(&o), z(&[(1, 1)]));
}

#[test]
fn a_restarted_sender_is_corrected_not_trusted() {
    let (mut sink, mut src, mut c, out, _) = mirror(8);
    sink.committed(z(&[(1, 1)]));
    sink.committed(z(&[(2, 1)]));
    drive(&mut sink, &mut src, &mut c);
    let _ = c.take(&out);
    // A new incarnation that has since committed one command on top of {1, 2, 9}.
    let mut sink = sink.restart(2, 0, z(&[(1, 1), (2, 1)]));
    sink.committed(z(&[(9, 1)]));
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(
        c.take(&out),
        z(&[(9, 1)]),
        "[restart] only the difference is applied"
    );
    assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 1 }));
}

/// Answers the receiver's latest request in `b` with a correction that echoes its token.
fn answer(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) -> Frame<u8> {
    let mut token = None;
    while let Ok(f) = b.try_recv() {
        if let Frame::Request(r) = f {
            token = Some(r.token);
        }
    }
    Frame::Correction(Correction {
        epoch,
        through,
        token: token.expect("a request was waiting"),
        target,
    })
}

fn reply(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) {
    let f = answer(b, epoch, through, target);
    b.try_send(f).unwrap();
}

#[test]
fn a_late_reply_from_a_dead_incarnation_changes_nothing() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 10);
    // Epoch 100 reads the first request; its reply is delayed. The sender restarts as 1001.
    let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
    src.connection_lost();
    src.poll(&mut c, 10);
    reply(&mut b, 1001, 1, z(&[(5, 1)]));
    src.poll(&mut c, 10);
    assert_eq!(
        src.stamp(),
        Some(Stamp {
            epoch: 1001,
            seq: 1
        })
    );
    b.try_send(late).unwrap();
    b.try_send(Frame::Delta(Delta {
        epoch: 1001,
        from: 1,
        to: 2,
        delta: z(&[(6, 1)]),
    }))
    .unwrap();
    src.poll(&mut c, 10);
    c.step();
    assert_eq!(
        src.stamp(),
        Some(Stamp {
            epoch: 1001,
            seq: 2
        }),
        "[late] the reply of an incarnation never joined does not move the receiver back"
    );
    assert_eq!(
        c.take(&o),
        z(&[(5, 1), (6, 1)]),
        "[late] and 1001 is not shut out"
    );
}

#[test]
fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 10);
    reply(&mut b, 1, 1, Z::new());
    src.poll(&mut c, 10);
    assert!(!src.stale());
    let wrong = Frame::Correction(Correction {
        epoch: 2,
        through: 9,
        token: 77,
        target: z(&[(3, 1)]),
    });
    b.try_send(wrong).unwrap();
    src.poll(&mut c, 10);
    assert_eq!(
        src.stamp(),
        Some(Stamp { epoch: 1, seq: 1 }),
        "[foreign] not moved"
    );
    assert!(src.stale(), "[foreign] but it asks");
}

#[test]
fn only_the_reply_to_the_latest_request_may_join_a_new_incarnation() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 10);
    let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
    src.connection_lost();
    src.poll(&mut c, 10);
    b.try_send(late).unwrap();
    src.poll(&mut c, 10);
    assert_eq!(
        src.stamp(),
        None,
        "[token] the reply to the first request is not the answer"
    );
    reply(&mut b, 1001, 1, z(&[(5, 1)]));
    src.poll(&mut c, 10);
    assert_eq!(
        src.stamp(),
        Some(Stamp {
            epoch: 1001,
            seq: 1
        })
    );
}

#[test]
fn an_answer_older_than_the_watermark_does_not_clear_stale() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 10);
    reply(&mut b, 1, 3, Z::new());
    src.poll(&mut c, 10);
    src.connection_lost();
    src.poll(&mut c, 10);
    reply(&mut b, 1, 1, Z::new());
    src.poll(&mut c, 10);
    assert_eq!(
        src.stamp(),
        Some(Stamp { epoch: 1, seq: 3 }),
        "[older] never moves back"
    );
    assert!(
        src.stale(),
        "[older] and it does not say the receiver is current"
    );
}

#[test]
fn traffic_never_goes_down_across_a_restart_of_either_end() {
    let (mut sink, mut src, mut c, _, _) = mirror(8);
    sink.committed(z(&[(1, 1)]));
    drive(&mut sink, &mut src, &mut c);
    let (s, r) = (sink.traffic(), src.traffic());
    assert!(s > 0 && r > 0, "[setup] frames moved both ways");
    let sink = sink.restart(2, 0, Z::new());
    let mut c2 = Circuit::new();
    let (i2, _) = c2.input::<u8>();
    let src = src.restart(i2);
    assert_eq!(
        (sink.traffic(), src.traffic()),
        (s, r),
        "[restart] counters kept"
    );
}

#[test]
fn a_new_epoch_frame_re_asks_once_after_a_lost_request() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(64);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 8);
    assert!(
        matches!(b.try_recv(), Ok(Frame::Request(_))),
        "[first] the request"
    );
    // The sender read it and restarted before answering.
    let mut asks = 0;
    for n in 0..5u64 {
        b.try_send(Frame::Delta(Delta {
            epoch: 2,
            from: n,
            to: n + 1,
            delta: Z::new(),
        }))
        .unwrap();
        src.poll(&mut c, 8);
        while let Ok(Frame::Request(_)) = b.try_recv() {
            asks += 1;
        }
    }
    assert_eq!(
        asks, 1,
        "[reask] once for the new epoch, not once per frame"
    );
}

#[test]
fn an_answer_at_the_watermark_says_nothing_newer_and_clears_stale() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 8);
    reply(&mut b, 1, 1, Z::new());
    src.poll(&mut c, 8);
    src.connection_lost();
    src.poll(&mut c, 8);
    reply(&mut b, 1, 1, Z::new());
    src.poll(&mut c, 8);
    assert!(
        !src.stale(),
        "[dup-answer] through == watermark clears stale"
    );
}

#[test]
fn trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let _o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 8);
    reply(&mut b, 1, 1, Z::new());
    src.poll(&mut c, 8);
    src.connection_lost();
    src.poll(&mut c, 8);
    let reply = answer(&mut b, 1, 1, Z::new());
    // A gap seen after the request went out: the reply was cut before it.
    b.try_send(Frame::Delta(Delta {
        epoch: 1,
        from: 4,
        to: 5,
        delta: Z::new(),
    }))
    .unwrap();
    b.try_send(reply).unwrap();
    src.poll(&mut c, 8);
    assert!(src.stale(), "[marks] the reply predates the gap");
}

#[test]
fn a_poll_inside_a_transaction_is_refused() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = c.output(&s);
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 8);
    reply(&mut b, 1, 1, z(&[(1, 1)]));
    // Outside a transaction a rollback of the consumer's step keeps what was polled.
    src.poll(&mut c, 8);
    c.begin();
    c.step();
    c.rollback();
    c.step();
    assert_eq!(
        c.take(&o),
        z(&[(1, 1)]),
        "[txn] a step rolled back after poll loses nothing"
    );
    c.begin();
    let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| src.poll(&mut c, 8)));
    assert!(
        r.is_err(),
        "[txn] a poll inside a transaction would be dropped by rollback"
    );
}

#[test]
#[should_panic(expected = "at least one item")]
fn a_zero_capacity_pipe_is_refused() {
    let _ = channel_pipes::<u8>(0);
}

#[test]
fn a_committed_that_overflows_leaves_the_sink_as_it_was() {
    let (mut sink, mut src, mut c, _, _) = mirror(8);
    sink.committed(z(&[(1, i64::MAX)]));
    let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        sink.committed(z(&[(2, 1), (1, 1)]))
    }));
    assert!(r.is_err(), "the weight overflows");
    assert_eq!(
        sink.stamp(),
        Stamp { epoch: 1, seq: 1 },
        "[overflow] no stamp consumed"
    );
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(
        src.held(),
        &z(&[(1, i64::MAX)]),
        "[overflow] and the key 2 did not slip in"
    );
}

#[test]
fn an_unreadable_frame_asks_for_a_correction() {
    let (a, mut b) = channel_pipes::<String>(8);
    let mut c = Circuit::new();
    let (i, s) = c.input::<u8>();
    let o = c.output(&s);
    let mut src = EdgeSource::new(Text, a, i);
    b.try_send("D 1 0 1 1:1".to_string()).unwrap();
    src.poll(&mut c, 10);
    assert!(
        src.stale(),
        "[initial] a new receiver asks for a correction"
    );
    let request = b.try_recv().unwrap();
    let token = request.split(' ').nth(1).unwrap();
    b.try_send(format!("C 1 1 {token} 1:1")).unwrap();
    src.poll(&mut c, 10);
    assert!(!src.stale());
    b.try_send("garbage".to_string()).unwrap();
    src.poll(&mut c, 10);
    assert!(src.stale(), "[unreadable] a frame that does not decode");
    c.step();
    assert_eq!(c.take(&o), z(&[(1, 1)]));
}

#[test]
fn a_rolled_back_command_does_not_cross() {
    let (mut sink, mut src, mut c, out, _) = mirror(8);
    let (mut p, p_in, p_out) = producer();
    // The host's command fails and rolls back; the sink is not asked.
    p.begin();
    p.push(&p_in, z(&[(1, 1)]));
    p.step();
    p.rollback();
    // The retry is a different command and commits.
    p.begin();
    p.push(&p_in, z(&[(2, 1)]));
    p.step();
    p.commit();
    assert_eq!(
        sink.committed_from(&mut p, &p_out),
        Stamp { epoch: 1, seq: 1 }
    );
    drive(&mut sink, &mut src, &mut c);
    assert_eq!(
        c.take(&out),
        z(&[(2, 1)]),
        "[retry] only the committed command crossed"
    );
}

#[test]
#[should_panic(expected = "only after it commits")]
fn taking_a_delta_inside_a_transaction_is_refused() {
    let (mut sink, ..) = mirror(8);
    let (mut p, p_in, p_out) = producer();
    p.begin();
    p.push(&p_in, z(&[(1, 1)]));
    p.step();
    sink.committed_from(&mut p, &p_out);
}

#[test]
fn a_circuit_on_each_of_two_threads_equals_one_circuit() {
    const N: u64 = 300;
    let deltas: Vec<Z> = (0..N)
        .map(|n| z(&[((n * 7 % 14) as u8, 1 - (n % 3) as i64), ((n % 5) as u8, 1)]))
        .collect();
    let (mut one, one_in, one_out) = single();
    let mut want = Counts::new();
    for d in &deltas {
        one.push(&one_in, d.clone());
        one.step();
        want += one.take(&one_out);
    }
    let (a, b) = channel_pipes::<Frame<u8>>(2);
    let finished = Arc::new(AtomicUsize::new(0));
    let fin = finished.clone();
    let producer_thread = thread::spawn(move || {
        let (mut p, p_in, p_out) = producer();
        let mut sink = EdgeSink::new(Direct, a, 9);
        for (n, d) in deltas.iter().enumerate() {
            p.begin();
            p.push(&p_in, d.clone());
            p.step();
            p.commit();
            sink.committed_from(&mut p, &p_out);
            if n % 3 != 0 {
                sink.pump();
            }
            if n % 7 == 0 {
                thread::yield_now();
            }
        }
        while fin.load(Ordering::SeqCst) == 0 {
            sink.pump();
            thread::yield_now();
        }
    });
    let (mut c, c_in, c_out) = consumer();
    let mut src = EdgeSource::new(Direct, b, c_in);
    let mut got = Counts::new();
    let mut spins = 0u64;
    while src.stamp() != Some(Stamp { epoch: 9, seq: N }) {
        spins += 1;
        assert!(spins < 50_000_000, "[threads] the consumer never caught up");
        src.poll(&mut c, 3);
        c.step();
        got += c.take(&c_out);
        if spins.is_multiple_of(5) {
            thread::yield_now();
        }
    }
    c.step();
    got += c.take(&c_out);
    finished.store(1, Ordering::SeqCst);
    producer_thread.join().unwrap();
    assert_eq!(
        got, want,
        "[threads] the consumer after the producer thread finishes"
    );
}

#[test]
fn a_poll_reads_at_most_the_frames_it_is_given() {
    let (a, mut b) = channel_pipes::<Frame<u8>>(8);
    let mut c = Circuit::new();
    let (i, _) = c.input::<u8>();
    let mut src = EdgeSource::new(Direct, a, i);
    src.poll(&mut c, 0);
    let Ok(Frame::Request(request)) = b.try_recv() else {
        panic!("a request")
    };
    for through in 0..3 {
        let target = Z::new();
        let frame = Frame::Correction(Correction {
            epoch: 1,
            through,
            token: request.token,
            target,
        });
        b.try_send(frame).unwrap();
    }
    assert_eq!(src.poll(&mut c, 2), 2, "[budget] two of three");
    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 1 }));
    assert_eq!(src.poll(&mut c, 2), 1, "[budget] the rest");
    assert_eq!(src.stamp(), Some(Stamp { epoch: 1, seq: 2 }));
}