pipecrab-runtime 0.7.0

Runtime-agnostic async orchestration for pipecrab: Inbound, Outbound. Built on futures.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//! Tests for the priority mailbox.
//!
//! Contract: FIFO within a lane; `sys` preempts a backed-up `data` lane; the
//! sys direction tag is carried through untouched; data lane always yields
//! `Received::Data`. The preemption is exercised in both directions a system
//! frame travels — an `Interrupt` going down and an `Error` going up — since
//! fast upstream failure depends on the error jumping the data backlog.
//!
//! Lane-close semantics (one lane closed, both closed) live as unit tests in
//! `src/inbound.rs`: a single `Outbound` owns both senders, so `link` cannot
//! close one lane on its own.

use futures::executor::block_on;
use pipecrab_core::{DataFrame, Direction, SystemFrame, Transcript};
use pipecrab_runtime::{Received, link};

#[test]
fn interrupt_preempts_backed_up_data() {
    block_on(async {
        let (out, mut inb) = link(16);
        for i in 0..8 {
            out.send_data(Transcript::user_final(i.to_string()).into())
                .await
                .unwrap();
        }
        out.send_system(Direction::Down, SystemFrame::Interrupt)
            .await
            .unwrap();

        let r = inb.recv().await.unwrap();
        assert!(
            matches!(r, Received::Sys(Direction::Down, SystemFrame::Interrupt)),
            "interrupt must jump the backlog, got {r:?}",
        );
    });
}

#[test]
fn fatal_error_propagates_upstream_ahead_of_data() {
    block_on(async {
        let (out, mut inb) = link(16);
        for i in 0..8 {
            out.send_data(Transcript::user_final(i.to_string()).into())
                .await
                .unwrap();
        }
        out.send_system(
            Direction::Up,
            SystemFrame::Error {
                message: "inference exploded".into(),
                fatal: true,
            },
        )
        .await
        .unwrap();

        match inb.recv().await.unwrap() {
            Received::Sys(Direction::Up, SystemFrame::Error { message, .. }) => {
                assert_eq!(message, "inference exploded".into());
            }
            other => panic!("expected Sys(Up, Error), got {other:?}"),
        }
    });
}

#[test]
fn data_lane_is_fifo() {
    block_on(async {
        let (out, mut inb) = link(16);
        for i in 0..4 {
            out.send_data(Transcript::user_final(i.to_string()).into())
                .await
                .unwrap();
        }
        for i in 0..4 {
            match inb.recv().await.unwrap() {
                Received::Data(DataFrame::Transcript(s)) => {
                    assert_eq!(s.text, i.to_string().into())
                }
                other => panic!("expected Data(Transcript({i})), got {other:?}"),
            }
        }
    });
}

#[test]
fn data_lane_is_always_downstream() {
    block_on(async {
        let (out, mut inb) = link(16);
        out.send_data(Transcript::user_final("a").into())
            .await
            .unwrap();
        out.send_data(Transcript::user_final("b").into())
            .await
            .unwrap();
        assert!(matches!(inb.recv().await.unwrap(), Received::Data(_)));
        assert!(matches!(inb.recv().await.unwrap(), Received::Data(_)));
    });
}

#[test]
fn dropping_the_outbound_closes_both_lanes() {
    block_on(async {
        let (out, mut inb) = link(16);
        out.send_data(Transcript::user_final("buffered").into())
            .await
            .unwrap();
        drop(out);

        // Buffered frames are still served after the sender side is gone…
        match inb.recv().await.unwrap() {
            Received::Data(DataFrame::Transcript(s)) => {
                assert_eq!(s.text, "buffered".into())
            }
            other => panic!("buffered frame must survive sender drop, got {other:?}"),
        }
        // …and only then does recv signal shutdown.
        assert!(
            inb.recv().await.is_none(),
            "closed lanes must signal shutdown via None"
        );
    });
}