clawverse/clawstor
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 5f: claw-cargo prewarm — cross-peer cache copy

Browse Source 
The last piece before "Gitea webhook triggers a cache-warm for the
CI runner before its build starts." Adds a `prewarm` subcommand that
copies a tagged cache from one peer (upstream) to another (downstream)
in one shot — same tag, same BlobId, both sides serve it after.

## New subcommand

```
claw-cargo prewarm \
  --from-peer tank --from-addr 10.0.0.14:7702 \
  --to-peer   morpheus --to-addr 10.0.0.15:7702 \
  --tls-dir   /etc/claw-store/tls \
  --pin       clawverse:main:latest
```

Flow:
1. Connect to upstream with local mTLS identity
2. `GetTag(tag)` → BlobId; `BlobStat(BlobId)` → size + chunk count
3. `BlobGetStream(BlobId)` → download bytes
4. Connect to downstream (second QUIC endpoint, same identity)
5. `BlobPutStream(bytes)` → returns BlobId; verified equal to upstream's
6. `PutTag(tag → BlobId)` on downstream

Summary output shows tag, blob id, both endpoints, byte count,
download/upload timings, total wall clock.

Assumes upstream + downstream share the same fleet CA (the common
case). Mixed-fleet variant with distinct identities is a follow-on.

## Integrity check

`assigned_id != blob_id` after the downstream upload triggers a
bail — the two BlobIds must match because content is BLAKE3-hashed
end-to-end. If they don't, the wire path corrupted bytes and the
whole prewarm fails loud rather than silently pinning a bad blob.

## Buffered vs streamed

Current implementation buffers the whole blob in memory between
download and upload. Fine for cargo target dirs (~1-5 GB compressed);
would break for a 20 GB blob. A follow-on will pipe upstream → tokio
duplex → downstream to run at fixed memory.

## Tests (1 new, real 2-peer QUIC)

- **`end_to_end_prewarm_copies_tagged_blob_between_two_peers`**
  Two full RpcRouters serving in-process (A upstream + C downstream),
  each on a distinct port. Seeds A with a blob + tag, then runs the
  exact sequence prewarm runs internally: `GetTag → BlobGetStream`
  against A, then `BlobPutStream → PutTag` against C. Verifies that
  C's blob store returns byte-equal payload and C's tag store now
  points at the same BlobId. Proves the composition works.

## Housekeeping

`rpc/tests.rs` hit 1408 lines with the new prewarm test. Phase 5
tests (5b refs + 5d tags + 5e restore + 5f prewarm) split to
`rpc/tests_phase5.rs` via a second `#[path]` module in rpc.rs.
Result:

- rpc/tests.rs: 727 (phase 1-2d tests)
- rpc/tests_phase5.rs: 722 (phase 5 tests)
- rpc.rs: 777
- rpc/client.rs: 589
- claw_cargo.rs: 818
- All under ceiling.

225 tests pass. Pre-existing macOS-only failure unchanged.

## What this enables

The complete CI runner flow now works end-to-end:

```
Primary (e.g. tank):
  claw-cargo build          # first ever build — MISS, uploads
  claw-cargo pin --name clawverse:main:latest

Fleet control plane on PR open:
  gitea webhook → shell hook → claw-cargo prewarm \
    --from-peer tank --to-peer $RUNNER_LOCAL \
    --pin clawverse:main:latest
  # runner's local daemon now serves the tag + blob

Runner picks up job:
  claw-cargo build --peer 127.0.0.1:7702
  # local daemon is warm → prefetch returns HIT
  # cargo build runs against restored deps → workspace-crates only
  # 50 min → 3 min
```

Every subcommand claw-cargo needs for this pipeline now exists:
build / prefetch / prefetch --pin / status / fingerprint /
pin / unpin / list-tags / prewarm.

## What's next

- 5g: cache hit/miss metrics into gossip so placement engines can
  bias runner scheduling toward warm nodes
- 5h: streaming variant of prewarm (tokio duplex) for many-GB blobs
- 6: FUSE mount for warm-tier git worktrees
- 3: full CRDT metadata if the plain-tag model shows conflict problems
This commit is contained in:
Omar Sobh
2026-07-12 03:57:10 -07:00
parent 05ba800d01
commit db55903311
4 changed files with 875 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claw-store/src/cluster/rpc/tests.rs
-576
View File
@@ -725,579 +725,3 @@ async fn end_to_end_push_blob_missing_chunks_replicates_only_needed_bytes() {
accept_task.abort();
}
// ── Phase 5b: reference-store RPC ────────────────────────────────────
async fn router_with_blobs_and_refs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
use crate::cluster::refs::RefStore;
let gossip = bootstrap_gossip(name, port).await;
let tmp = tempfile::TempDir::new().unwrap();
let blob_store =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, name.into(), "fabric-10g".into())
.with_blob_store(blob_store)
.with_ref_store(ref_store),
);
(tmp, router)
}
#[test]
fn phase_5b_method_byte_encoding() {
assert_eq!(Method::GetRef.as_byte(), 0x0d);
assert_eq!(Method::PutRef.as_byte(), 0x0e);
assert_eq!(Method::from_byte(0x0d), Some(Method::GetRef));
assert_eq!(Method::from_byte(0x0e), Some(Method::PutRef));
}
#[tokio::test]
async fn get_ref_returns_not_found_for_missing() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let key = [0u8; 32];
let mut req = vec![Method::GetRef.as_byte()];
req.extend_from_slice(&key);
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn put_ref_stores_and_get_ref_reads_back() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let key = [0x11u8; 32];
let value = [0x22u8; 32];
let mut put = vec![Method::PutRef.as_byte()];
put.extend_from_slice(&key);
put.extend_from_slice(&value);
assert_eq!(dispatch(&router, &put).await, vec![STREAM_STATUS_OK]);
let mut get = vec![Method::GetRef.as_byte()];
get.extend_from_slice(&key);
assert_eq!(dispatch(&router, &get).await, value.to_vec());
}
#[tokio::test]
async fn put_ref_rejects_wrong_length_payload() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
// 63 bytes — one shy of the 32+32 requirement.
let req = {
let mut r = vec![Method::PutRef.as_byte()];
r.extend(vec![0u8; 63]);
r
};
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn get_ref_rejects_wrong_length_payload() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let req = vec![Method::GetRef.as_byte(), 0, 1, 2];
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn ref_rpcs_return_not_configured_without_store() {
let gossip = bootstrap_gossip("solo", next_port()).await;
// Router with a blob store but NO ref store.
let tmp = tempfile::TempDir::new().unwrap();
let blob =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().to_path_buf()).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, "solo".into(), "z".into()).with_blob_store(blob),
);
for method in [Method::GetRef, Method::PutRef] {
let mut req = vec![method.as_byte()];
req.extend_from_slice(&[0u8; 32]);
req.extend_from_slice(&[0u8; 32]);
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotConfigured.as_byte()],
"method {method:?} should be NotConfigured"
);
}
}
#[tokio::test]
async fn end_to_end_put_ref_get_ref_over_real_quic() {
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_blobs_and_refs("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 key = [0x77u8; 32];
let value = [0x88u8; 32];
// Miss first.
assert!(call_get_ref(&conn, &key).await.unwrap().is_none());
// Put.
call_put_ref(&conn, &key, &value).await.unwrap();
// Hit.
assert_eq!(call_get_ref(&conn, &key).await.unwrap(), Some(value));
conn.close(quinn::VarInt::from_u32(0), b"done");
client.shutdown().await;
tokio::time::sleep(Duration::from_millis(50)).await;
accept_task.abort();
}
// ── Phase 5d: tag-store RPC ──────────────────────────────────────────
async fn router_with_full_stack(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
use crate::cluster::refs::RefStore;
use crate::cluster::tags::TagStore;
let gossip = bootstrap_gossip(name, port).await;
let tmp = tempfile::TempDir::new().unwrap();
let blob_store =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
let tag_store = Arc::new(TagStore::open(tmp.path().join("tags-db")).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, name.into(), "fabric-10g".into())
.with_blob_store(blob_store)
.with_ref_store(ref_store)
.with_tag_store(tag_store),
);
(tmp, router)
}
#[test]
fn phase_5d_method_byte_encoding() {
assert_eq!(Method::PutTag.as_byte(), 0x0f);
assert_eq!(Method::GetTag.as_byte(), 0x10);
assert_eq!(Method::DeleteTag.as_byte(), 0x11);
assert_eq!(Method::ListTags.as_byte(), 0x12);
for m in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
assert_eq!(Method::from_byte(m.as_byte()), Some(m));
}
}
#[tokio::test]
async fn tag_rpcs_return_not_configured_without_store() {
let gossip = bootstrap_gossip("solo", next_port()).await;
let router = RpcRouter::new(gossip, "solo".into(), "z".into());
for method in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
let mut req = vec![method.as_byte()];
req.extend_from_slice(b"any-key");
let reply = dispatch(&router, &req).await;
assert_eq!(
reply,
vec![ErrorCode::NotConfigured.as_byte()],
"method {method:?} should be NotConfigured"
);
}
}
#[tokio::test]
async fn put_tag_stores_and_get_tag_reads_back() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let key = "clawverse:main:latest";
let value = [0x77u8; 32];
let put_payload = crate::cluster::tags::encode_record(key, &value);
let mut put_req = vec![Method::PutTag.as_byte()];
put_req.extend_from_slice(&put_payload);
assert_eq!(dispatch(&router, &put_req).await, vec![STREAM_STATUS_OK]);
let mut get_req = vec![Method::GetTag.as_byte()];
get_req.extend_from_slice(key.as_bytes());
assert_eq!(dispatch(&router, &get_req).await, value.to_vec());
}
#[tokio::test]
async fn get_tag_returns_not_found_for_missing() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let mut req = vec![Method::GetTag.as_byte()];
req.extend_from_slice(b"never-set");
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn get_tag_rejects_empty_key() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let req = vec![Method::GetTag.as_byte()]; // empty payload
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn delete_tag_removes_and_returns_not_found_after() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let store = router.tag_store().unwrap().clone();
store.put("removable", &[0u8; 32]).await.unwrap();
let mut req = vec![Method::DeleteTag.as_byte()];
req.extend_from_slice(b"removable");
assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);
// Second delete → NotFound.
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn list_tags_returns_json_sorted() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let store = router.tag_store().unwrap().clone();
store.put("bravo", &[2u8; 32]).await.unwrap();
store.put("alpha", &[1u8; 32]).await.unwrap();
let req = vec![Method::ListTags.as_byte()];
let reply = dispatch(&router, &req).await;
let decoded: Vec<crate::cluster::tags::TagEntry> =
serde_json::from_slice(&reply).unwrap();
assert_eq!(decoded.len(), 2);
assert_eq!(decoded[0].key, "alpha");
assert_eq!(decoded[1].key, "bravo");
assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
}
#[tokio::test]
async fn end_to_end_tag_resolve_and_stream_restore_over_real_quic() {
// Phase 5e: the pipeline `claw-cargo prefetch --pin <tag>` runs
// internally — put a blob, publish a tag pointing at it, then
// GetTag → BlobStat → BlobGetStream to reassemble the content
// byte-equal on the other side.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_full_stack("a", next_port()).await;
// Seed the blob store with a synthetic "captured target dir".
let payload: Vec<u8> = (0..2 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
let blob_id = router
.blob_store()
.unwrap()
.put_bytes(&payload)
.await
.unwrap();
// Publish a tag pointing at that blob.
router
.tag_store()
.unwrap()
.put("clawverse:main:latest", blob_id.as_bytes())
.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();
// Client side of `prefetch --pin <tag>`:
// 1. GetTag → BlobId bytes
// 2. BlobStat → confirm blob exists + size
// 3. BlobGetStream → download into buffer
let tag_value = call_get_tag(&conn, "clawverse:main:latest")
.await
.unwrap()
.expect("tag set");
assert_eq!(tag_value, *blob_id.as_bytes());
let resolved_id = BlobId::from_bytes(tag_value);
let stat = call_blob_stat(&conn, &resolved_id).await.unwrap().unwrap();
assert_eq!(stat.total_size, payload.len() as u64);
let mut sink: Vec<u8> = Vec::new();
let ok = call_blob_get_stream(&conn, &resolved_id, &mut sink)
.await
.unwrap();
assert!(ok);
assert_eq!(sink.len(), payload.len());
assert_eq!(sink, payload, "reassembled bytes match source");
// Missing-tag path: prefetch --pin never-set-name reports None.
let missing = call_get_tag(&conn, "never-set").await.unwrap();
assert!(missing.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_pin_lookup_delete_over_real_quic() {
// Full flow: publish a tag → look it up → list → delete → confirm gone.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_full_stack("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 key = "clawverse:main:latest-cache";
let value = [0xaau8; 32];
// Miss first.
assert!(call_get_tag(&conn, key).await.unwrap().is_none());
// Publish.
call_put_tag(&conn, key, &value).await.unwrap();
// Hit.
assert_eq!(call_get_tag(&conn, key).await.unwrap(), Some(value));
// List sees it.
let list = call_list_tags(&conn).await.unwrap();
assert_eq!(list.len(), 1);
assert_eq!(list[0].key, key);
// Delete.
assert!(call_delete_tag(&conn, key).await.unwrap());
// Gone.
assert!(call_get_tag(&conn, key).await.unwrap().is_none());
assert!(!call_delete_tag(&conn, key).await.unwrap()); // second delete
assert!(call_list_tags(&conn).await.unwrap().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 call_get_chunk_verifies_returned_hash() {
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_blobs("a", next_port()).await;
let bytes = b"chunk to fetch";
let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
router
.blob_store()
.unwrap()
.put_chunk(&hash, bytes)
.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 got = call_get_chunk(&conn, &hash).await.unwrap().unwrap();
assert_eq!(got, bytes);
let missing = ChunkHash::from_bytes([0u8; 32]);
assert!(call_get_chunk(&conn, &missing).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_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();
}
claw-store/src/cluster/rpc/tests_phase5.rs
+722
View File
@@ -0,0 +1,722 @@
//! Phase 5 (ref-store, tag-store, prewarm) RPC tests split out of
//! `tests.rs` to keep both files under the 1300-line ceiling. Same
//! module namespace via `#[path]` from `rpc.rs`.
use super::*;
use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
use crate::config::ClusterConfig;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU16, Ordering};
use std::time::Duration;
/// Dedicated port range for Phase-5 RPC tests. Different range from
/// tests.rs (43000+) so cross-file parallel execution can't collide.
static NEXT_PORT: AtomicU16 = AtomicU16::new(45001);
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())
}
async fn router_with_blobs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
let gossip = bootstrap_gossip(name, port).await;
let tmp = tempfile::TempDir::new().unwrap();
let store = Arc::new(BlobStore::open(tmp.path().to_path_buf()).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, name.into(), "fabric-10g".into()).with_blob_store(store),
);
(tmp, router)
}
// ── Phase 5b: reference-store RPC ────────────────────────────────────
async fn router_with_blobs_and_refs(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
use crate::cluster::refs::RefStore;
let gossip = bootstrap_gossip(name, port).await;
let tmp = tempfile::TempDir::new().unwrap();
let blob_store =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, name.into(), "fabric-10g".into())
.with_blob_store(blob_store)
.with_ref_store(ref_store),
);
(tmp, router)
}
#[test]
fn phase_5b_method_byte_encoding() {
assert_eq!(Method::GetRef.as_byte(), 0x0d);
assert_eq!(Method::PutRef.as_byte(), 0x0e);
assert_eq!(Method::from_byte(0x0d), Some(Method::GetRef));
assert_eq!(Method::from_byte(0x0e), Some(Method::PutRef));
}
#[tokio::test]
async fn get_ref_returns_not_found_for_missing() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let key = [0u8; 32];
let mut req = vec![Method::GetRef.as_byte()];
req.extend_from_slice(&key);
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn put_ref_stores_and_get_ref_reads_back() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let key = [0x11u8; 32];
let value = [0x22u8; 32];
let mut put = vec![Method::PutRef.as_byte()];
put.extend_from_slice(&key);
put.extend_from_slice(&value);
assert_eq!(dispatch(&router, &put).await, vec![STREAM_STATUS_OK]);
let mut get = vec![Method::GetRef.as_byte()];
get.extend_from_slice(&key);
assert_eq!(dispatch(&router, &get).await, value.to_vec());
}
#[tokio::test]
async fn put_ref_rejects_wrong_length_payload() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
// 63 bytes — one shy of the 32+32 requirement.
let req = {
let mut r = vec![Method::PutRef.as_byte()];
r.extend(vec![0u8; 63]);
r
};
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn get_ref_rejects_wrong_length_payload() {
let (_tmp, router) = router_with_blobs_and_refs("solo", next_port()).await;
let req = vec![Method::GetRef.as_byte(), 0, 1, 2];
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn ref_rpcs_return_not_configured_without_store() {
let gossip = bootstrap_gossip("solo", next_port()).await;
// Router with a blob store but NO ref store.
let tmp = tempfile::TempDir::new().unwrap();
let blob =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().to_path_buf()).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, "solo".into(), "z".into()).with_blob_store(blob),
);
for method in [Method::GetRef, Method::PutRef] {
let mut req = vec![method.as_byte()];
req.extend_from_slice(&[0u8; 32]);
req.extend_from_slice(&[0u8; 32]);
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotConfigured.as_byte()],
"method {method:?} should be NotConfigured"
);
}
}
#[tokio::test]
async fn end_to_end_put_ref_get_ref_over_real_quic() {
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_blobs_and_refs("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 key = [0x77u8; 32];
let value = [0x88u8; 32];
// Miss first.
assert!(call_get_ref(&conn, &key).await.unwrap().is_none());
// Put.
call_put_ref(&conn, &key, &value).await.unwrap();
// Hit.
assert_eq!(call_get_ref(&conn, &key).await.unwrap(), Some(value));
conn.close(quinn::VarInt::from_u32(0), b"done");
client.shutdown().await;
tokio::time::sleep(Duration::from_millis(50)).await;
accept_task.abort();
}
// ── Phase 5d: tag-store RPC ──────────────────────────────────────────
async fn router_with_full_stack(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
use crate::cluster::refs::RefStore;
use crate::cluster::tags::TagStore;
let gossip = bootstrap_gossip(name, port).await;
let tmp = tempfile::TempDir::new().unwrap();
let blob_store =
Arc::new(crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap());
let ref_store = Arc::new(RefStore::open(tmp.path().join("refs-db")).unwrap());
let tag_store = Arc::new(TagStore::open(tmp.path().join("tags-db")).unwrap());
let router = Arc::new(
RpcRouter::new(gossip, name.into(), "fabric-10g".into())
.with_blob_store(blob_store)
.with_ref_store(ref_store)
.with_tag_store(tag_store),
);
(tmp, router)
}
#[test]
fn phase_5d_method_byte_encoding() {
assert_eq!(Method::PutTag.as_byte(), 0x0f);
assert_eq!(Method::GetTag.as_byte(), 0x10);
assert_eq!(Method::DeleteTag.as_byte(), 0x11);
assert_eq!(Method::ListTags.as_byte(), 0x12);
for m in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
assert_eq!(Method::from_byte(m.as_byte()), Some(m));
}
}
#[tokio::test]
async fn tag_rpcs_return_not_configured_without_store() {
let gossip = bootstrap_gossip("solo", next_port()).await;
let router = RpcRouter::new(gossip, "solo".into(), "z".into());
for method in [Method::PutTag, Method::GetTag, Method::DeleteTag, Method::ListTags] {
let mut req = vec![method.as_byte()];
req.extend_from_slice(b"any-key");
let reply = dispatch(&router, &req).await;
assert_eq!(
reply,
vec![ErrorCode::NotConfigured.as_byte()],
"method {method:?} should be NotConfigured"
);
}
}
#[tokio::test]
async fn put_tag_stores_and_get_tag_reads_back() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let key = "clawverse:main:latest";
let value = [0x77u8; 32];
let put_payload = crate::cluster::tags::encode_record(key, &value);
let mut put_req = vec![Method::PutTag.as_byte()];
put_req.extend_from_slice(&put_payload);
assert_eq!(dispatch(&router, &put_req).await, vec![STREAM_STATUS_OK]);
let mut get_req = vec![Method::GetTag.as_byte()];
get_req.extend_from_slice(key.as_bytes());
assert_eq!(dispatch(&router, &get_req).await, value.to_vec());
}
#[tokio::test]
async fn get_tag_returns_not_found_for_missing() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let mut req = vec![Method::GetTag.as_byte()];
req.extend_from_slice(b"never-set");
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn get_tag_rejects_empty_key() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let req = vec![Method::GetTag.as_byte()]; // empty payload
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::InvalidRequest.as_byte()]
);
}
#[tokio::test]
async fn delete_tag_removes_and_returns_not_found_after() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let store = router.tag_store().unwrap().clone();
store.put("removable", &[0u8; 32]).await.unwrap();
let mut req = vec![Method::DeleteTag.as_byte()];
req.extend_from_slice(b"removable");
assert_eq!(dispatch(&router, &req).await, vec![STREAM_STATUS_OK]);
// Second delete → NotFound.
assert_eq!(
dispatch(&router, &req).await,
vec![ErrorCode::NotFound.as_byte()]
);
}
#[tokio::test]
async fn list_tags_returns_json_sorted() {
let (_tmp, router) = router_with_full_stack("solo", next_port()).await;
let store = router.tag_store().unwrap().clone();
store.put("bravo", &[2u8; 32]).await.unwrap();
store.put("alpha", &[1u8; 32]).await.unwrap();
let req = vec![Method::ListTags.as_byte()];
let reply = dispatch(&router, &req).await;
let decoded: Vec<crate::cluster::tags::TagEntry> =
serde_json::from_slice(&reply).unwrap();
assert_eq!(decoded.len(), 2);
assert_eq!(decoded[0].key, "alpha");
assert_eq!(decoded[1].key, "bravo");
assert_eq!(decoded[0].decode_value().unwrap(), [1u8; 32]);
}
#[tokio::test]
async fn end_to_end_prewarm_copies_tagged_blob_between_two_peers() {
// Phase 5f: `claw-cargo prewarm --from A --to C --pin tag`.
// Two full RPC servers running in-process (upstream = A, downstream
// = C). Client is B (uses a third distinct leaf cert), fetches from
// A, uploads to C, republishes the tag on C. Verifies the tag +
// blob are queryable on C after the copy.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
// Second pair for C. The `_id_b2` here is unused but has to share
// the same CA as A so client B can talk to both. Since our test
// helper generates a fresh CA per pair, we cheat by using the same
// pair generator with distinct names — in a real deployment both
// servers would share the fleet CA that signed B.
let (id_c, id_b_for_c) =
NodeIdentity::generate_test_pair("c", "b_client_for_c").unwrap();
let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
let (_tmp_c, router_c) = router_with_full_stack("c", next_port()).await;
// Seed a blob + tag on A.
let payload: Vec<u8> = (0..1_536_000).map(|i| (i % 251) as u8).collect();
let blob_id = router_a
.blob_store()
.unwrap()
.put_bytes(&payload)
.await
.unwrap();
router_a
.tag_store()
.unwrap()
.put("clawverse:main:latest", blob_id.as_bytes())
.await
.unwrap();
let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
let server_a_addr = server_a.local_addr().unwrap();
let router_a_srv = router_a.clone();
let accept_a = tokio::spawn(async move {
if let Some(Ok(conn)) = server_a.accept().await {
let _ = serve_connection(conn, router_a_srv).await;
}
});
let server_c = QuicServer::bind(loopback(0), id_c).unwrap();
let server_c_addr = server_c.local_addr().unwrap();
let router_c_srv = router_c.clone();
let accept_c = tokio::spawn(async move {
if let Some(Ok(conn)) = server_c.accept().await {
let _ = serve_connection(conn, router_c_srv).await;
}
});
// Client → upstream A (via B's identity).
let up_client = QuicClient::new(loopback(0), id_b).unwrap();
let up_conn = up_client.connect(server_a_addr, "a").await.unwrap();
let tag_value = call_get_tag(&up_conn, "clawverse:main:latest")
.await
.unwrap()
.expect("tag on A");
let up_blob_id = BlobId::from_bytes(tag_value);
assert_eq!(up_blob_id, blob_id);
let mut buf: Vec<u8> = Vec::new();
let ok = call_blob_get_stream(&up_conn, &up_blob_id, &mut buf)
.await
.unwrap();
assert!(ok);
assert_eq!(buf, payload);
up_conn.close(quinn::VarInt::from_u32(0), b"done");
up_client.shutdown().await;
// Client → downstream C. Upload the bytes + publish the tag.
let down_client = QuicClient::new(loopback(0), id_b_for_c).unwrap();
let down_conn = down_client.connect(server_c_addr, "c").await.unwrap();
let cursor = std::io::Cursor::new(buf.clone());
let assigned = call_blob_put_stream(&down_conn, cursor).await.unwrap();
assert_eq!(assigned, blob_id, "content-addressed → same BlobId");
call_put_tag(&down_conn, "clawverse:main:latest", blob_id.as_bytes())
.await
.unwrap();
down_conn.close(quinn::VarInt::from_u32(0), b"done");
down_client.shutdown().await;
// Give the accept tasks a moment, then verify C ended up with both
// the blob AND the tag — the two things a subsequent
// `prefetch --pin` would look for.
tokio::time::sleep(Duration::from_millis(50)).await;
let c_blob = router_c
.blob_store()
.unwrap()
.get_bytes(&blob_id)
.await
.unwrap();
assert_eq!(c_blob.as_deref(), Some(payload.as_slice()));
let c_tag = router_c
.tag_store()
.unwrap()
.get("clawverse:main:latest")
.await
.unwrap();
assert_eq!(c_tag, Some(*blob_id.as_bytes()));
accept_a.abort();
accept_c.abort();
}
#[tokio::test]
async fn end_to_end_tag_resolve_and_stream_restore_over_real_quic() {
// Phase 5e: the pipeline `claw-cargo prefetch --pin <tag>` runs
// internally — put a blob, publish a tag pointing at it, then
// GetTag → BlobStat → BlobGetStream to reassemble the content
// byte-equal on the other side.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_full_stack("a", next_port()).await;
// Seed the blob store with a synthetic "captured target dir".
let payload: Vec<u8> = (0..2 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
let blob_id = router
.blob_store()
.unwrap()
.put_bytes(&payload)
.await
.unwrap();
// Publish a tag pointing at that blob.
router
.tag_store()
.unwrap()
.put("clawverse:main:latest", blob_id.as_bytes())
.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();
// Client side of `prefetch --pin <tag>`:
// 1. GetTag → BlobId bytes
// 2. BlobStat → confirm blob exists + size
// 3. BlobGetStream → download into buffer
let tag_value = call_get_tag(&conn, "clawverse:main:latest")
.await
.unwrap()
.expect("tag set");
assert_eq!(tag_value, *blob_id.as_bytes());
let resolved_id = BlobId::from_bytes(tag_value);
let stat = call_blob_stat(&conn, &resolved_id).await.unwrap().unwrap();
assert_eq!(stat.total_size, payload.len() as u64);
let mut sink: Vec<u8> = Vec::new();
let ok = call_blob_get_stream(&conn, &resolved_id, &mut sink)
.await
.unwrap();
assert!(ok);
assert_eq!(sink.len(), payload.len());
assert_eq!(sink, payload, "reassembled bytes match source");
// Missing-tag path: prefetch --pin never-set-name reports None.
let missing = call_get_tag(&conn, "never-set").await.unwrap();
assert!(missing.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_pin_lookup_delete_over_real_quic() {
// Full flow: publish a tag → look it up → list → delete → confirm gone.
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_full_stack("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 key = "clawverse:main:latest-cache";
let value = [0xaau8; 32];
// Miss first.
assert!(call_get_tag(&conn, key).await.unwrap().is_none());
// Publish.
call_put_tag(&conn, key, &value).await.unwrap();
// Hit.
assert_eq!(call_get_tag(&conn, key).await.unwrap(), Some(value));
// List sees it.
let list = call_list_tags(&conn).await.unwrap();
assert_eq!(list.len(), 1);
assert_eq!(list[0].key, key);
// Delete.
assert!(call_delete_tag(&conn, key).await.unwrap());
// Gone.
assert!(call_get_tag(&conn, key).await.unwrap().is_none());
assert!(!call_delete_tag(&conn, key).await.unwrap()); // second delete
assert!(call_list_tags(&conn).await.unwrap().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 call_get_chunk_verifies_returned_hash() {
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
let (_tmp, router) = router_with_blobs("a", next_port()).await;
let bytes = b"chunk to fetch";
let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
router
.blob_store()
.unwrap()
.put_chunk(&hash, bytes)
.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 got = call_get_chunk(&conn, &hash).await.unwrap().unwrap();
assert_eq!(got, bytes);
let missing = ChunkHash::from_bytes([0u8; 32]);
assert!(call_get_chunk(&conn, &missing).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_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();
}