diff --git a/Cargo.lock b/Cargo.lock index 7ccb434..5f45069 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,6 +79,15 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "atomic-polyfill" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4" +dependencies = [ + "critical-section", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -179,6 +188,12 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.1" @@ -243,6 +258,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -282,6 +306,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + [[package]] name = "crypto-common" version = "0.1.7" @@ -365,6 +395,18 @@ dependencies = [ "crypto-common 0.2.2", ] +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "env_filter" version = "2.0.0" @@ -507,6 +549,15 @@ dependencies = [ "r-efi 6.0.0", ] +[[package]] +name = "hash32" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67" +dependencies = [ + "byteorder", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -529,6 +580,20 @@ dependencies = [ "foldhash", ] +[[package]] +name = "heapless" +version = "0.7.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" +dependencies = [ + "atomic-polyfill", + "hash32", + "rustc_version", + "serde", + "spin", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.5.0" @@ -954,6 +1019,19 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "heapless", + "serde", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1125,6 +1203,15 @@ dependencies = [ "walkdir", ] +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "ryu" version = "1.0.23" @@ -1169,8 +1256,9 @@ version = "0.1.0" dependencies = [ "anyhow", "log", + "postcard", "serde", - "serde_json", + "serde_bytes", "tokio", ] @@ -1186,6 +1274,7 @@ dependencies = [ "seedmirror-core", "serde_json", "tokio", + "walkdir", ] [[package]] @@ -1195,6 +1284,12 @@ dependencies = [ "anyhow", ] +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.229" @@ -1205,6 +1300,16 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.229" @@ -1321,6 +1426,21 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] + +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index d47118a..0ae9ced 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,14 @@ env_logger = "0.11.8" log = "0.4.27" # Message serialization/deserialization serde = { version = "1.0.219", features = ["derive"] } +serde_bytes = "0.11" serde_json = "1.0.143" # Async tokio = { version = "1.47.1", features = ["full"] } +# Directory walking +walkdir = "2.5.0" + +# Release build with debug symbols +[profile.profiling] +inherits = "release" +debug = 1 diff --git a/README.md b/README.md index c6c731a..be7473b 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,10 @@ # seedmirror -seedmirror is a utility to monitor remote filesystem changes and automatically synchronize them to a local directory using rsync over ssh. +seedmirror is a utility to monitor remote filesystem changes and automatically synchronize them to a local directory over ssh. ## running (binary) - openssh 6.7+ (for unix domain socket forwarding support, client and server) -- rsync 3.2.3+ (for --mkpath flag, client only) Download the latest release [here](https://github.com/voidiz/seedmirror/releases) or [build](BUILDING.md) the binaries yourself. diff --git a/crates/seedmirror-client/src/cli.rs b/crates/seedmirror-client/src/cli.rs index a750ec4..9bfce7e 100644 --- a/crates/seedmirror-client/src/cli.rs +++ b/crates/seedmirror-client/src/cli.rs @@ -22,11 +22,11 @@ pub(crate) struct Args { value_parser = Self::parse_path_mapping, action = clap::ArgAction::Append )] - pub path_mappings: Vec<(PathBuf, PathBuf)>, + pub path_mappings: Vec, - /// Perform full sync of remote directory upon connecting. - #[arg(long, default_value_t = true)] - pub initial_sync: bool, + /// Disable the full sync of every mapped path upon connecting. + #[arg(long)] + pub no_initial_sync: bool, /// Preview all file changes through logs. No actual syncing of files (or full sync) will be /// done. @@ -41,10 +41,6 @@ pub(crate) struct Args { #[arg(long, default_value_os_t = PathBuf::from("/tmp/forwarded-seedmirror-server.sock"))] pub local_socket_path: PathBuf, - /// Additional rsync flags. Specify multiple times for multiple flags. - #[arg(long = "extra-rsync-flag")] - pub extra_rsync_flags: Vec, - /// Whether the GUI HTTP server is enabled. #[cfg(feature = "gui")] #[arg(long, default_value_t = false)] @@ -56,17 +52,45 @@ pub(crate) struct Args { pub http_addr: String, } +#[derive(Clone, Debug)] +pub(crate) struct PathMapping { + /// Remote root to watch and list (trailing slash normalized away). + pub remote: PathBuf, + + /// Local destination root. + pub local: PathBuf, + + /// The base path used to calculate the local path from the remote path. + /// It is `remote` for `src/`, the parent of `remote` for `src`. + pub strip: PathBuf, +} + impl Args { - fn parse_path_mapping(s: &str) -> clap::error::Result<(PathBuf, PathBuf), String> { - let parts: Vec<_> = s.split(':').collect(); - if parts.len() != 2 { + fn parse_path_mapping(s: &str) -> clap::error::Result { + let parts = s.split(':').collect::>(); + let [remote_str, local_str] = parts[..] else { return Err("expected :".into()); - } + }; + + let remote_path = Self::parse_absolute_path(remote_str)?; + let local_path = Self::parse_absolute_path(local_str)?; - let remote_path = Self::parse_absolute_path(parts[0])?; - let local_path = Self::parse_absolute_path(parts[1])?; + let strip = if remote_str.ends_with('/') { + remote_path.clone() + } else { + // Unreachable for `/`. Since it ends with `/` it falls under the branch above. + // `parent()` is `None` only for root. + remote_path + .parent() + .map(|p| p.to_path_buf()) + .ok_or_else(|| format!("cannot place filesystem root into target: {s}"))? + }; - Ok((remote_path, local_path)) + Ok(PathMapping { + remote: remote_path, + local: local_path, + strip, + }) } fn parse_absolute_path(s: &str) -> clap::error::Result { diff --git a/crates/seedmirror-client/src/command.rs b/crates/seedmirror-client/src/command.rs deleted file mode 100644 index 2a10fb0..0000000 --- a/crates/seedmirror-client/src/command.rs +++ /dev/null @@ -1,149 +0,0 @@ -use std::{ffi::OsStr, io::ErrorKind, iter::once, process::Stdio}; - -use tokio::{ - io::{AsyncReadExt, BufReader}, - process::Command, -}; - -/// Run the given command until completion and return the stdout. -pub(crate) async fn run_with_output(cmd: &str, args: I) -> anyhow::Result -where - I: AsRef<[S]>, - S: AsRef, -{ - let cmdline = format_cmdline(cmd, args.as_ref()); - log::debug!("running cmd: `{cmdline}`"); - - let child = Command::new(cmd) - .args(args.as_ref()) - .kill_on_drop(true) - .output() - .await - .map_err(|e| match e.kind() { - ErrorKind::NotFound => { - anyhow::anyhow!( - "executable `{cmd}` was not found on PATH while running `{cmdline}`" - ) - } - _ => anyhow::anyhow!("failed to spawn command `{cmdline}`: {e:#}"), - })?; - - let stdout = String::from_utf8_lossy(&child.stdout); - let stderr = String::from_utf8_lossy(&child.stderr); - - if !child.status.success() { - let code_str = match child.status.code() { - Some(code) => format!("exit code {code}"), - None => "terminated by signal".to_string(), - }; - - anyhow::bail!( - "cmd `{cmdline}` failed ({code_str}), stderr: `{stderr}`, stdout: `{stdout}`" - ); - } - - log::debug!("cmd: `{cmdline}`, stderr: `{stderr}`, stdout: `{stdout}`"); - - Ok(stdout.into_owned()) -} - -/// Run the given command until completion and apply `f` to each line of streaming stdout. -pub(crate) async fn run_with_streaming_output( - cmd: &str, - args: I, - mut f: F, -) -> anyhow::Result<()> -where - I: AsRef<[S]>, - S: AsRef, - F: FnMut(&str), -{ - let cmdline = format_cmdline(cmd, args.as_ref()); - log::debug!("streaming cmd: `{cmdline}`"); - - let mut child = Command::new(cmd) - .args(args.as_ref()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true) - .spawn() - .map_err(|e| match e.kind() { - ErrorKind::NotFound => { - anyhow::anyhow!( - "executable `{cmd}` was not found on PATH while running `{cmdline}`" - ) - } - _ => anyhow::anyhow!("failed to spawn command `{cmdline}`: {e:#}"), - })?; - - let stdout = child.stdout.take().expect("stdout should not be taken"); - let mut stderr = child.stderr.take().expect("stderr should not be taken"); - - let stderr_handle = tokio::spawn(async move { - let mut stderr_str = String::new(); - stderr.read_to_string(&mut stderr_str).await?; - Ok::(stderr_str) - }); - - let mut stdout_reader = BufReader::new(stdout); - - let mut buffer = Vec::new(); - let mut byte_buf = [0u8; 1024]; - - while let Ok(bytes_read) = stdout_reader.read(&mut byte_buf).await { - if bytes_read == 0 { - break; // EOF - } - - for &byte in &byte_buf[..bytes_read] { - if byte == b'\r' || byte == b'\n' { - if !buffer.is_empty() { - let chunk = String::from_utf8_lossy(&buffer); - f(&chunk); - buffer.clear(); - } - } else { - buffer.push(byte); - } - } - } - - // Flush any leftover trailing bytes - if !buffer.is_empty() { - let chunk = String::from_utf8_lossy(&buffer); - f(&chunk); - } - - let status = child.wait().await?; - let stderr_str = stderr_handle - .await - .map_err(|e| anyhow::anyhow!("failed to join stderr reader task: {e:#}"))??; - - if !status.success() { - let code_str = match status.code() { - Some(code) => format!("exit code {code}"), - None => "terminated by signal".to_string(), - }; - - anyhow::bail!("streaming cmd `{cmdline}` failed ({code_str}), stderr: `{stderr_str}`"); - } - - log::debug!("streaming cmd: `{cmdline}`, stderr: `{stderr_str}`"); - - Ok(()) -} - -fn format_cmdline(cmd: &str, args: I) -> String -where - I: IntoIterator, - S: AsRef, -{ - let lossy_args = args - .into_iter() - .map(|arg| arg.as_ref().to_string_lossy().to_string()); - - once(cmd.to_string()) - .chain(lossy_args) - .collect::>() - .join(" ") -} diff --git a/crates/seedmirror-client/src/dispatch.rs b/crates/seedmirror-client/src/dispatch.rs new file mode 100644 index 0000000..86a62b2 --- /dev/null +++ b/crates/seedmirror-client/src/dispatch.rs @@ -0,0 +1,427 @@ +use std::{ + collections::{HashMap, VecDeque}, + fs::remove_file, + path::{Path, PathBuf}, + pin::Pin, + process::Stdio, + time::Duration, +}; + +use anyhow::Context; +use seedmirror_core::message::{ClientMessage, FetchFailure, FileMeta, ServerMessage}; +use tokio::{ + io::{AsyncBufReadExt, BufReader}, + net::{ + UnixStream, + unix::{OwnedReadHalf, OwnedWriteHalf}, + }, + process::{Child, Command}, + time::sleep, +}; + +use crate::{ + cli::{Args, PathMapping}, + fetch::{Fetch, FetchOutcome, human_bytes, local_matches, prepare_sync, set_syncing_path}, + state::StateBrokerTx, +}; + +type Task = Pin> + Send>>; + +pub(crate) fn init_remote_watcher(args: &Args, state_tx: StateBrokerTx) -> anyhow::Result { + if args.local_socket_path.try_exists()? { + remove_file(&args.local_socket_path).with_context(|| { + format!( + "failed to remove existing socket: {:?}", + args.local_socket_path + ) + })?; + } + + let mut ssh_child = Command::new("ssh") + .kill_on_drop(true) + .arg(&args.ssh_hostname) + .arg("-nNT") + .arg("-o") + .arg("ServerAliveInterval=60") + .arg("-o") + .arg("ServerAliveCountMax=3") + .arg("-L") + .arg(format!( + "{}:{}", + args.local_socket_path.to_string_lossy(), + args.socket_path.to_string_lossy() + )) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .with_context(|| "failed to spawn ssh")?; + + if let Some(stderr) = ssh_child.stderr.take() { + tokio::spawn(async move { + let mut lines = BufReader::new(stderr).lines(); + + while let Ok(Some(line)) = lines.next_line().await { + log::error!("ssh stderr: {line}"); + } + }); + } + + let remote_watcher = new_remote_watcher(args.clone(), state_tx, ssh_child); + Ok(Box::pin(remote_watcher)) +} + +/// Read half of the socket to the server. +struct Reader { + stream: BufReader, + read_buf: Vec, +} + +impl Reader { + fn new(stream: OwnedReadHalf) -> Self { + Self { + stream: BufReader::new(stream), + read_buf: Vec::new(), + } + } + + async fn next(&mut self) -> anyhow::Result> { + ServerMessage::read_from_reader(&mut self.stream, &mut self.read_buf).await + } +} + +/// Write half of the socket to the server. +struct Writer { + stream: OwnedWriteHalf, + write_buf: Vec, +} + +impl Writer { + fn new(stream: OwnedWriteHalf) -> Self { + Self { + stream, + write_buf: Vec::new(), + } + } + + async fn send(&mut self, msg: &ClientMessage) -> anyhow::Result<()> { + msg.write_to_stream(&mut self.stream, &mut self.write_buf) + .await + } +} + +enum Pending { + None, + Listing, + Fetching(Box), +} + +struct State { + args: Args, + state_tx: StateBrokerTx, + writer: Writer, + + /// Paths that still need syncing, latest version per path. + work: HashMap, + + /// Path mappings that have yet to be synced. + mappings: VecDeque, + + /// The request currently in flight. + pending: Pending, +} + +impl State { + fn new(args: Args, state_tx: StateBrokerTx, writer: Writer) -> Self { + let mappings = if args.no_initial_sync { + VecDeque::new() + } else { + args.path_mappings.iter().cloned().collect() + }; + + Self { + args, + state_tx, + writer, + work: HashMap::new(), + mappings, + pending: Pending::None, + } + } + + fn take_fetch(&mut self) -> Option { + match std::mem::replace(&mut self.pending, Pending::None) { + Pending::Fetching(fetch) => Some(*fetch), + other => { + self.pending = other; + None + } + } + } + + fn pop_work(&mut self) -> Option { + let path = self.work.keys().next()?.clone(); + self.work.remove(&path) + } + + /// Handle queued sync jobs, either based on incoming `FileUpdate`s in watched directories from + /// the server, or initial sync. + async fn next(&mut self) -> anyhow::Result<()> { + // Don't do anything if we're in the middle of a request + if !matches!(self.pending, Pending::None) { + return Ok(()); + } + + loop { + if let Some(meta) = self.pop_work() { + if self.begin_sync(meta).await? { + return Ok(()); + } + + continue; + } + + if let Some(mapping) = self.mappings.pop_front() { + self.pending = Pending::Listing; + self.writer + .send(&ClientMessage::ListDir { + path: mapping.remote, + }) + .await?; + + return Ok(()); + } + + return Ok(()); + } + } + + /// Sync remote `meta` with local fs. + async fn begin_sync(&mut self, meta: FileMeta) -> anyhow::Result { + match prepare_sync(&meta, &self.args.path_mappings, self.args.dry_run).await? { + Some(local) => self.begin_fetch(meta, local).await, + None => Ok(false), + } + } + + async fn begin_fetch(&mut self, meta: FileMeta, local: PathBuf) -> anyhow::Result { + if local_matches(&local, &meta)? { + self.work.remove(&meta.path); + return Ok(false); + } + + if self.args.dry_run { + log::info!( + "would sync remote {:?} to local {local:?} (size: {})", + meta.path, + human_bytes(meta.size) + ); + + self.work.remove(&meta.path); + return Ok(false); + } + + log::info!( + "syncing remote {:?} to local {local:?} (size: {})", + meta.path, + human_bytes(meta.size) + ); + + set_syncing_path(&self.state_tx, &meta, &local); + + let path = meta.path.clone(); + self.pending = Pending::Fetching(Box::new(Fetch::new(meta, local))); + self.writer.send(&ClientMessage::FetchFile { path }).await?; + + Ok(true) + } + + async fn finish_fetch( + &mut self, + path: PathBuf, + local: PathBuf, + outcome: FetchOutcome, + ) -> anyhow::Result<()> { + // Check if there's a newer version of the file that came in after we started the transfer + let newer = match outcome { + FetchOutcome::Done { size, mtime_nanos } => match self.work.remove(&path) { + Some(newer) if newer.size != size || newer.mtime_nanos != mtime_nanos => { + Some(newer) + } + _ => None, + }, + FetchOutcome::Stale => self.work.remove(&path), + }; + + self.pending = Pending::None; + + // Refetch if there was a newer version + if let Some(newer) = newer { + self.begin_fetch(newer, local).await?; + } + + Ok(()) + } + + async fn dispatch(&mut self, msg: ServerMessage<'_>) -> anyhow::Result<()> { + match msg { + ServerMessage::FileUpdated { meta } => self.on_file_updated(meta), + ServerMessage::ListDirResponse { entries } => self.on_listing(entries), + ServerMessage::ListDirFailed { reason } => self.on_list_failed(reason), + ServerMessage::FileHeader { size, mtime_nanos } => { + self.on_header(size, mtime_nanos).await + } + ServerMessage::FileChunk { data } => self.on_chunk(data).await, + ServerMessage::FileEof => self.on_eof().await, + ServerMessage::FetchFailed { kind, reason } => self.on_fetch_failed(kind, reason).await, + _ => anyhow::bail!("unexpected message after handshake"), + } + } + + fn on_file_updated(&mut self, meta: FileMeta) -> anyhow::Result<()> { + log::debug!("queued update for {:?} ({} bytes)", meta.path, meta.size); + self.work.insert(meta.path.clone(), meta); + Ok(()) + } + + fn on_listing(&mut self, entries: Vec) -> anyhow::Result<()> { + match &self.pending { + Pending::Listing => {} + _ => anyhow::bail!("received ListDirResponse with no active listing"), + } + + self.pending = Pending::None; + + for entry in entries { + self.work.insert(entry.path.clone(), entry); + } + + Ok(()) + } + + fn on_list_failed(&mut self, reason: String) -> anyhow::Result<()> { + match &self.pending { + Pending::Listing => {} + _ => anyhow::bail!("received ListDirFailed with no active listing"), + } + + anyhow::bail!("list dir failed: {reason}"); + } + + async fn on_header(&mut self, size: u64, mtime_nanos: u64) -> anyhow::Result<()> { + let Pending::Fetching(fetch) = &mut self.pending else { + anyhow::bail!("received FileHeader with no active fetch"); + }; + + fetch.on_header(size, mtime_nanos).await + } + + async fn on_chunk(&mut self, data: &[u8]) -> anyhow::Result<()> { + let Pending::Fetching(fetch) = &mut self.pending else { + anyhow::bail!("received FileChunk with no active fetch"); + }; + + fetch.on_chunk(data, &self.state_tx).await + } + + async fn on_eof(&mut self) -> anyhow::Result<()> { + let Some(fetch) = self.take_fetch() else { + anyhow::bail!("received FileEof with no active fetch"); + }; + + let path = fetch.meta().path.clone(); + let local = fetch.local().to_path_buf(); + let outcome = fetch.on_eof(&self.state_tx).await?; + self.finish_fetch(path, local, outcome).await + } + + async fn on_fetch_failed(&mut self, kind: FetchFailure, reason: String) -> anyhow::Result<()> { + let Some(fetch) = self.take_fetch() else { + anyhow::bail!("received FetchFailed with no active fetch"); + }; + + let path = fetch.meta().path.clone(); + let local = fetch.local().to_path_buf(); + fetch.discard().await; + + match kind { + FetchFailure::Stale => { + log::info!("server aborted fetch for {path:?}: {reason}"); + self.finish_fetch(path, local, FetchOutcome::Stale).await + } + FetchFailure::Unavailable => { + anyhow::bail!("server could not serve {path:?}: {reason}"); + } + } + } +} + +async fn run(reader: &mut Reader, state: &mut State) -> anyhow::Result<()> { + loop { + state.next().await?; + + let msg = reader.next().await?; + state.dispatch(msg).await?; + } +} + +async fn handshake(args: &Args, reader: &mut Reader, writer: &mut Writer) -> anyhow::Result<()> { + writer + .send(&ClientMessage::ConnectionRequest { + watched_paths: args + .path_mappings + .iter() + .map(|mapping| mapping.remote.clone()) + .collect(), + }) + .await?; + + match reader.next().await? { + ServerMessage::Connected => { + log::debug!("received `Connected` answer from server"); + Ok(()) + } + ServerMessage::ConnectionFailed { reason } => { + anyhow::bail!("connection failed: {reason}"); + } + _ => anyhow::bail!("unexpected message during handshake"), + } +} + +async fn new_remote_watcher( + args: Args, + state_tx: StateBrokerTx, + mut ssh_child: Child, +) -> anyhow::Result<()> { + let local_socket_path = &args.local_socket_path; + + tokio::select! { + _ = wait_for_file(local_socket_path) => {}, + // File will never exist if the ssh process fails and exits. + status = ssh_child.wait() => { + anyhow::bail!("ssh exited: {:?}", status?) + } + } + + log::info!("connecting to {local_socket_path:?}"); + let stream = UnixStream::connect(&local_socket_path) + .await + .with_context(|| format!("failed to connect to socket at {local_socket_path:?}"))?; + log::info!("connected to {local_socket_path:?}"); + + let (read_half, write_half) = stream.into_split(); + let mut reader = Reader::new(read_half); + let mut writer = Writer::new(write_half); + handshake(&args, &mut reader, &mut writer).await?; + + let mut state = State::new(args, state_tx, writer); + run(&mut reader, &mut state).await +} + +async fn wait_for_file(path: &Path) { + // TODO: Use file watcher at some point + while !path.exists() { + log::info!("waiting for {path:?} to be created"); + sleep(Duration::from_millis(100)).await; + } +} diff --git a/crates/seedmirror-client/src/fetch.rs b/crates/seedmirror-client/src/fetch.rs new file mode 100644 index 0000000..84f19f9 --- /dev/null +++ b/crates/seedmirror-client/src/fetch.rs @@ -0,0 +1,343 @@ +use std::{ + io::ErrorKind, + path::{Path, PathBuf}, + time::{Instant, UNIX_EPOCH}, +}; + +use anyhow::Context; +use seedmirror_core::message::FileMeta; +use tokio::io::AsyncWriteExt; + +use crate::{ + cli::PathMapping, + file::{cleanup_partial, finalize_tmp, tmp_path}, + state::{StateBrokerMessage, StateBrokerTx}, +}; + +pub(crate) struct Fetch { + /// Remote information about the file being fetched. + meta: FileMeta, + + /// Path to local file where the file is being written to. + local: PathBuf, + + /// File handle of the local file being written to. + file: Option, + + /// Size received in file header, more recent than `meta.path`. + size: u64, + + /// Modified time received in file header, more recent than `meta.mtime_nanos`. + mtime_nanos: u64, + + /// Amount of bytes written so far. + written: u64, + + /// Timestamp at which the transfer started. Set when we receive the file header from the + /// server. + started: Option, + + /// File transfer percentage (0-100). + last_percentage: Option, +} + +pub(crate) enum FetchOutcome { + /// File fully transferred. + Done { size: u64, mtime_nanos: u64 }, + + /// File changed on the server mid-transfer. + Stale, +} + +impl Fetch { + pub(crate) fn new(meta: FileMeta, local: PathBuf) -> Self { + Self { + meta, + local, + file: None, + size: 0, + mtime_nanos: 0, + written: 0, + started: None, + last_percentage: None, + } + } + + pub(crate) fn meta(&self) -> &FileMeta { + &self.meta + } + + pub(crate) fn local(&self) -> &Path { + &self.local + } + + pub(crate) async fn on_header(&mut self, size: u64, mtime_nanos: u64) -> anyhow::Result<()> { + if self.file.is_some() { + anyhow::bail!("redundant header during fetch of {:?}", self.meta.path); + } + + if let Some(parent) = self.local.parent() + && !parent.as_os_str().is_empty() + { + tokio::fs::create_dir_all(parent) + .await + .with_context(|| format!("failed to create parent dir {parent:?}"))?; + } + + let tmp = tmp_path(&self.local); + self.file = Some( + tokio::fs::File::create(&tmp) + .await + .with_context(|| format!("failed to create temp file {tmp:?}"))?, + ); + + self.size = size; + self.mtime_nanos = mtime_nanos; + self.started = Some(Instant::now()); + + Ok(()) + } + + pub(crate) async fn on_chunk( + &mut self, + data: &[u8], + progress: &StateBrokerTx, + ) -> anyhow::Result<()> { + let write_result = { + let Some(file) = self.file.as_mut() else { + anyhow::bail!("received file chunk before header for {:?}", self.meta.path); + }; + + file.write_all(data).await + }; + + if let Err(e) = write_result { + log::error!("failed to write temp file {:?}: {e:#}", self.local); + cleanup_partial_logged(&self.local).await; + return Err(e.into()); + } + + self.written += data.len() as u64; + report_progress( + progress, + self.written, + self.size, + self.started.unwrap_or_else(Instant::now), + &mut self.last_percentage, + ); + + Ok(()) + } + + pub(crate) async fn on_eof(mut self, progress: &StateBrokerTx) -> anyhow::Result { + let Some(file) = self.file.take() else { + cleanup_partial_logged(&self.local).await; + anyhow::bail!("received file EOF before header for {:?}", self.meta.path); + }; + + if self.written != self.size { + drop(file); + cleanup_partial_logged(&self.local).await; + anyhow::bail!( + "fetch for {:?} size mismatch: got {} of {} bytes", + self.meta.path, + self.written, + self.size + ); + } + + if let Err(e) = finalize_tmp(file, &self.local, self.mtime_nanos).await { + cleanup_partial_logged(&self.local).await; + return Err(e).with_context(|| format!("failed to install {:?}", self.local)); + } + + report_progress( + progress, + self.size, + self.size, + self.started.unwrap_or_else(Instant::now), + &mut self.last_percentage, + ); + + Ok(FetchOutcome::Done { + size: self.size, + mtime_nanos: self.mtime_nanos, + }) + } + + /// Discard the local partially transferred file. + pub(crate) async fn discard(mut self) { + drop(self.file.take()); + cleanup_partial_logged(&self.local).await; + } +} + +async fn cleanup_partial_logged(local: &Path) { + if let Err(e) = cleanup_partial(local).await { + log::warn!("{e:#}"); + } +} + +/// Determine what needs to be done to sync remote `meta` to the local fs based on `mappings`. +/// Returns the local path to fetch, or `None` when there is nothing to transfer. +pub(crate) async fn prepare_sync( + meta: &FileMeta, + mappings: &[PathMapping], + dry_run: bool, +) -> anyhow::Result> { + let mapping = best_prefix_match(&meta.path, mappings) + .cloned() + .with_context(|| format!("received file meta for unwatched path: {:?}", meta.path))?; + + let local = mapping + .local + .join(meta.path.strip_prefix(&mapping.strip).with_context(|| { + format!( + "failed to relativize {:?} under {:?}", + meta.path, mapping.strip + ) + })?); + + match std::fs::symlink_metadata(&local) { + Ok(md) => { + let is_dir = md.is_dir() && !md.file_type().is_symlink(); + if is_dir != meta.is_dir { + if dry_run { + log::info!("would replace mismatched {local:?}"); + return Ok(None); + } + + if is_dir { + std::fs::remove_dir_all(&local) + .with_context(|| format!("failed to remove dir {local:?}"))?; + } else { + std::fs::remove_file(&local) + .with_context(|| format!("failed to remove file {local:?}"))?; + } + } + } + Err(e) if e.kind() == ErrorKind::NotFound => {} + Err(e) => anyhow::bail!("failed to stat local {local:?}: {e:#}"), + } + + if meta.is_dir { + if dry_run { + log::info!("would create dir {local:?}"); + return Ok(None); + } + + tokio::fs::create_dir_all(&local) + .await + .with_context(|| format!("failed to create dir {local:?}"))?; + + return Ok(None); + } + + Ok(Some(local)) +} + +/// Returns true if `local` matches remote `meta`. +pub(crate) fn local_matches(local: &Path, meta: &FileMeta) -> anyhow::Result { + let md = match std::fs::metadata(local) { + Ok(md) => md, + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(false), + Err(e) => anyhow::bail!("failed to stat local {local:?}: {e:#}"), + }; + + if md.is_dir() || meta.is_dir { + return Ok(md.is_dir() && meta.is_dir); + } + + Ok(md.len() == meta.size && local_mtime_nanos(&md)? == meta.mtime_nanos) +} + +pub(crate) fn set_syncing_path(state_tx: &StateBrokerTx, meta: &FileMeta, local: &Path) { + if let Err(e) = state_tx.try_send(StateBrokerMessage::SetSyncingPath { + remote_file_path: meta.path.to_string_lossy().into_owned(), + local_file_path: local.to_string_lossy().into_owned(), + }) { + log::error!("failed to send sync state update: {e:?}"); + } +} + +fn local_mtime_nanos(md: &std::fs::Metadata) -> anyhow::Result { + Ok(md + .modified() + .with_context(|| "failed to read local mtime")? + .duration_since(UNIX_EPOCH) + .map_err(|e| anyhow::anyhow!("local mtime before epoch: {e:?}"))? + .as_nanos() as u64) +} + +fn report_progress( + state_tx: &StateBrokerTx, + written: u64, + total: u64, + started: Instant, + last_percentage: &mut Option, +) { + let percentage = written + .saturating_mul(100) + .checked_div(total) + .map(|p| p.min(100) as u8) + .unwrap_or(100); + + if *last_percentage == Some(percentage) { + return; + } + + *last_percentage = Some(percentage); + + let elapsed = started.elapsed().as_secs_f64().max(0.001); + let speed = written as f64 / elapsed; + let remaining = if written >= total { + 0 + } else { + ((total - written) as f64 / speed.max(1.0)) as u64 + }; + + let _ = state_tx.try_send(StateBrokerMessage::SetSyncProgress { + transferred: human_bytes(written), + progress: percentage, + transfer_speed: format!("{}/s", human_bytes(speed as u64)), + remaining: fmt_hms(remaining), + }); +} + +pub(crate) fn human_bytes(n: u64) -> String { + const UNITS: &[&str] = &["B", "KiB", "MiB", "GiB", "TiB"]; + + let mut v = n as f64; + let mut unit = 0; + while v >= 1024.0 && unit + 1 < UNITS.len() { + v /= 1024.0; + unit += 1; + } + + if unit == 0 { + format!("{n}B") + } else { + format!("{v:.1}{}", UNITS[unit]) + } +} + +fn fmt_hms(total_secs: u64) -> String { + format!( + "{}:{:02}:{:02}", + total_secs / 3600, + (total_secs / 60) % 60, + total_secs % 60 + ) +} + +/// Returns the mapping whose remote root best matches `remote_file_path`, +/// i.e. the longest-prefix (most shared parent directories) match. +fn best_prefix_match<'a>( + remote_file_path: &'a Path, + mappings: &'a [PathMapping], +) -> Option<&'a PathMapping> { + mappings + .iter() + .filter(|mapping| remote_file_path.starts_with(&mapping.remote)) + .max_by_key(|mapping| mapping.remote.components().count()) +} diff --git a/crates/seedmirror-client/src/file.rs b/crates/seedmirror-client/src/file.rs new file mode 100644 index 0000000..a6c26c1 --- /dev/null +++ b/crates/seedmirror-client/src/file.rs @@ -0,0 +1,60 @@ +use std::{ + io::ErrorKind, + path::{Path, PathBuf}, + time::{Duration, UNIX_EPOCH}, +}; + +use anyhow::Context; + +/// Generates a temp file path next to the destination. +pub(crate) fn tmp_path(local: &Path) -> PathBuf { + let mut tmp = local.as_os_str().to_owned(); + tmp.push(".seedmirror-partial"); + PathBuf::from(tmp) +} + +/// Clean up a partially transferred file. +pub(crate) async fn cleanup_partial(local: &Path) -> anyhow::Result<()> { + match tokio::fs::remove_file(tmp_path(local)).await { + Ok(()) => Ok(()), + Err(e) if e.kind() == ErrorKind::NotFound => Ok(()), + Err(e) => Err(e).with_context(|| format!("failed to remove partial file for {local:?}")), + } +} + +/// Rename and set metadata on partially transferred file. Called when the file transfer is +/// completed. +pub(crate) async fn finalize_tmp( + tmp_file: tokio::fs::File, + local: &Path, + mtime_nanos: u64, +) -> anyhow::Result<()> { + let tmp = tmp_path(local); + let mtime = UNIX_EPOCH + Duration::from_nanos(mtime_nanos); + + tmp_file + .sync_all() + .await + .with_context(|| format!("failed to sync temp file {tmp:?}"))?; + + drop(tmp_file); + + let file = std::fs::File::options() + .write(true) + .open(&tmp) + .with_context(|| format!("failed to reopen temp file {tmp:?}"))?; + + file.set_modified(mtime) + .with_context(|| format!("failed to set mtime on {tmp:?}"))?; + + file.sync_all() + .with_context(|| format!("failed to sync temp file {tmp:?}"))?; + + drop(file); + + tokio::fs::rename(&tmp, local) + .await + .with_context(|| format!("failed to move {tmp:?} to {local:?}"))?; + + Ok(()) +} diff --git a/crates/seedmirror-client/src/http.rs b/crates/seedmirror-client/src/http.rs index 5e4c814..635f563 100644 --- a/crates/seedmirror-client/src/http.rs +++ b/crates/seedmirror-client/src/http.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use axum::{ Router, body::Body, @@ -134,7 +135,10 @@ pub(crate) async fn init_http_server( state_bcast, }); - let listener = TcpListener::bind(&args.http_addr).await?; + let listener = TcpListener::bind(&args.http_addr) + .await + .with_context(|| format!("failed to bind http server to {}", args.http_addr))?; + log::info!("websocket server running on {}", args.http_addr); Ok(Box::pin(async move { diff --git a/crates/seedmirror-client/src/main.rs b/crates/seedmirror-client/src/main.rs index 5e7492e..0fba5aa 100644 --- a/crates/seedmirror-client/src/main.rs +++ b/crates/seedmirror-client/src/main.rs @@ -6,18 +6,18 @@ use tokio::{ task::JoinSet, }; -use crate::{state::init_state_broker, transfer::init_remote_watcher, workqueue::Workqueue}; +use crate::{state::init_state_broker, dispatch::init_remote_watcher}; mod cli; -mod command; +mod dispatch; +mod fetch; +mod file; #[cfg(feature = "gui")] mod http; mod state; mod task; -mod transfer; -mod workqueue; async fn run() -> anyhow::Result<()> { let args = cli::Args::parse(); @@ -26,7 +26,6 @@ async fn run() -> anyhow::Result<()> { let (state_updater, state_tx, state_bcast) = init_state_broker(); - let queue = Workqueue::new(); let mut set = JoinSet::new(); set.spawn(state_updater); @@ -35,7 +34,7 @@ async fn run() -> anyhow::Result<()> { set.spawn(http::init_http_server(&args, state_tx.clone(), state_bcast).await?); } - set.spawn(init_remote_watcher(&args, queue, state_tx)?); + set.spawn(init_remote_watcher(&args, state_tx)?); tokio::select! { _ = sigterm.recv() => { diff --git a/crates/seedmirror-client/src/transfer.rs b/crates/seedmirror-client/src/transfer.rs deleted file mode 100644 index 3aec72c..0000000 --- a/crates/seedmirror-client/src/transfer.rs +++ /dev/null @@ -1,356 +0,0 @@ -use std::{ - fs::remove_file, - path::{Path, PathBuf}, - pin::Pin, - process::Stdio, - time::Duration, -}; - -use anyhow::Context; -use seedmirror_core::message::Message; -use tokio::{ - io::{AsyncBufReadExt, BufReader}, - net::UnixStream, - process::{Child, Command}, - time::sleep, -}; - -use crate::{ - cli::Args, - command::{run_with_output, run_with_streaming_output}, - state::{StateBrokerMessage, StateBrokerTx}, - workqueue::Workqueue, -}; - -type Task = Pin> + Send>>; - -pub(crate) fn init_remote_watcher( - args: &Args, - workqueue: Workqueue, - state_tx: StateBrokerTx, -) -> anyhow::Result { - if args.local_socket_path.try_exists()? { - remove_file(&args.local_socket_path).with_context(|| { - format!( - "failed to remove existing socket: {:?}", - args.local_socket_path - ) - })?; - } - - let mut ssh_child = Command::new("ssh") - .kill_on_drop(true) - .arg(&args.ssh_hostname) - .arg("-nNT") - .arg("-o") - .arg("ServerAliveInterval=60") - .arg("-o") - .arg("ServerAliveCountMax=3") - .arg("-L") - .arg(format!( - "{}:{}", - args.local_socket_path.to_string_lossy(), - args.socket_path.to_string_lossy() - )) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::piped()) - .spawn() - .with_context(|| "failed to spawn ssh")?; - - if let Some(stderr) = ssh_child.stderr.take() { - tokio::spawn(async move { - let mut lines = BufReader::new(stderr).lines(); - - while let Ok(Some(line)) = lines.next_line().await { - log::error!("ssh stderr: {line}"); - } - }); - } - - let remote_watcher = new_remote_watcher(args.clone(), workqueue, state_tx, ssh_child); - Ok(Box::pin(remote_watcher)) -} - -struct RemoteWatcher { - /// Program arguments. - args: Args, - - /// Queue for sync tasks. - workqueue: Workqueue, - - /// Channel for sending state updates. - state_tx: StateBrokerTx, -} - -impl RemoteWatcher { - pub(crate) fn new(args: Args, workqueue: Workqueue, state_tx: StateBrokerTx) -> Self { - Self { - args, - workqueue, - state_tx, - } - } - - async fn handle_message(&mut self, msg: Message) -> anyhow::Result<()> { - match msg { - Message::Connected => { - log::debug!("received `Connected` answer from server ",); - if self.args.initial_sync { - self.workqueue - .push( - "__full_sync".to_string(), - full_sync(self.args.clone(), self.state_tx.clone()), - ) - .await?; - } - } - Message::FileUpdated { path } => { - let id = path.to_string_lossy().into_owned(); - self.workqueue - .push( - id, - sync_file(self.args.clone(), path, self.state_tx.clone()), - ) - .await?; - } - Message::ConnectionFailed { reason } => { - anyhow::bail!("connection failed: {reason}"); - } - - // Client messages - Message::ConnectionRequest { .. } => (), - }; - - Ok(()) - } -} - -async fn new_remote_watcher( - args: Args, - workqueue: Workqueue, - state_tx: StateBrokerTx, - mut ssh_child: Child, -) -> anyhow::Result<()> { - let local_socket_path = &args.local_socket_path; - - tokio::select! { - _ = wait_for_file(local_socket_path) => {}, - // File will never exist if the ssh process fails and exits. - status = ssh_child.wait() => { - anyhow::bail!("ssh exited: {:?}", status?) - } - } - - log::info!("connecting to {local_socket_path:?}"); - let mut stream = UnixStream::connect(&local_socket_path) - .await - .with_context(|| format!("failed to connect to socket at {local_socket_path:?}"))?; - log::info!("connected to {local_socket_path:?}"); - - let req = Message::ConnectionRequest { - watched_paths: args - .path_mappings - .iter() - .map(|(remote, _local)| remote.clone()) - .collect(), - }; - req.write_to_stream(&mut stream).await?; - - let mut watcher = RemoteWatcher::new(args, workqueue, state_tx); - let mut reader = BufReader::new(stream); - - loop { - let msg = Message::read_from_reader(&mut reader).await?; - watcher.handle_message(msg).await?; - } -} - -async fn wait_for_file(path: &Path) { - // TODO: Use file watcher at some point - while !path.exists() { - log::info!("waiting for {path:?} to be created"); - sleep(Duration::from_millis(100)).await; - } -} - -async fn full_sync(args: Args, state_tx: StateBrokerTx) -> anyhow::Result<()> { - log::info!("performing full sync..."); - - for (remote_path, local_path) in &args.path_mappings { - let (rsync_dry_run_cmd, rsync_dry_run_args) = - construct_rsync_cmd(&args, remote_path, local_path, true); - let dry_run_output = run_with_output(rsync_dry_run_cmd, rsync_dry_run_args).await?; - - let fs_entries = dry_run_output.lines().collect::>(); - let fs_entries_amount = fs_entries.len(); - if fs_entries_amount == 0 { - log::info!("no difference between remote {remote_path:?} and local {local_path:?}"); - continue; - } - - let diff_msg = format!( - "found difference between remote {remote_path:?} and local {local_path:?}. syncing {fs_entries_amount} filesystem entries" - ); - if args.dry_run { - log::info!("{diff_msg}: {fs_entries:?}"); - continue; - } - - log::info!("{diff_msg}"); - - let (rsync_cmd, rsync_args) = construct_rsync_cmd(&args, remote_path, local_path, false); - - run_with_streaming_output(rsync_cmd, rsync_args, |chunk| { - handle_rsync_output_chunk(&state_tx, remote_path, local_path, chunk, true); - }) - .await?; - } - - log::info!("full sync done"); - Ok(()) -} - -async fn sync_file( - args: Args, - remote_file_path: PathBuf, - state_tx: StateBrokerTx, -) -> anyhow::Result<()> { - let (remote_path, local_path) = best_prefix_match(&remote_file_path, &args.path_mappings).ok_or(anyhow::anyhow!( - "found no watched remote path that matches the incoming remote file: {remote_file_path:?}" - ))?; - - let relative_path = remote_file_path.strip_prefix(remote_path)?; - let local_file_path = local_path.join(relative_path); - let (rsync_cmd, rsync_args) = - construct_rsync_cmd(&args, &remote_file_path, &local_file_path, false); - - if args.dry_run { - log::info!(r#"syncing remote {remote_file_path:?} to local {local_file_path:?}"#); - return Ok(()); - } - - run_with_streaming_output(rsync_cmd, rsync_args, |chunk| { - handle_rsync_output_chunk(&state_tx, &remote_file_path, &local_file_path, chunk, false); - }) - .await?; - - Ok(()) -} - -fn construct_rsync_cmd<'a>( - args: &'a Args, - remote_path: &'a Path, - local_path: &'a Path, - dry_run: bool, -) -> (&'a str, Vec) { - let ssh_hostname = &args.ssh_hostname; - let mut rsync_args = vec![ - "-ahz".to_string(), - "--progress".to_string(), - "--partial".to_string(), - "--mkpath".to_string(), // automatically create destination path - r#"--out-format=%n"#.to_string(), - ]; - - rsync_args.extend(args.extra_rsync_flags.iter().cloned()); - rsync_args.push(format!( - "{}:{}", - ssh_hostname, - remote_path.to_string_lossy() - )); - rsync_args.push(local_path.to_string_lossy().to_string()); - - if dry_run { - rsync_args.push("-n".to_string()); - } - - ("rsync", rsync_args) -} - -fn handle_rsync_output_chunk<'a>( - state_tx: &StateBrokerTx, - remote_path: &'a Path, - local_path: &'a Path, - chunk: &str, - is_full_sync: bool, -) { - log::debug!("rsync output chunk: {chunk}"); - - let msg = parse_rsync_output_chunk(remote_path, local_path, chunk, is_full_sync); - let Some(msg) = msg else { - return; - }; - - if let StateBrokerMessage::SetSyncingPath { - remote_file_path, - local_file_path, - } = &msg - { - log::info!(r#"syncing remote {remote_file_path:?} to local {local_file_path:?}"#); - } - - if let Err(e) = state_tx.try_send(msg) { - log::error!("failed to send rsync state update: {e:?}"); - } -} - -fn parse_rsync_output_chunk<'a>( - remote_path: &'a Path, - local_path: &'a Path, - chunk: &str, - is_full_sync: bool, -) -> Option { - let parts = chunk.split_whitespace().collect::>(); - match parts.as_slice() { - // prevent parsing "receiving incremental file list" as SyncProgress - ["receiving", ..] => None, - // rsync sometimes puts stuff like (xfr#1, to-chk=4/7) at the end, hence the .. - [transferred, progress_str, transfer_speed, remaining, ..] - if let Some(progress) = parse_progress(progress_str) => - { - Some(StateBrokerMessage::SetSyncProgress { - transferred: transferred.to_string(), - progress, - transfer_speed: transfer_speed.to_string(), - remaining: remaining.to_string(), - }) - } - _ => { - let remote_file_path = if is_full_sync { - remote_path.join(chunk).to_string_lossy().to_string() - } else { - // If not full sync, `remote_path` is already the full path (since we sync files one - // by one in the case of sync_file). - remote_path.to_string_lossy().to_string() - }; - - let local_file_path = if is_full_sync { - local_path.join(chunk).to_string_lossy().to_string() - } else { - local_path.to_string_lossy().to_string() - }; - - Some(StateBrokerMessage::SetSyncingPath { - remote_file_path, - local_file_path, - }) - } - } -} - -fn parse_progress(progress_str: &str) -> Option { - progress_str.strip_suffix("%").and_then(|v| v.parse().ok()) -} - -/// Returns the mapping that best matches `remote_file_path` based on the remote path with the -/// longest prefix (amount of shared parent directories). -fn best_prefix_match<'a>( - remote_file_path: &'a Path, - mappings: &'a [(PathBuf, PathBuf)], -) -> Option<&'a (PathBuf, PathBuf)> { - mappings - .iter() - .filter(|(remote, _local)| remote_file_path.starts_with(remote)) - .max_by_key(|(remote, _local)| remote.components().count()) -} diff --git a/crates/seedmirror-client/src/workqueue.rs b/crates/seedmirror-client/src/workqueue.rs deleted file mode 100644 index 0c6c29d..0000000 --- a/crates/seedmirror-client/src/workqueue.rs +++ /dev/null @@ -1,52 +0,0 @@ -use std::{collections::HashSet, pin::Pin, sync::Arc}; - -use tokio::sync::{Mutex, mpsc}; - -type Task = (String, BoxFutureResult); -type BoxFutureResult = Pin> + Send + Sync>>; - -pub(crate) struct Workqueue { - sender: mpsc::UnboundedSender, - active: Arc>>, -} - -impl Workqueue { - pub(crate) fn new() -> Self { - let (tx, rx) = mpsc::unbounded_channel::(); - let active = Arc::new(Mutex::new(HashSet::new())); - tokio::spawn(Self::start(rx, active.clone())); - - Self { sender: tx, active } - } - - pub(crate) async fn push(&self, id: String, fut: Fut) -> anyhow::Result<()> - where - Fut: Future> + Send + Sync + 'static, - { - let mut active = self.active.lock().await; - if active.contains(&id) { - // Already queued/running - log::debug!("skipping task `{id}` since it already exists"); - return Ok(()); - } - - active.insert(id.clone()); - self.sender.send((id, Box::pin(fut)))?; - - Ok(()) - } - - async fn start( - mut rx: mpsc::UnboundedReceiver, - active_worker: Arc>>, - ) { - while let Some((id, task)) = rx.recv().await { - if let Err(e) = task.await { - log::error!("task `{id}` failed: {e:#}"); - } - - let mut active = active_worker.lock().await; - active.remove(&id); - } - } -} diff --git a/crates/seedmirror-core/Cargo.toml b/crates/seedmirror-core/Cargo.toml index 4547be5..0df7045 100644 --- a/crates/seedmirror-core/Cargo.toml +++ b/crates/seedmirror-core/Cargo.toml @@ -6,6 +6,7 @@ edition = "2024" [dependencies] anyhow.workspace = true log.workspace = true +postcard = { version = "1.1.3", features = ["use-std"] } serde.workspace = true -serde_json.workspace = true +serde_bytes.workspace = true tokio.workspace = true diff --git a/crates/seedmirror-core/src/message.rs b/crates/seedmirror-core/src/message.rs index 6c7f1e3..da6a824 100644 --- a/crates/seedmirror-core/src/message.rs +++ b/crates/seedmirror-core/src/message.rs @@ -1,74 +1,196 @@ +/// Message protocol used to communicate between seedmirror-client and seedmirror-server. Includes +/// sending file chunks. The protocol is serial, so there are no IDs. For example, if the client +/// sends `FetchFile`, all file-related messages from the server will be in response to that +/// request. use anyhow::Context; use serde::{Deserialize, Serialize}; -use std::{io::ErrorKind, path::PathBuf}; -use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; +use std::{fmt::Debug, path::PathBuf}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; -#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Hash)] -#[serde(tag = "message")] -pub enum Message { +/// 1 MiB +const MAX_MESSAGE_BYTES: usize = 1024 * 1024; + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub enum ClientMessage { + /// Sent by the client upon connecting to the server. ConnectionRequest { /// List of paths to watch. watched_paths: Vec, }, + /// Sent by the client to recursively walk `path` on the server. If `path` is a file rather than + /// a directory, the response contains just that file's entry. + ListDir { path: PathBuf }, + + /// Sent by the client to fetch a specific file. This generally happens when the client has done + /// a `ListDir` and it identifies mismatches in `mtime` + `size`. + FetchFile { path: PathBuf }, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub enum FetchFailure { + /// The file changed on the server mid-transfer. This is generally followed by a new + /// `FileUpdated` from the server. + Stale, + + /// Any other error. + Unavailable, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub enum ServerMessage<'a> { /// Sent by the server to acknowledge a `ConnectionRequest`. Connected, /// Sent by the server when a `ConnectionRequest` fails. ConnectionFailed { reason: String }, - /// Sent when a file is updated. - FileUpdated { - /// Full (absolute) updated path. - path: PathBuf, + /// Sent by the server when a file in a monitored directory is updated. + FileUpdated { meta: FileMeta }, + + /// Sent by the server in response to `ListDir`. + ListDirResponse { entries: Vec }, + + /// Sent by the server when a `ListDir` request fails. + ListDirFailed { reason: String }, + + /// Sent by the server when a `FetchFile` request fails before or during + /// transfer. + FetchFailed { + /// Why the transfer was aborted. + kind: FetchFailure, + reason: String, }, + + /// Sent by the server in response to `FetchFile`. + FileHeader { + /// Size of the file, used to determine the chunk size + size: u64, + + /// Modified time, unix timestamp, nanoseconds. Shared between client and server when a file + /// is fully transferred to detect outdated files. + mtime_nanos: u64, + }, + + /// Sent by the server following `FileHeader`. All following `FileChunk`s must belong to the file + /// corresponding to the preceding `FileHeader`. + FileChunk { + #[serde(with = "serde_bytes")] + data: &'a [u8], + }, + + /// Sent by the server following the last `FileChunk`. If this message is not received at the + /// end of the file, it is safe for the client to delete the partial file. Conversely, when the + /// message is received, the file transfer should be seen as completed. + FileEof, +} + +/// Shared length-prefixed write logic. `buf` is used to avoid allocating a buffer for every `msg`. +async fn write_inner( + msg: &T, + mut stream: impl AsyncWriteExt + Unpin, + buf: &mut Vec, +) -> anyhow::Result<()> +where + T: Serialize + Debug, +{ + buf.clear(); + postcard::to_io(msg, &mut *buf).with_context(|| "failed to serialize message")?; + if buf.len() > MAX_MESSAGE_BYTES { + anyhow::bail!( + "message too large: {} bytes (max {MAX_MESSAGE_BYTES})", + buf.len() + ); + } + + async { + stream.write_all(&(buf.len() as u32).to_le_bytes()).await?; + stream.write_all(buf).await?; + stream.flush().await?; + Ok::<_, std::io::Error>(()) + } + .await + .map_err(|e| anyhow::anyhow!(e).context("failed writing to socket"))?; + + Ok(()) +} + +/// Deserialize length-prefixed `T` from `reader`, reusing `buf`. +async fn read_inner<'a, T, R>(reader: &mut R, buf: &'a mut Vec) -> anyhow::Result +where + T: Deserialize<'a>, + R: AsyncReadExt + Unpin, +{ + let mut len_bytes = [0u8; 4]; + reader + .read_exact(&mut len_bytes) + .await + .with_context(|| "failed to read message length (was the connection closed?)")?; + let content_length = u32::from_le_bytes(len_bytes) as usize; + if content_length > MAX_MESSAGE_BYTES { + anyhow::bail!("message too large: {content_length} bytes (max {MAX_MESSAGE_BYTES})"); + } + + buf.clear(); + buf.resize(content_length, 0); + reader + .read_exact(buf) + .await + .with_context(|| "failed to read message body")?; + + postcard::from_bytes(buf).with_context(|| "failed to deserialize message") } -impl Message { - /// Returns true if the connection is broken and should be terminated. +impl ClientMessage { pub async fn write_to_stream( &self, - mut stream: impl AsyncWriteExt + Unpin, - ) -> anyhow::Result { - let json = format!("{}\n", serde_json::to_string_pretty(self)?); - let content_length = json.len(); - let payload = format!("{content_length}\n{json}"); - let write_result = stream.write_all(payload.as_bytes()).await; - - if let Err(e) = write_result { - match e.kind() { - ErrorKind::BrokenPipe => { - return Ok(true); - } - _ => { - return Err(anyhow::anyhow!(e).context("failed writing to socket")); - } - } - } - - Ok(false) + stream: impl AsyncWriteExt + Unpin, + buf: &mut Vec, + ) -> anyhow::Result<()> { + write_inner(self, stream, buf).await } - pub async fn read_from_reader(reader: &mut R) -> anyhow::Result + pub async fn read_from_reader(reader: &mut R, buf: &mut Vec) -> anyhow::Result where - R: AsyncReadExt + AsyncBufReadExt + Unpin, + R: AsyncReadExt + Unpin, { - let mut line = String::new(); - reader.read_line(&mut line).await?; + let msg: Self = read_inner(reader, buf).await?; + log::debug!("received client message: {msg:?}"); + Ok(msg) + } +} - let content_length: usize = line.trim().parse().with_context( - || "failed to read server message content length (was the connection closed?)", - )?; +impl<'a> ServerMessage<'a> { + pub async fn write_to_stream( + &self, + stream: impl AsyncWriteExt + Unpin, + buf: &mut Vec, + ) -> anyhow::Result<()> { + write_inner(self, stream, buf).await + } - let mut msg_bytes = vec![0; content_length]; - reader - .read_exact(&mut msg_bytes) - .await - .with_context(|| "failed to read message from server")?; + pub async fn read_from_reader( + reader: &mut R, + buf: &'a mut Vec, + ) -> anyhow::Result> + where + R: AsyncReadExt + Unpin, + { + let msg: ServerMessage<'a> = read_inner(reader, buf).await?; - log::debug!("received message: {}", String::from_utf8_lossy(&msg_bytes)); - let msg: Message = serde_json::from_slice(&msg_bytes)?; + // Skip chunk payloads to reduce noise + if !matches!(msg, ServerMessage::FileChunk { .. }) { + log::debug!("received server message: {msg:?}"); + } Ok(msg) } } + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct FileMeta { + pub path: PathBuf, + pub size: u64, + pub mtime_nanos: u64, + pub is_dir: bool, +} diff --git a/crates/seedmirror-server/Cargo.toml b/crates/seedmirror-server/Cargo.toml index adb13a4..a033deb 100644 --- a/crates/seedmirror-server/Cargo.toml +++ b/crates/seedmirror-server/Cargo.toml @@ -13,3 +13,4 @@ notify = { version = "8.2.0", features = ["serde"] } seedmirror-core = { path = "../seedmirror-core" } serde_json.workspace = true tokio.workspace = true +walkdir.workspace = true diff --git a/crates/seedmirror-server/src/connection.rs b/crates/seedmirror-server/src/connection.rs index e6ee351..574fefa 100644 --- a/crates/seedmirror-server/src/connection.rs +++ b/crates/seedmirror-server/src/connection.rs @@ -1,15 +1,110 @@ use anyhow::Context; use notify::{RecursiveMode, Watcher}; -use seedmirror_core::message::Message; +use seedmirror_core::message::{ClientMessage, FetchFailure, FileMeta, ServerMessage}; +use std::{path::Path, time::UNIX_EPOCH}; use tokio::{ fs::remove_file, - io::BufReader, + io::{AsyncReadExt, BufReader}, net::{UnixListener, UnixStream}, sync::broadcast, task::JoinSet, }; -use crate::{cli::Args, informer, watcher}; +use crate::{ + cli::Args, + informer::{self, BroadcastMessage}, + watcher, +}; + +/// File content bytes per `FileChunk`. +// TODO: Make it variable depending on file size? +const FETCH_CHUNK_BYTES: usize = 64 * 1024; + +/// How often to check whether a file has updated during transfer. +const FETCH_STAT_EVERY_CHUNKS: usize = 16; + +enum FetchOutcome { + /// File fully transferred. + Complete, + + /// File changed mid-transfer. + Stale, +} + +/// Returns file handle and metadata for file at `path`. +async fn prepare_fetch(path: &Path) -> anyhow::Result<(tokio::fs::File, u64, u64)> { + let meta = tokio::fs::metadata(path) + .await + .with_context(|| format!("failed to stat {path:?}"))?; + + if meta.is_dir() { + anyhow::bail!("fetch for {path:?} refused: is a directory"); + } + + let mtime_nanos = mtime_nanos_of(&meta, path)?; + let file = tokio::fs::File::open(path) + .await + .with_context(|| format!("failed to open {path:?}"))?; + + Ok((file, meta.len(), mtime_nanos)) +} + +async fn stream_chunks( + path: &Path, + size: u64, + mtime_nanos: u64, + file: &mut tokio::fs::File, + write_stream: &mut (impl tokio::io::AsyncWriteExt + Unpin), + write_buf: &mut Vec, +) -> anyhow::Result { + let mut chunk_buf = vec![0u8; FETCH_CHUNK_BYTES]; + let mut chunks_since_check = 0usize; + + loop { + let chunk_size = file + .read(&mut chunk_buf) + .await + .with_context(|| format!("failed to read {path:?}"))?; + + if chunk_size == 0 { + break; + } + + let chunk = ServerMessage::FileChunk { + data: &chunk_buf[..chunk_size], + }; + + chunk.write_to_stream(&mut *write_stream, write_buf).await?; + + chunks_since_check += 1; + if chunks_since_check < FETCH_STAT_EVERY_CHUNKS { + continue; + } + + chunks_since_check = 0; + + // Check whether file has updated + let cur = tokio::fs::metadata(path) + .await + .with_context(|| format!("fetch for {path:?} failed: file vanished"))?; + + let cur_mtime = mtime_nanos_of(&cur, path)?; + if cur.len() != size || cur_mtime != mtime_nanos { + return Ok(FetchOutcome::Stale); + } + } + + let cur = tokio::fs::metadata(path) + .await + .with_context(|| format!("fetch for {path:?} failed: file vanished before Eof"))?; + + let cur_mtime = mtime_nanos_of(&cur, path)?; + if cur.len() != size || cur_mtime != mtime_nanos { + return Ok(FetchOutcome::Stale); + } + + Ok(FetchOutcome::Complete) +} pub(crate) async fn connection_manager(args: Args) -> anyhow::Result<()> { if let Err(e) = connection_manager_inner(args).await { @@ -45,14 +140,32 @@ async fn connection_manager_inner(args: Args) -> anyhow::Result<()> { async fn connection_handler(args: Args, stream: UnixStream) { if let Err(e) = connection_handler_inner(args, stream).await { - log::error!("connection handler failed: {e:#}"); + if is_peer_disconnected(&e) { + log::info!("client disconnected: {e:#}"); + } else { + log::error!("connection handler failed: {e:#}"); + } } } +fn is_peer_disconnected(e: &anyhow::Error) -> bool { + e.chain().any(|cause| { + cause.downcast_ref::().is_some_and(|io| { + matches!( + io.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::UnexpectedEof + ) + }) + }) +} + async fn connection_handler_inner(args: Args, mut stream: UnixStream) -> anyhow::Result<()> { log::info!("established socket connection with client"); - let (server_msg_tx, mut server_msg_rx) = broadcast::channel::(100); + // TODO: Figure out how to handle a big queue of fs events + let (server_msg_tx, mut server_msg_rx) = broadcast::channel::(65536); // Watcher will be shut down on drop let (mut watcher, notify_rx) = watcher::create_watcher().await?; @@ -60,39 +173,41 @@ async fn connection_handler_inner(args: Args, mut stream: UnixStream) -> anyhow: let mut set = JoinSet::new(); set.spawn(informer::notify_handler(args, notify_rx, server_msg_tx)); + let mut read_buf = Vec::new(); + let mut write_buf = Vec::new(); + loop { tokio::select! { res = server_msg_rx.recv() => { - match handle_server_msg(res, &mut stream).await { - Ok(true) => break, - Ok(false) => (), - Err(e) => anyhow::bail!(e), - }; + handle_server_msg(res, &mut stream, &mut write_buf).await?; } Ok(_) = stream.readable() => { - match handle_client_msg(&mut watcher, &mut stream).await { - Ok(true) => break, - Ok(false) => (), - Err(e) => anyhow::bail!(e), - }; + handle_client_msg(&mut watcher, &mut stream, &mut read_buf, &mut write_buf).await?; } } } - - log::info!("connection broken, terminating it..."); - Ok(()) } -/// Returns true if the connection should be terminated. async fn handle_server_msg( - res: Result, + res: Result, stream: &mut UnixStream, -) -> anyhow::Result { + write_buf: &mut Vec, +) -> anyhow::Result<()> { match res { - Ok(msg) => { - if msg.write_to_stream(stream).await? { - return Ok(true); - } + Ok(BroadcastMessage::FileUpdated { path }) => { + let meta = match build_file_meta(&path).await { + Ok(meta) => meta, + Err(e) => { + log::info!( + "skipping FileUpdated for {path:?}, file most likely removed: {e:#}" + ); + + return Ok(()); + } + }; + + let msg = ServerMessage::FileUpdated { meta }; + msg.write_to_stream(stream, write_buf).await?; } Err(broadcast::error::RecvError::Lagged(skipped)) => { log::warn!("receiving too many filesystem events, skipping {skipped} event(s)"); @@ -100,54 +215,194 @@ async fn handle_server_msg( Err(e) => anyhow::bail!("recv on filesystem event broadcast channel failed: {e:#}"), } - Ok(false) + Ok(()) } -/// Returns true if the connection should be terminated. async fn handle_client_msg( watcher: &mut impl Watcher, stream: &mut UnixStream, -) -> anyhow::Result { + read_buf: &mut Vec, + write_buf: &mut Vec, +) -> anyhow::Result<()> { let (read_stream, mut write_stream) = tokio::io::split(&mut *stream); let mut reader = BufReader::new(read_stream); - let msg = match Message::read_from_reader(&mut reader).await { - Ok(msg) => msg, - Err(e) => { - log::debug!( - "error when reading message from client (probably due to terminated connection): {e:#}" - ); - - return Ok(true); - } - }; + let msg = ClientMessage::read_from_reader(&mut reader, read_buf).await?; - #[allow(clippy::single_match)] match msg { // TODO: Exchange version information to ensure client and server match - Message::ConnectionRequest { watched_paths } => { + ClientMessage::ConnectionRequest { watched_paths } => { for path in watched_paths { let watch_res = watcher .watch(&path, RecursiveMode::Recursive) .with_context(|| format!("failed to watch path `{}`", path.to_string_lossy())); if let Err(e) = watch_res { - Message::ConnectionFailed { + ServerMessage::ConnectionFailed { reason: format!("{e:#}"), } - .write_to_stream(&mut write_stream) + .write_to_stream(&mut write_stream, write_buf) .await?; anyhow::bail!(e); } } - Message::Connected - .write_to_stream(&mut write_stream) + ServerMessage::Connected + .write_to_stream(&mut write_stream, write_buf) .await?; } - _ => (), + ClientMessage::ListDir { path } => match list_dir(&path).await { + Ok(entries) => { + let resp = ServerMessage::ListDirResponse { entries }; + resp.write_to_stream(&mut write_stream, write_buf).await?; + } + Err(e) => { + let reason = format!("failed to list dir {path:?}: {e:#}"); + log::warn!("{reason}"); + + let msg = ServerMessage::ListDirFailed { reason }; + msg.write_to_stream(&mut write_stream, write_buf).await?; + } + }, + ClientMessage::FetchFile { path } => { + serve_fetch(&path, &mut write_stream, write_buf).await?; + } + } + + Ok(()) +} + +/// Streams `FileHeader` + `FileChunk`s + `FileEof` for `path`. +async fn serve_fetch( + path: &Path, + write_stream: &mut (impl tokio::io::AsyncWriteExt + Unpin), + write_buf: &mut Vec, +) -> anyhow::Result<()> { + let (mut file, size, mtime_nanos) = match prepare_fetch(path).await { + Ok(v) => v, + Err(e) => { + return fail_fetch( + FetchFailure::Unavailable, + format!("{e:#}, skipping fetch without Eof"), + write_stream, + write_buf, + ) + .await; + } + }; + + let header = ServerMessage::FileHeader { size, mtime_nanos }; + + header + .write_to_stream(&mut *write_stream, write_buf) + .await?; + + match stream_chunks(path, size, mtime_nanos, &mut file, write_stream, write_buf).await { + Ok(FetchOutcome::Complete) => {} + Ok(FetchOutcome::Stale) => { + return fail_fetch( + FetchFailure::Stale, + format!("file changed mid-transfer for {path:?}, discarding partial"), + write_stream, + write_buf, + ) + .await; + } + Err(e) => { + return fail_fetch( + FetchFailure::Unavailable, + format!("{e:#}, aborting fetch without Eof"), + write_stream, + write_buf, + ) + .await; + } + } + + ServerMessage::FileEof + .write_to_stream(&mut *write_stream, write_buf) + .await?; + + log::debug!("fetched {path:?}: {size} bytes"); + Ok(()) +} + +async fn fail_fetch( + kind: FetchFailure, + reason: String, + write_stream: &mut (impl tokio::io::AsyncWriteExt + Unpin), + write_buf: &mut Vec, +) -> anyhow::Result<()> { + log::warn!("{reason}"); + + let msg = ServerMessage::FetchFailed { kind, reason }; + msg.write_to_stream(&mut *write_stream, write_buf).await?; + Ok(()) +} + +/// Recursively walks `path`, returning a `FileMeta` per entry (including `path` itself). +async fn list_dir(path: &Path) -> anyhow::Result> { + tokio::fs::metadata(path) + .await + .with_context(|| format!("failed to stat list root {path:?}"))?; + + // spawn_blocking since walk_dir is blocking + let owned = path.to_path_buf(); + tokio::task::spawn_blocking(move || walk_dir(&owned)) + .await + .with_context(|| format!("blocking walk task failed for {path:?}"))? +} + +fn walk_dir(path: &Path) -> anyhow::Result> { + let mut entries = Vec::new(); + + for entry in walkdir::WalkDir::new(path).follow_links(false) { + let entry = entry.with_context(|| format!("failed to walk under {path:?}"))?; + + // TODO: Support symlinks? + if entry.file_type().is_symlink() { + log::warn!("skipping symlink {:?}: not supported", entry.path()); + continue; + } + + let meta = entry + .metadata() + .with_context(|| format!("failed to stat {:?}", entry.path()))?; + + let mtime_nanos = mtime_nanos_of(&meta, entry.path())?; + + entries.push(FileMeta { + path: entry.path().to_path_buf(), + size: meta.len(), + mtime_nanos, + is_dir: meta.is_dir(), + }); } - Ok(false) + Ok(entries) +} + +async fn build_file_meta(path: &Path) -> anyhow::Result { + let md = tokio::fs::metadata(path) + .await + .with_context(|| format!("failed to stat path: {path:?}"))?; + + let mtime_nanos = mtime_nanos_of(&md, path)?; + + Ok(FileMeta { + path: path.to_path_buf(), + size: md.len(), + mtime_nanos, + is_dir: md.is_dir(), + }) +} + +fn mtime_nanos_of(meta: &std::fs::Metadata, path: &Path) -> anyhow::Result { + Ok(meta + .modified() + .with_context(|| format!("failed to read mtime: {path:?}"))? + .duration_since(UNIX_EPOCH) + .map_err(|e| anyhow::anyhow!("mtime before epoch for {path:?}: {e:?}"))? + .as_nanos() as u64) } diff --git a/crates/seedmirror-server/src/informer.rs b/crates/seedmirror-server/src/informer.rs index d0223da..960260d 100644 --- a/crates/seedmirror-server/src/informer.rs +++ b/crates/seedmirror-server/src/informer.rs @@ -5,11 +5,15 @@ use std::{ use anyhow::Context; use notify::Event; -use seedmirror_core::message::Message; use tokio::{sync::broadcast, task::JoinHandle, time::sleep}; use crate::{cli::Args, watcher::NotifyEventReceiver}; +#[derive(Clone, Debug)] +pub(crate) enum BroadcastMessage { + FileUpdated { path: PathBuf }, +} + struct NotifyHandler { args: Args, @@ -17,7 +21,7 @@ struct NotifyHandler { notify_rx: NotifyEventReceiver, /// Broadcast channel used to inform clients of updated files. - server_msg_tx: broadcast::Sender, + server_msg_tx: broadcast::Sender, /// Ongoing event handlers for file updates. event_handlers: HashMap>, @@ -27,7 +31,7 @@ impl NotifyHandler { fn new( args: Args, notify_rx: NotifyEventReceiver, - server_msg_tx: broadcast::Sender, + server_msg_tx: broadcast::Sender, ) -> Self { Self { args, @@ -56,9 +60,8 @@ impl NotifyHandler { }, Ok(msg) = msg_rx.recv() => { // Clean up the event handler when the message has been sent - if let Message::FileUpdated { path } = msg { - self.event_handlers.remove(&path); - } + let BroadcastMessage::FileUpdated { path } = msg; + self.event_handlers.remove(&path); } } } @@ -71,22 +74,9 @@ impl NotifyHandler { let absolute_path = path::absolute(path) .with_context(|| format!("failed to resolve path: {path:?}"))?; - #[allow(clippy::single_match)] match event.kind { notify::EventKind::Create(_) | notify::EventKind::Modify(_) => { - let mut path = absolute_path.clone(); - - // Push an empty component to the path to add a trailing slash. This is - // important for rsync to treat it as a directory so that - // `rsync ` - // synchronizes the state of `` with `` instead of placing - // `` inside ``. - if path.is_dir() { - path.push(""); - } - - let msg = Message::FileUpdated { path }; - self.queue_notify_message(&absolute_path, msg); + self.queue_notify_message(&absolute_path); } notify::EventKind::Remove(_) => { self.abort_event_handler(&absolute_path); @@ -98,19 +88,23 @@ impl NotifyHandler { Ok(()) } - fn queue_notify_message(&mut self, path: &Path, msg: Message) { + fn queue_notify_message(&mut self, path: &Path) { self.abort_event_handler(path); let sync_delay = self.args.sync_delay; let msg_tx = self.server_msg_tx.clone(); + let path = path.to_path_buf(); + self.event_handlers.insert( - path.to_path_buf(), + path.clone(), tokio::spawn(async move { sleep(sync_delay).await; // Spawn a separate task so it can't be canceled. Currently there aren't any yield // points, so it isn't strictly necessary right now. tokio::spawn(async move { + let msg = BroadcastMessage::FileUpdated { path }; + let inner = || -> anyhow::Result<()> { log::info!("broadcasting message: {msg:?}"); msg_tx.send(msg.clone())?; @@ -135,7 +129,7 @@ impl NotifyHandler { pub(crate) async fn notify_handler( args: Args, rx: NotifyEventReceiver, - server_msg_tx: broadcast::Sender, + server_msg_tx: broadcast::Sender, ) { let state = NotifyHandler::new(args, rx, server_msg_tx); if let Err(e) = state.handle().await { diff --git a/crates/seedmirror-test/src/harness.rs b/crates/seedmirror-test/src/harness.rs new file mode 100644 index 0000000..b8869d9 --- /dev/null +++ b/crates/seedmirror-test/src/harness.rs @@ -0,0 +1,240 @@ +use std::{ + env, fs, + path::{Path, PathBuf}, + process::Command, + thread, + time::{Duration, Instant}, +}; + +use anyhow::Context; + +use crate::{path::dir_contains, process::ProcessGuard}; + +pub enum Entry { + /// An (empty) directory + Dir, + + /// A zero-byte file + Empty, + + /// A file with the given contents + Bytes(Vec), + + /// A sparse, zero-filled file of the given length + Zeros(u64), +} + +impl Entry { + pub fn dir() -> Self { + Self::Dir + } + + pub fn empty() -> Self { + Self::Empty + } + + pub fn bytes(bytes: impl Into>) -> Self { + Self::Bytes(bytes.into()) + } + + pub fn zeros(len: u64) -> Self { + Self::Zeros(len) + } + + fn write(&self, path: &Path) -> anyhow::Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).with_context(|| format!("failed to create {parent:?}"))?; + } + + match self { + Self::Dir => { + fs::create_dir_all(path).with_context(|| format!("failed to create {path:?}"))?; + } + Self::Empty => { + fs::write(path, []).with_context(|| format!("failed to write {path:?}"))?; + } + Self::Bytes(bytes) => { + fs::write(path, bytes).with_context(|| format!("failed to write {path:?}"))?; + } + Self::Zeros(len) => { + fs::File::create(path) + .and_then(|file| file.set_len(*len)) + .with_context(|| format!("failed to create {path:?}"))?; + } + } + + Ok(()) + } +} + +pub struct Fixture { + name: String, + source: Vec<(PathBuf, Entry)>, + target: Vec<(PathBuf, Entry)>, +} + +impl Fixture { + pub fn new(name: impl Into) -> Self { + Self { + name: name.into(), + source: Vec::new(), + target: Vec::new(), + } + } + + /// Add `entry` into the source directory at `path` (relative) + pub fn source(mut self, path: impl Into, entry: Entry) -> Self { + self.source.push((path.into(), entry)); + self + } + + /// Add `entry` into the target directory at `path` (relative) + pub fn target(mut self, path: impl Into, entry: Entry) -> Self { + self.target.push((path.into(), entry)); + self + } + + pub fn build(self) -> anyhow::Result { + let workspace_dir = Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .canonicalize() + .with_context(|| "failed to resolve workspace dir")?; + + let mut root = env::temp_dir(); + root.push(format!("seedmirror-test-{}", self.name)); + if root.exists() { + fs::remove_dir_all(&root).with_context(|| format!("failed to clear {root:?}"))?; + } + + // Make sure source/ is synced with target/ instead of putting source/ inside target/ + let src = root.join("source").join(""); + let dst = root.join("target").join(""); + fs::create_dir_all(&src).with_context(|| format!("failed to create {src:?}"))?; + fs::create_dir_all(&dst).with_context(|| format!("failed to create {dst:?}"))?; + + for (rel, entry) in &self.source { + entry.write(&src.join(rel))?; + } + for (rel, entry) in &self.target { + entry.write(&dst.join(rel))?; + } + + let target_paths = self.target.iter().map(|(rel, _)| rel.clone()).collect(); + let socket_path = root.join("seedmirror-server.sock"); + + Ok(SyncHarness { + workspace_dir, + tmp: TempDir(root), + src, + dst, + socket_path, + target_paths, + }) + } +} + +pub struct SyncHarness { + pub workspace_dir: PathBuf, + pub tmp: TempDir, + pub src: PathBuf, + pub dst: PathBuf, + pub socket_path: PathBuf, + target_paths: Vec, +} + +impl SyncHarness { + pub fn build(&self, profile: &str) -> anyhow::Result<()> { + let status = Command::new("cargo") + .current_dir(&self.workspace_dir) + .args(["build", "--profile", profile]) + .status() + .with_context(|| "failed to run cargo build")?; + + anyhow::ensure!(status.success(), "cargo build failed: {status}"); + Ok(()) + } + + pub fn bin(&self, profile: &str, name: &str) -> PathBuf { + self.workspace_dir.join("target").join(profile).join(name) + } + + pub fn server_cmd(&self, profile: &str) -> Command { + let mut cmd = Command::new(self.bin(profile, "seedmirror-server")); + cmd.current_dir(&self.workspace_dir) + .arg("--socket-path") + .arg(&self.socket_path) + .arg("--sync-delay") + .arg("100"); + + cmd + } + + pub fn client_cmd(&self, profile: &str) -> Command { + let mut cmd = Command::new(self.bin(profile, "seedmirror-client")); + cmd.current_dir(&self.workspace_dir) + .arg("--socket-path") + .arg(&self.socket_path) + .arg("--ssh-hostname") + .arg("localhost") + .arg("-p") + .arg(format!( + "{}:{}", + self.src.to_string_lossy(), + self.dst.to_string_lossy() + )); + + cmd + } + + pub fn spawn_and_sync(&self, profile: &str) -> anyhow::Result<(ProcessGuard, ProcessGuard)> { + let server = ProcessGuard::spawn(&mut self.server_cmd(profile))?; + let client = ProcessGuard::spawn(&mut self.client_cmd(profile))?; + self.wait_for_initial_sync()?; + Ok((server, client)) + } + + pub fn wait_for(&self, timeout: Duration, cond: impl FnMut() -> bool) -> anyhow::Result<()> { + let start = Instant::now(); + let mut cond = cond; + while !cond() { + if start.elapsed() > timeout { + anyhow::bail!("timed out after {timeout:?} waiting for condition"); + } + thread::sleep(Duration::from_millis(250)); + } + + Ok(()) + } + + /// Wait until the destination mirrors the source (all entries present) + pub fn wait_for_initial_sync(&self) -> anyhow::Result<()> { + self.wait_for(Duration::from_secs(60), || { + dir_contains(&self.src, &self.dst) + }) + } + + /// Assert that files which only existed locally were not removed + pub fn assert_target_preserved(&self) -> anyhow::Result<()> { + for rel in &self.target_paths { + let path = self.dst.join(rel); + anyhow::ensure!(path.exists(), "target-only path {path:?} was removed"); + } + + Ok(()) + } +} + +/// Owns a temp dir, removing it on drop +pub struct TempDir(PathBuf); + +impl TempDir { + pub fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} diff --git a/crates/seedmirror-test/src/lib.rs b/crates/seedmirror-test/src/lib.rs index 2571809..e3d9704 100644 --- a/crates/seedmirror-test/src/lib.rs +++ b/crates/seedmirror-test/src/lib.rs @@ -1,2 +1,3 @@ +pub mod harness; pub mod path; pub mod process; diff --git a/crates/seedmirror-test/src/path.rs b/crates/seedmirror-test/src/path.rs index b520a8d..6d466d6 100644 --- a/crates/seedmirror-test/src/path.rs +++ b/crates/seedmirror-test/src/path.rs @@ -2,33 +2,6 @@ use std::{fs, path::Path}; use anyhow::Context; -pub fn copy_recursive(src: &Path, dst: &Path) -> anyhow::Result<()> { - if !dst.exists() { - fs::create_dir_all(dst)?; - } - - if !src.is_dir() { - anyhow::bail!("src: {src:?} must be a directory"); - } - - for entry_res in fs::read_dir(src)? { - let entry = entry_res?.path(); - let dst_path = dst.join( - entry - .file_name() - .with_context(|| "invalid final path component".to_string())?, - ); - - if entry.is_dir() { - copy_recursive(&entry, &dst_path)?; - } else { - fs::copy(&entry, &dst_path)?; - } - } - - Ok(()) -} - pub fn assert_dst_contains_src(src: &Path, dst: &Path) -> anyhow::Result<()> { for entry_res in fs::read_dir(src)? { let entry = entry_res?.path(); @@ -53,3 +26,7 @@ pub fn assert_dst_contains_src(src: &Path, dst: &Path) -> anyhow::Result<()> { Ok(()) } + +pub fn dir_contains(src: &Path, dst: &Path) -> bool { + assert_dst_contains_src(src, dst).is_ok() +} diff --git a/crates/seedmirror-test/src/process.rs b/crates/seedmirror-test/src/process.rs index 1b70bfc..65aee5e 100644 --- a/crates/seedmirror-test/src/process.rs +++ b/crates/seedmirror-test/src/process.rs @@ -9,6 +9,10 @@ impl ProcessGuard { let child = cmd.spawn()?; Ok(Self { child }) } + + pub fn id(&self) -> u32 { + self.child.id() + } } impl Drop for ProcessGuard { diff --git a/crates/seedmirror-test/tests/profile_test.rs b/crates/seedmirror-test/tests/profile_test.rs new file mode 100644 index 0000000..204e041 --- /dev/null +++ b/crates/seedmirror-test/tests/profile_test.rs @@ -0,0 +1,80 @@ +use std::{ + fs, + path::Path, + process::Command, + thread, + time::{Duration, Instant}, +}; + +use anyhow::Context; +use seedmirror_test::{ + harness::{Entry, Fixture}, + process::ProcessGuard, +}; + +const BIG_FILE_BYTES: u64 = 1 << 30; + +#[test] +#[ignore = "requires perf and perf_event access"] +fn profile_full_sync() -> anyhow::Result<()> { + let fx = Fixture::new("profiling") + .source("subdir/initial.txt", Entry::empty()) + .source("subdir/test/big.bin", Entry::zeros(BIG_FILE_BYTES)) + .build()?; + + fx.build("profiling")?; + + let out_dir = fx.workspace_dir.join("target/profiling"); + let server_data = out_dir.join("seedmirror-server.perf.data"); + let client_data = out_dir.join("seedmirror-client.perf.data"); + let _ = fs::remove_file(&server_data); + let _ = fs::remove_file(&client_data); + + let server = ProcessGuard::spawn(&mut fx.server_cmd("profiling"))?; + let perf_server = spawn_perf(server.id(), &server_data)?; + + let client = ProcessGuard::spawn(&mut fx.client_cmd("profiling"))?; + let perf_client = spawn_perf(client.id(), &client_data)?; + + fx.wait_for_initial_sync()?; + + // Stop the profilers before their targets exit so the captures flush + drop(perf_client); + drop(perf_server); + drop(client); + drop(server); + + for (name, path) in [("server", &server_data), ("client", &client_data)] { + anyhow::ensure!( + wait_for_file(path, Duration::from_secs(10)), + "perf did not write the {name} profile to {path:?}" + ); + + println!("{name} profile ready: {path:?}"); + } + + Ok(()) +} + +fn spawn_perf(pid: u32, out: &Path) -> anyhow::Result { + let mut cmd = Command::new("perf"); + cmd.args(["record", "-F", "99", "-g", "--call-graph", "dwarf", "-o"]) + .arg(out) + .arg("-p") + .arg(pid.to_string()); + + ProcessGuard::spawn(&mut cmd).with_context(|| "failed to spawn perf (is it in PATH?)") +} + +fn wait_for_file(path: &Path, timeout: Duration) -> bool { + let start = Instant::now(); + while start.elapsed() < timeout { + if fs::metadata(path).map(|m| m.len() > 0).unwrap_or(false) { + return true; + } + + thread::sleep(Duration::from_millis(100)); + } + + false +} diff --git a/crates/seedmirror-test/tests/sync_test.rs b/crates/seedmirror-test/tests/sync_test.rs index ee8ebfd..43aaac8 100644 --- a/crates/seedmirror-test/tests/sync_test.rs +++ b/crates/seedmirror-test/tests/sync_test.rs @@ -1,111 +1,34 @@ -use std::{ - env, fs, - path::{Path, PathBuf}, - process::Command, - thread, - time::Duration, -}; +use std::{fs, time::Duration}; -use anyhow::Context; use seedmirror_test::{ - path::{assert_dst_contains_src, copy_recursive}, + harness::{Entry, Fixture}, + path::assert_dst_contains_src, process::ProcessGuard, }; -const SHARED_TEST_DIR: &str = "seedmirror-test"; - #[test] fn test_full_sync() -> anyhow::Result<()> { - let test_dir = TestDir::from("sync_test")?; - - // Empty component to add a trailing slash to ensure that the directories are synced, instead - // of placing the directory named "source" inside target - let src = test_dir.path.join("source").join(""); - let dst = test_dir.path.join("target").join(""); - - let expected_dst = test_dir.path.join("expected_target"); - let socket_path = test_dir.path.join("seedmirror-server.sock"); + let fx = Fixture::new("sync_test") + .source("subdir/initial.txt", Entry::empty()) + .target("existing_file.txt", Entry::empty()) + .build()?; - let _ = Command::new("cargo") - .current_dir(&test_dir.workspace_dir) - .arg("build") - .status()?; + fx.build("release")?; - let _server = ProcessGuard::spawn( - Command::new("target/debug/seedmirror-server") - .current_dir(&test_dir.workspace_dir) - .arg("--socket-path") - .arg(&socket_path) - .arg("--sync-delay") - .arg("100"), - )?; + let _server = ProcessGuard::spawn(&mut fx.server_cmd("release"))?; + let _client = ProcessGuard::spawn(&mut fx.client_cmd("release"))?; - // This will trigger a full sync - let _client = ProcessGuard::spawn( - Command::new("target/debug/seedmirror-client") - .current_dir(&test_dir.workspace_dir) - .arg("--socket-path") - .arg(&socket_path) - .arg("--ssh-hostname") - .arg("localhost") - .arg("-p") - .arg(format!( - "{}:{}", - src.to_string_lossy(), - dst.to_string_lossy() - )), - )?; + fx.wait_for_initial_sync()?; + assert_dst_contains_src(&fx.src, &fx.dst)?; - // Wait a second for the full sync to finish - thread::sleep(Duration::new(1, 0)); + // Check that existing files aren't removed + fx.assert_target_preserved()?; - // This will trigger sync for a single file - fs::write(src.join("new_file.txt"), "")?; - - // The sync delay is set to 100ms, so a short sleep is fine - thread::sleep(Duration::new(1, 0)); - assert_dst_contains_src(&expected_dst, &dst)?; + // A file created after the initial sync is picked up by the watcher + let new_file = "new_file.txt"; + fs::write(fx.src.join(new_file), b"")?; + fx.wait_for(Duration::from_secs(10), || fx.dst.join(new_file).exists())?; + assert_dst_contains_src(&fx.src, &fx.dst)?; Ok(()) } - -struct TestDir { - pub workspace_dir: PathBuf, - pub path: PathBuf, -} - -impl Drop for TestDir { - fn drop(&mut self) { - if self.path.exists() { - let _ = fs::remove_dir_all(&self.path); - } - } -} - -impl TestDir { - pub fn from(test_files_dir: &str) -> anyhow::Result { - let mut test_dir = env::temp_dir(); - test_dir.push(format!("{}-{}", SHARED_TEST_DIR, test_files_dir)); - - // file!() is relative to workspace, but CARGO_MANIFEST_DIR is the directory of the crate-level - // cargo manifest - let workspace_dir = - Path::new(&format!("{}/../..", env!("CARGO_MANIFEST_DIR"))).canonicalize()?; - - let mut test_file_path = workspace_dir - .join(file!()) - .parent() - .with_context(|| "invalid file path")? - .to_owned(); - - test_file_path.push("test_files"); - test_file_path.push(test_files_dir); - - copy_recursive(&test_file_path, &test_dir)?; - - Ok(Self { - workspace_dir, - path: test_dir, - }) - } -} diff --git a/crates/seedmirror-test/tests/test_files/sync_test/expected_target/existing_file.txt b/crates/seedmirror-test/tests/test_files/sync_test/expected_target/existing_file.txt deleted file mode 100644 index e69de29..0000000 diff --git a/crates/seedmirror-test/tests/test_files/sync_test/expected_target/new_file.txt b/crates/seedmirror-test/tests/test_files/sync_test/expected_target/new_file.txt deleted file mode 100644 index e69de29..0000000 diff --git a/crates/seedmirror-test/tests/test_files/sync_test/expected_target/subdir/initial.txt b/crates/seedmirror-test/tests/test_files/sync_test/expected_target/subdir/initial.txt deleted file mode 100644 index e69de29..0000000 diff --git a/crates/seedmirror-test/tests/test_files/sync_test/source/subdir/initial.txt b/crates/seedmirror-test/tests/test_files/sync_test/source/subdir/initial.txt deleted file mode 100644 index e69de29..0000000 diff --git a/crates/seedmirror-test/tests/test_files/sync_test/target/existing_file.txt b/crates/seedmirror-test/tests/test_files/sync_test/target/existing_file.txt deleted file mode 100644 index e69de29..0000000 diff --git a/docker/seedmirror-client/Dockerfile b/docker/seedmirror-client/Dockerfile index c25bc15..cc60255 100644 --- a/docker/seedmirror-client/Dockerfile +++ b/docker/seedmirror-client/Dockerfile @@ -34,7 +34,7 @@ FROM debian:trixie-slim AS final ARG APP_NAME RUN apt-get update && apt-get install -y \ - openssh-client rsync && \ + openssh-client && \ rm -rf /var/lib/apt/lists/* COPY --from=build /bin/$APP_NAME /bin/ diff --git a/docker/seedmirror-test/Dockerfile b/docker/seedmirror-test/Dockerfile index 9c2cab7..cbe3bda 100644 --- a/docker/seedmirror-test/Dockerfile +++ b/docker/seedmirror-test/Dockerfile @@ -8,10 +8,11 @@ ENV RUSTUP_HOME=/usr/local/rustup \ RUN apt-get update && apt-get install -y --no-install-recommends \ curl \ openssh-server \ - rsync \ build-essential \ git \ + just \ ca-certificates \ + linux-perf \ && rm -rf /var/lib/apt/lists/* RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain stable diff --git a/flake.lock b/flake.lock index e6a89c4..b51768c 100644 --- a/flake.lock +++ b/flake.lock @@ -16,9 +16,44 @@ "type": "github" } }, + "nixpkgs_2": { + "locked": { + "lastModified": 1744536153, + "narHash": "sha256-awS2zRgF4uTwrOKwwiJcByDzDOdo3Q1rPZbiHQg/N38=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "18dd725c29603f582cf1900e0d25f9f1063dbf11", + "type": "github" + }, + "original": { + "owner": "NixOS", + "ref": "nixpkgs-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, "root": { "inputs": { - "nixpkgs": "nixpkgs" + "nixpkgs": "nixpkgs", + "rust-overlay": "rust-overlay" + } + }, + "rust-overlay": { + "inputs": { + "nixpkgs": "nixpkgs_2" + }, + "locked": { + "lastModified": 1790234952, + "narHash": "sha256-tjQa1vDoHJir4lasZTakz/xgmsfB6SRjeQqaIbefFOk=", + "owner": "oxalica", + "repo": "rust-overlay", + "rev": "ed34466c59b766481248959a21a6e46cf0c60468", + "type": "github" + }, + "original": { + "owner": "oxalica", + "repo": "rust-overlay", + "type": "github" } } }, diff --git a/flake.nix b/flake.nix index fd5c82f..7fe9855 100644 --- a/flake.nix +++ b/flake.nix @@ -1,18 +1,40 @@ { - inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable"; + inputs = { + nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable"; + rust-overlay.url = "github:oxalica/rust-overlay"; + }; - outputs = {nixpkgs, ...}: let + outputs = { + nixpkgs, + rust-overlay, + ... + }: let system = "x86_64-linux"; - pkgs = import nixpkgs {inherit system;}; + pkgs = import nixpkgs { + inherit system; + overlays = [rust-overlay.overlays.default]; + }; + + rustToolchain = (pkgs.rust-bin.fromRustupToolchainFile ./rust-toolchain.toml) + .override { + extensions = ["clippy" "rustfmt"]; + targets = ["x86_64-unknown-linux-musl"]; + }; + + muslCC = pkgs.pkgsMusl.stdenv.cc; in { devShells.${system}.default = pkgs.mkShell { - packages = with pkgs; [ - cargo - rustc - clippy - pnpm - nodejs + packages = [ + rustToolchain + pkgs.pnpm + pkgs.nodejs + pkgs.just + pkgs.samply ]; + + CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_LINKER = "${muslCC}/bin/cc"; + CC_x86_64_unknown_linux_musl = "${muslCC}/bin/cc"; + CXX_x86_64_unknown_linux_musl = "${muslCC}/bin/c++"; }; }; } diff --git a/justfile b/justfile new file mode 100644 index 0000000..675a175 --- /dev/null +++ b/justfile @@ -0,0 +1,45 @@ +# Show all recipes +default: + @just --list + +# Build the client and server (release) +build: + cargo build --release + +# Build for x86_64-unknown-linux-musl (static) +build-musl: + cargo build --release --target x86_64-unknown-linux-musl + +# Build with symbols for profiling (binaries in target/profiling/) +build-profiling: + cargo build --profile profiling + +# Run the functional sync test +test: + cargo test -p seedmirror-test + +# Build the test container image +build-test-image: + docker build -t seedmirror-test ./docker/seedmirror-test + +# Build the client release image +build-client-image tag="latest": + docker build -f docker/seedmirror-client/Dockerfile -t seedmirror-client:{{tag}} . + +# Enter (or create) the test container. Build the image first +test-container: + ./test/run-container.sh + +# Run the ignored profiling test. Run this inside the test container +profile-test: + cargo test -p seedmirror-test --test profile_test -- --ignored --nocapture + +# Serve the server perf profile in the browser via samply +view-server port="3000": + samply import target/profiling/seedmirror-server.perf.data \ + --symbol-dir target/profiling --port {{port}} --no-open + +# Serve the client perf profile in the browser via samply +view-client port="3001": + samply import target/profiling/seedmirror-client.perf.data \ + --symbol-dir target/profiling --port {{port}} --no-open diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 398f3f0..6422209 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,2 +1,2 @@ [toolchain] -channel = "1.97.0" +channel = "1.98.1" diff --git a/test-container.sh b/test/run-container.sh similarity index 75% rename from test-container.sh rename to test/run-container.sh index 7f34b66..4154478 100755 --- a/test-container.sh +++ b/test/run-container.sh @@ -5,8 +5,7 @@ set -euo pipefail IMAGE_NAME="seedmirror-test" CONTAINER_NAME="seedmirror-test" -echo "Building Docker image ${IMAGE_NAME}..." -docker build -t "${IMAGE_NAME}" ./docker/seedmirror-test +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" if [[ $(docker ps -q -f name=^/${CONTAINER_NAME}$) ]]; then echo "Attaching to existing container '${CONTAINER_NAME}'..." @@ -18,6 +17,8 @@ echo "Starting new container '${CONTAINER_NAME}'..." docker run -it \ --rm \ --name "${CONTAINER_NAME}" \ - -v "$(pwd):/workspace" \ + -v "${REPO_ROOT}:/workspace" \ -p 8080:8080 \ + --cap-add=PERFMON \ + --cap-add=SYS_PTRACE \ "${IMAGE_NAME}"