feat: SSH transport — clawsync sync user@host:path without pre-running server
Adds generic framed I/O transport (FramedReader/FramedWriter over any AsyncRead/AsyncWrite) and a StreamPeer backed by boxed I/O, enabling two new modes: SSH client mode: clawsync sync <local> [user@]host:path Detects SSH syntax (@ or non-numeric port), spawns `ssh user@host clawsync serve-fs path --stdio`, and uses the child's stdin/stdout as the sync protocol channel. No pre-running server needed. stdio server mode (--stdio flag on serve-fs / serve-all): clawsync serve-fs <dir> --stdio Reads/writes the sync protocol on stdin/stdout instead of binding TCP. Suppresses the startup message (stdout is the protocol channel). Used automatically by SSH clients; also usable in scripts/containers. Transport layer (clawsync-transport): - New framed.rs: FramedReader<R>, FramedWriter<W>, StreamPeer, split halves - SyncPeer::Stream variant; into_pipe_halves() splits into StreamReadHalf/WriteHalf - PipeReadHalf::Stream, PipeWriteHalf::Stream for pipelined W=16 sessions CLI (clawsync-cli): - ParsedRemote enum; parse_remote_target() handles both TCP and SSH syntax - connect_remote() dispatches to TCP/QUIC or ssh_connect() - cmd_sync and cmd_watch use connect_remote() — SSH works transparently - cmd_serve_fs and cmd_serve_all gain --stdio flag - 2 integration tests: SSH (skipped without CLAWSYNC_TEST_SSH=1), stdio no-panic 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
75e2550d9e
commit
85b6edbdff
+158
-43
@@ -37,6 +37,7 @@ use std::sync::Arc;
|
||||
|
||||
use clawsync_fs::{FsSyncClient, FsSyncPullClient, FsSyncServer};
|
||||
use clawsync_hdf5::{DatasetManifest, ReceivedDataset, apply_received_payloads};
|
||||
use clawsync_transport::framed::StreamPeer;
|
||||
use clawsync_transport::peer::{PipeWriteHalf, SyncPeer};
|
||||
use clawsync_transport::protocol::SyncMessage;
|
||||
use clawsync_transport::quic::{QuicConfig, QuicServer, quic_connect};
|
||||
@@ -192,12 +193,16 @@ enum Commands {
|
||||
///
|
||||
/// Uses CDC-based delta transfer: only changed chunks are sent,
|
||||
/// immune to insertion/deletion staircase unlike rsync.
|
||||
///
|
||||
/// Remote formats:
|
||||
/// `host:port/path` — connect to a running `serve-fs` over TCP
|
||||
/// `[user@]host:path` — connect via SSH (spawns `clawsync serve-fs --stdio`)
|
||||
Sync {
|
||||
/// Local file or directory path.
|
||||
#[arg(value_name = "LOCAL_PATH")]
|
||||
local: PathBuf,
|
||||
/// Remote address and path: `host:port/remote/path`
|
||||
#[arg(value_name = "HOST:PORT/REMOTE_PATH")]
|
||||
/// Remote: `host:port/path` (TCP) or `[user@]host:path` (SSH)
|
||||
#[arg(value_name = "REMOTE")]
|
||||
remote: String,
|
||||
/// Delete files on the remote that don't exist locally.
|
||||
#[arg(long)]
|
||||
@@ -233,6 +238,12 @@ enum Commands {
|
||||
/// Use QUIC transport instead of TCP (self-signed TLS, for testing).
|
||||
#[arg(long)]
|
||||
quic: bool,
|
||||
/// Read/write the sync protocol on stdin/stdout instead of a TCP socket.
|
||||
///
|
||||
/// Used automatically by SSH clients that spawn this process remotely.
|
||||
/// In this mode no startup message is printed and --bind is ignored.
|
||||
#[arg(long)]
|
||||
stdio: bool,
|
||||
},
|
||||
|
||||
/// Sync a local HDF5 file to a remote server at dataset granularity.
|
||||
@@ -342,6 +353,9 @@ enum Commands {
|
||||
/// Use QUIC transport instead of TCP (self-signed TLS, for testing).
|
||||
#[arg(long)]
|
||||
quic: bool,
|
||||
/// Read/write the sync protocol on stdin/stdout (used by SSH clients).
|
||||
#[arg(long)]
|
||||
stdio: bool,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -417,6 +431,105 @@ fn parse_remote(remote: &str) -> Result<(SocketAddr, PathBuf)> {
|
||||
Ok((addr, PathBuf::from(path_str)))
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// SSH / TCP remote detection
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// A parsed remote target — either a TCP socket address or an SSH endpoint.
|
||||
enum ParsedRemote {
|
||||
/// Direct TCP connection: `host:port/relative-path`
|
||||
Tcp { addr: SocketAddr, _path: String },
|
||||
/// SSH-tunnelled connection: `[user@]host:remote-path`
|
||||
Ssh { user_host: String, path: String },
|
||||
}
|
||||
|
||||
/// Parse a remote string into either a TCP or SSH target.
|
||||
///
|
||||
/// Heuristic:
|
||||
/// * If the string contains `@` → SSH syntax (`[user@]host:path`)
|
||||
/// * Otherwise try TCP (`host:port/path`): if there is no `/` or the port
|
||||
/// segment is non-numeric → SSH syntax
|
||||
fn parse_remote_target(remote: &str) -> Result<ParsedRemote> {
|
||||
// Explicit SSH: [user@]host:path
|
||||
if remote.contains('@') {
|
||||
return Ok(parse_ssh_target(remote)?);
|
||||
}
|
||||
// Try TCP: host:port/path
|
||||
if let Some(slash) = remote.find('/') {
|
||||
let addr_str = &remote[..slash];
|
||||
if let Ok(addr) = addr_str.parse::<SocketAddr>() {
|
||||
return Ok(ParsedRemote::Tcp {
|
||||
addr,
|
||||
_path: remote[slash + 1..].to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
// Fallback: treat as SSH (no @, but port is non-numeric or no slash)
|
||||
parse_ssh_target(remote)
|
||||
}
|
||||
|
||||
fn parse_ssh_target(remote: &str) -> Result<ParsedRemote> {
|
||||
let colon = remote
|
||||
.find(':')
|
||||
.with_context(|| format!("SSH remote must be '[user@]host:path', got: {remote}"))?;
|
||||
let user_host = remote[..colon].to_string();
|
||||
let path = remote[colon + 1..].to_string();
|
||||
if user_host.is_empty() {
|
||||
anyhow::bail!("SSH remote must have a host before ':', got: {remote}");
|
||||
}
|
||||
Ok(ParsedRemote::Ssh { user_host, path })
|
||||
}
|
||||
|
||||
/// Connect to a remote using either TCP or SSH depending on `target`.
|
||||
async fn connect_remote(target: &ParsedRemote, quic: bool) -> Result<SyncPeer> {
|
||||
match target {
|
||||
ParsedRemote::Tcp { addr, .. } => {
|
||||
if quic {
|
||||
let cfg = QuicConfig::insecure()
|
||||
.with_context(|| "failed to build QUIC insecure config")?;
|
||||
Ok(SyncPeer::Quic(Arc::new(
|
||||
quic_connect(*addr, "localhost", cfg)
|
||||
.await
|
||||
.with_context(|| format!("QUIC connect to {addr} failed"))?,
|
||||
)))
|
||||
} else {
|
||||
Ok(SyncPeer::Tcp(
|
||||
TcpConnection::connect(*addr)
|
||||
.await
|
||||
.with_context(|| format!("TCP connect to {addr} failed"))?,
|
||||
))
|
||||
}
|
||||
}
|
||||
ParsedRemote::Ssh { user_host, path } => {
|
||||
if quic {
|
||||
anyhow::bail!("--quic is not supported with SSH transport; use TCP or omit --quic");
|
||||
}
|
||||
ssh_connect(user_host, path).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn `ssh <user_host> clawsync serve-fs <path> --stdio` and wrap the
|
||||
/// child's stdin/stdout as a `SyncPeer::Stream`.
|
||||
async fn ssh_connect(user_host: &str, path: &str) -> Result<SyncPeer> {
|
||||
use tokio::process::Command;
|
||||
use std::process::Stdio;
|
||||
|
||||
let child = Command::new("ssh")
|
||||
.arg(user_host)
|
||||
.arg("clawsync")
|
||||
.arg("serve-fs")
|
||||
.arg(path)
|
||||
.arg("--stdio")
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit())
|
||||
.spawn()
|
||||
.with_context(|| format!("failed to spawn ssh {user_host}"))?;
|
||||
|
||||
Ok(SyncPeer::Stream(StreamPeer::from_child(child)))
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Commands
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -1189,31 +1302,22 @@ async fn cmd_sync(
|
||||
exclude: Vec<String>,
|
||||
quic: bool,
|
||||
) -> Result<()> {
|
||||
let (addr, _remote_path) = parse_remote(&remote)?;
|
||||
let target = parse_remote_target(&remote)?;
|
||||
let excludes = build_glob_set(&exclude)?;
|
||||
|
||||
let transport_label = match (&target, quic) {
|
||||
(ParsedRemote::Ssh { .. }, _) => "SSH",
|
||||
(_, true) => "QUIC",
|
||||
_ => "TCP",
|
||||
};
|
||||
println!(
|
||||
"Syncing {} → {} ({}) ...",
|
||||
local.display(),
|
||||
addr,
|
||||
if quic { "QUIC" } else { "TCP" }
|
||||
remote,
|
||||
transport_label,
|
||||
);
|
||||
|
||||
let peer: SyncPeer = if quic {
|
||||
let cfg =
|
||||
QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
|
||||
SyncPeer::Quic(Arc::new(
|
||||
quic_connect(addr, "localhost", cfg)
|
||||
.await
|
||||
.with_context(|| format!("QUIC connect to {addr} failed"))?,
|
||||
))
|
||||
} else {
|
||||
SyncPeer::Tcp(
|
||||
TcpConnection::connect(addr)
|
||||
.await
|
||||
.with_context(|| format!("cannot connect to {addr}"))?,
|
||||
)
|
||||
};
|
||||
let peer = connect_remote(&target, quic).await?;
|
||||
|
||||
let mut client = FsSyncClient::new(peer, local, excludes, delete);
|
||||
if dry_run {
|
||||
@@ -1298,20 +1402,25 @@ async fn cmd_watch(
|
||||
use std::sync::mpsc as std_mpsc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
let (addr, _remote_path) = parse_remote(&remote)?;
|
||||
let target = parse_remote_target(&remote)?;
|
||||
let excludes = build_glob_set(&exclude)?;
|
||||
let debounce = Duration::from_millis(debounce_ms);
|
||||
|
||||
let transport_label = match (&target, quic) {
|
||||
(ParsedRemote::Ssh { .. }, _) => "SSH",
|
||||
(_, true) => "QUIC",
|
||||
_ => "TCP",
|
||||
};
|
||||
println!(
|
||||
"Watching {} → {} ({}), debounce {}ms ...",
|
||||
local.display(),
|
||||
addr,
|
||||
if quic { "QUIC" } else { "TCP" },
|
||||
remote,
|
||||
transport_label,
|
||||
debounce_ms
|
||||
);
|
||||
|
||||
// Run an initial sync before entering the watch loop.
|
||||
watch_sync_once(&local, &excludes, delete, addr, quic).await?;
|
||||
watch_sync_once(&local, &excludes, delete, &target, quic).await?;
|
||||
|
||||
// Set up OS file-system watcher.
|
||||
let (tx, rx) = std_mpsc::channel::<notify::Result<notify::Event>>();
|
||||
@@ -1348,7 +1457,7 @@ async fn cmd_watch(
|
||||
}
|
||||
}
|
||||
// Re-sync, printing errors without exiting so the watcher stays alive.
|
||||
if let Err(e) = watch_sync_once(&local, &excludes, delete, addr, quic).await {
|
||||
if let Err(e) = watch_sync_once(&local, &excludes, delete, &target, quic).await {
|
||||
eprintln!(" sync error: {e:#}");
|
||||
}
|
||||
}
|
||||
@@ -1360,24 +1469,10 @@ async fn watch_sync_once(
|
||||
local: &Path,
|
||||
excludes: &globset::GlobSet,
|
||||
delete: bool,
|
||||
addr: std::net::SocketAddr,
|
||||
target: &ParsedRemote,
|
||||
quic: bool,
|
||||
) -> Result<()> {
|
||||
let peer: SyncPeer = if quic {
|
||||
let cfg =
|
||||
QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
|
||||
SyncPeer::Quic(Arc::new(
|
||||
quic_connect(addr, "localhost", cfg)
|
||||
.await
|
||||
.with_context(|| format!("QUIC connect to {addr} failed"))?,
|
||||
))
|
||||
} else {
|
||||
SyncPeer::Tcp(
|
||||
TcpConnection::connect(addr)
|
||||
.await
|
||||
.with_context(|| format!("cannot connect to {addr}"))?,
|
||||
)
|
||||
};
|
||||
let peer = connect_remote(target, quic).await?;
|
||||
let stats = FsSyncClient::new(peer, local.to_path_buf(), excludes.clone(), delete)
|
||||
.run()
|
||||
.await?;
|
||||
@@ -1448,9 +1543,19 @@ async fn cmd_serve_fs(
|
||||
allow_delete: bool,
|
||||
exclude: Vec<String>,
|
||||
quic: bool,
|
||||
stdio: bool,
|
||||
) -> Result<()> {
|
||||
let excludes = build_glob_set(&exclude)?;
|
||||
|
||||
// stdio mode: single session over stdin/stdout — used by SSH clients.
|
||||
// No TCP binding, no startup message (stdout is the protocol channel).
|
||||
if stdio {
|
||||
let peer = SyncPeer::Stream(StreamPeer::from_stdio());
|
||||
let srv = FsSyncServer::new(peer, dir, excludes, allow_delete);
|
||||
srv.handle().await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if quic {
|
||||
let cfg =
|
||||
QuicConfig::self_signed().with_context(|| "failed to build QUIC self-signed config")?;
|
||||
@@ -1992,9 +2097,17 @@ async fn cmd_serve_all(
|
||||
allow_delete: bool,
|
||||
exclude: Vec<String>,
|
||||
quic: bool,
|
||||
stdio: bool,
|
||||
) -> Result<()> {
|
||||
let excludes = build_glob_set(&exclude)?;
|
||||
|
||||
if stdio {
|
||||
let peer = SyncPeer::Stream(StreamPeer::from_stdio());
|
||||
return handle_any_client(peer, dir, allow_delete, excludes)
|
||||
.await
|
||||
.map_err(Into::into);
|
||||
}
|
||||
|
||||
if quic {
|
||||
let cfg =
|
||||
QuicConfig::self_signed().with_context(|| "failed to build QUIC self-signed config")?;
|
||||
@@ -2133,7 +2246,8 @@ fn main() -> Result<()> {
|
||||
allow_delete,
|
||||
exclude,
|
||||
quic,
|
||||
} => cmd_serve_fs(dir, bind, allow_delete, exclude, quic).await,
|
||||
stdio,
|
||||
} => cmd_serve_fs(dir, bind, allow_delete, exclude, quic, stdio).await,
|
||||
Commands::Hdf5Sync {
|
||||
local,
|
||||
remote,
|
||||
@@ -2152,7 +2266,8 @@ fn main() -> Result<()> {
|
||||
allow_delete,
|
||||
exclude,
|
||||
quic,
|
||||
} => cmd_serve_all(dir, bind, allow_delete, exclude, quic).await,
|
||||
stdio,
|
||||
} => cmd_serve_all(dir, bind, allow_delete, exclude, quic, stdio).await,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -989,3 +989,104 @@ fn fs_watch_new_file_picked_up() {
|
||||
);
|
||||
assert_files_equal(&src.path().join("new.bin"), &dst.path().join("new.bin"));
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// SSH / stdio transport tests
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Run `clawsync sync <src> <user@localhost:dst_path>` where the SSH target is
|
||||
/// `localhost`. This spawns a real SSH connection, so it requires:
|
||||
/// - SSH server running on localhost
|
||||
/// - Password-less login (key-based auth) for the current user
|
||||
/// - `clawsync` binary on the remote PATH (or the same binary path used here)
|
||||
///
|
||||
/// Guarded by the `CLAWSYNC_TEST_SSH` environment variable; skip when absent.
|
||||
#[test]
|
||||
fn fs_sync_ssh_transport_cold_copy() {
|
||||
if std::env::var("CLAWSYNC_TEST_SSH").is_err() {
|
||||
eprintln!("Skipping fs_sync_ssh_transport_cold_copy (set CLAWSYNC_TEST_SSH=1 to enable)");
|
||||
return;
|
||||
}
|
||||
|
||||
let src = TempDir::new().unwrap();
|
||||
let dst = TempDir::new().unwrap();
|
||||
|
||||
fs::write(src.path().join("alpha.bin"), vec![0x11u8; 4096]).unwrap();
|
||||
fs::write(src.path().join("beta.bin"), vec![0x22u8; 8192]).unwrap();
|
||||
|
||||
// SSH syntax: user@localhost:absolute_path
|
||||
let user = std::env::var("USER").unwrap_or_else(|_| whoami());
|
||||
let remote = format!("{user}@localhost:{}", dst.path().display());
|
||||
|
||||
let output = Command::new(BIN)
|
||||
.arg("sync")
|
||||
.arg(src.path())
|
||||
.arg(&remote)
|
||||
.output()
|
||||
.expect("failed to spawn clawsync sync (SSH)");
|
||||
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"SSH sync failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
assert_dir_equal(src.path(), dst.path());
|
||||
}
|
||||
|
||||
fn whoami() -> String {
|
||||
std::process::Command::new("whoami")
|
||||
.output()
|
||||
.ok()
|
||||
.and_then(|o| String::from_utf8(o.stdout).ok())
|
||||
.map(|s| s.trim().to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string())
|
||||
}
|
||||
|
||||
/// Directly exercise the `--stdio` protocol path without SSH:
|
||||
/// spawn `clawsync serve-fs <dst> --stdio` with piped stdin/stdout, then run
|
||||
/// `clawsync sync <src> localhost:9999/` against a real TCP server BUT
|
||||
/// instead use the simpler approach: verify `serve-fs --stdio` exits cleanly
|
||||
/// when the client closes the connection (smoke test).
|
||||
///
|
||||
/// This guards against regressions where `--stdio` panics or hangs on startup.
|
||||
#[test]
|
||||
fn fs_sync_stdio_mode_cold_copy() {
|
||||
let src = TempDir::new().unwrap();
|
||||
let dst = TempDir::new().unwrap();
|
||||
|
||||
fs::write(src.path().join("test.bin"), vec![0xABu8; 2048]).unwrap();
|
||||
|
||||
// Start serve-fs in stdio mode. Because `clawsync sync` spawns the server
|
||||
// through SSH in real use, we replicate that here by starting the server
|
||||
// as a child process with piped stdio AND running the client pointing at
|
||||
// the server's address via a short-circuit: we use the SSH syntax which
|
||||
// will call `ssh localhost clawsync serve-fs ... --stdio`.
|
||||
//
|
||||
// Instead, directly test that `clawsync serve-fs --stdio` starts cleanly
|
||||
// and processes one session by using `clawsync sync` with SSH syntax to
|
||||
// localhost (same as above). Since that requires SSH, just verify the
|
||||
// subprocess doesn't panic on startup by sending a graceful close.
|
||||
let mut child = Command::new(BIN)
|
||||
.arg("serve-fs")
|
||||
.arg(dst.path())
|
||||
.arg("--stdio")
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit())
|
||||
.spawn()
|
||||
.expect("failed to spawn serve-fs --stdio");
|
||||
|
||||
// Drop stdin immediately — the server should detect EOF and exit cleanly
|
||||
// rather than panicking.
|
||||
drop(child.stdin.take());
|
||||
drop(child.stdout.take());
|
||||
|
||||
let status = child.wait().expect("serve-fs --stdio did not exit");
|
||||
// Exit with an error is OK here (EOF on protocol = transport error),
|
||||
// but it must NOT panic (exit code 101 on Rust panic).
|
||||
assert_ne!(
|
||||
status.code(),
|
||||
Some(101),
|
||||
"serve-fs --stdio must not panic on EOF"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
//! Generic length-prefixed `SyncMessage` framing over any `AsyncRead`/`AsyncWrite`.
|
||||
//!
|
||||
//! The wire format is identical to [`crate::tcp`]:
|
||||
//! ```text
|
||||
//! [4 bytes LE length][body bytes]
|
||||
//! ```
|
||||
//!
|
||||
//! Using trait objects (`Box<dyn AsyncRead + …>`) lets both child-process stdio
|
||||
//! and terminal stdin/stdout share a single concrete type, which makes
|
||||
//! `SyncPeer::Stream` work for both SSH-spawned servers and `--stdio` mode.
|
||||
|
||||
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
|
||||
|
||||
use crate::error::TransportError;
|
||||
use crate::protocol::{MAX_FRAME_SIZE, SyncMessage};
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// FramedReader / FramedWriter
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// The read half: deserialises length-prefixed `SyncMessage` frames.
|
||||
pub struct FramedReader<R: AsyncRead + Unpin + Send> {
|
||||
inner: R,
|
||||
}
|
||||
|
||||
impl<R: AsyncRead + Unpin + Send> FramedReader<R> {
|
||||
pub fn new(inner: R) -> Self {
|
||||
Self { inner }
|
||||
}
|
||||
|
||||
pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
|
||||
let mut len_buf = [0u8; 4];
|
||||
self.inner.read_exact(&mut len_buf).await?;
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
||||
let mut body = vec![0u8; len];
|
||||
self.inner.read_exact(&mut body).await?;
|
||||
SyncMessage::from_bytes(&body).map_err(TransportError::Deserialization)
|
||||
}
|
||||
}
|
||||
|
||||
/// The write half: serialises `SyncMessage` frames with a 4-byte LE length prefix.
|
||||
pub struct FramedWriter<W: AsyncWrite + Unpin + Send> {
|
||||
inner: W,
|
||||
}
|
||||
|
||||
impl<W: AsyncWrite + Unpin + Send> FramedWriter<W> {
|
||||
pub fn new(inner: W) -> Self {
|
||||
Self { inner }
|
||||
}
|
||||
|
||||
pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
|
||||
let body = msg.to_bytes().map_err(TransportError::Serialization)?;
|
||||
let len = body.len() as u32;
|
||||
self.inner.write_all(&len.to_le_bytes()).await?;
|
||||
self.inner.write_all(&body).await?;
|
||||
self.inner.flush().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn shutdown(&mut self) -> Result<(), TransportError> {
|
||||
self.inner.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// StreamPeer — boxed I/O peer (child-process pipes or stdin/stdout)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type BoxReader = Box<dyn AsyncRead + Unpin + Send>;
|
||||
type BoxWriter = Box<dyn AsyncWrite + Unpin + Send>;
|
||||
|
||||
/// A `SyncMessage` peer backed by arbitrary boxed `AsyncRead` + `AsyncWrite`.
|
||||
///
|
||||
/// Used for:
|
||||
/// * **SSH mode** — child-process `ChildStdout` / `ChildStdin`
|
||||
/// * **`--stdio` mode** — `tokio::io::stdin()` / `tokio::io::stdout()`
|
||||
pub struct StreamPeer {
|
||||
reader: FramedReader<BoxReader>,
|
||||
writer: FramedWriter<BoxWriter>,
|
||||
/// Keeps a spawned child process alive for the lifetime of the peer.
|
||||
_child: Option<tokio::process::Child>,
|
||||
}
|
||||
|
||||
impl StreamPeer {
|
||||
/// Wrap a spawned child process (takes ownership, keeps it alive).
|
||||
pub fn from_child(mut child: tokio::process::Child) -> Self {
|
||||
let stdout: BoxReader = Box::new(
|
||||
child
|
||||
.stdout
|
||||
.take()
|
||||
.expect("child must have piped stdout"),
|
||||
);
|
||||
let stdin: BoxWriter = Box::new(
|
||||
child
|
||||
.stdin
|
||||
.take()
|
||||
.expect("child must have piped stdin"),
|
||||
);
|
||||
Self {
|
||||
reader: FramedReader::new(stdout),
|
||||
writer: FramedWriter::new(stdin),
|
||||
_child: Some(child),
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrap this process's own stdin/stdout (for `--stdio` server mode).
|
||||
pub fn from_stdio() -> Self {
|
||||
Self {
|
||||
reader: FramedReader::new(Box::new(tokio::io::stdin())),
|
||||
writer: FramedWriter::new(Box::new(tokio::io::stdout())),
|
||||
_child: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
|
||||
self.writer.send(msg).await
|
||||
}
|
||||
|
||||
pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
|
||||
self.reader.recv().await
|
||||
}
|
||||
|
||||
pub async fn shutdown(&mut self) -> Result<(), TransportError> {
|
||||
self.writer.shutdown().await
|
||||
}
|
||||
|
||||
/// Split into independent halves for pipelined I/O.
|
||||
pub fn into_split(self) -> (StreamReadHalf, StreamWriteHalf) {
|
||||
(StreamReadHalf(self.reader), StreamWriteHalf(self.writer))
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Split halves
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
pub struct StreamReadHalf(pub(crate) FramedReader<BoxReader>);
|
||||
|
||||
impl StreamReadHalf {
|
||||
pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
|
||||
self.0.recv().await
|
||||
}
|
||||
}
|
||||
|
||||
pub struct StreamWriteHalf(pub(crate) FramedWriter<BoxWriter>);
|
||||
|
||||
impl StreamWriteHalf {
|
||||
pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
|
||||
self.0.send(msg).await
|
||||
}
|
||||
|
||||
pub async fn shutdown(&mut self) -> Result<(), TransportError> {
|
||||
self.0.shutdown().await
|
||||
}
|
||||
}
|
||||
@@ -1,14 +1,16 @@
|
||||
//! ClawSync transport backends: TCP, QUIC, and mmap.
|
||||
//! ClawSync transport backends: TCP, QUIC, mmap, and stream (stdio / process).
|
||||
//!
|
||||
//! - [`protocol`]: `SyncMessage` wire protocol + length-prefixed framing
|
||||
//! - [`tcp`]: Async TCP transport (tokio)
|
||||
//! - [`quic`]: QUIC transport (quinn 0.11, TLS 1.3)
|
||||
//! - [`mmap`]: Memory-mapped local transport — zero-copy same-node IPC
|
||||
//! - [`framed`]: Generic length-prefixed framing over any `AsyncRead`/`AsyncWrite`
|
||||
//! - [`error`] — [`TransportError`]
|
||||
|
||||
#![deny(unsafe_code)]
|
||||
|
||||
pub mod error;
|
||||
pub mod framed;
|
||||
pub mod mmap;
|
||||
pub mod peer;
|
||||
pub mod protocol;
|
||||
@@ -16,6 +18,7 @@ pub mod quic;
|
||||
pub mod tcp;
|
||||
|
||||
pub use error::TransportError;
|
||||
pub use framed::StreamPeer;
|
||||
pub use mmap::{DEFAULT_CAPACITY, MmapChannel, MmapReceiver, MmapSender};
|
||||
pub use peer::{PipeReadHalf, PipeWriteHalf, SyncPeer};
|
||||
pub use protocol::SyncMessage;
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::error::TransportError;
|
||||
use crate::framed::{StreamPeer, StreamReadHalf, StreamWriteHalf};
|
||||
use crate::mmap::{MmapReceiver, MmapSender};
|
||||
use crate::protocol::SyncMessage;
|
||||
use crate::quic::QuicConnection;
|
||||
@@ -24,7 +25,7 @@ use crate::tcp::{TcpConnection, TcpReadHalf, TcpWriteHalf};
|
||||
// SyncPeer
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// A unified connection handle for a single sync session (TCP, QUIC, or mmap).
|
||||
/// A unified connection handle for a single sync session (TCP, QUIC, mmap, or stream).
|
||||
pub enum SyncPeer {
|
||||
Tcp(TcpConnection),
|
||||
Quic(Arc<QuicConnection>),
|
||||
@@ -35,6 +36,9 @@ pub enum SyncPeer {
|
||||
send_ch: MmapSender,
|
||||
recv_ch: MmapReceiver,
|
||||
},
|
||||
/// Generic boxed-I/O peer — covers SSH child-process pipes and `--stdio`
|
||||
/// mode (server reads/writes on its own stdin/stdout).
|
||||
Stream(StreamPeer),
|
||||
}
|
||||
|
||||
impl SyncPeer {
|
||||
@@ -45,6 +49,7 @@ impl SyncPeer {
|
||||
SyncPeer::Mmap { send_ch, .. } => {
|
||||
tokio::task::block_in_place(|| send_ch.send(msg))
|
||||
}
|
||||
SyncPeer::Stream(s) => s.send(msg).await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,6 +60,7 @@ impl SyncPeer {
|
||||
SyncPeer::Mmap { recv_ch, .. } => {
|
||||
tokio::task::block_in_place(|| recv_ch.recv())
|
||||
}
|
||||
SyncPeer::Stream(s) => s.recv().await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -66,9 +72,15 @@ impl SyncPeer {
|
||||
Ok(())
|
||||
}
|
||||
SyncPeer::Mmap { .. } => Ok(()),
|
||||
SyncPeer::Stream(mut s) => s.shutdown().await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the inner `StreamPeer` if this is a `Stream` variant, otherwise `None`.
|
||||
pub fn as_stream_peer(&self) -> Option<&StreamPeer> {
|
||||
if let SyncPeer::Stream(s) = self { Some(s) } else { None }
|
||||
}
|
||||
|
||||
/// Return a cloned `Arc` if this is a QUIC connection, otherwise `None`.
|
||||
///
|
||||
/// This is used by servers to keep the QUIC connection alive after the
|
||||
@@ -97,6 +109,10 @@ impl SyncPeer {
|
||||
SyncPeer::Mmap { send_ch, recv_ch } => {
|
||||
(PipeReadHalf::Mmap(recv_ch), PipeWriteHalf::Mmap(send_ch))
|
||||
}
|
||||
SyncPeer::Stream(s) => {
|
||||
let (r, w) = s.into_split();
|
||||
(PipeReadHalf::Stream(r), PipeWriteHalf::Stream(w))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -110,6 +126,7 @@ pub enum PipeWriteHalf {
|
||||
Tcp(TcpWriteHalf),
|
||||
Quic(Arc<QuicConnection>),
|
||||
Mmap(MmapSender),
|
||||
Stream(StreamWriteHalf),
|
||||
}
|
||||
|
||||
impl PipeWriteHalf {
|
||||
@@ -118,6 +135,7 @@ impl PipeWriteHalf {
|
||||
PipeWriteHalf::Tcp(h) => h.send(msg).await,
|
||||
PipeWriteHalf::Quic(c) => c.send(msg).await,
|
||||
PipeWriteHalf::Mmap(tx) => tokio::task::block_in_place(|| tx.send(msg)),
|
||||
PipeWriteHalf::Stream(w) => w.send(msg).await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -129,6 +147,7 @@ impl PipeWriteHalf {
|
||||
Ok(())
|
||||
}
|
||||
PipeWriteHalf::Mmap(_) => Ok(()),
|
||||
PipeWriteHalf::Stream(mut w) => w.shutdown().await,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -138,6 +157,7 @@ pub enum PipeReadHalf {
|
||||
Tcp(TcpReadHalf),
|
||||
Quic(Arc<QuicConnection>),
|
||||
Mmap(MmapReceiver),
|
||||
Stream(StreamReadHalf),
|
||||
}
|
||||
|
||||
impl PipeReadHalf {
|
||||
@@ -146,6 +166,7 @@ impl PipeReadHalf {
|
||||
PipeReadHalf::Tcp(h) => h.recv().await,
|
||||
PipeReadHalf::Quic(c) => c.recv().await,
|
||||
PipeReadHalf::Mmap(rx) => tokio::task::block_in_place(|| rx.recv()),
|
||||
PipeReadHalf::Stream(r) => r.recv().await,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user