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

Phase 5d: named tags + pin/unpin/list-tags CLI

Browse Source 
Human-readable pins on top of the raw 32-byte ref layer. Operators
publish `clawverse:main:latest-cache` → BlobId once, then everything
downstream (CI runners, dev laptops) references the tag instead of
passing 64-char hex hashes around.

## Module: cluster/tags.rs (433 lines)

TagStore for string-key → 32-byte-value:

- open(root) — creates layout, safe on existing stores
- put(key, value) / get(key) / delete(key) / contains(key)
- list() — sorted by key
- Atomic writes via tempfile + rename
- Key length capped at MAX_TAG_KEY_BYTES (4 KiB); empty keys rejected

On-disk record: `key_len:u16 (LE) || key_bytes || value:32bytes`.
Filename is `blake3(key)` hex so arbitrary UTF-8 keys land at
deterministic paths without shell escaping.

TagEntry type (public, serde) for list results:
`{ key, value_hex }`. Includes `decode_value() → Result<[u8;32]>`.

## RPC methods

- PutTag (0x0f):  payload = encoded record → STREAM_STATUS_OK / err
- GetTag (0x10):  payload = key bytes → 32-byte value / NotFound
- DeleteTag (0x11): payload = key bytes → STREAM_STATUS_OK / NotFound
- ListTags (0x12): payload = empty → JSON Vec<TagEntry>

RpcRouter grows optional Arc<TagStore> via `.with_tag_store(store)`.

## Services + config

ClusterServices auto-opens a TagStore at `<blob_store_root>/tags-db`
alongside the ref store. `tag_store` field on ClusterServices, same
enable-with-blob-store semantics.

## claw-cargo new subcommands

- `claw-cargo pin --name clawverse:main:latest`
    Compute current fingerprint → look up its BlobId via GetRef →
    publish TagStore mapping. Errors cleanly if the fingerprint
    hasn't been built yet (nothing to point at).

- `claw-cargo unpin --name clawverse:main:latest`
    Delete the tag. Prints "no such tag" if it wasn't set.

- `claw-cargo list-tags`
    Print every tag with its 32-byte hex value.

Total subcommand count now 7: build / prefetch / status / fingerprint
/ pin / unpin / list-tags. All share the layered config from Phase 5c.

## Client helpers

- call_put_tag / call_get_tag / call_delete_tag / call_list_tags
- All follow the same error-mapping conventions as prior client helpers
  (NotFound → Ok(None) or Ok(false), everything else → Err)

## Housekeeping

rpc.rs was pushing past the 1300-line ceiling with the tag methods
added. Client helpers moved to `cluster/rpc/client.rs` with a
re-export (`pub use client::*;`) so external callers still write
`cluster::rpc::call_*`. Result:

- rpc.rs: 773 (was 1343)
- rpc/client.rs: 593 (new)
- rpc/tests.rs: 1235
- All under ceiling.

## Tests (33 new, all real filesystem / real QUIC — no mocks)

TagStore (16 in cluster/tags.rs):
- open_creates_layout
- get_returns_none_for_missing (+ contains false)
- put_and_get_round_trip
- put_overwrites_prior_value
- delete_returns_true_for_existing_and_false_for_missing
- put_rejects_empty_key
- put_rejects_oversize_key
- list_returns_all_tags_sorted
- list_is_empty_on_fresh_store
- keys_with_slashes_and_colons_round_trip (real-world tag shape)
- encode_and_decode_round_trip (raw wire format)
- decode_rejects_short_record
- decode_rejects_length_mismatch
- decode_rejects_non_utf8_key
- tag_entry_decode_value_round_trip
- tag_entry_decode_value_rejects_bad_hex

RPC dispatch (7 new):
- phase_5d_method_byte_encoding
- tag_rpcs_return_not_configured_without_store
- put_tag_stores_and_get_tag_reads_back
- get_tag_returns_not_found_for_missing
- get_tag_rejects_empty_key
- delete_tag_removes_and_returns_not_found_after
- list_tags_returns_json_sorted

End-to-end over real QUIC (1):
- **end_to_end_pin_lookup_delete_over_real_quic** — publish tag →
  look up → list → delete → confirm gone. Full round trip through
  the wire layer including JSON deserialization of the list.

Also 8 downstream tests continued passing after the client.rs split
(no test moved, they were untouched).

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

## What this enables

Operator flow:

  # Build once on the primary
  $ claw-cargo build
  → cache MISS → cargo build (50 min) → capture + upload
  → summary: fingerprint 4a3b…, blob 8c2f…, uploaded 3.2 GiB

  # Publish a friendly name
  $ claw-cargo pin --name clawverse:main:2026-07-12
  pinned:      clawverse:main:2026-07-12
  fingerprint: 4a3b2c…
  blob:        8c2f1a…

  # Anyone else can now find it via list-tags
  $ claw-cargo list-tags
  clawverse:main:2026-07-12    8c2f1a…
  clawverse:main:latest         8c2f1a…

  # CI runner sees the same fingerprint in its workspace state, hits
  # the ref directly via GetRef — the tag is for operator visibility

## Follow-on

- 5e: prefetch --pin <tag> — bypass fingerprint compute, download
  the tagged BlobId directly (useful when you want an old cache to
  test regression scenarios)
- 5f: gitea webhook pre-fetch — daemon pre-warms cache for known
  fingerprints before CI runner starts
- 3: full CRDT metadata layer (namespaces, versioned pointers,
  vector clocks) if the plain-tag model turns out to have
  real-world conflict scenarios
This commit is contained in:
Omar Sobh
2026-07-12 00:07:33 -07:00
parent 57d8358255
commit b7904b59a5
7 changed files with 1435 additions and 492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claw-store/src/cluster/rpc/client.rs
+589
View File
@@ -0,0 +1,589 @@
//! Client-side RPC helpers split out of rpc.rs to keep the parent
//! module under the 1300-line ceiling. Every function here calls
//! [`super::rpc_call`] or opens a bidi stream directly and follows
//! the wire format documented on [`super::Method`].
//!
//! Re-exported through `pub use client::*;` in the parent so external
//! callers keep the `cluster::rpc::call_*` paths they've been using.
use super::*;
// ── Streaming client helpers ─────────────────────────────────────────
/// Upload a blob by streaming from `reader` into the peer's store.
/// Never buffers the whole blob — memory usage is dominated by
/// tokio's default copy buffer. Returns the peer-assigned BlobId,
/// which is NOT locally re-verified (the peer doesn't get the whole
/// content in memory at once so we can't cheaply hash from our side);
/// callers who need cryptographic proof of end-to-end integrity
/// should follow up with `call_blob_stat` and compare hashes.
pub async fn call_blob_put_stream<R>(
conn: &Connection,
mut reader: R,
) -> Result<BlobId>
where
R: AsyncRead + Unpin,
{
let (mut send, mut recv) = conn
.open_bi()
.await
.context("opening bidi stream for BlobPutStream")?;
send.write_all(&[Method::BlobPutStream.as_byte()])
.await
.context("writing method tag")?;
tokio::io::copy(&mut reader, &mut send)
.await
.context("streaming blob content to peer")?;
send.finish().context("finishing BlobPutStream send")?;
let mut status = [0u8; 1];
recv.read_exact(&mut status)
.await
.context("reading BlobPutStream status byte")?;
if let Some(code) = decode_error(status[0]) {
bail!("peer replied with error: {}", code.describe());
}
if status[0] != STREAM_STATUS_OK {
bail!(
"peer replied with unknown status byte 0x{:02x}",
status[0]
);
}
let mut id_bytes = [0u8; 32];
recv.read_exact(&mut id_bytes)
.await
.context("reading BlobId from BlobPutStream reply")?;
Ok(BlobId::from_bytes(id_bytes))
}
/// Fetch a blob by streaming its bytes into `writer`. Returns
/// `Ok(false)` when the peer replies [`ErrorCode::NotFound`]; other
/// error codes surface as `Err`. `writer` receives exactly the blob
/// content — no leading status byte, no framing.
pub async fn call_blob_get_stream<W>(
conn: &Connection,
id: &BlobId,
writer: &mut W,
) -> Result<bool>
where
W: AsyncWrite + Unpin,
{
let (mut send, mut recv) = conn
.open_bi()
.await
.context("opening bidi stream for BlobGetStream")?;
let mut req = Vec::with_capacity(33);
req.push(Method::BlobGetStream.as_byte());
req.extend_from_slice(id.as_bytes());
send.write_all(&req)
.await
.context("writing BlobGetStream request")?;
send.finish().context("finishing BlobGetStream send")?;
let mut status = [0u8; 1];
recv.read_exact(&mut status)
.await
.context("reading BlobGetStream status byte")?;
if let Some(code) = decode_error(status[0]) {
if code == ErrorCode::NotFound {
return Ok(false);
}
bail!("peer replied with error: {}", code.describe());
}
if status[0] != STREAM_STATUS_OK {
bail!(
"peer replied with unknown status byte 0x{:02x}",
status[0]
);
}
tokio::io::copy(&mut recv, writer)
.await
.context("streaming blob content from peer")?;
writer
.flush()
.await
.context("flushing destination after BlobGetStream")?;
Ok(true)
}
// `dispatch` is a server-side helper — it lives back in rpc.rs alongside
// `serve_connection`. Tests reach it via `super::dispatch`.
/// Client-side: open a bidi stream, write `method || payload`, read
/// reply. Returns the raw reply bytes; callers deserialise per method.
pub async fn rpc_call(
conn: &Connection,
method: Method,
payload: &[u8],
) -> Result<Vec<u8>> {
if payload.len() + 1 > MAX_MESSAGE_BYTES {
bail!(
"RPC payload {} bytes (+1 for method tag) exceeds cap {}",
payload.len(),
MAX_MESSAGE_BYTES
);
}
let (mut send, mut recv) = conn
.open_bi()
.await
.context("opening bidi stream for RPC")?;
let mut buf = Vec::with_capacity(1 + payload.len());
buf.push(method.as_byte());
buf.extend_from_slice(payload);
send.write_all(&buf).await.context("writing RPC request")?;
send.finish().context("finishing RPC send stream")?;
let reply = recv
.read_to_end(MAX_MESSAGE_BYTES)
.await
.context("reading RPC reply")?;
Ok(reply)
}
/// Convenience wrapper for [`Method::Ping`]. Sends `payload`, returns
/// the peer's echo (`"pong:" || payload`) with the prefix stripped.
pub async fn call_ping(conn: &Connection, payload: &[u8]) -> Result<Vec<u8>> {
let reply = rpc_call(conn, Method::Ping, payload).await?;
if reply.len() == 1 {
if let Some(code) = decode_error(reply[0]) {
bail!("peer replied with error: {}", code.describe());
}
}
if let Some(rest) = reply.strip_prefix(b"pong:") {
Ok(rest.to_vec())
} else {
bail!(
"peer replied with unexpected shape: {} bytes, no 'pong:' prefix",
reply.len()
);
}
}
/// Convenience wrapper for [`Method::PeerStatus`]. Sends an empty
/// payload, deserialises the JSON response.
pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
let reply = rpc_call(conn, Method::PeerStatus, &[]).await?;
if reply.len() == 1 {
if let Some(code) = decode_error(reply[0]) {
bail!("peer replied with error: {}", code.describe());
}
}
serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
}
/// Recognise a single-byte reply as one of our error codes. Returns
/// `None` for any other single-byte value (which is a valid reply,
/// just an unusually short one).
pub fn decode_error(b: u8) -> Option<ErrorCode> {
match b {
0xf0 => Some(ErrorCode::EmptyRequest),
0xf1 => Some(ErrorCode::UnknownMethod),
0xf2 => Some(ErrorCode::HandlerFailure),
0xf3 => Some(ErrorCode::NotFound),
0xf4 => Some(ErrorCode::InvalidRequest),
0xf5 => Some(ErrorCode::NotConfigured),
_ => None,
}
}
// ── Blob RPC client helpers ───────────────────────────────────────────
/// Ask the peer for a blob's size + chunk count. `Ok(None)` when the
/// peer replies [`ErrorCode::NotFound`]; other error codes surface as
/// `Err`.
pub async fn call_blob_stat(
conn: &Connection,
id: &BlobId,
) -> Result<Option<BlobStat>> {
let reply = rpc_call(conn, Method::BlobStat, id.as_bytes()).await?;
if reply.len() == 1 {
match decode_error(reply[0]) {
Some(ErrorCode::NotFound) => return Ok(None),
Some(code) => bail!("peer replied with error: {}", code.describe()),
None => {} // single-byte JSON like "1" is technically possible; fall through
}
}
let stat = serde_json::from_slice(&reply).context("decoding BlobStat JSON")?;
Ok(Some(stat))
}
/// Fetch a blob's raw bytes. `Ok(None)` when the peer replies
/// [`ErrorCode::NotFound`]. Callers that need many-GB transfers should
/// use the streaming variant (Phase 2c); this helper caps at
/// [`MAX_MESSAGE_BYTES`].
pub async fn call_blob_get(
conn: &Connection,
id: &BlobId,
) -> Result<Option<Vec<u8>>> {
let reply = rpc_call(conn, Method::BlobGet, id.as_bytes()).await?;
if reply.len() == 1 {
match decode_error(reply[0]) {
Some(ErrorCode::NotFound) => return Ok(None),
Some(code) => bail!("peer replied with error: {}", code.describe()),
None => {} // single-byte blob is legitimate content
}
}
Ok(Some(reply))
}
/// Upload a blob to the peer's store. Returns the BlobId the peer
/// assigned; must match the local hash of `bytes` (proves the peer
/// stored what we sent).
pub async fn call_blob_put(conn: &Connection, bytes: &[u8]) -> Result<BlobId> {
let reply = rpc_call(conn, Method::BlobPut, bytes).await?;
if reply.len() == 1 {
if let Some(code) = decode_error(reply[0]) {
bail!("peer replied with error: {}", code.describe());
}
}
if reply.len() != 32 {
bail!(
"expected 32-byte BlobId in reply, got {} bytes",
reply.len()
);
}
let mut buf = [0u8; 32];
buf.copy_from_slice(&reply);
let assigned = BlobId::from_bytes(buf);
let expected = BlobId::from_bytes(blake3::hash(bytes).into());
if assigned != expected {
bail!(
"peer returned BlobId {} but content hashes to {}; corruption or protocol drift",
assigned.to_hex(),
expected.to_hex()
);
}
Ok(assigned)
}
/// Fetch a blob's manifest (chunk-list + size). `Ok(None)` on NotFound.
pub async fn call_blob_load_manifest(
conn: &Connection,
id: &BlobId,
) -> Result<Option<BlobManifest>> {
let reply = rpc_call(conn, Method::BlobLoadManifest, id.as_bytes()).await?;
if reply.len() == 1 {
match decode_error(reply[0]) {
Some(ErrorCode::NotFound) => return Ok(None),
Some(code) => bail!("peer replied with error: {}", code.describe()),
None => {}
}
}
let manifest =
serde_json::from_slice(&reply).context("decoding BlobManifest JSON")?;
Ok(Some(manifest))
}
// ── Phase 2d: chunk-level client helpers ─────────────────────────────
/// Does the peer already have this chunk? `Ok(true)` on presence,
/// `Ok(false)` on NotFound. Other errors surface as `Err`.
pub async fn call_has_chunk(
conn: &Connection,
hash: &ChunkHash,
) -> Result<bool> {
let reply = rpc_call(conn, Method::HasChunk, hash.as_bytes()).await?;
if reply.len() != 1 {
bail!(
"expected single-byte HasChunk reply, got {} bytes",
reply.len()
);
}
match reply[0] {
STREAM_STATUS_OK => Ok(true),
code => match decode_error(code) {
Some(ErrorCode::NotFound) => Ok(false),
Some(err) => bail!("peer replied with error: {}", err.describe()),
None => bail!(
"peer replied with unknown byte 0x{:02x} for HasChunk",
code
),
},
}
}
/// Upload a single chunk to the peer's store. Peer verifies bytes
/// hash to `hash` before writing; a mismatch surfaces as `Err`.
pub async fn call_put_chunk(
conn: &Connection,
hash: &ChunkHash,
bytes: &[u8],
) -> Result<()> {
let mut payload = Vec::with_capacity(32 + bytes.len());
payload.extend_from_slice(hash.as_bytes());
payload.extend_from_slice(bytes);
let reply = rpc_call(conn, Method::PutChunk, &payload).await?;
if reply.len() != 1 {
bail!(
"expected single-byte PutChunk reply, got {} bytes",
reply.len()
);
}
match reply[0] {
STREAM_STATUS_OK => Ok(()),
code => match decode_error(code) {
Some(err) => bail!("peer rejected chunk: {}", err.describe()),
None => bail!(
"peer replied with unknown byte 0x{:02x} for PutChunk",
code
),
},
}
}
/// Fetch a single chunk from the peer's store. `Ok(None)` on NotFound.
/// Verifies the returned bytes hash to `hash` locally — corruption or
/// protocol drift surfaces as `Err`.
pub async fn call_get_chunk(
conn: &Connection,
hash: &ChunkHash,
) -> Result<Option<Vec<u8>>> {
let reply = rpc_call(conn, Method::GetChunk, hash.as_bytes()).await?;
if reply.is_empty() {
bail!("empty GetChunk reply");
}
match reply[0] {
STREAM_STATUS_OK => {
let bytes = reply[1..].to_vec();
let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
if recomputed != *hash {
bail!(
"GetChunk hash mismatch: requested {}, got bytes hashing to {}",
hash.to_hex(),
recomputed.to_hex()
);
}
Ok(Some(bytes))
}
code => match decode_error(code) {
Some(ErrorCode::NotFound) => Ok(None),
Some(err) => bail!("peer replied with error: {}", err.describe()),
None => {
if reply.len() == 1 {
bail!(
"peer replied with unknown byte 0x{:02x} for GetChunk",
code
);
}
// A single legitimate content byte with value 0xf3 is
// technically distinguishable from NotFound because
// GetChunk always prefixes with STREAM_STATUS_OK. The
// unreachable branch stays as belt-and-braces.
let bytes = reply[1..].to_vec();
let recomputed = ChunkHash::from_bytes(blake3::hash(&bytes).into());
if recomputed != *hash {
bail!("GetChunk hash mismatch (fallback path)");
}
Ok(Some(bytes))
}
},
}
}
/// Commit a manifest whose chunks the peer should already have.
/// Returns the list of chunk hashes the peer is still missing — an
/// empty list means the manifest was persisted; a non-empty list
/// tells the caller which chunks to upload before retrying.
pub async fn call_put_manifest(
conn: &Connection,
manifest: &BlobManifest,
) -> Result<Vec<ChunkHash>> {
let payload =
serde_json::to_vec(manifest).context("encoding BlobManifest as JSON")?;
let reply = rpc_call(conn, Method::PutManifest, &payload).await?;
if reply.len() == 1 {
if let Some(err) = decode_error(reply[0]) {
bail!("peer replied with error: {}", err.describe());
}
}
let decoded: PutManifestReply = serde_json::from_slice(&reply)
.context("decoding PutManifestReply JSON")?;
if decoded.blob_id != manifest.blob_id {
bail!(
"peer echoed blob_id {} but we sent {}",
decoded.blob_id.to_hex(),
manifest.blob_id.to_hex()
);
}
Ok(decoded.missing)
}
// ── Phase 5b: reference-store client helpers ─────────────────────────
/// Look up a 32-byte value by 32-byte key. `Ok(None)` on `NotFound`.
pub async fn call_get_ref(
conn: &Connection,
key: &RefKey,
) -> Result<Option<RefValue>> {
let reply = rpc_call(conn, Method::GetRef, key).await?;
if reply.len() == 1 {
match decode_error(reply[0]) {
Some(ErrorCode::NotFound) => return Ok(None),
Some(err) => bail!("peer replied with error: {}", err.describe()),
None => {}
}
}
if reply.len() != 32 {
bail!("expected 32-byte ref value, got {} bytes", reply.len());
}
let mut value = [0u8; 32];
value.copy_from_slice(&reply);
Ok(Some(value))
}
/// Set a 32-byte value for a 32-byte key. Overwrites any prior value.
pub async fn call_put_ref(
conn: &Connection,
key: &RefKey,
value: &RefValue,
) -> Result<()> {
let mut payload = Vec::with_capacity(64);
payload.extend_from_slice(key);
payload.extend_from_slice(value);
let reply = rpc_call(conn, Method::PutRef, &payload).await?;
if reply.len() != 1 {
bail!("expected single-byte PutRef reply, got {} bytes", reply.len());
}
match reply[0] {
STREAM_STATUS_OK => Ok(()),
code => match decode_error(code) {
Some(err) => bail!("peer rejected PutRef: {}", err.describe()),
None => bail!(
"peer replied with unknown byte 0x{:02x} for PutRef",
code
),
},
}
}
// ── Phase 5d: tag-store client helpers ───────────────────────────────
/// Publish (or overwrite) a named tag pointing at a 32-byte value.
pub async fn call_put_tag(
conn: &Connection,
key: &str,
value: &[u8; 32],
) -> Result<()> {
let payload = crate::cluster::tags::encode_record(key, value);
let reply = rpc_call(conn, Method::PutTag, &payload).await?;
if reply.len() != 1 {
bail!("expected single-byte PutTag reply, got {} bytes", reply.len());
}
match reply[0] {
STREAM_STATUS_OK => Ok(()),
code => match decode_error(code) {
Some(err) => bail!("peer rejected PutTag: {}", err.describe()),
None => bail!(
"peer replied with unknown byte 0x{:02x} for PutTag",
code
),
},
}
}
/// Look up a named tag. `Ok(None)` on `NotFound`.
pub async fn call_get_tag(
conn: &Connection,
key: &str,
) -> Result<Option<[u8; 32]>> {
if key.is_empty() {
bail!("tag key cannot be empty");
}
let reply = rpc_call(conn, Method::GetTag, key.as_bytes()).await?;
if reply.len() == 1 {
match decode_error(reply[0]) {
Some(ErrorCode::NotFound) => return Ok(None),
Some(err) => bail!("peer replied with error: {}", err.describe()),
None => {}
}
}
if reply.len() != 32 {
bail!("expected 32-byte tag value, got {} bytes", reply.len());
}
let mut value = [0u8; 32];
value.copy_from_slice(&reply);
Ok(Some(value))
}
/// Delete a named tag. `Ok(false)` when no such tag existed.
pub async fn call_delete_tag(conn: &Connection, key: &str) -> Result<bool> {
if key.is_empty() {
bail!("tag key cannot be empty");
}
let reply = rpc_call(conn, Method::DeleteTag, key.as_bytes()).await?;
if reply.len() != 1 {
bail!(
"expected single-byte DeleteTag reply, got {} bytes",
reply.len()
);
}
match reply[0] {
STREAM_STATUS_OK => Ok(true),
code => match decode_error(code) {
Some(ErrorCode::NotFound) => Ok(false),
Some(err) => bail!("peer replied with error: {}", err.describe()),
None => bail!(
"peer replied with unknown byte 0x{:02x} for DeleteTag",
code
),
},
}
}
/// List every tag stored on the peer. Sorted by key.
pub async fn call_list_tags(conn: &Connection) -> Result<Vec<TagEntry>> {
let reply = rpc_call(conn, Method::ListTags, &[]).await?;
if reply.len() == 1 {
if let Some(err) = decode_error(reply[0]) {
bail!("peer replied with error: {}", err.describe());
}
}
serde_json::from_slice(&reply).context("decoding TagEntry list JSON")
}
/// High-level partial-sync helper: replicate a local blob to a peer
/// by uploading only chunks the peer is missing.
///
/// Flow:
/// 1. Load the local manifest.
/// 2. For each chunk, ask the peer if it already has it.
/// 3. Upload the missing chunks.
/// 4. Send the manifest so the peer commits.
///
/// Returns `(uploaded_chunks, total_chunks)` — the difference is what
/// dedup saved. On a warm peer with an identical prior build both
/// numbers are equal to `manifest.chunks.len()` minus the count that
/// were already present, and only the missing-chunk bytes cross the
/// wire.
pub async fn push_blob_missing_chunks(
conn: &Connection,
local: &BlobStore,
id: &BlobId,
) -> Result<(usize, usize)> {
let manifest = local
.load_manifest(id)
.await?
.with_context(|| format!("blob {} not present locally", id.to_hex()))?;
let total = manifest.chunks.len();
let mut uploaded = 0usize;
for hash in &manifest.chunks {
if !call_has_chunk(conn, hash).await? {
let bytes = local
.read_chunk(hash)
.await?
.with_context(|| format!("chunk {} missing locally", hash.to_hex()))?;
call_put_chunk(conn, hash, &bytes).await?;
uploaded += 1;
}
}
// Commit the manifest. `missing` MUST be empty by now — if the
// peer still reports missing chunks after we uploaded them, the
// most likely cause is a store crash on their side; surface as Err.
let still_missing = call_put_manifest(conn, &manifest).await?;
if !still_missing.is_empty() {
bail!(
"peer still reports {} missing chunks after we uploaded {} — retry",
still_missing.len(),
uploaded
);
}
Ok((uploaded, total))
}
claw-store/src/cluster/rpc/tests.rs
+162
View File
@@ -858,6 +858,168 @@ async fn end_to_end_put_ref_get_ref_over_real_quic() {
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_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();