Add serve-all integration tests + fix push/pull agent-id routing
Integration tests (tests/serve_all.rs): 7 new tests verifying serve-all dispatches correctly for each protocol: - serve_all_routes_onion_push/pull — push 4 revisions, pull back - serve_all_routes_hdf5_sync — dataset-granular sync via serve-all - serve_all_routes_fs_sync — general file sync via serve-all - serve_all_multi_protocol_sequential — FS then HDF5 on same server - serve_all_fs_warm_noop — second sync is a no-op - serve_all_push_pull_round_trip — push N revisions, pull, count matches Bug fix (main.rs): cmd_push and cmd_pull were hardcoding agent_id as "push-client" / "pull-client" in their IbltManifest / ManifestRequest. serve-all uses the agent_id to derive the filename in the root directory, so these literal strings routed all pushes to the same stub file regardless of the remote path. Fixed by using remote_path.to_string_lossy() as the agent_id — this is the natural semantic (agent identifies the remote endpoint) and remains backward-compatible with the fixed-file serve command (which ignores agent_id entirely). Added onion_path_for_agent() helper in handle_any_client: lazily creates the base HDF5 stub file (8-byte magic) when the serve-all root doesn't yet contain the target file, mirroring the create-on-first-pull behavior in cmd_pull. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
osobh
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
e1e1bf0b2e
commit
a6df333dfe
@@ -449,7 +449,7 @@ async fn cmd_push(
|
||||
.collect();
|
||||
let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
|
||||
let iblt_manifest = IbltManifest {
|
||||
agent_id: "push-client".to_string(),
|
||||
agent_id: remote_path.to_string_lossy().into_owned(),
|
||||
file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
|
||||
revision_count: local_rev_numbers.len() as u64,
|
||||
head_revision: local_rev_numbers.last().copied().unwrap_or(0),
|
||||
@@ -613,7 +613,7 @@ async fn cmd_pull(
|
||||
|
||||
// Send manifest request
|
||||
peer.send(&SyncMessage::ManifestRequest {
|
||||
agent_id: "pull-client".to_string(),
|
||||
agent_id: remote_path.to_string_lossy().into_owned(),
|
||||
head_revision: local_rev_count.saturating_sub(1),
|
||||
revision_count: local_rev_count,
|
||||
})
|
||||
@@ -1612,6 +1612,27 @@ fn sanitize_agent_id(id: &str) -> String {
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Resolve (and lazily create) the HDF5 file path for an onion agent in the
|
||||
/// serve-all root directory.
|
||||
///
|
||||
/// The `agent_id` is sanitised into a safe filename component and the base
|
||||
/// HDF5 file is created with the minimal 8-byte magic if it does not yet exist.
|
||||
/// `OnionFile::create_auto` requires the base file to be present (even if
|
||||
/// empty) so it can detect the page size; the create-on-first-push path here
|
||||
/// mirrors what `cmd_pull` does for the local file.
|
||||
fn onion_path_for_agent(root: &Path, agent_id: &str) -> Result<PathBuf> {
|
||||
let safe_id = sanitize_agent_id(agent_id);
|
||||
let h5_path = root.join(&safe_id);
|
||||
if !h5_path.exists() {
|
||||
if let Some(parent) = h5_path.parent() {
|
||||
std::fs::create_dir_all(parent)?;
|
||||
}
|
||||
// Write minimal HDF5 magic so OnionFile::create_auto can open it.
|
||||
std::fs::write(&h5_path, b"\x89HDF\r\n\x1a\n")?;
|
||||
}
|
||||
Ok(h5_path)
|
||||
}
|
||||
|
||||
/// Dispatch one incoming connection to the correct protocol handler based on
|
||||
/// the first message received.
|
||||
///
|
||||
@@ -1628,8 +1649,7 @@ async fn handle_any_client(
|
||||
let first_msg = conn.recv().await?;
|
||||
match first_msg {
|
||||
SyncMessage::IbltRequest { ref sketch } => {
|
||||
let safe_id = sanitize_agent_id(&sketch.agent_id);
|
||||
let h5_path = root.join(format!("{safe_id}.claws"));
|
||||
let h5_path = onion_path_for_agent(&root, &sketch.agent_id)?;
|
||||
if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
|
||||
let _ = conn
|
||||
.send(&SyncMessage::Error {
|
||||
@@ -1640,8 +1660,7 @@ async fn handle_any_client(
|
||||
}
|
||||
}
|
||||
SyncMessage::ManifestRequest { ref agent_id, .. } => {
|
||||
let safe_id = sanitize_agent_id(agent_id);
|
||||
let h5_path = root.join(format!("{safe_id}.claws"));
|
||||
let h5_path = onion_path_for_agent(&root, agent_id)?;
|
||||
if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
|
||||
let _ = conn
|
||||
.send(&SyncMessage::Error {
|
||||
|
||||
@@ -0,0 +1,326 @@
|
||||
//! Integration tests for `clawsync serve-all` — the universal multi-protocol server.
|
||||
//!
|
||||
//! Each test starts a single `serve-all` server and connects to it with
|
||||
//! different client types (onion push/pull, HDF5-sync, fs-sync) verifying that
|
||||
//! the correct protocol is dispatched based on the first message.
|
||||
//!
|
||||
//! Run with:
|
||||
//! cargo test -p clawsync-cli --test serve_all -- --nocapture
|
||||
|
||||
use std::fs;
|
||||
use std::io::{BufRead, BufReader};
|
||||
use std::net::SocketAddr;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::time::Duration;
|
||||
|
||||
use clawhdf5::FileBuilder;
|
||||
use clawhdf5_onion::writer::OnionFile;
|
||||
use clawsync_hdf5::DatasetManifest;
|
||||
use tempfile::TempDir;
|
||||
|
||||
const BIN: &str = env!("CARGO_BIN_EXE_clawsync");
|
||||
const H5_MAGIC: &[u8] = b"\x89HDF\r\n\x1a\n";
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// serve-all harness
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
struct ServeAllServer {
|
||||
child: Child,
|
||||
pub addr: SocketAddr,
|
||||
pub dir: TempDir,
|
||||
}
|
||||
|
||||
impl ServeAllServer {
|
||||
/// Start `clawsync serve-all <dir> --bind 127.0.0.1:0 [--allow-delete]`
|
||||
/// and wait for the startup line.
|
||||
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");
|
||||
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");
|
||||
let stdout = child.stdout.take().unwrap();
|
||||
let mut reader = BufReader::new(stdout);
|
||||
|
||||
// Wait for: "ClawSync server listening on 127.0.0.1:PORT"
|
||||
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 did not print startup line in time; got: {line:?}");
|
||||
}
|
||||
}
|
||||
|
||||
let addr_str = line
|
||||
.trim()
|
||||
.rsplit("on ")
|
||||
.next()
|
||||
.expect("unexpected startup line")
|
||||
.trim();
|
||||
let addr: SocketAddr = addr_str
|
||||
.parse()
|
||||
.expect(&format!("could not parse bound addr from: {addr_str}"));
|
||||
|
||||
// Drain remaining stdout in a background thread.
|
||||
std::thread::spawn(move || for _ in reader.lines() {});
|
||||
|
||||
Self { child, addr, dir }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ServeAllServer {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Create a temp dir with an HDF5 file containing `n` onion revisions.
|
||||
fn make_versioned_h5(n: u8) -> (TempDir, PathBuf, OnionFile) {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let h5 = dir.path().join("data.h5");
|
||||
fs::write(&h5, H5_MAGIC).unwrap();
|
||||
let mut onion = OnionFile::create(&h5, 4096).unwrap();
|
||||
for i in 0..n {
|
||||
let mut s = onion.begin_session(None).unwrap();
|
||||
s.record_page(0, &vec![i; 4096]);
|
||||
onion.commit_session(s, Some(&format!("rev {i}"))).unwrap();
|
||||
}
|
||||
onion.flush().unwrap();
|
||||
(dir, h5, onion)
|
||||
}
|
||||
|
||||
/// Build an HDF5 file with the given f64 datasets.
|
||||
fn write_h5(path: &Path, datasets: &[(&str, Vec<f64>)]) {
|
||||
let mut b = FileBuilder::new();
|
||||
for (name, data) in datasets {
|
||||
b.create_dataset(name).with_f64_data(data);
|
||||
}
|
||||
let bytes = b.finish().unwrap();
|
||||
fs::write(path, bytes).unwrap();
|
||||
}
|
||||
|
||||
/// Run `clawsync push <local> <addr/remote>` and assert exit 0.
|
||||
fn run_push(local: &Path, addr: SocketAddr, remote_name: &str) {
|
||||
let remote = format!("{addr}/{remote_name}");
|
||||
let status = Command::new(BIN)
|
||||
.args(["push", local.to_str().unwrap(), &remote])
|
||||
.status()
|
||||
.expect("failed to run push");
|
||||
assert!(status.success(), "push exited with {status}");
|
||||
}
|
||||
|
||||
/// Run `clawsync pull <addr/remote> <local>` and assert exit 0.
|
||||
fn run_pull(addr: SocketAddr, remote_name: &str, local: &Path) {
|
||||
let remote = format!("{addr}/{remote_name}");
|
||||
let status = Command::new(BIN)
|
||||
.args(["pull", &remote, local.to_str().unwrap()])
|
||||
.status()
|
||||
.expect("failed to run pull");
|
||||
assert!(status.success(), "pull exited with {status}");
|
||||
}
|
||||
|
||||
/// Run `clawsync hdf5-sync <local> <addr/rel>` and assert exit 0.
|
||||
fn run_hdf5_sync(local: &Path, addr: SocketAddr, rel_path: &str) {
|
||||
let remote = format!("{addr}/{rel_path}");
|
||||
let status = Command::new(BIN)
|
||||
.args(["hdf5-sync", local.to_str().unwrap(), &remote])
|
||||
.status()
|
||||
.expect("failed to run hdf5-sync");
|
||||
assert!(status.success(), "hdf5-sync exited with {status}");
|
||||
}
|
||||
|
||||
/// Run `clawsync sync <src> <addr/>` and assert exit 0.
|
||||
fn run_fs_sync(src: &Path, addr: SocketAddr, delete: bool) {
|
||||
let remote = format!("{addr}/");
|
||||
let mut cmd = Command::new(BIN);
|
||||
cmd.args(["sync", src.to_str().unwrap(), &remote]);
|
||||
if delete {
|
||||
cmd.arg("--delete");
|
||||
}
|
||||
let status = cmd.status().expect("failed to run sync");
|
||||
assert!(status.success(), "sync exited with {status}");
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Tests
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Push onion revisions to a serve-all server.
|
||||
///
|
||||
/// The server should detect IbltRequest as the onion protocol and serve the
|
||||
/// file at `<root>/<agent-id>.claws`.
|
||||
#[test]
|
||||
fn serve_all_routes_onion_push() {
|
||||
let server = ServeAllServer::start(false);
|
||||
let (_src_dir, h5, _onion) = make_versioned_h5(3);
|
||||
|
||||
run_push(&h5, server.addr, "onion-data");
|
||||
|
||||
// Verify the serve-all root has the served file.
|
||||
let entries: Vec<_> = fs::read_dir(server.dir.path())
|
||||
.unwrap()
|
||||
.filter_map(|e| e.ok())
|
||||
.collect();
|
||||
assert!(
|
||||
!entries.is_empty(),
|
||||
"serve-all root should contain the served file"
|
||||
);
|
||||
}
|
||||
|
||||
/// Pull revisions via a serve-all server.
|
||||
#[test]
|
||||
fn serve_all_routes_onion_pull() {
|
||||
let server = ServeAllServer::start(false);
|
||||
let (_src_dir, src_h5, _onion) = make_versioned_h5(4);
|
||||
|
||||
// Push first so the server has revisions.
|
||||
run_push(&src_h5, server.addr, "onion-data");
|
||||
|
||||
// Pull to a fresh local file.
|
||||
let dst_dir = TempDir::new().unwrap();
|
||||
let dst_h5 = dst_dir.path().join("data.h5");
|
||||
run_pull(server.addr, "onion-data", &dst_h5);
|
||||
|
||||
// Destination should now have the same number of revisions.
|
||||
let dst_onion = OnionFile::open(&dst_h5).expect("failed to open pulled file");
|
||||
assert_eq!(
|
||||
dst_onion.revision_count(),
|
||||
4,
|
||||
"pulled file should have 4 revisions"
|
||||
);
|
||||
}
|
||||
|
||||
/// HDF5 dataset-granular sync through a serve-all server.
|
||||
#[test]
|
||||
fn serve_all_routes_hdf5_sync() {
|
||||
let server = ServeAllServer::start(false);
|
||||
|
||||
let src_dir = TempDir::new().unwrap();
|
||||
let src_h5 = src_dir.path().join("science.h5");
|
||||
write_h5(
|
||||
&src_h5,
|
||||
&[
|
||||
("temperature", vec![1.0, 2.0, 3.0]),
|
||||
("pressure", vec![101.0, 102.0]),
|
||||
],
|
||||
);
|
||||
|
||||
run_hdf5_sync(&src_h5, server.addr, "science.h5");
|
||||
|
||||
// Verify the served file now has the same datasets.
|
||||
let served = server.dir.path().join("science.h5");
|
||||
assert!(served.exists(), "served h5 file must exist after hdf5-sync");
|
||||
|
||||
let src_manifest = DatasetManifest::from_path(&src_h5).expect("failed to read src manifest");
|
||||
let srv_manifest = DatasetManifest::from_path(&served).expect("failed to read server manifest");
|
||||
assert_eq!(
|
||||
src_manifest.len(),
|
||||
srv_manifest.len(),
|
||||
"server must have same number of datasets as source"
|
||||
);
|
||||
for (path, entry) in &src_manifest.datasets {
|
||||
let srv = srv_manifest.get(path).expect(&format!("dataset {path} missing on server"));
|
||||
assert_eq!(entry.blake3, srv.blake3, "dataset {path} hash mismatch");
|
||||
}
|
||||
}
|
||||
|
||||
/// General file sync (any-type) through a serve-all server.
|
||||
#[test]
|
||||
fn serve_all_routes_fs_sync() {
|
||||
let server = ServeAllServer::start(false);
|
||||
|
||||
let src_dir = TempDir::new().unwrap();
|
||||
fs::write(src_dir.path().join("hello.txt"), b"hello, world").unwrap();
|
||||
fs::write(src_dir.path().join("data.bin"), vec![0xDE, 0xAD, 0xBE, 0xEF]).unwrap();
|
||||
|
||||
run_fs_sync(src_dir.path(), server.addr, false);
|
||||
|
||||
// Both files must exist in the serve-all root.
|
||||
let hello = server.dir.path().join("hello.txt");
|
||||
let data = server.dir.path().join("data.bin");
|
||||
assert!(hello.exists(), "hello.txt must be synced");
|
||||
assert!(data.exists(), "data.bin must be synced");
|
||||
assert_eq!(fs::read(&hello).unwrap(), b"hello, world");
|
||||
assert_eq!(fs::read(&data).unwrap(), [0xDE, 0xAD, 0xBE, 0xEF]);
|
||||
}
|
||||
|
||||
/// Two sequential clients of different protocols connect to the same serve-all
|
||||
/// port; both must complete successfully without the server crashing.
|
||||
#[test]
|
||||
fn serve_all_multi_protocol_sequential() {
|
||||
let server = ServeAllServer::start(false);
|
||||
|
||||
// First: FS sync.
|
||||
let src_dir = TempDir::new().unwrap();
|
||||
fs::write(src_dir.path().join("file.txt"), b"content").unwrap();
|
||||
run_fs_sync(src_dir.path(), server.addr, false);
|
||||
|
||||
// Second: HDF5 sync.
|
||||
let h5_dir = TempDir::new().unwrap();
|
||||
let h5 = h5_dir.path().join("metrics.h5");
|
||||
write_h5(&h5, &[("latency", vec![0.1, 0.2, 0.3])]);
|
||||
run_hdf5_sync(&h5, server.addr, "metrics.h5");
|
||||
|
||||
// Both outputs must be present in the serve-all root.
|
||||
assert!(server.dir.path().join("file.txt").exists());
|
||||
assert!(server.dir.path().join("metrics.h5").exists());
|
||||
}
|
||||
|
||||
/// A warm no-op FS sync (already in sync) through serve-all succeeds.
|
||||
#[test]
|
||||
fn serve_all_fs_warm_noop() {
|
||||
let server = ServeAllServer::start(false);
|
||||
|
||||
let src_dir = TempDir::new().unwrap();
|
||||
fs::write(src_dir.path().join("stable.txt"), b"unchanged").unwrap();
|
||||
|
||||
// First sync.
|
||||
run_fs_sync(src_dir.path(), server.addr, false);
|
||||
// Second sync — same content, should be a no-op.
|
||||
run_fs_sync(src_dir.path(), server.addr, false);
|
||||
|
||||
assert_eq!(
|
||||
fs::read(server.dir.path().join("stable.txt")).unwrap(),
|
||||
b"unchanged"
|
||||
);
|
||||
}
|
||||
|
||||
/// Push/pull round-trip through serve-all preserves revision data.
|
||||
#[test]
|
||||
fn serve_all_push_pull_round_trip() {
|
||||
let server = ServeAllServer::start(false);
|
||||
let (_src_dir, src_h5, src_onion) = make_versioned_h5(5);
|
||||
let src_rev_count = src_onion.revision_count();
|
||||
|
||||
run_push(&src_h5, server.addr, "round-trip");
|
||||
|
||||
let dst_dir = TempDir::new().unwrap();
|
||||
let dst_h5 = dst_dir.path().join("round_trip.h5");
|
||||
run_pull(server.addr, "round-trip", &dst_h5);
|
||||
|
||||
let dst_onion = OnionFile::open(&dst_h5).unwrap();
|
||||
assert_eq!(
|
||||
dst_onion.revision_count(),
|
||||
src_rev_count,
|
||||
"round-tripped file must have same revision count"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user