lernie 0.1.64

lernie: the operator seat — the window and wire client for a yog server
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

//! **One line to one engine** (yog's `docs/REMOTE.md` §13.4; DESIGN §4.40):
//! the stream a rung of the ladder handed back, how it is read, and what is
//! kept of it once the answer is in.
//!
//! **A punched line is held between asks, and a direct one is not.** First
//! contact over the commons costs seconds — an inbox write, a poll period, a
//! punch window — so the connection it buys is kept and reused, which is the
//! payer §10 of that document waited for. A direct dial costs milliseconds
//! and stays what it always was: one connection per ask.
//!
//! **The engine pings the silence and this end discards the ping.** Between
//! asks the engine writes `{"ping":true}` as a frame every 25 seconds of
//! nothing, to keep both NATs' mappings alive; it is never the start of a
//! reply stream, so [`read`] skips it wherever a frame is read. The engine
//! hangs up after two minutes of silence in all, so a held line is judged
//! gone **one ping before that bound** — the same arithmetic the engine's own
//! `pings_at` runs — rather than reused at the moment the far end is closing
//! it. Judged by the injected clock, so the suite crosses the bound by hand.
//!
//! **And a held line is asked whether it is still open before it is spoken
//! on.** Whatever the socket holds is drawn into rustls without consuming a
//! byte of plaintext — a FIN, a `close_notify`, a reset, or nothing but
//! pings still to be read — so a dropped connection re-enters the ladder at
//! the next ask instead of turning it into an act IN DOUBT. What that cannot
//! see is a peer that vanishes *after* the check, which is REMOTE §3's doubt
//! and is answered the way every lost reply is: by a read.

use std::io::{self, ErrorKind};
use std::net::TcpStream;
use std::time::{Duration, Instant};

use rustls::{ClientConnection, StreamOwned};
use serde_json::{Value, json};

use super::frame;

/// How long a read may find nothing, in all, before the engine is judged
/// gone — the engine's own bound, and the bound it hangs up on.
///
/// It is a bound on the **transport**, not on the wait. An ordinary ask is
/// answered at once; a follow-class read is answered by the engine at the
/// rate the thing being followed writes, so this has to sit comfortably
/// above that cadence or an idle tail would read as a dead channel.
pub(crate) const GONE: Duration = Duration::from_mins(2);
/// How often the engine pings a held line's silence.
pub(crate) const PING: Duration = Duration::from_secs(25);

/// The mTLS stream over one TCP connection.
pub(crate) type Tls = StreamOwned<ClientConnection, TcpStream>;

/// A connection a rung produced, with a request about to go down it.
pub(crate) struct Line {
    pub(crate) tls: Tls,
    /// What the engine stated it could spell on this connection's preface.
    pub(crate) spelled: u32,
    /// Whether the preface is still owed: a stream a rung just opened.
    pub(crate) fresh: bool,
    /// Whether the line is kept once the answer is in: a punched one is.
    pub(crate) kept: bool,
}

impl Line {
    /// A held line taken back up: its preface was exchanged when it was new.
    pub(crate) fn from_held(held: Held) -> Line {
        Line {
            tls: held.tls,
            spelled: held.spelled,
            fresh: false,
            kept: true,
        }
    }

    /// The answer is in: a kept line goes back to the entry's pool, timed
    /// from now; any other is dropped, which is the hang-up.
    pub(crate) fn give_back(self, key: &str, now: Instant) {
        if !self.kept {
            return;
        }
        let held = Held {
            tls: self.tls,
            spelled: self.spelled,
            heard: now,
        };
        crate::state::worked(key, |w| w.held.push(held));
    }
}

/// A line kept between asks.
pub(crate) struct Held {
    tls: Tls,
    spelled: u32,
    /// When the last answer on it was complete.
    heard: Instant,
}

impl Held {
    /// Why this line is no longer one the engine would answer on — past the
    /// silence bound with no ping to spare, or hung up at the far end — or
    /// `None` while it still is.
    pub(crate) fn gone(&mut self, now: Instant) -> Option<&'static str> {
        if now.saturating_duration_since(self.heard) + PING >= GONE {
            Some("past the silence bound")
        } else if open(&mut self.tls) {
            None
        } else {
            Some("closed at the far end")
        }
    }
}

/// Draw whatever the socket holds into rustls without reading a byte of
/// plaintext, and say whether the peer is still there. What rustls already
/// holds is judged first — a `close_notify` that rode in behind the last
/// answer is known before the socket is asked for more.
fn open(tls: &mut Tls) -> bool {
    let _ = tls.sock.set_nonblocking(true);
    let verdict = loop {
        match tls.conn.process_new_packets() {
            Ok(state) if state.peer_has_closed() => break false,
            Ok(_) => {}
            Err(_) => break false,
        }
        if !tls.conn.wants_read() {
            break true;
        }
        match tls.conn.read_tls(&mut tls.sock) {
            Ok(0) => break false,
            Ok(_) => {}
            Err(e) if e.kind() == ErrorKind::WouldBlock => break true,
            Err(_) => break false,
        }
    };
    let _ = tls.sock.set_nonblocking(false);
    verdict
}

/// Read one frame that is not a ping: `Some` a frame of the answer, `None`
/// the terminator. The engine's ping is discarded here, wherever a frame is
/// read, because it is never part of any answer — and counted into `pings`,
/// so the discard can be said.
pub(crate) fn read(r: &mut dyn io::Read, pings: &mut usize) -> io::Result<Option<Value>> {
    loop {
        match frame::read_value(r)? {
            Some(frame) if frame == ping() => *pings += 1,
            other => return Ok(other),
        }
    }
}

/// The frame the engine writes into a held line's silence.
pub(crate) fn ping() -> Value {
    json!({"ping": true})
}

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

    #[test]
    fn the_bounds_are_the_engines() {
        assert_eq!(GONE, Duration::from_mins(2));
        assert_eq!(PING, Duration::from_secs(25));
        assert_eq!(ping(), json!({"ping": true}));
    }

    #[test]
    fn a_read_skips_every_ping_and_hands_back_the_frame_or_the_end() {
        let mut bytes = Vec::new();
        frame::write_value(&mut bytes, &ping()).unwrap();
        frame::write_value(&mut bytes, &ping()).unwrap();
        frame::write_value(&mut bytes, &json!({"n": 1})).unwrap();
        frame::write_value(&mut bytes, &ping()).unwrap();
        frame::write_end(&mut bytes).unwrap();
        let mut cursor = std::io::Cursor::new(bytes);
        let mut pings = 0;
        assert_eq!(
            read(&mut cursor, &mut pings).unwrap(),
            Some(json!({"n": 1}))
        );
        assert_eq!(pings, 2);
        assert_eq!(read(&mut cursor, &mut pings).unwrap(), None);
        assert_eq!(pings, 3);
    }
}