Phase 2b: Blob RPC (Stat/Get/Put/LoadManifest)

Wires the Phase 2 content-addressed blob store onto the network via
four new methods on the existing RpcRouter. Combined with the mTLS +
gossip stack from Phase 1c-1e, peers can now exchange content-addressed
blobs over real QUIC. This is the substrate the fingerprint-keyed cargo
cache (Phase 5) sits on directly.

New methods:
- BlobStat        (0x03): payload = 32-byte BlobId, reply = JSON BlobStat
- BlobGet         (0x04): payload = 32-byte BlobId, reply = raw bytes
- BlobPut         (0x05): payload = raw bytes, reply = 32-byte BlobId
- BlobLoadManifest (0x06): payload = 32-byte BlobId, reply = JSON manifest

New error codes:
- NotFound (0xf3)      — the requested BlobId isn't in the local store
- InvalidRequest (0xf4) — e.g. non-32-byte payload for a hash-keyed method
- NotConfigured (0xf5) — Blob* called on a router without an attached store

Wire-format bump: MAX_MESSAGE_BYTES 16 KiB → 16 MiB so a single 4 MiB
chunk (plus JSON overhead) fits comfortably. Anything above 16 MiB
needs the streaming variants coming in Phase 2c.

Client helpers:
- call_blob_stat / call_blob_get / call_blob_put / call_blob_load_manifest
- All map ErrorCode::NotFound to Ok(None), other codes to Err.
- call_blob_put verifies the peer-assigned BlobId matches local blake3
  hash of the payload — corruption or protocol drift surfaces
  immediately instead of silently accepting a mismatched receipt.

Router changes:
- RpcRouter grows an Option<Arc<BlobStore>> via with_blob_store(store).
- handle() split into handle_outcome() → HandlerOutcome enum
  {Reply(bytes) | Error(ErrorCode)} for cleaner control flow across
  the growing method set.

Services / daemon:
- ClusterServices::start gains blob_store_root: Option<PathBuf>.
- ClusterConfig gains blob_store_root: Option<PathBuf>.
- daemon.rs reads it from cluster_cfg + passes through.
- New ClusterServices::blob_store_enabled() introspection.

Tests (11 new, all real — no mocks):

Router (7 new):
- method_round_trips_byte_encoding — updated for 6 methods
- error_code_describe_covers_all_variants — updated for 6 codes
- decode_error_covers_all_known_codes
- blob_rpcs_return_not_configured_without_store — all four Blob*
  methods return NotConfigured when the router lacks a store
- blob_stat_returns_not_found_for_missing
- blob_stat_returns_json_for_existing
- blob_stat_returns_invalid_request_for_bad_length
- blob_get_returns_content_bytes
- blob_put_stores_bytes_and_returns_hash — verifies BlobId matches
  independent local hash
- blob_load_manifest_returns_json_for_existing (2-chunk case)
- blob_load_manifest_returns_not_found_for_missing

End-to-end over real QUIC (2 new):
- end_to_end_blob_put_stat_get_over_real_quic — full 4-method loop
  (Put → Stat → Get → LoadManifest) + NotFound path
- end_to_end_multi_chunk_blob_over_real_quic — 6 MiB blob → 2 chunks,
  proves MAX_MESSAGE_BYTES bump took effect

Services (1 new):
- services_with_blob_store_serves_blob_rpc_end_to_end — cut CA, sign
  leaves, config includes blob_store_root, start ClusterServices,
  dial from B over persisted mTLS, put + get through the router,
  then independently verify bytes landed on A's on-disk store

Also fixed a parallel-test port collision: services `next_port()`
now increments by 2 so `port + 1` (the RPC bind) is reserved
alongside `port` (the gossip bind).

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

File sizes (all under 1300-line ceiling):
- cluster/rpc.rs: 1006
- cluster/services.rs: 535
- cluster/blob.rs: 802
- config.rs: 511
- daemon.rs: 265

Follow-on:
- 2c: streaming variants (AsyncRead/AsyncWrite) so a many-GB blob
  transfers without holding it in memory
- 2d: chunk-level RPC (BlobPutChunk / BlobGetChunk) so a receiver
  can request only chunks it's missing after LoadManifest
- Phase 5 (the killer feature) can now build on Phase 2b directly —
  fingerprint the target dir, PutBlob the compressed tarball, and
  next node calls GetBlob keyed by the same fingerprint hash.
This commit is contained in:
Omar Sobh
2026-07-11 22:42:17 -07:00
parent eab10005fd
commit f43b34ad15
5 changed files with 654 additions and 24 deletions
+14 -6
View File
@@ -137,6 +137,12 @@ pub struct ClusterConfig {
/// used; absent means "gossip only, no RPC" for now.
#[serde(default)]
pub tls: Option<ClusterTlsConfig>,
/// Optional local blob-store root (Phase 2). When set, the daemon
/// opens a content-addressed store at this path and serves it via
/// the Blob* RPC methods. Absent means the node participates in
/// gossip + peer-status but returns `NotConfigured` for Blob RPCs.
#[serde(default)]
pub blob_store_root: Option<PathBuf>,
}
/// Compute the default RPC address for a gossip address: same IP, port + 1.
@@ -392,9 +398,10 @@ tailscale_addr = "100.64.1.5:7701"
bind_lan: None,
bind_tailscale: None,
peers: vec![],
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
blob_store_root: None,
};
let err = cluster.validate().unwrap_err().to_string();
assert!(
@@ -423,9 +430,10 @@ tailscale_addr = "100.64.1.5:7701"
tailscale_addr: None,
},
],
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
blob_store_root: None,
};
let err = cluster.validate().unwrap_err().to_string();
assert!(