redclaw/clawsync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Apply rustfmt to entire workspace

Browse Source 
Runs cargo fmt --all; all 573 tests still passing, clippy still clean.
No logic changes — formatting only.

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
osobh
2026-04-04 20:31:33 -05:00
co-authored by Claude Sonnet 4.6
parent 3d524e2d63
commit 1c107fe58a
69 changed files with 2790 additions and 1325 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clawsync-transport/src/lib.rs
+2 -2
View File
@@ -16,8 +16,8 @@ pub mod quic;
pub mod tcp;
pub use error::TransportError;
pub use mmap::{MmapChannel, DEFAULT_CAPACITY};
pub use mmap::{DEFAULT_CAPACITY, MmapChannel};
pub use peer::{PipeReadHalf, PipeWriteHalf, SyncPeer};
pub use protocol::SyncMessage;
pub use tcp::{TcpConnection, TcpServer};
pub use quic::{QuicConfig, QuicConnection, QuicServer, quic_connect};
pub use tcp::{TcpConnection, TcpServer};
crates/clawsync-transport/src/mmap.rs
+33 -12
View File
@@ -26,12 +26,12 @@
//! - Capacity is fixed at creation time.
//! - No cross-machine support (file must be on a shared filesystem).
use memmap2::MmapMut;
use std::fs::OpenOptions;
use std::path::{Path, PathBuf};
use memmap2::MmapMut;
use crate::error::TransportError;
use crate::protocol::{SyncMessage, MAX_FRAME_SIZE};
use crate::protocol::{MAX_FRAME_SIZE, SyncMessage};
// ─────────────────────────────────────────────────────────────────────────────
// Constants
@@ -92,7 +92,11 @@ impl MmapChannel {
write_u64(&mut mmap, 24, capacity as u64); // capacity
mmap.flush().map_err(TransportError::Io)?;
Ok(Self { path: path.to_owned(), mmap, capacity })
Ok(Self {
path: path.to_owned(),
mmap,
capacity,
})
}
/// Open an existing channel file.
@@ -105,10 +109,16 @@ impl MmapChannel {
let mmap = unsafe { MmapMut::map_mut(&file).map_err(TransportError::Io)? };
if &mmap[..8] != MAGIC {
return Err(TransportError::Protocol("invalid mmap channel magic".into()));
return Err(TransportError::Protocol(
"invalid mmap channel magic".into(),
));
}
let capacity = read_u64(&mmap, 24) as usize;
Ok(Self { path: path.to_owned(), mmap, capacity })
Ok(Self {
path: path.to_owned(),
mmap,
capacity,
})
}
/// Send a message into the ring buffer.
@@ -194,8 +204,7 @@ impl MmapChannel {
write_u64(&mut self.mmap, 16, new_rh);
self.mmap.flush_range(0, 32).map_err(TransportError::Io)?;
SyncMessage::from_bytes(&body)
.map_err(TransportError::Protocol)
SyncMessage::from_bytes(&body).map_err(TransportError::Protocol)
}
/// Path to the backing file.
@@ -231,14 +240,20 @@ impl MmapChannel {
impl MmapSender {
pub fn open(path: &Path) -> Result<Self, TransportError> {
let ch = MmapChannel::open(path)?;
Ok(Self { _path: path.to_owned(), _capacity: ch.capacity })
Ok(Self {
_path: path.to_owned(),
_capacity: ch.capacity,
})
}
}
impl MmapReceiver {
pub fn open(path: &Path) -> Result<Self, TransportError> {
let ch = MmapChannel::open(path)?;
Ok(Self { _path: path.to_owned(), _capacity: ch.capacity })
Ok(Self {
_path: path.to_owned(),
_capacity: ch.capacity,
})
}
}
@@ -310,7 +325,9 @@ mod tests {
let (_f, path) = tmp_path();
let mut ch = MmapChannel::create(&path, DEFAULT_CAPACITY).unwrap();
let msg = SyncMessage::Error { message: "oops".into() };
let msg = SyncMessage::Error {
message: "oops".into(),
};
ch.send(&msg).unwrap();
let recv = ch.recv().unwrap();
assert!(matches!(recv, SyncMessage::Error { .. }));
@@ -324,9 +341,13 @@ mod tests {
ch.send(&SyncMessage::SyncComplete {
revisions_transferred: 100,
bytes_transferred: 4096 * 100,
}).unwrap();
})
.unwrap();
match ch.recv().unwrap() {
SyncMessage::SyncComplete { revisions_transferred, .. } => {
SyncMessage::SyncComplete {
revisions_transferred,
..
} => {
assert_eq!(revisions_transferred, 100);
}
other => panic!("unexpected: {other:?}"),
crates/clawsync-transport/src/peer.rs
+1 -3
View File
@@ -63,9 +63,7 @@ impl SyncPeer {
let (r, w) = conn.into_split();
(PipeReadHalf::Tcp(r), PipeWriteHalf::Tcp(w))
}
SyncPeer::Quic(arc) => {
(PipeReadHalf::Quic(arc.clone()), PipeWriteHalf::Quic(arc))
}
SyncPeer::Quic(arc) => (PipeReadHalf::Quic(arc.clone()), PipeWriteHalf::Quic(arc)),
}
}
}
crates/clawsync-transport/src/protocol.rs
+134 -61
View File
@@ -156,8 +156,8 @@ pub enum SyncMessage {
/// lacks (what to push). The client decodes `sketch` against its own
/// sketch to find revisions it needs to pull.
IbltResponse {
sketch: IbltManifest,
missing_from_remote: Vec<u64>,
sketch: IbltManifest,
missing_from_remote: Vec<u64>,
},
/// An error occurred; connection will be closed after this message.
@@ -167,7 +167,6 @@ pub enum SyncMessage {
//
// IMPORTANT: rkyv assigns discriminants positionally. These variants MUST
// remain appended after `Error` (index 8). Never insert or reorder.
/// Client announces it wants to sync a single file using CDC delta.
///
/// Carries the ordered CDC chunk descriptors of the client's version of
@@ -205,7 +204,6 @@ pub enum SyncMessage {
FsFileAck { path: String },
// ── FS directory tree sync (variants 13–16) ───────────────────────────────
/// Client sends a per-file manifest for an entire directory tree.
FsDirManifest {
/// Root path on the client (informational).
@@ -227,15 +225,15 @@ pub enum SyncMessage {
/// Server reports that all changes have been applied.
FsDirComplete {
files_added: u32,
files_modified: u32,
files_removed: u32,
files_added: u32,
files_modified: u32,
files_removed: u32,
bytes_transferred: u64,
},
/// Dry-run: reports what *would* change without applying anything.
FsDirDryRun {
would_add: Vec<String>,
would_add: Vec<String>,
would_modify: Vec<String>,
would_remove: Vec<String>,
},
@@ -244,7 +242,6 @@ pub enum SyncMessage {
//
// IMPORTANT: rkyv assigns discriminants positionally. These variants MUST
// remain appended after `FsDirDryRun` (index 16). Never insert or reorder.
/// Client sends its per-dataset manifest and the remote h5 path to sync.
///
/// The server compares this manifest against its own to find which datasets
@@ -288,9 +285,9 @@ pub enum SyncMessage {
/// Server reports that the HDF5 sync is complete.
Hdf5SyncComplete {
datasets_added: u32,
datasets_added: u32,
datasets_modified: u32,
datasets_removed: u32,
datasets_removed: u32,
bytes_transferred: u64,
},
}
@@ -310,8 +307,7 @@ impl SyncMessage {
pub fn from_bytes(bytes: &[u8]) -> Result<Self, String> {
let mut aligned = rkyv::util::AlignedVec::<16>::with_capacity(bytes.len());
aligned.extend_from_slice(bytes);
rkyv::from_bytes::<SyncMessage, rkyv::rancor::Error>(&aligned)
.map_err(|e| e.to_string())
rkyv::from_bytes::<SyncMessage, rkyv::rancor::Error>(&aligned).map_err(|e| e.to_string())
}
/// Encode to a length-prefixed frame: `[u32 LE length][message bytes]`.
@@ -333,10 +329,16 @@ impl SyncMessage {
}
let len = u32::from_le_bytes(data[..4].try_into().unwrap()) as usize;
if len > MAX_FRAME_SIZE {
return Err(format!("frame too large: {len} bytes (max {MAX_FRAME_SIZE})"));
return Err(format!(
"frame too large: {len} bytes (max {MAX_FRAME_SIZE})"
));
}
if data.len() < 4 + len {
return Err(format!("incomplete frame: have {} bytes, need {}", data.len(), 4 + len));
return Err(format!(
"incomplete frame: have {} bytes, need {}",
data.len(),
4 + len
));
}
let msg = Self::from_bytes(&data[4..4 + len])?;
Ok((msg, 4 + len))
@@ -377,7 +379,11 @@ mod tests {
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::ManifestRequest { agent_id, head_revision, revision_count } => {
SyncMessage::ManifestRequest {
agent_id,
head_revision,
revision_count,
} => {
assert_eq!(agent_id, "test-agent");
assert_eq!(head_revision, 5);
assert_eq!(revision_count, 6);
@@ -419,7 +425,10 @@ mod tests {
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::SyncComplete { revisions_transferred, bytes_transferred } => {
SyncMessage::SyncComplete {
revisions_transferred,
bytes_transferred,
} => {
assert_eq!(revisions_transferred, 10);
assert_eq!(bytes_transferred, 40960);
}
@@ -429,7 +438,9 @@ mod tests {
#[test]
fn error_message_roundtrip() {
let msg = SyncMessage::Error { message: "something went wrong".to_string() };
let msg = SyncMessage::Error {
message: "something went wrong".to_string(),
};
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
@@ -440,21 +451,23 @@ mod tests {
#[test]
fn iblt_request_roundtrip() {
use clawsync_onion::manifest::IbltManifest;
use clawsync_onion::iblt::IbltSketch;
use clawsync_onion::manifest::IbltManifest;
// Build a minimal IbltManifest
let sketch = IbltSketch::from_keys(&[0u64, 1, 2], 0xABCD);
let iblt_manifest = IbltManifest {
agent_id: "test".to_string(),
file_blake3: [0u8; 32],
agent_id: "test".to_string(),
file_blake3: [0u8; 32],
revision_count: 3,
head_revision: 2,
head_blake3: [0u8; 32],
last_write: 0.0,
sketch: sketch.to_bytes(),
sketch_cells: sketch.cell_count() as u32,
head_revision: 2,
head_blake3: [0u8; 32],
last_write: 0.0,
sketch: sketch.to_bytes(),
sketch_cells: sketch.cell_count() as u32,
};
let msg = SyncMessage::IbltRequest {
sketch: iblt_manifest,
};
let msg = SyncMessage::IbltRequest { sketch: iblt_manifest };
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
assert!(matches!(recovered, SyncMessage::IbltRequest { .. }));
@@ -462,18 +475,18 @@ mod tests {
#[test]
fn iblt_response_roundtrip() {
use clawsync_onion::manifest::IbltManifest;
use clawsync_onion::iblt::IbltSketch;
use clawsync_onion::manifest::IbltManifest;
let sketch = IbltSketch::from_keys(&[5u64, 6], 0xDEAD);
let iblt_manifest = IbltManifest {
agent_id: "srv".to_string(),
file_blake3: [1u8; 32],
agent_id: "srv".to_string(),
file_blake3: [1u8; 32],
revision_count: 2,
head_revision: 6,
head_blake3: [2u8; 32],
last_write: 1.0,
sketch: sketch.to_bytes(),
sketch_cells: sketch.cell_count() as u32,
head_revision: 6,
head_blake3: [2u8; 32],
last_write: 1.0,
sketch: sketch.to_bytes(),
sketch_cells: sketch.cell_count() as u32,
};
let msg = SyncMessage::IbltResponse {
sketch: iblt_manifest,
@@ -482,7 +495,10 @@ mod tests {
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::IbltResponse { missing_from_remote, .. } => {
SyncMessage::IbltResponse {
missing_from_remote,
..
} => {
assert_eq!(missing_from_remote, vec![7, 8, 9]);
}
_ => panic!("wrong variant"),
@@ -496,9 +512,18 @@ mod tests {
let msg = SyncMessage::FsCdcRequest {
path: "data/weights.bin".to_string(),
chunk_hashes: vec![
FsChunkHash { hash: 0xDEAD, length: 65536 },
FsChunkHash { hash: 0xBEEF, length: 32768 },
FsChunkHash { hash: 0xCAFE, length: 65536 },
FsChunkHash {
hash: 0xDEAD,
length: 65536,
},
FsChunkHash {
hash: 0xBEEF,
length: 32768,
},
FsChunkHash {
hash: 0xCAFE,
length: 65536,
},
],
};
let bytes = msg.to_bytes().unwrap();
@@ -522,7 +547,10 @@ mod tests {
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::FsCdcNeed { path, needed_indices } => {
SyncMessage::FsCdcNeed {
path,
needed_indices,
} => {
assert_eq!(path, "model.bin");
assert_eq!(needed_indices, vec![0, 2]);
}
@@ -535,18 +563,36 @@ mod tests {
let msg = SyncMessage::FsCdcData {
path: "cfg.bin".to_string(),
chunk_order: vec![
FsChunkHash { hash: 0x11, length: 65536 },
FsChunkHash { hash: 0x22, length: 65536 },
FsChunkHash {
hash: 0x11,
length: 65536,
},
FsChunkHash {
hash: 0x22,
length: 65536,
},
],
chunks: vec![
FsChunk { hash: 0x11, data: vec![0xAAu8; 100], compressed: true },
FsChunk { hash: 0x22, data: vec![0xBBu8; 200], compressed: false },
FsChunk {
hash: 0x11,
data: vec![0xAAu8; 100],
compressed: true,
},
FsChunk {
hash: 0x22,
data: vec![0xBBu8; 200],
compressed: false,
},
],
};
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::FsCdcData { path, chunk_order, chunks } => {
SyncMessage::FsCdcData {
path,
chunk_order,
chunks,
} => {
assert_eq!(path, "cfg.bin");
assert_eq!(chunk_order.len(), 2);
assert_eq!(chunks.len(), 2);
@@ -560,10 +606,14 @@ mod tests {
#[test]
fn fs_file_ack_roundtrip() {
let msg = SyncMessage::FsFileAck { path: "embeddings.bin".to_string() };
let msg = SyncMessage::FsFileAck {
path: "embeddings.bin".to_string(),
};
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
assert!(matches!(recovered, SyncMessage::FsFileAck { ref path } if path == "embeddings.bin"));
assert!(
matches!(recovered, SyncMessage::FsFileAck { ref path } if path == "embeddings.bin")
);
}
#[test]
@@ -571,8 +621,18 @@ mod tests {
let msg = SyncMessage::FsDirManifest {
local_root: "/home/user/data".to_string(),
entries: vec![
FsManifestEntry { path: "a.bin".to_string(), blake3: [1u8; 32], size: 4096, mtime: 1000 },
FsManifestEntry { path: "b.bin".to_string(), blake3: [2u8; 32], size: 8192, mtime: 2000 },
FsManifestEntry {
path: "a.bin".to_string(),
blake3: [1u8; 32],
size: 4096,
mtime: 1000,
},
FsManifestEntry {
path: "b.bin".to_string(),
blake3: [2u8; 32],
size: 8192,
mtime: 2000,
},
],
};
let bytes = msg.to_bytes().unwrap();
@@ -590,19 +650,23 @@ mod tests {
#[test]
fn fs_dir_need_roundtrip() {
let msg = SyncMessage::FsDirNeed {
needed_files: vec![
FsFileNeed {
path: "weights.bin".to_string(),
server_chunks: vec![FsChunkHash { hash: 0xAA, length: 65536 }],
expected_blake3: [0xABu8; 32],
},
],
needed_files: vec![FsFileNeed {
path: "weights.bin".to_string(),
server_chunks: vec![FsChunkHash {
hash: 0xAA,
length: 65536,
}],
expected_blake3: [0xABu8; 32],
}],
to_delete: vec!["old_model.bin".to_string()],
};
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::FsDirNeed { needed_files, to_delete } => {
SyncMessage::FsDirNeed {
needed_files,
to_delete,
} => {
assert_eq!(needed_files.len(), 1);
assert_eq!(needed_files[0].path, "weights.bin");
assert_eq!(to_delete, vec!["old_model.bin"]);
@@ -614,12 +678,19 @@ mod tests {
#[test]
fn fs_dir_complete_roundtrip() {
let msg = SyncMessage::FsDirComplete {
files_added: 2, files_modified: 3, files_removed: 1, bytes_transferred: 1_048_576,
files_added: 2,
files_modified: 3,
files_removed: 1,
bytes_transferred: 1_048_576,
};
let bytes = msg.to_bytes().unwrap();
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
match recovered {
SyncMessage::FsDirComplete { files_added, bytes_transferred, .. } => {
SyncMessage::FsDirComplete {
files_added,
bytes_transferred,
..
} => {
assert_eq!(files_added, 2);
assert_eq!(bytes_transferred, 1_048_576);
}
@@ -637,8 +708,10 @@ mod tests {
// We just verify the roundtrip still produces the exact same ManifestRequest —
// any reordering would cause a deserialization error here.
let recovered = SyncMessage::from_bytes(&bytes).unwrap();
assert!(matches!(recovered, SyncMessage::ManifestRequest { .. }),
"ManifestRequest discriminant changed — rkyv variant ordering was modified");
assert!(
matches!(recovered, SyncMessage::ManifestRequest { .. }),
"ManifestRequest discriminant changed — rkyv variant ordering was modified"
);
}
#[test]
crates/clawsync-transport/src/quic.rs
+25 -19
View File
@@ -49,11 +49,8 @@ impl QuicConfig {
.map_err(|e| TransportError::Tls(e.to_string()))?;
// Server config
let server_config = ServerConfig::with_single_cert(
vec![cert_der.clone()],
key_der,
)
.map_err(|e| TransportError::Tls(e.to_string()))?;
let server_config = ServerConfig::with_single_cert(vec![cert_der.clone()], key_der)
.map_err(|e| TransportError::Tls(e.to_string()))?;
// Client config: trust only our self-signed cert
let mut roots = rustls::RootCertStore::empty();
@@ -63,7 +60,10 @@ impl QuicConfig {
let client_config = ClientConfig::with_root_certificates(Arc::new(roots))
.map_err(|e| TransportError::Tls(e.to_string()))?;
Ok(Self { server_config, client_config })
Ok(Self {
server_config,
client_config,
})
}
}
@@ -83,9 +83,7 @@ impl QuicConnection {
/// Send a `SyncMessage` on a new unidirectional QUIC stream.
pub async fn send(&self, msg: &SyncMessage) -> Result<(), TransportError> {
let body = msg
.to_bytes()
.map_err(TransportError::Serialization)?;
let body = msg.to_bytes().map_err(TransportError::Serialization)?;
let mut send = self
.conn
@@ -101,7 +99,8 @@ impl QuicConnection {
send.write_all(&body)
.await
.map_err(|e| TransportError::Io(e.into()))?;
send.finish().map_err(|e| TransportError::Quic(e.to_string()))?;
send.finish()
.map_err(|e| TransportError::Quic(e.to_string()))?;
Ok(())
}
@@ -147,10 +146,12 @@ pub struct QuicServer {
impl QuicServer {
/// Bind a QUIC server endpoint.
pub async fn bind(addr: SocketAddr, config: QuicConfig) -> Result<Self, TransportError> {
let endpoint = Endpoint::server(config.server_config, addr)
.map_err(TransportError::Io)?;
let endpoint = Endpoint::server(config.server_config, addr).map_err(TransportError::Io)?;
let local_addr = endpoint.local_addr()?;
Ok(Self { endpoint, local_addr })
Ok(Self {
endpoint,
local_addr,
})
}
/// Accept the next incoming QUIC connection.
@@ -177,8 +178,8 @@ pub async fn quic_connect(
server_name: &str,
config: QuicConfig,
) -> Result<QuicConnection, TransportError> {
let mut endpoint = Endpoint::client("0.0.0.0:0".parse().unwrap())
.map_err(TransportError::Io)?;
let mut endpoint =
Endpoint::client("0.0.0.0:0".parse().unwrap()).map_err(TransportError::Io)?;
endpoint.set_default_client_config(config.client_config);
let conn = endpoint
@@ -217,10 +218,15 @@ mod tests {
roots.add(cert_der).unwrap();
let cli_cfg = ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();
let server = QuicServer::bind(any_addr(), QuicConfig {
server_config: srv_cfg,
client_config: cli_cfg.clone(),
}).await.unwrap();
let server = QuicServer::bind(
any_addr(),
QuicConfig {
server_config: srv_cfg,
client_config: cli_cfg.clone(),
},
)
.await
.unwrap();
let addr = server.local_addr;
let server_task = tokio::spawn(async move {
crates/clawsync-transport/src/tcp.rs
+33 -14
View File
@@ -7,11 +7,11 @@
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use crate::error::TransportError;
use crate::protocol::{SyncMessage, MAX_FRAME_SIZE};
use crate::protocol::{MAX_FRAME_SIZE, SyncMessage};
// ─────────────────────────────────────────────────────────────────────────────
// TcpConnection — async read/write of SyncMessages
@@ -37,9 +37,7 @@ impl TcpConnection {
/// Send a `SyncMessage` over the connection.
pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
let body = msg
.to_bytes()
.map_err(TransportError::Serialization)?;
let body = msg.to_bytes().map_err(TransportError::Serialization)?;
let len = body.len() as u32;
self.stream.write_all(&len.to_le_bytes()).await?;
self.stream.write_all(&body).await?;
@@ -65,8 +63,7 @@ impl TcpConnection {
let mut body = vec![0u8; len];
self.stream.read_exact(&mut body).await?;
SyncMessage::from_bytes(&body)
.map_err(TransportError::Deserialization)
SyncMessage::from_bytes(&body).map_err(TransportError::Deserialization)
}
/// Gracefully shut down the write half of the connection.
@@ -81,7 +78,10 @@ impl TcpConnection {
/// pipelined send/receive (sliding window) without holding a single lock.
pub fn into_split(self) -> (TcpReadHalf, TcpWriteHalf) {
let (read_half, write_half) = self.stream.into_split();
(TcpReadHalf { half: read_half }, TcpWriteHalf { half: write_half })
(
TcpReadHalf { half: read_half },
TcpWriteHalf { half: write_half },
)
}
}
@@ -102,7 +102,10 @@ impl TcpReadHalf {
let len = u32::from_le_bytes(len_buf) as usize;
if len > MAX_FRAME_SIZE {
return Err(TransportError::FrameTooLarge { size: len, max: MAX_FRAME_SIZE });
return Err(TransportError::FrameTooLarge {
size: len,
max: MAX_FRAME_SIZE,
});
}
let mut body = vec![0u8; len];
@@ -149,7 +152,10 @@ impl TcpServer {
pub async fn bind(addr: SocketAddr) -> Result<Self, TransportError> {
let listener = TcpListener::bind(addr).await?;
let local_addr = listener.local_addr()?;
Ok(Self { listener, local_addr })
Ok(Self {
listener,
local_addr,
})
}
/// Accept the next incoming connection.
@@ -186,7 +192,10 @@ mod tests {
});
let mut client = TcpConnection::connect(addr).await.unwrap();
client.send(&SyncMessage::Ack { revision: 42 }).await.unwrap();
client
.send(&SyncMessage::Ack { revision: 42 })
.await
.unwrap();
let reply = client.recv().await.unwrap();
assert!(matches!(reply, SyncMessage::Ack { revision: 42 }));
@@ -208,7 +217,10 @@ mod tests {
let mut client = TcpConnection::connect(addr).await.unwrap();
for rev in 0u64..5 {
client.send(&SyncMessage::Ack { revision: rev }).await.unwrap();
client
.send(&SyncMessage::Ack { revision: rev })
.await
.unwrap();
}
server_task.await.unwrap();
}
@@ -233,7 +245,10 @@ mod tests {
client.send(&msg).await.unwrap();
let reply = client.recv().await.unwrap();
match reply {
SyncMessage::SyncComplete { revisions_transferred, bytes_transferred } => {
SyncMessage::SyncComplete {
revisions_transferred,
bytes_transferred,
} => {
assert_eq!(revisions_transferred, 7);
assert_eq!(bytes_transferred, 28672);
}
@@ -263,7 +278,11 @@ mod tests {
client.send(&msg).await.unwrap();
let reply = client.recv().await.unwrap();
match reply {
SyncMessage::ManifestRequest { agent_id, head_revision, .. } => {
SyncMessage::ManifestRequest {
agent_id,
head_revision,
..
} => {
assert_eq!(agent_id, "my-agent");
assert_eq!(head_revision, 3);
}
crates/clawsync-transport/tests/e2e_quic_sync.rs
+54 -20
View File
@@ -12,7 +12,7 @@ use clawhdf5_onion::writer::OnionFile;
use clawsync_onion::{differ::diff_revisions, manifest::ClawSyncManifest, merger::merge_packets};
use clawsync_transport::{
protocol::SyncMessage,
quic::{quic_connect, QuicConfig, QuicServer},
quic::{QuicConfig, QuicServer, quic_connect},
};
use tempfile::NamedTempFile;
@@ -62,8 +62,7 @@ fn make_quic_configs() -> (QuicConfig, QuicConfig) {
let mut roots = rustls::RootCertStore::empty();
roots.add(srv_cert_der).unwrap();
let client_config =
quinn::ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();
let client_config = quinn::ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();
// Client-side QuicConfig requires a server_config field; generate a fresh
// cert+key pair so there is no key/cert mismatch.
@@ -71,8 +70,14 @@ fn make_quic_configs() -> (QuicConfig, QuicConfig) {
let cli_server_config =
quinn::ServerConfig::with_single_cert(vec![cli_cert_der], cli_key_der).unwrap();
let srv_cfg = QuicConfig { server_config: srv_server_config, client_config: client_config.clone() };
let cli_cfg = QuicConfig { server_config: cli_server_config, client_config };
let srv_cfg = QuicConfig {
server_config: srv_server_config,
client_config: client_config.clone(),
};
let cli_cfg = QuicConfig {
server_config: cli_server_config,
client_config,
};
(srv_cfg, cli_cfg)
}
@@ -99,16 +104,20 @@ async fn run_quic_push(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
match msg {
SyncMessage::ManifestRequest { .. } => {
let dst_manifest = ClawSyncManifest::from_onion("dst", &dst, H5_BASE);
conn.send(&SyncMessage::ManifestResponse { manifest: dst_manifest })
.await
.unwrap();
conn.send(&SyncMessage::ManifestResponse {
manifest: dst_manifest,
})
.await
.unwrap();
}
SyncMessage::LayerPacket { packet } => {
let rev = packet.revision;
merge_packets(&mut dst, vec![packet], false).unwrap();
packets_received += 1;
conn.send(&SyncMessage::Ack { revision: rev }).await.unwrap();
conn.send(&SyncMessage::Ack { revision: rev })
.await
.unwrap();
}
SyncMessage::SyncComplete { .. } => {
@@ -125,7 +134,9 @@ async fn run_quic_push(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
});
// ── Client logic ─────────────────────────────────────────────────────────
let client = quic_connect(server_addr, "localhost", cli_cfg).await.unwrap();
let client = quic_connect(server_addr, "localhost", cli_cfg)
.await
.unwrap();
client
.send(&SyncMessage::ManifestRequest {
@@ -152,7 +163,10 @@ async fn run_quic_push(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
for packet in packets {
let rev = packet.revision;
client.send(&SyncMessage::LayerPacket { packet }).await.unwrap();
client
.send(&SyncMessage::LayerPacket { packet })
.await
.unwrap();
match client.recv().await.unwrap() {
SyncMessage::Ack { revision } => assert_eq!(revision, rev),
other => panic!("client: expected Ack, got {other:?}"),
@@ -228,7 +242,11 @@ async fn quic_revision_hashes_match_after_sync() {
assert_eq!(src_revisions.len(), dst_revisions.len());
for (s, d) in src_revisions.iter().zip(dst_revisions.iter()) {
assert_eq!(s.blake3_hex, d.blake3_hex, "rev {} hash mismatch", s.revision);
assert_eq!(
s.blake3_hex, d.blake3_hex,
"rev {} hash mismatch",
s.revision
);
assert_eq!(s.annotation, d.annotation);
}
}
@@ -269,9 +287,11 @@ async fn run_quic_pull(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
};
let srv_manifest = ClawSyncManifest::from_onion("srv", &src, H5_BASE);
conn.send(&SyncMessage::ManifestResponse { manifest: srv_manifest })
.await
.unwrap();
conn.send(&SyncMessage::ManifestResponse {
manifest: srv_manifest,
})
.await
.unwrap();
if src.revision_count() > client_rev_count {
let remote_head = if client_rev_count == 0 {
@@ -284,7 +304,9 @@ async fn run_quic_pull(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
let mut bytes_sent = 0u64;
for packet in packets {
bytes_sent += packet.page_data_size() as u64;
conn.send(&SyncMessage::LayerPacket { packet }).await.unwrap();
conn.send(&SyncMessage::LayerPacket { packet })
.await
.unwrap();
match conn.recv().await.unwrap() {
SyncMessage::Ack { .. } => {}
other => panic!("server: expected Ack, got {other:?}"),
@@ -310,7 +332,9 @@ async fn run_quic_pull(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
});
// ── Client logic ─────────────────────────────────────────────────────────
let client = quic_connect(server_addr, "localhost", cli_cfg).await.unwrap();
let client = quic_connect(server_addr, "localhost", cli_cfg)
.await
.unwrap();
let local_rev_count = dst.revision_count();
client
@@ -344,10 +368,16 @@ async fn run_quic_pull(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
match client.recv().await.unwrap() {
SyncMessage::LayerPacket { packet } => {
let rev = packet.revision;
client.send(&SyncMessage::Ack { revision: rev }).await.unwrap();
client
.send(&SyncMessage::Ack { revision: rev })
.await
.unwrap();
packets.push(packet);
}
SyncMessage::SyncComplete { revisions_transferred, .. } => {
SyncMessage::SyncComplete {
revisions_transferred,
..
} => {
merge_packets(&mut dst, packets, true).unwrap();
server_handle.await.unwrap();
return (revisions_transferred, dst);
@@ -412,7 +442,11 @@ async fn quic_pull_revision_hashes_match() {
assert_eq!(srv_revisions.len(), cli_revisions.len());
for (s, c) in srv_revisions.iter().zip(cli_revisions.iter()) {
assert_eq!(s.blake3_hex, c.blake3_hex, "rev {} hash mismatch", s.revision);
assert_eq!(
s.blake3_hex, c.blake3_hex,
"rev {} hash mismatch",
s.revision
);
assert_eq!(s.annotation, c.annotation);
}
}
crates/clawsync-transport/tests/e2e_sync.rs
+30 -23
View File
@@ -18,11 +18,7 @@
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use clawhdf5_onion::writer::OnionFile;
use clawsync_onion::{
differ::diff_revisions,
manifest::ClawSyncManifest,
merger::merge_packets,
};
use clawsync_onion::{differ::diff_revisions, manifest::ClawSyncManifest, merger::merge_packets};
use clawsync_transport::{
protocol::SyncMessage,
tcp::{TcpConnection, TcpServer},
@@ -45,9 +41,7 @@ fn make_onion(n: u8) -> (NamedTempFile, OnionFile) {
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.commit_session(s, Some(&format!("rev {i}"))).unwrap();
}
(tmp, onion)
}
@@ -57,10 +51,7 @@ fn make_onion(n: u8) -> (NamedTempFile, OnionFile) {
// number of LayerPackets actually transferred.
// ─────────────────────────────────────────────────────────────────────────────
async fn run_push(
src: OnionFile,
mut dst: OnionFile,
) -> (u64, OnionFile) {
async fn run_push(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
// Bind the server; the OS picks a free port.
let server = TcpServer::bind(localhost_any()).await.unwrap();
let server_addr = server.local_addr;
@@ -79,8 +70,7 @@ async fn run_push(
match msg {
SyncMessage::ManifestRequest { head_revision, .. } => {
// Build our local manifest and send it back.
let dst_manifest =
ClawSyncManifest::from_onion("dst", &dst, H5_BASE);
let dst_manifest = ClawSyncManifest::from_onion("dst", &dst, H5_BASE);
conn.send(&SyncMessage::ManifestResponse {
manifest: dst_manifest,
})
@@ -163,7 +153,10 @@ async fn run_push(
.unwrap();
let (server_received, dst_final) = server_handle.await.unwrap();
assert_eq!(server_received, total_sent, "server count must match client send count");
assert_eq!(
server_received, total_sent,
"server count must match client send count"
);
(total_sent, dst_final)
}
@@ -287,7 +280,7 @@ async fn large_push_correct_packet_count() {
// ─────────────────────────────────────────────────────────────────────────────
async fn run_pull(
src: OnionFile, // server-side onion (richer)
src: OnionFile, // server-side onion (richer)
mut dst: OnionFile, // client-side onion (missing revisions)
) -> (u64, OnionFile) {
let server = TcpServer::bind(localhost_any()).await.unwrap();
@@ -305,9 +298,11 @@ async fn run_pull(
// 2. Send back the server manifest.
let srv_manifest = ClawSyncManifest::from_onion("srv", &src, H5_BASE);
conn.send(&SyncMessage::ManifestResponse { manifest: srv_manifest })
.await
.unwrap();
conn.send(&SyncMessage::ManifestResponse {
manifest: srv_manifest,
})
.await
.unwrap();
// 3. Server has more revisions — push the delta to the client.
if src.revision_count() > client_rev_count {
@@ -321,7 +316,9 @@ async fn run_pull(
let mut bytes_sent = 0u64;
for packet in packets {
bytes_sent += packet.page_data_size() as u64;
conn.send(&SyncMessage::LayerPacket { packet }).await.unwrap();
conn.send(&SyncMessage::LayerPacket { packet })
.await
.unwrap();
match conn.recv().await.unwrap() {
SyncMessage::Ack { .. } => {}
other => panic!("server: expected Ack, got {other:?}"),
@@ -381,10 +378,16 @@ async fn run_pull(
match client.recv().await.unwrap() {
SyncMessage::LayerPacket { packet } => {
let rev = packet.revision;
client.send(&SyncMessage::Ack { revision: rev }).await.unwrap();
client
.send(&SyncMessage::Ack { revision: rev })
.await
.unwrap();
packets.push(packet);
}
SyncMessage::SyncComplete { revisions_transferred, .. } => {
SyncMessage::SyncComplete {
revisions_transferred,
..
} => {
merge_packets(&mut dst, packets, true).unwrap();
server_handle.await.unwrap();
return (revisions_transferred, dst);
@@ -449,7 +452,11 @@ async fn pull_revision_hashes_match() {
assert_eq!(srv_revisions.len(), cli_revisions.len());
for (s, c) in srv_revisions.iter().zip(cli_revisions.iter()) {
assert_eq!(s.blake3_hex, c.blake3_hex, "rev {} hash mismatch", s.revision);
assert_eq!(
s.blake3_hex, c.blake3_hex,
"rev {} hash mismatch",
s.revision
);
assert_eq!(s.annotation, c.annotation);
}
}