Add serve-all universal server + QuicSyncBackend for agent
serve-all (clawsync-cli): - New ServeAll command listens on one port and dispatches on the first message: IbltRequest/ManifestRequest → onion revision protocol (file path derived from agent_id: root/<sanitized-id>.claws), Hdf5ManifestRequest → HDF5 dataset protocol, FsDirManifest → FS general sync. - Extracted handle_client_msg and handle_hdf5_client_msg inner functions that take an already-received first SyncMessage; handle_client/handle_hdf5_client become thin wrappers that call recv() then delegate. - Added FsSyncServer::handle_with_first_message() for the same pattern on the FS side (FsSyncServer owns the peer so handle_any_client consumes it there). - agent_id sanitization strips path-unsafe characters before using as filename. QuicSyncBackend (clawsync-agent): - Mirrors TcpSyncBackend exactly but establishes a QuicConnection via quic_connect(addr, "localhost", QuicConfig::self_signed()) per call. - Implements push (IBLT pre-flight + pipelined LayerPackets) and pull (ManifestRequest + packet drain + merge) via the same protocol logic. - Exported from clawsync_agent as QuicSyncBackend. 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
f1bdf921be
commit
e1e1bf0b2e
@@ -13,6 +13,7 @@ use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
|
||||
use clawsync_onion::merger::merge_packets;
|
||||
use clawsync_onion::selector::SyncSelector;
|
||||
use clawsync_transport::protocol::SyncMessage;
|
||||
use clawsync_transport::quic::{QuicConfig, quic_connect};
|
||||
use clawsync_transport::tcp::TcpConnection;
|
||||
|
||||
use crate::error::AgentSyncError;
|
||||
@@ -260,6 +261,199 @@ impl SyncBackend for TcpSyncBackend {
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// QUIC implementation
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// `SyncBackend` implementation over a QUIC connection (TLS 1.3, self-signed).
|
||||
///
|
||||
/// Each call to `push` or `pull` establishes a fresh QUIC connection using a
|
||||
/// self-signed certificate (suitable for testing and same-LAN deployments).
|
||||
/// For production use, construct a `QuicConfig` with a trusted certificate and
|
||||
/// pass it to `QuicSyncBackend::with_config`.
|
||||
pub struct QuicSyncBackend {
|
||||
remote_addr: SocketAddr,
|
||||
agent_id: String,
|
||||
}
|
||||
|
||||
impl QuicSyncBackend {
|
||||
/// Create a new backend connecting to `remote_addr` using a self-signed
|
||||
/// TLS certificate.
|
||||
pub fn new(remote_addr: SocketAddr, agent_id: impl Into<String>) -> Self {
|
||||
Self {
|
||||
remote_addr,
|
||||
agent_id: agent_id.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl SyncBackend for QuicSyncBackend {
|
||||
async fn push(
|
||||
&self,
|
||||
local_path: &Path,
|
||||
selector: &SyncSelector,
|
||||
) -> Result<SyncStats, AgentSyncError> {
|
||||
let local_onion = OnionFile::open(local_path).map_err(AgentSyncError::Onion)?;
|
||||
let h5_base = std::fs::read(local_path).map_err(AgentSyncError::Io)?;
|
||||
let local_manifest = ClawSyncManifest::from_onion(&self.agent_id, &local_onion, &h5_base);
|
||||
|
||||
let config = QuicConfig::self_signed().map_err(AgentSyncError::Transport)?;
|
||||
let conn = quic_connect(self.remote_addr, "localhost", config)
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
|
||||
// ── IBLT pre-flight ───────────────────────────────────────────────
|
||||
let local_rev_numbers: Vec<u64> = local_onion
|
||||
.list_revisions()
|
||||
.iter()
|
||||
.map(|s| s.revision)
|
||||
.collect();
|
||||
let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
|
||||
let iblt_manifest = IbltManifest {
|
||||
agent_id: self.agent_id.clone(),
|
||||
file_blake3: local_manifest.file_blake3,
|
||||
revision_count: local_manifest.revision_count,
|
||||
head_revision: local_manifest.head_revision,
|
||||
head_blake3: local_manifest.head_blake3,
|
||||
last_write: local_manifest.last_write,
|
||||
sketch_cells: sketch.cell_count() as u32,
|
||||
sketch: sketch.to_bytes(),
|
||||
};
|
||||
conn.send(&SyncMessage::IbltRequest {
|
||||
sketch: iblt_manifest,
|
||||
})
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
|
||||
let missing_from_remote = match conn.recv().await.map_err(AgentSyncError::Transport)? {
|
||||
SyncMessage::IbltResponse {
|
||||
missing_from_remote,
|
||||
..
|
||||
} => missing_from_remote,
|
||||
SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
|
||||
other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
|
||||
};
|
||||
|
||||
let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
|
||||
.map_err(AgentSyncError::SyncOnion)?;
|
||||
let packets =
|
||||
clawsync_onion::selector::filter_packets(all_packets, selector, &local_onion)
|
||||
.map_err(AgentSyncError::SyncOnion)?;
|
||||
|
||||
let total = packets.len() as u64;
|
||||
let mut bytes_sent = 0u64;
|
||||
|
||||
for packet in packets {
|
||||
let sz = packet.page_data_size() as u64;
|
||||
conn.send(&SyncMessage::LayerPacket { packet })
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
match conn.recv().await.map_err(AgentSyncError::Transport)? {
|
||||
SyncMessage::Ack { .. } => {
|
||||
bytes_sent += sz;
|
||||
}
|
||||
SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
|
||||
other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
|
||||
}
|
||||
}
|
||||
|
||||
conn.send(&SyncMessage::SyncComplete {
|
||||
revisions_transferred: total,
|
||||
bytes_transferred: bytes_sent,
|
||||
})
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
conn.close();
|
||||
|
||||
Ok(SyncStats {
|
||||
revisions_transferred: total,
|
||||
bytes_transferred: bytes_sent,
|
||||
revisions_skipped: 0,
|
||||
})
|
||||
}
|
||||
|
||||
async fn pull(&self, local_path: &Path) -> Result<SyncStats, AgentSyncError> {
|
||||
let local_rev_count = OnionFile::open(local_path)
|
||||
.map(|o| o.revision_count())
|
||||
.unwrap_or(0);
|
||||
|
||||
let config = QuicConfig::self_signed().map_err(AgentSyncError::Transport)?;
|
||||
let conn = quic_connect(self.remote_addr, "localhost", config)
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
|
||||
conn.send(&SyncMessage::ManifestRequest {
|
||||
agent_id: self.agent_id.clone(),
|
||||
head_revision: local_rev_count.saturating_sub(1),
|
||||
revision_count: local_rev_count,
|
||||
})
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
|
||||
let server_rev_count = match conn.recv().await.map_err(AgentSyncError::Transport)? {
|
||||
SyncMessage::ManifestResponse { manifest } => manifest.revision_count,
|
||||
SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
|
||||
other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
|
||||
};
|
||||
|
||||
if server_rev_count <= local_rev_count {
|
||||
conn.send(&SyncMessage::SyncComplete {
|
||||
revisions_transferred: 0,
|
||||
bytes_transferred: 0,
|
||||
})
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
return Ok(SyncStats::default());
|
||||
}
|
||||
|
||||
let mut local_onion = if local_path.exists() {
|
||||
OnionFile::open(local_path)
|
||||
.or_else(|_| OnionFile::create_auto(local_path))
|
||||
.map_err(AgentSyncError::Onion)?
|
||||
} else {
|
||||
std::fs::write(local_path, b"\x89HDF\r\n\x1a\n").map_err(AgentSyncError::Io)?;
|
||||
OnionFile::create_auto(local_path).map_err(AgentSyncError::Onion)?
|
||||
};
|
||||
|
||||
let mut packets = Vec::new();
|
||||
loop {
|
||||
match conn.recv().await.map_err(AgentSyncError::Transport)? {
|
||||
SyncMessage::LayerPacket { packet } => {
|
||||
let rev = packet.revision;
|
||||
conn.send(&SyncMessage::Ack { revision: rev })
|
||||
.await
|
||||
.map_err(AgentSyncError::Transport)?;
|
||||
packets.push(packet);
|
||||
}
|
||||
SyncMessage::SyncComplete {
|
||||
revisions_transferred,
|
||||
bytes_transferred,
|
||||
} => {
|
||||
let stats = merge_packets(&mut local_onion, packets, true)
|
||||
.map_err(AgentSyncError::SyncOnion)?;
|
||||
return Ok(SyncStats {
|
||||
revisions_transferred,
|
||||
bytes_transferred,
|
||||
revisions_skipped: stats.revisions_skipped,
|
||||
});
|
||||
}
|
||||
SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
|
||||
other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn local_manifest(&self, local_path: &Path) -> Result<ClawSyncManifest, AgentSyncError> {
|
||||
let onion = OnionFile::open(local_path).map_err(AgentSyncError::Onion)?;
|
||||
let h5_base = std::fs::read(local_path).map_err(AgentSyncError::Io)?;
|
||||
Ok(ClawSyncManifest::from_onion(
|
||||
&self.agent_id,
|
||||
&onion,
|
||||
&h5_base,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Tests
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
//!
|
||||
//! ## Modules
|
||||
//!
|
||||
//! - [`backend`]: `SyncBackend` trait + `TcpSyncBackend` implementation
|
||||
//! - [`backend`]: `SyncBackend` trait + `TcpSyncBackend` / `QuicSyncBackend` implementations
|
||||
//! - [`onion_memory`]: `OnionMemory` — revision-aware wrapper around `HDF5Memory`
|
||||
//! - [`scheduler`]: `SyncScheduler` — autonomous push after each flush
|
||||
//! - [`negotiator`]: Peer capability negotiation
|
||||
@@ -32,7 +32,7 @@ pub mod negotiator;
|
||||
pub mod onion_memory;
|
||||
pub mod scheduler;
|
||||
|
||||
pub use backend::{PeerCapabilities, SyncBackend, SyncStats, TcpSyncBackend};
|
||||
pub use backend::{PeerCapabilities, QuicSyncBackend, SyncBackend, SyncStats, TcpSyncBackend};
|
||||
pub use error::AgentSyncError;
|
||||
pub use negotiator::{SessionCapabilities, assert_compatible, negotiate};
|
||||
pub use onion_memory::{MemorySnapshot, OnionMemory};
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
//! ```
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::path::PathBuf;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use clap::{Parser, Subcommand};
|
||||
@@ -263,6 +263,29 @@ enum Commands {
|
||||
#[arg(long)]
|
||||
quic: bool,
|
||||
},
|
||||
|
||||
/// Universal server: handles onion-revision, HDF5-dataset, and FS-sync
|
||||
/// clients on a single port by dispatching on the first message.
|
||||
///
|
||||
/// Onion clients are served files rooted at `<dir>/<agent-id>.claws`;
|
||||
/// HDF5 and FS clients use `<dir>` as their root.
|
||||
ServeAll {
|
||||
/// Root directory for all served files and sub-directories.
|
||||
#[arg(value_name = "DIR")]
|
||||
dir: PathBuf,
|
||||
/// Address to listen on (default: 0.0.0.0:9996).
|
||||
#[arg(long, default_value = "0.0.0.0:9996")]
|
||||
bind: SocketAddr,
|
||||
/// Allow clients to delete files / datasets from the served directory.
|
||||
#[arg(long)]
|
||||
allow_delete: bool,
|
||||
/// Exclude paths matching this glob pattern from FS sync (may be repeated).
|
||||
#[arg(long, value_name = "GLOB")]
|
||||
exclude: Vec<String>,
|
||||
/// Use QUIC transport instead of TCP (self-signed TLS, for testing).
|
||||
#[arg(long)]
|
||||
quic: bool,
|
||||
},
|
||||
}
|
||||
|
||||
/// Sub-commands for `clawsync branch`.
|
||||
@@ -748,8 +771,15 @@ async fn cmd_serve(h5_path: PathBuf, bind: SocketAddr, quic: bool) -> Result<()>
|
||||
/// as a fallback for old clients.
|
||||
///
|
||||
/// Works with either `SyncPeer::Tcp` or `SyncPeer::Quic` — transport-agnostic.
|
||||
async fn handle_client(conn: &mut SyncPeer, h5_path: &PathBuf) -> Result<()> {
|
||||
match conn.recv().await? {
|
||||
async fn handle_client(conn: &mut SyncPeer, h5_path: &Path) -> Result<()> {
|
||||
let first = conn.recv().await?;
|
||||
handle_client_msg(conn, h5_path, first).await
|
||||
}
|
||||
|
||||
/// Inner dispatch: process an already-received first message for the onion
|
||||
/// revision protocol. Called by both `handle_client` and `handle_any_client`.
|
||||
async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage) -> Result<()> {
|
||||
match msg {
|
||||
// ── IBLT pre-flight (push direction only) ─────────────────────────
|
||||
SyncMessage::IbltRequest {
|
||||
sketch: client_iblt,
|
||||
@@ -1366,9 +1396,22 @@ async fn handle_hdf5_client(
|
||||
conn: &mut SyncPeer,
|
||||
serve_dir: &std::path::Path,
|
||||
allow_delete: bool,
|
||||
) -> Result<()> {
|
||||
let first = conn.recv().await?;
|
||||
handle_hdf5_client_msg(conn, serve_dir, allow_delete, first).await
|
||||
}
|
||||
|
||||
/// Inner dispatch for the HDF5 dataset-granular protocol, processing an
|
||||
/// already-received first message. Called by both `handle_hdf5_client` and
|
||||
/// `handle_any_client`.
|
||||
async fn handle_hdf5_client_msg(
|
||||
conn: &mut SyncPeer,
|
||||
serve_dir: &Path,
|
||||
allow_delete: bool,
|
||||
msg: SyncMessage,
|
||||
) -> Result<()> {
|
||||
// Receive the client's manifest.
|
||||
let (remote_path, _client_file_blake3, client_entries, want_delete) = match conn.recv().await? {
|
||||
let (remote_path, _client_file_blake3, client_entries, want_delete) = match msg {
|
||||
SyncMessage::Hdf5ManifestRequest {
|
||||
remote_path,
|
||||
file_blake3,
|
||||
@@ -1550,6 +1593,135 @@ async fn cmd_serve_hdf5(
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Universal multi-protocol server
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Sanitise an agent ID so it is safe to use as a filename component.
|
||||
///
|
||||
/// Allows alphanumerics, `-`, `_`, and `.`; replaces everything else with `_`.
|
||||
fn sanitize_agent_id(id: &str) -> String {
|
||||
id.chars()
|
||||
.map(|c| {
|
||||
if c.is_alphanumeric() || c == '-' || c == '_' || c == '.' {
|
||||
c
|
||||
} else {
|
||||
'_'
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Dispatch one incoming connection to the correct protocol handler based on
|
||||
/// the first message received.
|
||||
///
|
||||
/// - `IbltRequest` / `ManifestRequest` → onion-revision protocol;
|
||||
/// file path = `root/<agent-id>.claws`
|
||||
/// - `Hdf5ManifestRequest` → HDF5 dataset-granular protocol; root dir = `root`
|
||||
/// - `FsDirManifest` → FS general file sync; root dir = `root`
|
||||
async fn handle_any_client(
|
||||
mut conn: SyncPeer,
|
||||
root: PathBuf,
|
||||
allow_delete: bool,
|
||||
excludes: globset::GlobSet,
|
||||
) -> Result<()> {
|
||||
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"));
|
||||
if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
|
||||
let _ = conn
|
||||
.send(&SyncMessage::Error {
|
||||
message: e.to_string(),
|
||||
})
|
||||
.await;
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
SyncMessage::ManifestRequest { ref agent_id, .. } => {
|
||||
let safe_id = sanitize_agent_id(agent_id);
|
||||
let h5_path = root.join(format!("{safe_id}.claws"));
|
||||
if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
|
||||
let _ = conn
|
||||
.send(&SyncMessage::Error {
|
||||
message: e.to_string(),
|
||||
})
|
||||
.await;
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
SyncMessage::Hdf5ManifestRequest { .. } => {
|
||||
if let Err(e) = handle_hdf5_client_msg(&mut conn, &root, allow_delete, first_msg).await
|
||||
{
|
||||
let _ = conn
|
||||
.send(&SyncMessage::Error {
|
||||
message: e.to_string(),
|
||||
})
|
||||
.await;
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
SyncMessage::FsDirManifest { .. } => {
|
||||
// FsSyncServer takes ownership of the connection.
|
||||
let srv = FsSyncServer::new(conn, root, excludes, allow_delete);
|
||||
srv.handle_with_first_message(first_msg)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
||||
}
|
||||
other => anyhow::bail!("serve-all: unrecognized first message: {other:?}"),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Serve a root directory on a single port, accepting connections from any
|
||||
/// ClawSync client (onion revision, HDF5 dataset, or general file sync).
|
||||
async fn cmd_serve_all(
|
||||
dir: PathBuf,
|
||||
bind: SocketAddr,
|
||||
allow_delete: bool,
|
||||
exclude: Vec<String>,
|
||||
quic: bool,
|
||||
) -> Result<()> {
|
||||
let excludes = build_glob_set(&exclude)?;
|
||||
|
||||
if quic {
|
||||
let cfg =
|
||||
QuicConfig::self_signed().with_context(|| "failed to build QUIC self-signed config")?;
|
||||
let server = QuicServer::bind(bind, cfg)
|
||||
.await
|
||||
.with_context(|| format!("cannot bind QUIC to {bind}"))?;
|
||||
println!("ClawSync server listening on {}", server.local_addr);
|
||||
loop {
|
||||
let conn = server.accept().await?;
|
||||
let peer = SyncPeer::Quic(Arc::new(conn));
|
||||
let root = dir.clone();
|
||||
let excl = excludes.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = handle_any_client(peer, root, allow_delete, excl).await {
|
||||
eprintln!("serve-all QUIC client error: {e}");
|
||||
}
|
||||
});
|
||||
}
|
||||
} else {
|
||||
let server = TcpServer::bind(bind)
|
||||
.await
|
||||
.with_context(|| format!("cannot bind to {bind}"))?;
|
||||
println!("ClawSync server listening on {}", server.local_addr);
|
||||
loop {
|
||||
let (conn, peer_addr) = server.accept().await?;
|
||||
let peer = SyncPeer::Tcp(conn);
|
||||
let root = dir.clone();
|
||||
let excl = excludes.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = handle_any_client(peer, root, allow_delete, excl).await {
|
||||
eprintln!("serve-all client {peer_addr} error: {e}");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn format_timestamp(ts: f64) -> String {
|
||||
use chrono::{DateTime, Utc};
|
||||
let secs = ts as i64;
|
||||
@@ -1639,6 +1811,13 @@ fn main() -> Result<()> {
|
||||
allow_delete,
|
||||
quic,
|
||||
} => cmd_serve_hdf5(dir, bind, allow_delete, quic).await,
|
||||
Commands::ServeAll {
|
||||
dir,
|
||||
bind,
|
||||
allow_delete,
|
||||
exclude,
|
||||
quic,
|
||||
} => cmd_serve_all(dir, bind, allow_delete, exclude, quic).await,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -249,9 +249,25 @@ impl FsSyncServer {
|
||||
}
|
||||
|
||||
/// Handle one complete client session.
|
||||
///
|
||||
/// Receives the first message from the client, verifies it is a
|
||||
/// `FsDirManifest`, and runs the full sync protocol.
|
||||
pub async fn handle(mut self) -> Result<SyncStats, FsSyncError> {
|
||||
let first = self.conn.recv().await?;
|
||||
self.handle_with_first_message(first).await
|
||||
}
|
||||
|
||||
/// Like `handle`, but processes an already-received first message.
|
||||
///
|
||||
/// Use this when the caller has already called `recv()` on the connection
|
||||
/// (e.g., a universal multiplexed server that peeks at the first message to
|
||||
/// decide which protocol to route to).
|
||||
pub async fn handle_with_first_message(
|
||||
mut self,
|
||||
first_msg: SyncMessage,
|
||||
) -> Result<SyncStats, FsSyncError> {
|
||||
// Receive client's directory manifest.
|
||||
let client_entries = match self.conn.recv().await? {
|
||||
let client_entries = match first_msg {
|
||||
SyncMessage::FsDirManifest { entries, .. } => entries,
|
||||
other => {
|
||||
return Err(FsSyncError::Protocol(format!(
|
||||
|
||||
Reference in New Issue
Block a user