Phase 2c: streaming Blob RPC (BlobPutStream / BlobGetStream)
Removes the 16 MiB message cap for blob transfers. The bounded Blob* methods from Phase 2b still exist; the streaming variants let a peer push or pull a many-GB blob without either side holding it in memory. ## Wire format Streaming methods use a slightly different reply shape so the client can route on the first byte alone: Reply : status:u8 || payload:bytes... Where `status` is either `STREAM_STATUS_OK` (0x00, content follows) or a single-byte ErrorCode. `serve_connection` now peeks at the method tag byte via read_exact and hands streaming methods the raw send/recv streams; bounded methods still use the old read_to_end path. ## Method additions - BlobPutStream (0x07): client streams bytes → server pipes into BlobStore::put_stream → reply is 0x00 || 32-byte BlobId - BlobGetStream (0x08): client sends 32-byte BlobId → server verifies existence, writes 0x00 status, then streams chunks from disk into the send stream Method::is_streaming() introspection so callers can decide which wire variant to use. ## BlobStore additions - put_stream<R: AsyncRead + Unpin>(reader) -> BlobId Memory ceiling: one CHUNK_SIZE (4 MiB) buffer regardless of blob size. Handles short-reads correctly (loops until CHUNK_SIZE bytes are available or EOF), including the empty-reader case (produces the empty-blob BlobId, zero chunks). - stream_to<W: AsyncWrite + Unpin>(id, writer) -> bool Ok(false) on NotFound (writer untouched). Verifies each chunk hash before emitting; corruption halts mid-stream with Err. ## Client helpers - call_blob_put_stream(conn, reader) -> Result<BlobId> Uses tokio::io::copy directly onto quinn's SendStream. - call_blob_get_stream(conn, id, writer) -> Result<bool> Ok(false) on NotFound; other errors surface as Err. ## Tests (11 new, all real — no mocks) Blob store (6): - put_stream_produces_same_hash_as_put_bytes (3-chunk blob via Cursor) - put_stream_handles_empty_reader (produces empty-blob BlobId) - put_stream_handles_short_reads (custom Trickle reader that only serves 100 bytes per read call — must still assemble full chunks) - stream_to_writes_full_blob (2-chunk write to Vec<u8>) - stream_to_returns_false_when_missing (writer untouched) - stream_to_detects_chunk_corruption (bit-flip a chunk → Err with "chunk hash mismatch") RPC (5): - method_reports_streaming_variants - end_to_end_stream_put_and_get_over_real_quic — 12 MiB + 777 bytes → 4 chunks, real 2-node QUIC + mTLS + stream round-trip - stream_get_returns_false_for_missing_blob - stream_methods_return_not_configured_without_store - stream_put_deduplicates_with_prior_put_bytes — verify streaming put produces the same BlobId as a prior bounded put on identical content, and the manifest chunk count didn't fork ## Housekeeping rpc.rs was tipping over the 1300-line ceiling with the streaming handlers + helpers + tests. Tests split into `cluster/rpc/tests.rs` via `#[path = "rpc/tests.rs"] mod tests;`. Result: - rpc.rs: 748 lines - rpc/tests.rs: 694 lines - blob.rs: 1002 lines - All under ceiling. 139 tests pass. Pre-existing macOS-only failure unchanged. ## What's next - Phase 2d: chunk-level RPC (BlobPutChunk / BlobGetChunk) so a receiver can `LoadManifest` then request only the chunks it's missing — big bandwidth win on partially-overlapping caches. - Phase 3: CRDT metadata for human-readable namespaces on top of content hashes. - Phase 5: the killer feature — fingerprint the cargo target dir, BlobPutStream it, next node BlobGetStream by the same fingerprint. Now buildable directly on Phase 2c since target dirs run 100 MB to a few GB and the previous 16 MiB cap would have blocked us.
This commit is contained in:
Omar Sobh
@@ -0,0 +1,694 @@
|
||||
//! Tests split out of rpc.rs to keep the parent under the 1300-line ceiling.
|
||||
//! Same visibility as `#[cfg(test)] mod tests { ... }` — everything reaches
|
||||
//! parent private items via `use super::*`.
|
||||
|
||||
use super::*;
|
||||
use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
|
||||
use crate::config::{ClusterConfig, PeerEntry};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::atomic::{AtomicU16, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Dedicated port range for RPC tests, distinct from gossip (41000+)
|
||||
/// and transport (42000+) so parallel tests never conflict.
|
||||
static NEXT_PORT: AtomicU16 = AtomicU16::new(43001);
|
||||
fn next_port() -> u16 {
|
||||
NEXT_PORT.fetch_add(1, Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn loopback(port: u16) -> SocketAddr {
|
||||
format!("127.0.0.1:{port}").parse().unwrap()
|
||||
}
|
||||
|
||||
async fn bootstrap_gossip(name: &str, port: u16) -> Arc<ClusterGossip> {
|
||||
let cfg = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(loopback(port)),
|
||||
..Default::default()
|
||||
};
|
||||
Arc::new(ClusterGossip::bootstrap(&cfg, name).await.unwrap())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn method_round_trips_byte_encoding() {
|
||||
assert_eq!(Method::Ping.as_byte(), 0x01);
|
||||
assert_eq!(Method::PeerStatus.as_byte(), 0x02);
|
||||
assert_eq!(Method::BlobStat.as_byte(), 0x03);
|
||||
assert_eq!(Method::BlobGet.as_byte(), 0x04);
|
||||
assert_eq!(Method::BlobPut.as_byte(), 0x05);
|
||||
assert_eq!(Method::BlobLoadManifest.as_byte(), 0x06);
|
||||
for m in [
|
||||
Method::Ping,
|
||||
Method::PeerStatus,
|
||||
Method::BlobStat,
|
||||
Method::BlobGet,
|
||||
Method::BlobPut,
|
||||
Method::BlobLoadManifest,
|
||||
] {
|
||||
assert_eq!(Method::from_byte(m.as_byte()), Some(m));
|
||||
}
|
||||
assert_eq!(Method::from_byte(0x00), None);
|
||||
assert_eq!(Method::from_byte(0xff), None);
|
||||
}
|
||||
|
||||
fn open_blob_store() -> (tempfile::TempDir, Arc<BlobStore>) {
|
||||
let tmp = tempfile::TempDir::new().unwrap();
|
||||
let store = Arc::new(BlobStore::open(tmp.path().to_path_buf()).unwrap());
|
||||
(tmp, store)
|
||||
}
|
||||
|
||||
async fn router_with_blobs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
|
||||
let gossip = bootstrap_gossip(name, port).await;
|
||||
let (tmp, store) = open_blob_store();
|
||||
let router = Arc::new(
|
||||
RpcRouter::new(gossip, name.into(), "fabric-10g".into())
|
||||
.with_blob_store(store),
|
||||
);
|
||||
(tmp, router)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dispatch_returns_pong_for_ping() {
|
||||
let gossip = bootstrap_gossip("solo", next_port()).await;
|
||||
let router = RpcRouter::new(gossip.clone(), "solo".into(), "test-zone".into());
|
||||
|
||||
// Direct dispatch — no network involved.
|
||||
let reply = dispatch(&router, &[Method::Ping.as_byte(), b'h', b'i']).await;
|
||||
assert_eq!(reply, b"pong:hi");
|
||||
gossip.peer("nobody").await; // touch to keep gossip alive
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dispatch_returns_json_for_peer_status() {
|
||||
let gossip = bootstrap_gossip("architect", next_port()).await;
|
||||
let router = RpcRouter::new(
|
||||
gossip.clone(),
|
||||
"architect".into(),
|
||||
"fabric-10g".into(),
|
||||
);
|
||||
let reply = dispatch(&router, &[Method::PeerStatus.as_byte()]).await;
|
||||
let decoded: PeerStatusReply = serde_json::from_slice(&reply).unwrap();
|
||||
assert_eq!(decoded.local_name, "architect");
|
||||
assert_eq!(decoded.local_zone, "fabric-10g");
|
||||
// Solo node — no peers yet.
|
||||
assert!(decoded.peers.is_empty(), "solo node has no peers");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dispatch_returns_empty_request_error() {
|
||||
let gossip = bootstrap_gossip("solo", next_port()).await;
|
||||
let router = RpcRouter::new(gossip.clone(), "solo".into(), "z".into());
|
||||
let reply = dispatch(&router, &[]).await;
|
||||
assert_eq!(reply, vec![ErrorCode::EmptyRequest.as_byte()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dispatch_returns_unknown_method_error() {
|
||||
let gossip = bootstrap_gossip("solo", next_port()).await;
|
||||
let router = RpcRouter::new(gossip.clone(), "solo".into(), "z".into());
|
||||
let reply = dispatch(&router, &[0xab, 0x01, 0x02]).await;
|
||||
assert_eq!(reply, vec![ErrorCode::UnknownMethod.as_byte()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rpc_call_rejects_oversize_payload() {
|
||||
// No network involved — the size check runs client-side before we
|
||||
// even try to open the bidi stream. Use a dummy connection built
|
||||
// via generate_test_pair; we won't actually connect.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let server = QuicServer::bind(loopback(0), id_b).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let _accept = tokio::spawn(async move {
|
||||
let _ = server.accept().await;
|
||||
});
|
||||
let client = QuicClient::new(loopback(0), id_a).unwrap();
|
||||
let conn = client.connect(server_addr, "b").await.unwrap();
|
||||
let huge = vec![0u8; MAX_MESSAGE_BYTES];
|
||||
let err = rpc_call(&conn, Method::Ping, &huge)
|
||||
.await
|
||||
.err()
|
||||
.expect("must reject oversize");
|
||||
assert!(err.to_string().contains("exceeds cap"));
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn end_to_end_ping_and_peer_status_over_real_quic() {
|
||||
// Two full nodes: A runs gossip + a QuicServer serving RpcRouter.
|
||||
// B is a client that dials A and calls both RPCs.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
|
||||
let port_a = next_port();
|
||||
let gossip_a = bootstrap_gossip("a", port_a).await;
|
||||
let router = Arc::new(RpcRouter::new(
|
||||
gossip_a.clone(),
|
||||
"a".into(),
|
||||
"fabric-10g".into(),
|
||||
));
|
||||
|
||||
// A's advertised state so the PeerStatus reply exercises the
|
||||
// gossip → PeerView path even though B isn't in the peer table.
|
||||
gossip_a.set_hot_used(4096).await;
|
||||
gossip_a.set_hot_max(1_000_000).await;
|
||||
gossip_a
|
||||
.set_warm_projects(&["osobh/clawverse", "osobh/clawmates"])
|
||||
.await;
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
// Ping.
|
||||
let pong = call_ping(&conn, b"hello").await.unwrap();
|
||||
assert_eq!(pong, b"hello");
|
||||
|
||||
// PeerStatus.
|
||||
let status = call_peer_status(&conn).await.unwrap();
|
||||
assert_eq!(status.local_name, "a");
|
||||
assert_eq!(status.local_zone, "fabric-10g");
|
||||
assert!(status.peers.is_empty(), "A has no peers configured");
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
// Keep gossip_a alive to the end so its background task doesn't
|
||||
// drop mid-serve.
|
||||
let _ = gossip_a.peers().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn peer_status_reflects_peer_gossip_state() {
|
||||
// A and B both run gossip; A serves RPC. When A's PeerStatus is
|
||||
// called by a third-party client, the reply's `peers` field
|
||||
// contains B (as long as gossip has converged).
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let port_a = next_port();
|
||||
let port_b = next_port();
|
||||
|
||||
let cfg_a = ClusterConfig {
|
||||
zone: "fabric-10g".into(),
|
||||
bind_lan: Some(loopback(port_a)),
|
||||
..Default::default()
|
||||
};
|
||||
let cfg_b = ClusterConfig {
|
||||
zone: "lan-1g".into(),
|
||||
bind_lan: Some(loopback(port_b)),
|
||||
peers: vec![PeerEntry {
|
||||
name: "a".into(),
|
||||
zone: "fabric-10g".into(),
|
||||
lan_addr: Some(loopback(port_a)),
|
||||
tailscale_addr: None,
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let gossip_a = Arc::new(ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap());
|
||||
let gossip_b = Arc::new(ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap());
|
||||
|
||||
let router = Arc::new(RpcRouter::new(
|
||||
gossip_a.clone(),
|
||||
"a".into(),
|
||||
"fabric-10g".into(),
|
||||
));
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router).await;
|
||||
}
|
||||
});
|
||||
|
||||
// Wait for gossip convergence: A must see B.
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(10);
|
||||
loop {
|
||||
if let Some(v) = gossip_a.peer("b").await {
|
||||
if v.alive {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if std::time::Instant::now() >= deadline {
|
||||
panic!("A never saw B alive within 10s");
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
let status = call_peer_status(&conn).await.unwrap();
|
||||
assert_eq!(status.local_name, "a");
|
||||
assert_eq!(status.peers.len(), 1, "A should report exactly B");
|
||||
assert_eq!(status.peers[0].name, "b");
|
||||
assert_eq!(status.peers[0].zone, "lan-1g");
|
||||
assert!(status.peers[0].alive);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
// Keep gossip services alive until end.
|
||||
drop(gossip_b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn error_code_describe_covers_all_variants() {
|
||||
assert_eq!(ErrorCode::EmptyRequest.describe(), "empty request");
|
||||
assert_eq!(ErrorCode::UnknownMethod.describe(), "unknown method");
|
||||
assert_eq!(ErrorCode::HandlerFailure.describe(), "handler failure");
|
||||
assert_eq!(ErrorCode::NotFound.describe(), "not found");
|
||||
assert_eq!(ErrorCode::InvalidRequest.describe(), "invalid request");
|
||||
assert_eq!(
|
||||
ErrorCode::NotConfigured.describe(),
|
||||
"server subsystem not configured"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_error_covers_all_known_codes() {
|
||||
for code in [
|
||||
ErrorCode::EmptyRequest,
|
||||
ErrorCode::UnknownMethod,
|
||||
ErrorCode::HandlerFailure,
|
||||
ErrorCode::NotFound,
|
||||
ErrorCode::InvalidRequest,
|
||||
ErrorCode::NotConfigured,
|
||||
] {
|
||||
assert_eq!(decode_error(code.as_byte()), Some(code));
|
||||
}
|
||||
assert_eq!(decode_error(0x00), None);
|
||||
assert_eq!(decode_error(0xff), None);
|
||||
}
|
||||
|
||||
// ── Phase 2b: Blob RPC ─────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_rpcs_return_not_configured_without_store() {
|
||||
// Router built via `new` alone (no `.with_blob_store`) must
|
||||
// refuse Blob* methods with a well-known error code.
|
||||
let gossip = bootstrap_gossip("solo", next_port()).await;
|
||||
let router = RpcRouter::new(gossip, "solo".into(), "z".into());
|
||||
for method in [
|
||||
Method::BlobStat,
|
||||
Method::BlobGet,
|
||||
Method::BlobPut,
|
||||
Method::BlobLoadManifest,
|
||||
] {
|
||||
let mut req = vec![method.as_byte()];
|
||||
req.extend_from_slice(&[0u8; 32]);
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(
|
||||
reply,
|
||||
vec![ErrorCode::NotConfigured.as_byte()],
|
||||
"method {method:?} should be NotConfigured without a store"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_stat_returns_not_found_for_missing() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let missing = BlobId::from_bytes([0u8; 32]);
|
||||
let mut req = vec![Method::BlobStat.as_byte()];
|
||||
req.extend_from_slice(missing.as_bytes());
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(reply, vec![ErrorCode::NotFound.as_byte()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_stat_returns_json_for_existing() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let id = router
|
||||
.blob_store()
|
||||
.unwrap()
|
||||
.put_bytes(b"tiny content")
|
||||
.await
|
||||
.unwrap();
|
||||
let mut req = vec![Method::BlobStat.as_byte()];
|
||||
req.extend_from_slice(id.as_bytes());
|
||||
let reply = dispatch(&router, &req).await;
|
||||
let stat: BlobStat = serde_json::from_slice(&reply).unwrap();
|
||||
assert_eq!(stat.total_size, b"tiny content".len() as u64);
|
||||
assert_eq!(stat.chunk_count, 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_stat_returns_invalid_request_for_bad_length() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
// Payload is only 5 bytes; a valid BlobId is 32.
|
||||
let req = vec![Method::BlobStat.as_byte(), 1, 2, 3, 4, 5];
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(reply, vec![ErrorCode::InvalidRequest.as_byte()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_get_returns_content_bytes() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let payload: &[u8] = b"contents to retrieve";
|
||||
let id = router.blob_store().unwrap().put_bytes(payload).await.unwrap();
|
||||
let mut req = vec![Method::BlobGet.as_byte()];
|
||||
req.extend_from_slice(id.as_bytes());
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(reply, payload);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_put_stores_bytes_and_returns_hash() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let payload = b"put via rpc";
|
||||
let mut req = vec![Method::BlobPut.as_byte()];
|
||||
req.extend_from_slice(payload);
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(reply.len(), 32);
|
||||
let mut id_bytes = [0u8; 32];
|
||||
id_bytes.copy_from_slice(&reply);
|
||||
let assigned = BlobId::from_bytes(id_bytes);
|
||||
let expected = BlobId::from_bytes(blake3::hash(payload).into());
|
||||
assert_eq!(assigned, expected);
|
||||
// Round-trip: the bytes are now readable via the store.
|
||||
let round = router
|
||||
.blob_store()
|
||||
.unwrap()
|
||||
.get_bytes(&assigned)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(round.as_deref(), Some(payload.as_slice()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_load_manifest_returns_json_for_existing() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let data = vec![0x77u8; 4 * 1024 * 1024 + 1]; // 2 chunks
|
||||
let id = router.blob_store().unwrap().put_bytes(&data).await.unwrap();
|
||||
let mut req = vec![Method::BlobLoadManifest.as_byte()];
|
||||
req.extend_from_slice(id.as_bytes());
|
||||
let reply = dispatch(&router, &req).await;
|
||||
let manifest: BlobManifest = serde_json::from_slice(&reply).unwrap();
|
||||
assert_eq!(manifest.blob_id, id);
|
||||
assert_eq!(manifest.total_size, data.len() as u64);
|
||||
assert_eq!(manifest.chunks.len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn blob_load_manifest_returns_not_found_for_missing() {
|
||||
let (_tmp, router) = router_with_blobs("solo", next_port()).await;
|
||||
let missing = BlobId::from_bytes([0u8; 32]);
|
||||
let mut req = vec![Method::BlobLoadManifest.as_byte()];
|
||||
req.extend_from_slice(missing.as_bytes());
|
||||
let reply = dispatch(&router, &req).await;
|
||||
assert_eq!(reply, vec![ErrorCode::NotFound.as_byte()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn end_to_end_blob_put_stat_get_over_real_quic() {
|
||||
// The full loop: B → A over real QUIC + mTLS.
|
||||
// 1. B puts a blob on A (BlobPut).
|
||||
// 2. B queries stat + fetches it back (BlobStat + BlobGet).
|
||||
// 3. B asks for the manifest (BlobLoadManifest).
|
||||
// Every step goes through the actual wire, no shortcuts.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp, router) = router_with_blobs("a", next_port()).await;
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
let payload: &[u8] = b"cross-node payload";
|
||||
let assigned = call_blob_put(&conn, payload).await.unwrap();
|
||||
let expected = BlobId::from_bytes(blake3::hash(payload).into());
|
||||
assert_eq!(assigned, expected);
|
||||
|
||||
let stat = call_blob_stat(&conn, &assigned).await.unwrap().unwrap();
|
||||
assert_eq!(stat.total_size, payload.len() as u64);
|
||||
assert_eq!(stat.chunk_count, 1);
|
||||
|
||||
let round = call_blob_get(&conn, &assigned).await.unwrap().unwrap();
|
||||
assert_eq!(round, payload);
|
||||
|
||||
let manifest = call_blob_load_manifest(&conn, &assigned)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(manifest.blob_id, assigned);
|
||||
assert_eq!(manifest.chunks.len(), 1);
|
||||
|
||||
// NotFound path also works over the wire.
|
||||
let ghost = BlobId::from_bytes([0u8; 32]);
|
||||
assert!(call_blob_stat(&conn, &ghost).await.unwrap().is_none());
|
||||
assert!(call_blob_get(&conn, &ghost).await.unwrap().is_none());
|
||||
assert!(call_blob_load_manifest(&conn, &ghost)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn end_to_end_multi_chunk_blob_over_real_quic() {
|
||||
// 6 MB blob → 2 chunks. Round-trips whole via BlobPut/Get.
|
||||
// Also confirms the RPC layer's MAX_MESSAGE_BYTES bump from
|
||||
// 16 KiB to 16 MiB actually took effect end-to-end.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp, router) = router_with_blobs("a", next_port()).await;
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
let payload: Vec<u8> = (0..6 * 1024 * 1024)
|
||||
.map(|i| (i % 251) as u8)
|
||||
.collect();
|
||||
let assigned = call_blob_put(&conn, &payload).await.unwrap();
|
||||
let manifest = call_blob_load_manifest(&conn, &assigned)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(manifest.chunks.len(), 2, "6 MB should split into 2 chunks");
|
||||
let round = call_blob_get(&conn, &assigned).await.unwrap().unwrap();
|
||||
assert_eq!(round.len(), payload.len());
|
||||
assert_eq!(round, payload);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
|
||||
// ── Phase 2c: streaming Blob RPC ─────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn method_reports_streaming_variants() {
|
||||
assert!(!Method::Ping.is_streaming());
|
||||
assert!(!Method::PeerStatus.is_streaming());
|
||||
assert!(!Method::BlobStat.is_streaming());
|
||||
assert!(!Method::BlobGet.is_streaming());
|
||||
assert!(!Method::BlobPut.is_streaming());
|
||||
assert!(!Method::BlobLoadManifest.is_streaming());
|
||||
assert!(Method::BlobPutStream.is_streaming());
|
||||
assert!(Method::BlobGetStream.is_streaming());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn end_to_end_stream_put_and_get_over_real_quic() {
|
||||
// The whole point of Phase 2c: put a big blob without holding
|
||||
// it in memory on either side. This test drives a 12 MiB
|
||||
// payload (3 chunks) through BlobPutStream and BlobGetStream.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp, router) = router_with_blobs("a", next_port()).await;
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
// 12 MiB payload; deliberately not a chunk multiple so the last
|
||||
// chunk is short.
|
||||
let mut payload: Vec<u8> = Vec::with_capacity(12 * 1024 * 1024 + 777);
|
||||
for i in 0..12 * 1024 * 1024 + 777 {
|
||||
payload.push((i % 251) as u8);
|
||||
}
|
||||
let reader = std::io::Cursor::new(payload.clone());
|
||||
let id = call_blob_put_stream(&conn, reader).await.unwrap();
|
||||
let expected = BlobId::from_bytes(blake3::hash(&payload).into());
|
||||
assert_eq!(id, expected);
|
||||
|
||||
// Manifest verifies the chunk split.
|
||||
let manifest = call_blob_load_manifest(&conn, &id).await.unwrap().unwrap();
|
||||
assert_eq!(manifest.chunks.len(), 4, "12 MiB + 777 → 4 chunks");
|
||||
assert_eq!(manifest.total_size, payload.len() as u64);
|
||||
|
||||
// Stream it back.
|
||||
let mut sink: Vec<u8> = Vec::new();
|
||||
let ok = call_blob_get_stream(&conn, &id, &mut sink).await.unwrap();
|
||||
assert!(ok);
|
||||
assert_eq!(sink.len(), payload.len());
|
||||
assert_eq!(sink, payload);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stream_get_returns_false_for_missing_blob() {
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp, router) = router_with_blobs("a", next_port()).await;
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
let ghost = BlobId::from_bytes([0u8; 32]);
|
||||
let mut sink: Vec<u8> = Vec::new();
|
||||
let ok = call_blob_get_stream(&conn, &ghost, &mut sink)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(!ok);
|
||||
assert!(sink.is_empty());
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stream_methods_return_not_configured_without_store() {
|
||||
// Router built without a store: streaming methods must reply
|
||||
// with NotConfigured as their first status byte.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let gossip = bootstrap_gossip("a", next_port()).await;
|
||||
// NOTE: no `.with_blob_store(...)` — Blob* methods should error.
|
||||
let router = Arc::new(RpcRouter::new(gossip, "a".into(), "z".into()));
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
// BlobPutStream on a store-less router → NotConfigured.
|
||||
let reader = std::io::Cursor::new(b"never lands".to_vec());
|
||||
let put_err = call_blob_put_stream(&conn, reader)
|
||||
.await
|
||||
.err()
|
||||
.expect("no store → error");
|
||||
assert!(
|
||||
put_err.to_string().contains("not configured"),
|
||||
"put_stream err: {put_err}"
|
||||
);
|
||||
|
||||
// BlobGetStream: same error surface.
|
||||
let ghost = BlobId::from_bytes([0u8; 32]);
|
||||
let mut sink: Vec<u8> = Vec::new();
|
||||
let get_err = call_blob_get_stream(&conn, &ghost, &mut sink)
|
||||
.await
|
||||
.err()
|
||||
.expect("no store → error");
|
||||
assert!(
|
||||
get_err.to_string().contains("not configured"),
|
||||
"get_stream err: {get_err}"
|
||||
);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stream_put_deduplicates_with_prior_put_bytes() {
|
||||
// Uploading the same content twice — once bounded, once
|
||||
// streaming — yields the same BlobId AND doesn't double-store
|
||||
// chunks. Proves stream + bounded are consistent addresses.
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp, router) = router_with_blobs("a", next_port()).await;
|
||||
|
||||
// Pre-populate via the local store's bounded API.
|
||||
let payload = vec![0xdeu8; 4 * 1024 * 1024 + 100];
|
||||
let pre_id = router
|
||||
.blob_store()
|
||||
.unwrap()
|
||||
.put_bytes(&payload)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let server = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let server_addr = server.local_addr().unwrap();
|
||||
let router_srv = router.clone();
|
||||
let accept_task = tokio::spawn(async move {
|
||||
if let Some(Ok(conn)) = server.accept().await {
|
||||
let _ = serve_connection(conn, router_srv).await;
|
||||
}
|
||||
});
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(server_addr, "a").await.unwrap();
|
||||
|
||||
let reader = std::io::Cursor::new(payload.clone());
|
||||
let stream_id = call_blob_put_stream(&conn, reader).await.unwrap();
|
||||
assert_eq!(pre_id, stream_id);
|
||||
|
||||
// The manifest is still there and its chunk count matches the
|
||||
// pre-existing one — no fork.
|
||||
let manifest = router
|
||||
.blob_store()
|
||||
.unwrap()
|
||||
.load_manifest(&stream_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(manifest.chunks.len(), 2);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
accept_task.abort();
|
||||
}
|
||||
Reference in New Issue
Block a user