feat: --verbose per-file progress for sync + serve-all QUIC integration tests
- Add ProgressEvent enum (TotalFiles / FileAcked) to clawsync-fs - Add FsSyncClient::with_progress() builder accepting an UnboundedSender<ProgressEvent>; emits one TotalFiles before RTT2 and one FileAcked per completed file - Add --verbose / -v flag to `clawsync sync`: spawns a background printer that reports each synced file and byte count to stderr as acks arrive - Add ServeAllServerQuic harness to serve_all integration tests - Add 3 QUIC integration tests for serve-all: cold-copy, warm-noop, incremental; exercises the wait_for_peer_close drain path in handle_any_client Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
ad479324ca
commit
b8365a4d41
@@ -205,6 +205,9 @@ enum Commands {
|
|||||||
/// Show what would be transferred without actually transferring anything.
|
/// Show what would be transferred without actually transferring anything.
|
||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
dry_run: bool,
|
dry_run: bool,
|
||||||
|
/// Print each file as it is transferred.
|
||||||
|
#[arg(short, long)]
|
||||||
|
verbose: bool,
|
||||||
/// Exclude paths matching this glob pattern (may be repeated).
|
/// Exclude paths matching this glob pattern (may be repeated).
|
||||||
#[arg(long, value_name = "GLOB")]
|
#[arg(long, value_name = "GLOB")]
|
||||||
exclude: Vec<String>,
|
exclude: Vec<String>,
|
||||||
@@ -1131,6 +1134,7 @@ async fn cmd_sync(
|
|||||||
remote: String,
|
remote: String,
|
||||||
delete: bool,
|
delete: bool,
|
||||||
dry_run: bool,
|
dry_run: bool,
|
||||||
|
verbose: bool,
|
||||||
exclude: Vec<String>,
|
exclude: Vec<String>,
|
||||||
quic: bool,
|
quic: bool,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
@@ -1164,7 +1168,32 @@ async fn cmd_sync(
|
|||||||
if dry_run {
|
if dry_run {
|
||||||
client = client.with_dry_run();
|
client = client.with_dry_run();
|
||||||
}
|
}
|
||||||
|
// If verbose, spawn a background printer that reports each file as it completes.
|
||||||
|
let _progress_task = if verbose && !dry_run {
|
||||||
|
use clawsync_fs::ProgressEvent;
|
||||||
|
use tokio::sync::mpsc;
|
||||||
|
let (tx, mut rx) = mpsc::unbounded_channel::<ProgressEvent>();
|
||||||
|
client = client.with_progress(tx);
|
||||||
|
let task = tokio::spawn(async move {
|
||||||
|
while let Some(ev) = rx.recv().await {
|
||||||
|
match ev {
|
||||||
|
ProgressEvent::TotalFiles(n) => {
|
||||||
|
eprintln!(" Transferring {n} file(s)...");
|
||||||
|
}
|
||||||
|
ProgressEvent::FileAcked { path, bytes } => {
|
||||||
|
eprintln!(" ✓ {path} ({bytes} B)");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
Some(task)
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
let stats = client.run().await?;
|
let stats = client.run().await?;
|
||||||
|
if let Some(t) = _progress_task {
|
||||||
|
let _ = t.await;
|
||||||
|
}
|
||||||
|
|
||||||
if dry_run {
|
if dry_run {
|
||||||
println!("Dry-run (no changes applied):");
|
println!("Dry-run (no changes applied):");
|
||||||
@@ -1861,9 +1890,10 @@ fn main() -> Result<()> {
|
|||||||
remote,
|
remote,
|
||||||
delete,
|
delete,
|
||||||
dry_run,
|
dry_run,
|
||||||
|
verbose,
|
||||||
exclude,
|
exclude,
|
||||||
quic,
|
quic,
|
||||||
} => cmd_sync(local, remote, delete, dry_run, exclude, quic).await,
|
} => cmd_sync(local, remote, delete, dry_run, verbose, exclude, quic).await,
|
||||||
Commands::ServeFs {
|
Commands::ServeFs {
|
||||||
dir,
|
dir,
|
||||||
bind,
|
bind,
|
||||||
|
|||||||
@@ -324,3 +324,138 @@ fn serve_all_push_pull_round_trip() {
|
|||||||
"round-tripped file must have same revision count"
|
"round-tripped file must have same revision count"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// QUIC transport tests for serve-all
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Harness for `clawsync serve-all --quic`.
|
||||||
|
struct ServeAllServerQuic {
|
||||||
|
child: Child,
|
||||||
|
pub addr: SocketAddr,
|
||||||
|
pub dir: TempDir,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ServeAllServerQuic {
|
||||||
|
fn start(allow_delete: bool) -> Self {
|
||||||
|
let dir = TempDir::new().expect("failed to create temp dir");
|
||||||
|
let mut cmd = Command::new(BIN);
|
||||||
|
cmd.arg("serve-all")
|
||||||
|
.arg(dir.path())
|
||||||
|
.arg("--bind")
|
||||||
|
.arg("127.0.0.1:0")
|
||||||
|
.arg("--quic");
|
||||||
|
if allow_delete {
|
||||||
|
cmd.arg("--allow-delete");
|
||||||
|
}
|
||||||
|
cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
|
||||||
|
|
||||||
|
let mut child = cmd.spawn().expect("failed to spawn clawsync serve-all --quic");
|
||||||
|
let stdout = child.stdout.take().unwrap();
|
||||||
|
let mut reader = BufReader::new(stdout);
|
||||||
|
|
||||||
|
let deadline = std::time::Instant::now() + Duration::from_secs(10);
|
||||||
|
let mut line = String::new();
|
||||||
|
loop {
|
||||||
|
line.clear();
|
||||||
|
reader.read_line(&mut line).expect("failed to read startup line");
|
||||||
|
if line.contains("ClawSync server listening on") {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if std::time::Instant::now() > deadline {
|
||||||
|
panic!("serve-all --quic did not print startup line in time; got: {line:?}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let addr_str = line.trim().rsplit("on ").next().unwrap().trim();
|
||||||
|
let addr: SocketAddr = addr_str
|
||||||
|
.parse()
|
||||||
|
.expect(&format!("could not parse bound addr from: {addr_str}"));
|
||||||
|
|
||||||
|
std::thread::spawn(move || for _ in reader.lines() {});
|
||||||
|
|
||||||
|
Self { child, addr, dir }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for ServeAllServerQuic {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
let _ = self.child.kill();
|
||||||
|
let _ = self.child.wait();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn run_fs_sync_quic(src: &Path, addr: SocketAddr) {
|
||||||
|
let remote = format!("{addr}/");
|
||||||
|
let status = Command::new(BIN)
|
||||||
|
.args(["sync", src.to_str().unwrap(), &remote, "--quic"])
|
||||||
|
.status()
|
||||||
|
.expect("failed to run sync --quic");
|
||||||
|
assert!(status.success(), "sync --quic exited with {status}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Cold copy via QUIC through the universal serve-all server.
|
||||||
|
#[test]
|
||||||
|
fn serve_all_quic_fs_cold_copy() {
|
||||||
|
let server = ServeAllServerQuic::start(false);
|
||||||
|
|
||||||
|
let src = TempDir::new().unwrap();
|
||||||
|
fs::write(src.path().join("quic_file.txt"), b"quic cold copy test").unwrap();
|
||||||
|
fs::write(src.path().join("binary.bin"), vec![0x42u8; 1024]).unwrap();
|
||||||
|
|
||||||
|
run_fs_sync_quic(src.path(), server.addr);
|
||||||
|
|
||||||
|
let dst_txt = server.dir.path().join("quic_file.txt");
|
||||||
|
let dst_bin = server.dir.path().join("binary.bin");
|
||||||
|
assert!(dst_txt.exists(), "quic_file.txt must be synced via QUIC");
|
||||||
|
assert!(dst_bin.exists(), "binary.bin must be synced via QUIC");
|
||||||
|
assert_eq!(fs::read(&dst_txt).unwrap(), b"quic cold copy test");
|
||||||
|
assert_eq!(fs::read(&dst_bin).unwrap(), vec![0x42u8; 1024]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Warm no-op via QUIC: re-syncing identical content transfers nothing
|
||||||
|
/// and exits 0.
|
||||||
|
#[test]
|
||||||
|
fn serve_all_quic_fs_warm_noop() {
|
||||||
|
let server = ServeAllServerQuic::start(false);
|
||||||
|
|
||||||
|
// Seed server directory with the same content as the source.
|
||||||
|
let src = TempDir::new().unwrap();
|
||||||
|
let content = b"identical content for noop";
|
||||||
|
fs::write(src.path().join("noop.txt"), content).unwrap();
|
||||||
|
fs::write(server.dir.path().join("noop.txt"), content).unwrap();
|
||||||
|
|
||||||
|
// Second sync should be a warm no-op.
|
||||||
|
run_fs_sync_quic(src.path(), server.addr);
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
fs::read(server.dir.path().join("noop.txt")).unwrap(),
|
||||||
|
content
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Incremental sync via QUIC: modifying one file transfers only that file.
|
||||||
|
#[test]
|
||||||
|
fn serve_all_quic_fs_incremental() {
|
||||||
|
let server = ServeAllServerQuic::start(false);
|
||||||
|
|
||||||
|
let src = TempDir::new().unwrap();
|
||||||
|
|
||||||
|
// Put both files on server already.
|
||||||
|
fs::write(src.path().join("unchanged.bin"), vec![0x11u8; 512]).unwrap();
|
||||||
|
fs::write(src.path().join("changed.bin"), vec![0xAAu8; 512]).unwrap();
|
||||||
|
run_fs_sync_quic(src.path(), server.addr);
|
||||||
|
|
||||||
|
// Modify one file; re-sync.
|
||||||
|
fs::write(src.path().join("changed.bin"), vec![0xBBu8; 512]).unwrap();
|
||||||
|
run_fs_sync_quic(src.path(), server.addr);
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
fs::read(server.dir.path().join("changed.bin")).unwrap(),
|
||||||
|
vec![0xBBu8; 512]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
fs::read(server.dir.path().join("unchanged.bin")).unwrap(),
|
||||||
|
vec![0x11u8; 512]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|||||||
@@ -27,4 +27,4 @@ pub mod manifest;
|
|||||||
pub mod session;
|
pub mod session;
|
||||||
|
|
||||||
pub use error::FsSyncError;
|
pub use error::FsSyncError;
|
||||||
pub use session::{FsSyncClient, FsSyncServer, SyncStats};
|
pub use session::{FsSyncClient, FsSyncServer, ProgressEvent, SyncStats};
|
||||||
|
|||||||
@@ -25,6 +25,26 @@ use crate::manifest::FsManifest;
|
|||||||
|
|
||||||
const PIPELINE_WINDOW: usize = 16;
|
const PIPELINE_WINDOW: usize = 16;
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// ProgressEvent
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Per-file progress notification emitted by `FsSyncClient` during a sync
|
||||||
|
/// session. Callers that want live progress should pass an
|
||||||
|
/// `UnboundedSender<ProgressEvent>` via [`FsSyncClient::with_progress`].
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum ProgressEvent {
|
||||||
|
/// Total number of files that will be transferred (sent once, before RTT 2).
|
||||||
|
TotalFiles(usize),
|
||||||
|
/// A single file has been fully transferred and acknowledged by the server.
|
||||||
|
FileAcked {
|
||||||
|
/// Relative path of the file (forward-slash separators).
|
||||||
|
path: String,
|
||||||
|
/// Number of literal chunk bytes that were sent over the wire.
|
||||||
|
bytes: u64,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
// ─────────────────────────────────────────────────────────────────────────────
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
// Stats
|
// Stats
|
||||||
// ─────────────────────────────────────────────────────────────────────────────
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
@@ -59,6 +79,7 @@ pub struct FsSyncClient {
|
|||||||
excludes: GlobSet,
|
excludes: GlobSet,
|
||||||
_delete: bool,
|
_delete: bool,
|
||||||
dry_run: bool,
|
dry_run: bool,
|
||||||
|
progress_tx: Option<tokio::sync::mpsc::UnboundedSender<ProgressEvent>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FsSyncClient {
|
impl FsSyncClient {
|
||||||
@@ -69,6 +90,7 @@ impl FsSyncClient {
|
|||||||
excludes,
|
excludes,
|
||||||
_delete: delete,
|
_delete: delete,
|
||||||
dry_run: false,
|
dry_run: false,
|
||||||
|
progress_tx: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -79,6 +101,19 @@ impl FsSyncClient {
|
|||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Enable per-file progress notifications.
|
||||||
|
///
|
||||||
|
/// The caller receives a [`ProgressEvent::TotalFiles`] before RTT 2 starts,
|
||||||
|
/// then one [`ProgressEvent::FileAcked`] per transferred file as acks arrive.
|
||||||
|
/// The sender is dropped when `run()` returns.
|
||||||
|
pub fn with_progress(
|
||||||
|
mut self,
|
||||||
|
tx: tokio::sync::mpsc::UnboundedSender<ProgressEvent>,
|
||||||
|
) -> Self {
|
||||||
|
self.progress_tx = Some(tx);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// Execute the full directory sync protocol.
|
/// Execute the full directory sync protocol.
|
||||||
pub async fn run(mut self) -> Result<SyncStats, FsSyncError> {
|
pub async fn run(mut self) -> Result<SyncStats, FsSyncError> {
|
||||||
// ── RTT 1: build local manifest and send it ──────────────────────────
|
// ── RTT 1: build local manifest and send it ──────────────────────────
|
||||||
@@ -162,6 +197,10 @@ impl FsSyncClient {
|
|||||||
// The semaphore (W=16) limits in-flight FsCdcData messages.
|
// The semaphore (W=16) limits in-flight FsCdcData messages.
|
||||||
|
|
||||||
let total = needed_files.len();
|
let total = needed_files.len();
|
||||||
|
let progress_tx = self.progress_tx.take();
|
||||||
|
if let Some(tx) = &progress_tx {
|
||||||
|
let _ = tx.send(ProgressEvent::TotalFiles(total));
|
||||||
|
}
|
||||||
let (mut read_half, mut write_half) = self.conn.into_pipe_halves();
|
let (mut read_half, mut write_half) = self.conn.into_pipe_halves();
|
||||||
let semaphore = Arc::new(Semaphore::new(PIPELINE_WINDOW));
|
let semaphore = Arc::new(Semaphore::new(PIPELINE_WINDOW));
|
||||||
// Channel carries wire bytes per file (for stats tracking).
|
// Channel carries wire bytes per file (for stats tracking).
|
||||||
@@ -222,9 +261,16 @@ impl FsSyncClient {
|
|||||||
let mut bytes_transferred = 0u64;
|
let mut bytes_transferred = 0u64;
|
||||||
for _ in 0..total {
|
for _ in 0..total {
|
||||||
match read_half.recv().await? {
|
match read_half.recv().await? {
|
||||||
SyncMessage::FsFileAck { .. } => {
|
SyncMessage::FsFileAck { path: acked_path } => {
|
||||||
semaphore.add_permits(1);
|
semaphore.add_permits(1);
|
||||||
bytes_transferred += meta_rx.recv().await.expect("meta closed early");
|
let bytes = meta_rx.recv().await.expect("meta closed early");
|
||||||
|
bytes_transferred += bytes;
|
||||||
|
if let Some(tx) = &progress_tx {
|
||||||
|
let _ = tx.send(ProgressEvent::FileAcked {
|
||||||
|
path: acked_path,
|
||||||
|
bytes,
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
SyncMessage::Error { message } => return Err(FsSyncError::Protocol(message)),
|
SyncMessage::Error { message } => return Err(FsSyncError::Protocol(message)),
|
||||||
other => {
|
other => {
|
||||||
|
|||||||
Reference in New Issue
Block a user