remote/protocol/
mod.rs

1//! Remote copy protocol definitions for source-destination communication.
2//!
3//! # Protocol Overview
4//!
5//! The remote copy protocol uses TCP for communication between source and destination.
6//! The source listens on two ports: a control port for bidirectional messages and a
7//! data port for file transfers. Both sides exchange messages to coordinate directory
8//! creation, file transfers, and completion.
9//!
10//! See `docs/remote_protocol.md` for the full protocol specification.
11//!
12//! # Message Flow
13//!
14//! ```text
15//! Source                              Destination
16//!   |                                      |
17//!   |  ---- Directory(root, meta) -------> |  Create root, store metadata
18//!   |  ---- Directory(child, meta) ------> |  Create child, store metadata
19//!   |  ---- Symlink(...) ----------------> |  Create symlink
20//!   |  ---- DirStructureComplete --------> |  Structure complete
21//!   |                                      |
22//!   |  <--- DirectoryCreated(root) ------- |
23//!   |  <--- DirectoryCreated(child) ------ |
24//!   |                                      |
25//!   |  ~~~~ File(f, total=N) ~~~~~~~~~~~~> |  Write file, track count
26//!   |  ~~~~ File(...) ~~~~~~~~~~~~~~~~~~-> |  ...
27//!   |                                      |  All files done → apply metadata
28//!   |                                      |
29//!   |  <--- DestinationDone -------------- |  Close send side
30//!   |  (close send side)                   |  (detect EOF)
31//!   |  (detect EOF)                        |  Close connection
32//! ```
33//!
34//! # Error Communication
35//!
36//! The protocol uses asymmetric error communication:
37//! - **Source → Destination**: Must communicate failures (FileSkipped, SymlinkSkipped)
38//!   so destination can track file counts correctly
39//! - **Destination → Source**: Does NOT communicate failures. Destination handles
40//!   errors locally and source continues sending the full structure.
41//!
42//! # Shutdown Sequence
43//!
44//! Shutdown is coordinated through TCP connection closure:
45//! 1. Destination sends `DestinationDone` and closes its send side
46//! 2. Source detects EOF on recv, closes its send side
47//! 3. Destination detects EOF on recv, closes connection
48
49use serde::{Deserialize, Serialize};
50use std::os::unix::fs::MetadataExt;
51use std::os::unix::prelude::PermissionsExt;
52
53#[derive(Clone, Debug, Deserialize, Serialize)]
54pub struct Metadata {
55    pub mode: u32,
56    pub uid: u32,
57    pub gid: u32,
58    pub atime: i64,
59    pub mtime: i64,
60    pub atime_nsec: i64,
61    pub mtime_nsec: i64,
62}
63
64impl common::preserve::Metadata for Metadata {
65    fn uid(&self) -> u32 {
66        self.uid
67    }
68    fn gid(&self) -> u32 {
69        self.gid
70    }
71    fn atime(&self) -> i64 {
72        self.atime
73    }
74    fn atime_nsec(&self) -> i64 {
75        self.atime_nsec
76    }
77    fn mtime(&self) -> i64 {
78        self.mtime
79    }
80    fn mtime_nsec(&self) -> i64 {
81        self.mtime_nsec
82    }
83    fn permissions(&self) -> std::fs::Permissions {
84        std::fs::Permissions::from_mode(self.mode)
85    }
86}
87
88impl common::preserve::Metadata for &Metadata {
89    fn uid(&self) -> u32 {
90        (*self).uid()
91    }
92    fn gid(&self) -> u32 {
93        (*self).gid()
94    }
95    fn atime(&self) -> i64 {
96        (*self).atime()
97    }
98    fn atime_nsec(&self) -> i64 {
99        (*self).atime_nsec()
100    }
101    fn mtime(&self) -> i64 {
102        (*self).mtime()
103    }
104    fn mtime_nsec(&self) -> i64 {
105        (*self).mtime_nsec()
106    }
107    fn permissions(&self) -> std::fs::Permissions {
108        (*self).permissions()
109    }
110}
111
112impl From<&std::fs::Metadata> for Metadata {
113    fn from(metadata: &std::fs::Metadata) -> Self {
114        Metadata {
115            mode: metadata.mode(),
116            uid: metadata.uid(),
117            gid: metadata.gid(),
118            atime: metadata.atime(),
119            mtime: metadata.mtime(),
120            atime_nsec: metadata.atime_nsec(),
121            mtime_nsec: metadata.mtime_nsec(),
122        }
123    }
124}
125
126/// File header sent on unidirectional streams, followed by raw file data.
127#[derive(Debug, Deserialize, Serialize)]
128pub struct File {
129    pub src: std::path::PathBuf,
130    pub dst: std::path::PathBuf,
131    pub size: u64,
132    pub metadata: Metadata,
133    pub is_root: bool,
134}
135
136/// Wrapper that includes size for comparison purposes.
137#[derive(Debug)]
138pub struct FileMetadata<'a> {
139    pub metadata: &'a Metadata,
140    pub size: u64,
141}
142
143impl<'a> common::preserve::Metadata for FileMetadata<'a> {
144    fn uid(&self) -> u32 {
145        self.metadata.uid()
146    }
147    fn gid(&self) -> u32 {
148        self.metadata.gid()
149    }
150    fn atime(&self) -> i64 {
151        self.metadata.atime()
152    }
153    fn atime_nsec(&self) -> i64 {
154        self.metadata.atime_nsec()
155    }
156    fn mtime(&self) -> i64 {
157        self.metadata.mtime()
158    }
159    fn mtime_nsec(&self) -> i64 {
160        self.metadata.mtime_nsec()
161    }
162    fn permissions(&self) -> std::fs::Permissions {
163        self.metadata.permissions()
164    }
165    fn size(&self) -> u64 {
166        self.size
167    }
168}
169
170/// Messages sent from source to destination on the control stream.
171#[derive(Debug, Deserialize, Serialize)]
172pub enum SourceMessage {
173    /// Create directory, store metadata, and declare entry counts for completion tracking.
174    /// Sent during directory tree traversal in depth-first order. Source pre-reads the
175    /// directory children before sending, so counts are known at send time.
176    Directory {
177        src: std::path::PathBuf,
178        dst: std::path::PathBuf,
179        metadata: Metadata,
180        is_root: bool,
181        /// total child entries (files + directories + symlinks) for completion tracking
182        entry_count: usize,
183        /// number of child files, echoed back via `DirectoryCreated` for file sending
184        file_count: usize,
185        /// whether to keep this directory if it ends up empty after filtering
186        keep_if_empty: bool,
187    },
188    /// Create symlink with metadata.
189    Symlink {
190        src: std::path::PathBuf,
191        dst: std::path::PathBuf,
192        target: std::path::PathBuf,
193        metadata: Metadata,
194        is_root: bool,
195    },
196    /// Signal that all directories and symlinks have been sent.
197    /// Required before destination can send `DestinationDone`.
198    /// `has_root_item` indicates whether a root file/directory/symlink will be sent.
199    /// When false (dry-run or filtered root), destination can mark root as complete.
200    DirStructureComplete { has_root_item: bool },
201    /// Notify destination that a file failed to send.
202    /// Counts as a processed entry for the parent directory's completion tracking.
203    FileSkipped {
204        src: std::path::PathBuf,
205        dst: std::path::PathBuf,
206    },
207    /// Notify destination that a symlink failed to read.
208    /// If `is_root` is true, this signals that root processing is complete (even if failed).
209    /// Non-root skipped symlinks count as a processed entry for the parent directory.
210    SymlinkSkipped { src_dst: SrcDst, is_root: bool },
211}
212
213#[derive(Clone, Debug, Deserialize, Serialize)]
214pub struct SrcDst {
215    pub src: std::path::PathBuf,
216    pub dst: std::path::PathBuf,
217}
218
219/// Messages sent from destination to source on the control stream.
220#[derive(Clone, Debug, Deserialize, Serialize)]
221pub enum DestinationMessage {
222    /// Confirm directory created, request file transfers.
223    /// `file_count` is echoed back from the `Directory` message so source knows
224    /// how many files to send from this directory.
225    DirectoryCreated {
226        src: std::path::PathBuf,
227        dst: std::path::PathBuf,
228        file_count: usize,
229    },
230    /// Signal destination has finished all operations.
231    /// Initiates graceful shutdown via stream closure.
232    DestinationDone,
233}
234
235#[derive(Clone, Debug, Deserialize, Serialize)]
236pub struct RcpdConfig {
237    pub verbose: u8,
238    pub fail_early: bool,
239    pub max_workers: usize,
240    pub max_blocking_threads: usize,
241    pub max_open_files: Option<usize>,
242    pub ops_throttle: usize,
243    pub iops_throttle: usize,
244    pub chunk_size: usize,
245    // common::copy::Settings
246    pub dereference: bool,
247    pub overwrite: bool,
248    pub overwrite_compare: String,
249    pub overwrite_filter: Option<String>,
250    pub ignore_existing: bool,
251    pub debug_log_prefix: Option<String>,
252    /// Port ranges for TCP connections (e.g., "8000-8999,9000-9999")
253    pub port_ranges: Option<String>,
254    pub progress: bool,
255    pub progress_delay: Option<String>,
256    pub remote_copy_conn_timeout_sec: u64,
257    /// Network profile for buffer sizing
258    pub network_profile: crate::NetworkProfile,
259    /// Buffer size for file transfers (defaults to profile-specific value)
260    pub buffer_size: Option<usize>,
261    /// Maximum concurrent connections in the pool
262    pub max_connections: usize,
263    /// Multiplier for pending file writes (max pending = max_connections × multiplier)
264    pub pending_writes_multiplier: usize,
265    /// Chrome trace output prefix for profiling
266    pub chrome_trace_prefix: Option<String>,
267    /// Flamegraph output prefix for profiling
268    pub flamegraph_prefix: Option<String>,
269    /// Log level for profiling (default: trace when profiling is enabled)
270    pub profile_level: Option<String>,
271    /// Enable tokio-console
272    pub tokio_console: bool,
273    /// Port for tokio-console server
274    pub tokio_console_port: Option<u16>,
275    /// Enable TLS encryption (default: true)
276    pub encryption: bool,
277    /// Master's certificate fingerprint for client authentication (when encryption enabled)
278    pub master_cert_fingerprint: Option<CertFingerprint>,
279}
280
281impl RcpdConfig {
282    pub fn to_args(&self) -> Vec<String> {
283        let mut args = vec![
284            format!("--max-workers={}", self.max_workers),
285            format!("--max-blocking-threads={}", self.max_blocking_threads),
286            format!("--ops-throttle={}", self.ops_throttle),
287            format!("--iops-throttle={}", self.iops_throttle),
288            format!("--chunk-size={}", self.chunk_size),
289            format!("--overwrite-compare={}", self.overwrite_compare),
290        ];
291        if self.verbose > 0 {
292            args.push(format!("-{}", "v".repeat(self.verbose as usize)));
293        }
294        if self.fail_early {
295            args.push("--fail-early".to_string());
296        }
297        if let Some(v) = self.max_open_files {
298            args.push(format!("--max-open-files={v}"));
299        }
300        if self.dereference {
301            args.push("--dereference".to_string());
302        }
303        if self.overwrite {
304            args.push("--overwrite".to_string());
305            if let Some(ref filter) = self.overwrite_filter {
306                args.push(format!("--overwrite-filter={filter}"));
307            }
308        }
309        if self.ignore_existing {
310            args.push("--ignore-existing".to_string());
311        }
312        if let Some(ref prefix) = self.debug_log_prefix {
313            args.push(format!("--debug-log-prefix={prefix}"));
314        }
315        if let Some(ref ranges) = self.port_ranges {
316            args.push(format!("--port-ranges={ranges}"));
317        }
318        if self.progress {
319            args.push("--progress".to_string());
320        }
321        if let Some(ref delay) = self.progress_delay {
322            args.push(format!("--progress-delay={delay}"));
323        }
324        args.push(format!(
325            "--remote-copy-conn-timeout-sec={}",
326            self.remote_copy_conn_timeout_sec
327        ));
328        // network profile
329        args.push(format!("--network-profile={}", self.network_profile));
330        // tcp tuning (only if set)
331        if let Some(v) = self.buffer_size {
332            args.push(format!("--buffer-size={v}"));
333        }
334        args.push(format!("--max-connections={}", self.max_connections));
335        args.push(format!(
336            "--pending-writes-multiplier={}",
337            self.pending_writes_multiplier
338        ));
339        // profiling options (only add --profile-level when profiling is enabled)
340        let profiling_enabled =
341            self.chrome_trace_prefix.is_some() || self.flamegraph_prefix.is_some();
342        if let Some(ref prefix) = self.chrome_trace_prefix {
343            args.push(format!("--chrome-trace={prefix}"));
344        }
345        if let Some(ref prefix) = self.flamegraph_prefix {
346            args.push(format!("--flamegraph={prefix}"));
347        }
348        if profiling_enabled {
349            if let Some(ref level) = self.profile_level {
350                args.push(format!("--profile-level={level}"));
351            }
352        }
353        if self.tokio_console {
354            args.push("--tokio-console".to_string());
355        }
356        if let Some(port) = self.tokio_console_port {
357            args.push(format!("--tokio-console-port={port}"));
358        }
359        if !self.encryption {
360            args.push("--no-encryption".to_string());
361        }
362        if let Some(fp) = self.master_cert_fingerprint {
363            args.push(format!(
364                "--master-cert-fp={}",
365                crate::tls::fingerprint_to_hex(&fp)
366            ));
367        }
368        args
369    }
370}
371
372#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
373pub enum RcpdRole {
374    Source,
375    Destination,
376}
377
378impl std::fmt::Display for RcpdRole {
379    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
380        match self {
381            RcpdRole::Source => write!(f, "source"),
382            RcpdRole::Destination => write!(f, "destination"),
383        }
384    }
385}
386
387impl std::str::FromStr for RcpdRole {
388    type Err = anyhow::Error;
389    fn from_str(s: &str) -> Result<Self, Self::Err> {
390        match s.to_lowercase().as_str() {
391            "source" => Ok(RcpdRole::Source),
392            "destination" | "dest" => Ok(RcpdRole::Destination),
393            _ => Err(anyhow::anyhow!("invalid role: {}", s)),
394        }
395    }
396}
397
398#[derive(Clone, Debug, Deserialize, Serialize)]
399pub struct TracingHello {
400    pub role: RcpdRole,
401    /// true for tracing/progress connection, false for control connection
402    pub is_tracing: bool,
403}
404
405/// TLS certificate fingerprint (SHA-256 of DER-encoded certificate).
406pub type CertFingerprint = [u8; 32];
407
408#[derive(Clone, Debug, Deserialize, Serialize)]
409pub enum MasterHello {
410    Source {
411        src: std::path::PathBuf,
412        dst: std::path::PathBuf,
413        /// Destination's TLS certificate fingerprint (None if encryption disabled)
414        dest_cert_fingerprint: Option<CertFingerprint>,
415        /// Filter settings for include/exclude patterns (source-side filtering)
416        filter: Option<common::filter::FilterSettings>,
417        /// Dry-run mode for previewing operations
418        dry_run: Option<common::config::DryRunMode>,
419    },
420    Destination {
421        /// TCP address for control connection to source
422        source_control_addr: std::net::SocketAddr,
423        /// TCP address for data connections to source
424        source_data_addr: std::net::SocketAddr,
425        server_name: String,
426        preserve: common::preserve::Settings,
427        /// Source's TLS certificate fingerprint (None if encryption disabled)
428        source_cert_fingerprint: Option<CertFingerprint>,
429    },
430}
431
432#[derive(Clone, Debug, Deserialize, Serialize)]
433pub struct SourceMasterHello {
434    /// TCP address for control connection (bidirectional messages)
435    pub control_addr: std::net::SocketAddr,
436    /// TCP address for data connections (file transfers)
437    pub data_addr: std::net::SocketAddr,
438    pub server_name: String,
439}
440
441// re-export RuntimeStats from common for convenience
442pub use common::RuntimeStats;
443
444#[derive(Clone, Debug, Deserialize, Serialize)]
445pub enum RcpdResult {
446    Success {
447        message: String,
448        summary: common::copy::Summary,
449        runtime_stats: common::RuntimeStats,
450    },
451    Failure {
452        error: String,
453        summary: common::copy::Summary,
454        runtime_stats: common::RuntimeStats,
455    },
456}
remote/protocol/mod.rs

remote/protocol/
mod.rs
