ripsync_core/net/
mod.rs

1//! Remote sync over a single duplex byte stream (ssh, or an in-process pipe).
2//!
3//! Both peers are ripsync. One side is the **initiator** (the locally-invoked
4//! `ripsync`, which spawns `ssh host ripsync --server`); the other is the
5//! **responder** (`ripsync --server`). After a versioned handshake the connection
6//! carries the receiver-driven lock-step protocol described in [`proto`].
7//!
8//! Role mapping:
9//! * [`proto::Role::Push`] — local → remote: local is the **sender**, the remote
10//!   server is the **receiver**.
11//! * [`proto::Role::Pull`] — remote → local: the remote server is the sender,
12//!   local is the receiver.
13
14pub mod proto;
15pub mod receiver;
16pub mod sender;
17pub mod server;
18pub mod transport;
19
20use std::io::{Read, Write};
21use std::path::Path;
22
23use crate::filter::Filter;
24use crate::net::proto::{Init, Msg, NetOptions, PROTO_VERSION, Role};
25use crate::net::transport::{recv_msg, send_msg};
26use crate::report::{Reporter, Stats};
27use crate::{Error, Result, RunControl};
28
29pub use proto::MAGIC;
30pub use receiver::run_receiver;
31pub use sender::run_sender;
32pub use transport::{IoDuplex, duplex_pair};
33
34/// Perform the initiator side of the handshake: send magic+version, check the
35/// server's hello, send [`Init`], await the ack.
36fn handshake_initiator<C: Read + Write>(
37    conn: &mut C,
38    role: Role,
39    root: &Path,
40    options: NetOptions,
41) -> Result<()> {
42    let mut hello = Vec::with_capacity(11);
43    hello.extend_from_slice(MAGIC);
44    hello.extend_from_slice(&PROTO_VERSION.to_be_bytes());
45    conn.write_all(&hello)?;
46    conn.flush()?;
47
48    match recv_msg(conn)? {
49        Msg::ServerHello { version, ok } => {
50            if !ok || version != PROTO_VERSION {
51                return Err(Error::Protocol(format!(
52                    "remote ripsync protocol v{version} is incompatible (need v{PROTO_VERSION})"
53                )));
54            }
55        }
56        other => {
57            return Err(Error::Protocol(format!(
58                "expected ServerHello, got {other:?}"
59            )));
60        }
61    }
62
63    send_msg(
64        conn,
65        &Msg::Init(Init {
66            role,
67            root: root.to_path_buf(),
68            options,
69        }),
70    )?;
71    match recv_msg(conn)? {
72        Msg::InitAck => Ok(()),
73        other => Err(Error::Protocol(format!("expected InitAck, got {other:?}"))),
74    }
75}
76
77/// Perform the responder side of the handshake: validate magic+version, send the
78/// hello, read [`Init`], ack it.
79pub(crate) fn handshake_responder<C: Read + Write>(conn: &mut C) -> Result<Init> {
80    let mut buf = [0u8; 11];
81    conn.read_exact(&mut buf)?;
82    let magic_ok = &buf[..7] == MAGIC.as_slice();
83    let version = u32::from_be_bytes([buf[7], buf[8], buf[9], buf[10]]);
84    let ok = magic_ok && version == PROTO_VERSION;
85
86    send_msg(
87        conn,
88        &Msg::ServerHello {
89            version: PROTO_VERSION,
90            ok,
91        },
92    )?;
93    if !ok {
94        return Err(Error::Protocol(format!(
95            "bad handshake (magic_ok={magic_ok}, version={version})"
96        )));
97    }
98
99    match recv_msg(conn)? {
100        Msg::Init(init) => {
101            send_msg(conn, &Msg::InitAck)?;
102            Ok(init)
103        }
104        other => Err(Error::Protocol(format!("expected Init, got {other:?}"))),
105    }
106}
107
108/// Drive the responder side over an established connection: perform the
109/// responder handshake, then play the opposite role to the initiator. Used by
110/// `ripsync --server` (over stdio) and by tests (over an in-process pipe).
111///
112/// # Errors
113///
114/// Returns a handshake, protocol, walk, or I/O error.
115pub fn run_responder<C: Read + Write, R: Reporter>(
116    conn: &mut C,
117    filter: &Filter,
118    threads: usize,
119    control: &RunControl,
120    reporter: &R,
121) -> Result<Stats> {
122    let init = handshake_responder(conn)?;
123    match init.role {
124        // Initiator pushed to us: we are the receiver.
125        Role::Push => run_receiver(
126            conn,
127            &init.root,
128            filter,
129            init.options,
130            threads,
131            control,
132            reporter,
133        ),
134        // Initiator pulled from us: we are the sender.
135        Role::Pull => run_sender(conn, &init.root, filter, init.options, threads, control),
136    }
137}
138
139/// Drive a remote sync from the local (initiator) side over an established
140/// connection. `local_root` is the local path; `remote_root` is the path on the
141/// peer. Returns the final [`Stats`] (for push, these come back from the remote
142/// receiver via [`Msg::Finished`]).
143///
144/// # Errors
145///
146/// Returns a handshake, protocol, walk, or I/O error.
147#[allow(clippy::too_many_arguments)]
148pub fn run_initiator<C: Read + Write, R: Reporter>(
149    conn: &mut C,
150    role: Role,
151    local_root: &Path,
152    remote_root: &Path,
153    options: NetOptions,
154    filter: &Filter,
155    threads: usize,
156    control: &RunControl,
157    reporter: &R,
158) -> Result<Stats> {
159    handshake_initiator(conn, role, remote_root, options)?;
160    match role {
161        Role::Push => run_sender(conn, local_root, filter, options, threads, control),
162        Role::Pull => run_receiver(
163            conn, local_root, filter, options, threads, control, reporter,
164        ),
165    }
166}
ripsync_core/net/mod.rs

ripsync_core/net/
mod.rs
