ripsync_core/net/
transport.rs

1//! Framing and byte-stream plumbing for the remote protocol.
2//!
3//! A *connection* is any value that is both [`Read`] and [`Write`]: an
4//! [`IoDuplex`] over an ssh child's stdout/stdin, over the server's stdin/stdout,
5//! or over the in-process [`duplex_pair`] used by tests and the local self-test.
6//!
7//! Messages are framed as `[u64 little-endian length][bincode bytes]`. A 4 GiB
8//! frame cap bounds peak allocation (whole-file transfers above that are out of
9//! scope for v1 — the delta path keeps large mostly-unchanged files cheap).
10//!
11//! The cap is 256 MiB.
12
13use std::collections::VecDeque;
14use std::io::{self, Read, Write};
15use std::sync::{Arc, Condvar, Mutex};
16
17use crate::net::proto::Msg;
18use crate::{Error, Result};
19
20/// Maximum accepted frame length (256 MiB). Guards against a corrupt or hostile
21/// length prefix forcing an unbounded allocation.
22pub const MAX_FRAME: u64 = 256 * 1024 * 1024;
23
24/// Serialize and write one framed message, flushing the stream.
25///
26/// # Errors
27///
28/// Returns [`Error::Protocol`] if the message cannot be serialized or exceeds
29/// [`MAX_FRAME`], or an I/O error if the write fails.
30pub fn send_msg<W: Write>(w: &mut W, msg: &Msg) -> Result<()> {
31    let bytes = bincode::serialize(msg).map_err(|e| Error::Protocol(format!("serialize: {e}")))?;
32    let len = bytes.len() as u64;
33    if len > MAX_FRAME {
34        return Err(Error::Protocol(format!("frame too large: {len} bytes")));
35    }
36    w.write_all(&len.to_le_bytes())?;
37    w.write_all(&bytes)?;
38    w.flush()?;
39    Ok(())
40}
41
42/// Read one framed message.
43///
44/// # Errors
45///
46/// Returns [`Error::Protocol`] on a bad/oversized frame or a deserialization
47/// failure, or an I/O error (including EOF mid-frame) if the read fails.
48pub fn recv_msg<R: Read>(r: &mut R) -> Result<Msg> {
49    let mut len_buf = [0u8; 8];
50    r.read_exact(&mut len_buf)?;
51    let len = u64::from_le_bytes(len_buf);
52    if len > MAX_FRAME {
53        return Err(Error::Protocol(format!("frame too large: {len} bytes")));
54    }
55    let len_usize = usize::try_from(len)
56        .map_err(|_| Error::Protocol("frame length overflows usize".to_string()))?;
57    let mut buf = vec![0u8; len_usize];
58    r.read_exact(&mut buf)?;
59    bincode::deserialize(&buf).map_err(|e| Error::Protocol(format!("deserialize: {e}")))
60}
61
62/// A bidirectional stream built from a separate reader and writer. Implements
63/// both [`Read`] and [`Write`] by delegating to each half.
64pub struct IoDuplex<R: Read, W: Write> {
65    /// The read half (e.g. ssh child stdout, or server stdin).
66    pub r: R,
67    /// The write half (e.g. ssh child stdin, or server stdout).
68    pub w: W,
69}
70
71impl<R: Read, W: Write> IoDuplex<R, W> {
72    /// Pair a reader and a writer into one duplex stream.
73    pub fn new(r: R, w: W) -> Self {
74        Self { r, w }
75    }
76}
77
78impl<R: Read, W: Write> Read for IoDuplex<R, W> {
79    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
80        self.r.read(buf)
81    }
82}
83
84impl<R: Read, W: Write> Write for IoDuplex<R, W> {
85    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
86        self.w.write(buf)
87    }
88    fn flush(&mut self) -> io::Result<()> {
89        self.w.flush()
90    }
91}
92
93/// Shared state of one in-process byte channel.
94struct PipeState {
95    buf: VecDeque<u8>,
96    closed: bool,
97}
98
99/// The write half of an in-process byte pipe.
100pub struct PipeWriter {
101    inner: Arc<(Mutex<PipeState>, Condvar)>,
102}
103
104/// The read half of an in-process byte pipe.
105pub struct PipeReader {
106    inner: Arc<(Mutex<PipeState>, Condvar)>,
107}
108
109fn new_pipe() -> (PipeWriter, PipeReader) {
110    let inner = Arc::new((
111        Mutex::new(PipeState {
112            buf: VecDeque::new(),
113            closed: false,
114        }),
115        Condvar::new(),
116    ));
117    (
118        PipeWriter {
119            inner: Arc::clone(&inner),
120        },
121        PipeReader { inner },
122    )
123}
124
125impl Write for PipeWriter {
126    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
127        let (lock, cvar) = &*self.inner;
128        let mut state = lock.lock().unwrap();
129        state.buf.extend(buf.iter().copied());
130        cvar.notify_all();
131        Ok(buf.len())
132    }
133    fn flush(&mut self) -> io::Result<()> {
134        Ok(())
135    }
136}
137
138impl Drop for PipeWriter {
139    fn drop(&mut self) {
140        let (lock, cvar) = &*self.inner;
141        let mut state = lock.lock().unwrap();
142        state.closed = true;
143        cvar.notify_all();
144    }
145}
146
147impl Read for PipeReader {
148    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
149        if buf.is_empty() {
150            return Ok(0);
151        }
152        let (lock, cvar) = &*self.inner;
153        let mut state = lock.lock().unwrap();
154        loop {
155            if !state.buf.is_empty() {
156                let n = state.buf.len().min(buf.len());
157                for slot in buf.iter_mut().take(n) {
158                    *slot = state.buf.pop_front().unwrap();
159                }
160                return Ok(n);
161            }
162            if state.closed {
163                return Ok(0); // EOF
164            }
165            state = cvar.wait(state).unwrap();
166        }
167    }
168}
169
170/// A connected pair of in-process duplex endpoints. Whatever endpoint A writes,
171/// endpoint B reads, and vice versa. Used to drive a full sender/receiver
172/// exchange in two threads without spawning a process or touching a socket.
173#[must_use]
174pub fn duplex_pair() -> (
175    IoDuplex<PipeReader, PipeWriter>,
176    IoDuplex<PipeReader, PipeWriter>,
177) {
178    let (a_to_b_w, a_to_b_r) = new_pipe();
179    let (b_to_a_w, b_to_a_r) = new_pipe();
180    let end_a = IoDuplex::new(b_to_a_r, a_to_b_w);
181    let end_b = IoDuplex::new(a_to_b_r, b_to_a_w);
182    (end_a, end_b)
183}
ripsync_core/net/transport.rs

ripsync_core/net/
transport.rs
