ripsync_core/net/
receiver.rs

1//! The **receiver**: the side that holds the destination tree.
2//!
3//! It reads the sender's file list, creates directories and symlinks directly,
4//! and for each file decides skip / whole / delta. It drives the exchange — one
5//! [`Request`] then one [`Data`] response at a time — applies each result with a
6//! containment-checked atomic write, mirrors deletions, and finally sends
7//! [`Msg::Finished`] to end the session.
8
9use std::collections::{HashMap, HashSet};
10use std::io::{Read, Write};
11use std::path::{Path, PathBuf};
12
13use filetime::FileTime;
14
15use crate::delta::{Signature, apply};
16use crate::filter::Filter;
17use crate::meta::{
18    canonical_root, check_relative, contained_target, set_mode, set_mtime, set_symlink_mtime,
19};
20use crate::net::proto::{Data, Msg, NetEntry, NetKind, NetOptions, Request};
21use crate::net::transport::{recv_msg, send_msg};
22use crate::plan::Action;
23use crate::report::{Event, Reporter, RunPhase, Stats};
24use crate::walk::{Entry, walk_controlled};
25use crate::{Error, Result, RunControl};
26
27/// Run the receiver half over `conn` for the destination tree at `root`.
28///
29/// # Errors
30///
31/// Returns a walk, protocol, or I/O error, or [`Error::Cancelled`] on cancellation.
32#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
33pub fn run_receiver<C: Read + Write, R: Reporter>(
34    conn: &mut C,
35    root: &Path,
36    filter: &Filter,
37    options: NetOptions,
38    threads: usize,
39    control: &RunControl,
40    reporter: &R,
41) -> Result<Stats> {
42    reporter.event(Event::Phase(RunPhase::Copying));
43    let root_canon = canonical_root(root)?;
44
45    // 1. Receive the sender's full file list (parents precede children).
46    let mut src_entries: Vec<NetEntry> = Vec::new();
47    loop {
48        control.checkpoint()?;
49        match recv_msg(conn)? {
50            Msg::Entry(e) => src_entries.push(e),
51            Msg::ListDone => break,
52            Msg::Error(e) => return Err(Error::Protocol(format!("peer error: {e}"))),
53            other => {
54                return Err(Error::Protocol(format!(
55                    "receiver: expected entry/list-done, got {other:?}"
56                )));
57            }
58        }
59    }
60
61    // 2. Walk our own tree once for skip and delete decisions.
62    let existing = walk_controlled(root, threads, filter, control)?;
63    let existing_map: HashMap<PathBuf, Entry> = existing
64        .iter()
65        .map(|e| (e.rel.clone(), e.clone()))
66        .collect();
67    let src_set: HashSet<&PathBuf> = src_entries.iter().map(|e| &e.rel).collect();
68
69    let mut stats = Stats::default();
70
71    // 3. Apply every source entry, in order.
72    for e in &src_entries {
73        control.checkpoint()?;
74        if check_relative(&e.rel).is_err() {
75            fail(
76                reporter,
77                &mut stats,
78                &e.rel,
79                "path escapes destination root",
80            );
81            continue;
82        }
83        let mtime = FileTime::from_unix_time(e.mtime_s, e.mtime_ns);
84        let existing_entry = existing_map.get(&e.rel);
85        match &e.kind {
86            NetKind::Dir => {
87                let full = root_canon.join(&e.rel);
88                if let Err(err) = std::fs::create_dir_all(&full) {
89                    fail(reporter, &mut stats, &e.rel, &err.to_string());
90                    continue;
91                }
92                apply_meta(&full, e.mode, mtime, options, false);
93                let action = bump(&mut stats, existing_entry.is_some());
94                reporter.event(Event::DirDone {
95                    rel: e.rel.clone(),
96                    action,
97                });
98            }
99            NetKind::Symlink(target) => {
100                let dst = match ensure_and_contain(&root_canon, &e.rel) {
101                    Ok(p) => p,
102                    Err(err) => {
103                        fail(reporter, &mut stats, &e.rel, &err.to_string());
104                        continue;
105                    }
106                };
107                // Remove existing entry; must handle directory (remove_file silently fails on dirs)
108                if let Ok(meta) = std::fs::symlink_metadata(&dst) {
109                    if meta.is_dir() {
110                        let _ = std::fs::remove_dir_all(&dst);
111                    } else {
112                        let _ = std::fs::remove_file(&dst);
113                    }
114                }
115                // Security: reject symlink targets that escape the destination root
116                #[cfg(unix)]
117                {
118                    use std::path::Component;
119                    if target.is_absolute()
120                        || target.components().any(|c| c == Component::ParentDir)
121                    {
122                        fail(
123                            reporter,
124                            &mut stats,
125                            &e.rel,
126                            "symlink target escapes destination root",
127                        );
128                        continue;
129                    }
130                }
131                if let Err(err) = symlink_create(target, &dst) {
132                    fail(reporter, &mut stats, &e.rel, &err.to_string());
133                    continue;
134                }
135                if options.preserve_mtime {
136                    let _ = set_symlink_mtime(&dst, mtime);
137                }
138                let action = bump(&mut stats, existing_entry.is_some());
139                reporter.event(Event::SymlinkDone {
140                    rel: e.rel.clone(),
141                    action,
142                });
143            }
144            NetKind::File => {
145                let is_existing_file = existing_entry.is_some_and(Entry::is_file);
146                let unchanged = !options.checksum
147                    && is_existing_file
148                    && existing_entry.is_some_and(|x| {
149                        x.len == e.len
150                            && x.mtime.unix_seconds() == e.mtime_s
151                            && x.mtime.nanoseconds() == e.mtime_ns
152                    });
153                if unchanged {
154                    stats.skipped += 1;
155                    reporter.event(Event::Skipped { rel: e.rel.clone() });
156                    continue;
157                }
158                let dst = match ensure_and_contain(&root_canon, &e.rel) {
159                    Ok(p) => p,
160                    Err(err) => {
161                        fail(reporter, &mut stats, &e.rel, &err.to_string());
162                        continue;
163                    }
164                };
165                let Some(new_bytes) = fetch_bytes(conn, e, &dst, is_existing_file, options)? else {
166                    fail(
167                        reporter,
168                        &mut stats,
169                        &e.rel,
170                        "source unreadable or delta failed",
171                    );
172                    continue;
173                };
174                if let Err(err) = write_atomic(&dst, &new_bytes, e.mode, mtime, options) {
175                    fail(reporter, &mut stats, &e.rel, &err.to_string());
176                    continue;
177                }
178                stats.bytes += new_bytes.len() as u64;
179                let action = bump(&mut stats, is_existing_file);
180                reporter.event(Event::FileDone {
181                    rel: e.rel.clone(),
182                    action,
183                    bytes: new_bytes.len() as u64,
184                });
185            }
186        }
187    }
188
189    // 4. Mirror deletions (deepest paths first so directories empty before removal).
190    if options.delete {
191        let mut victims: Vec<&Entry> = existing
192            .iter()
193            .filter(|x| !src_set.contains(&x.rel))
194            .collect();
195        victims.sort_by(|a, b| b.rel.cmp(&a.rel));
196        for victim in victims {
197            control.checkpoint()?;
198            let full = root_canon.join(&victim.rel);
199            let removed = if victim.is_dir() {
200                std::fs::remove_dir(&full).or_else(|_| std::fs::remove_dir_all(&full))
201            } else {
202                std::fs::remove_file(&full)
203            };
204            if removed.is_ok() {
205                stats.deleted += 1;
206                reporter.event(Event::Deleted {
207                    rel: victim.rel.clone(),
208                });
209            }
210        }
211    }
212
213    // 5. Terminate the session: the sender returns these stats to its caller.
214    send_msg(conn, &Msg::Finished(stats.clone()))?;
215    Ok(stats)
216}
217
218/// Request and receive the new contents for one file: a delta against the local
219/// copy when possible, otherwise the whole file. Returns `None` if the sender
220/// could not supply it or the delta failed to apply.
221fn fetch_bytes<C: Read + Write>(
222    conn: &mut C,
223    e: &NetEntry,
224    dst: &Path,
225    is_existing_file: bool,
226    options: NetOptions,
227) -> Result<Option<Vec<u8>>> {
228    let use_delta = !options.whole_file && is_existing_file;
229    if use_delta {
230        let old = std::fs::read(dst).unwrap_or_default();
231        let sig = Signature::compute(&old, None);
232        send_msg(conn, &Msg::Request(Request::Delta(e.rel.clone(), sig)))?;
233        match recv_msg(conn)? {
234            Msg::Data(Data::Delta(d)) => Ok(apply(&old, &d).map_err(|e| tracing::debug!("failed to apply delta: {e}")).ok()),
235            Msg::Data(Data::Whole { bytes, zstd }) => Ok(unpack(bytes, zstd)),
236            Msg::Data(Data::NotFound) => Ok(None),
237            Msg::Error(e) => Err(Error::Protocol(format!("peer error: {e}"))),
238            other => Err(Error::Protocol(format!("receiver: unexpected {other:?}"))),
239        }
240    } else {
241        send_msg(conn, &Msg::Request(Request::Whole(e.rel.clone())))?;
242        match recv_msg(conn)? {
243            Msg::Data(Data::Whole { bytes, zstd }) => Ok(unpack(bytes, zstd)),
244            Msg::Data(Data::NotFound) => Ok(None),
245            Msg::Error(e) => Err(Error::Protocol(format!("peer error: {e}"))),
246            other => Err(Error::Protocol(format!("receiver: unexpected {other:?}"))),
247        }
248    }
249}
250
251/// Decode a whole-file payload, decompressing when it was zstd-packed. Returns
252/// `None` if decompression fails (treated as a per-entry error upstream).
253fn unpack(bytes: Vec<u8>, zstd: bool) -> Option<Vec<u8>> {
254    if zstd {
255        zstd::decode_all(bytes.as_slice()).map_err(|e| tracing::debug!("failed to decompress: {e}")).ok()
256    } else {
257        Some(bytes)
258    }
259}
260
261/// Report a per-entry failure and bump the error tally; the run continues.
262fn fail<R: Reporter>(reporter: &R, stats: &mut Stats, rel: &Path, error: &str) {
263    stats.errors += 1;
264    reporter.event(Event::Failed {
265        rel: rel.to_path_buf(),
266        error: error.to_string(),
267    });
268}
269
270/// Count a created/updated entry and return the action taken.
271fn bump(stats: &mut Stats, existed: bool) -> Action {
272    if existed {
273        stats.updated += 1;
274        Action::Update
275    } else {
276        stats.copied += 1;
277        Action::Copy
278    }
279}
280
281/// Apply mode/mtime to a just-written path, honoring the preserve options.
282fn apply_meta(path: &Path, mode: u32, mtime: FileTime, options: NetOptions, _is_file: bool) {
283    if options.preserve_mode {
284        let _ = set_mode(path, mode);
285    }
286    if options.preserve_mtime {
287        let _ = set_mtime(path, mtime);
288    }
289}
290
291/// Ensure the parent directory of `rel` exists, then return the containment-checked
292/// concrete path to write.
293fn ensure_and_contain(root_canon: &Path, rel: &Path) -> Result<PathBuf> {
294    let target = root_canon.join(rel);
295    if let Some(parent) = target.parent() {
296        std::fs::create_dir_all(parent).map_err(|e| Error::io(parent, e))?;
297    }
298    contained_target(root_canon, &target)
299}
300
301/// Write `bytes` to a temp file beside `dst`, set metadata, then atomically rename.
302fn write_atomic(
303    dst: &Path,
304    bytes: &[u8],
305    mode: u32,
306    mtime: FileTime,
307    options: NetOptions,
308) -> Result<()> {
309    let parent = dst
310        .parent()
311        .ok_or_else(|| Error::Containment(dst.to_path_buf()))?;
312    let tmp = parent.join(format!(".ripsync.tmp.{:016x}", rand::random::<u64>()));
313    std::fs::write(&tmp, bytes).map_err(|e| Error::io(&tmp, e))?;
314    apply_meta(&tmp, mode, mtime, options, true);
315    
316    // Bug 1: Remove directory if it exists before atomic replace
317    if let Ok(meta) = std::fs::symlink_metadata(dst) {
318        if meta.is_dir() {
319            if let Err(e) = std::fs::remove_dir_all(dst) {
320                // Log warning if reporter available, otherwise ignore
321                eprintln!("Warning: failed to remove directory {:?}: {}", dst, e);
322            }
323        }
324    }
325    
326    if let Err(e) = atomic_replace(&tmp, dst) {
327        let _ = std::fs::remove_file(&tmp);
328        return Err(Error::io(dst, e));
329    }
330    Ok(())
331}
332
333/// Atomically replace `dst` with `tmp`.
334#[cfg(not(windows))]
335fn atomic_replace(tmp: &Path, dst: &Path) -> std::io::Result<()> {
336    std::fs::rename(tmp, dst)
337}
338
339#[cfg(windows)]
340fn atomic_replace(tmp: &Path, dst: &Path) -> std::io::Result<()> {
341    crate::io::windows::replace_file(tmp, dst)
342}
343
344/// Create a symlink at `link` pointing to `target` (verbatim).
345#[cfg(unix)]
346fn symlink_create(target: &Path, link: &Path) -> std::io::Result<()> {
347    std::os::unix::fs::symlink(target, link)
348}
349
350#[cfg(windows)]
351fn symlink_create(target: &Path, link: &Path) -> std::io::Result<()> {
352    // Best effort; needs Developer Mode or privilege. Treat the link as a file link.
353    std::os::windows::fs::symlink_file(target, link)
354}
ripsync_core/net/receiver.rs

ripsync_core/net/
receiver.rs
