Phase 3c + 3e: stamped tags + namespaced ref keys — closes Phase 3 #42
@@ -48,7 +48,7 @@ use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
|
||||
use crate::cluster::rpc::{
|
||||
call_blob_get_parallel, call_blob_get_stream, call_blob_put_stream, call_blob_stat,
|
||||
call_delete_tag, call_get_metrics, call_get_ref_versioned, call_peer_status,
|
||||
call_put_ref_versioned, prewarm_missing_chunks_between_parallel,
|
||||
call_put_ref_versioned, call_put_tag_versioned, prewarm_missing_chunks_between_parallel,
|
||||
call_get_ref, call_get_tag, call_list_tags, call_put_ref, call_put_tag,
|
||||
};
|
||||
use crate::cluster::transport::{NodeIdentity, QuicClient};
|
||||
@@ -184,6 +184,15 @@ struct PeerArgs {
|
||||
/// Workspace root. Defaults to the current directory.
|
||||
#[arg(long)]
|
||||
workspace: Option<PathBuf>,
|
||||
/// Phase 3e (2026-07-13): opt-in namespace for the ref key. When
|
||||
/// set, the fingerprint is combined with the namespace via
|
||||
/// `blake3("clawstor.ns.v1" || namespace || fp)` before use — so
|
||||
/// two runners on different namespaces (e.g. `clawverse/main` vs
|
||||
/// `clawverse/pr-42`) don't collide on the same fingerprint.
|
||||
/// Empty (the default) preserves the pre-3e behavior for
|
||||
/// backward compat.
|
||||
#[arg(long)]
|
||||
namespace: Option<String>,
|
||||
}
|
||||
|
||||
/// Args for the local-only `fingerprint` subcommand.
|
||||
@@ -321,16 +330,15 @@ async fn connect_peer(cfg: &ResolvedClientConfig) -> Result<(QuicClient, quinn::
|
||||
/// garbage-collected blob).
|
||||
async fn peer_lookup(
|
||||
conn: &quinn::Connection,
|
||||
fp: &Fingerprint,
|
||||
key: &crate::cluster::refs::RefKey,
|
||||
) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
|
||||
let key = *fp.as_bytes();
|
||||
// Phase 3b (2026-07-13): read stamped refs first so cross-runner
|
||||
// ref-forwarding uses the CRDT-merge path. Fall back to legacy
|
||||
// unstamped refs for any pre-Phase-3a data still on disk.
|
||||
let value_bytes: crate::cluster::refs::RefValue =
|
||||
match call_get_ref_versioned(conn, &key).await? {
|
||||
match call_get_ref_versioned(conn, key).await? {
|
||||
Some(s) => s.value,
|
||||
None => match call_get_ref(conn, &key).await? {
|
||||
None => match call_get_ref(conn, key).await? {
|
||||
Some(v) => v,
|
||||
None => return Ok(None),
|
||||
},
|
||||
@@ -374,7 +382,8 @@ async fn cmd_status(args: PeerArgs) -> Result<()> {
|
||||
println!("tls: {}", tls_dir.display());
|
||||
|
||||
let (client, conn) = connect_peer(&resolved).await?;
|
||||
match peer_lookup(&conn, &fp).await? {
|
||||
let key = ref_key_for(&args.namespace, &fp);
|
||||
match peer_lookup(&conn, &key).await? {
|
||||
Some((blob_id, stat)) => {
|
||||
println!("cache: HIT");
|
||||
println!(" blob: {}", blob_id);
|
||||
@@ -412,7 +421,10 @@ async fn cmd_prefetch(args: PrefetchArgs) -> Result<()> {
|
||||
return Ok(());
|
||||
}
|
||||
},
|
||||
None => peer_lookup(&conn, &fp).await?,
|
||||
None => {
|
||||
let key = ref_key_for(&args.peer.namespace, &fp);
|
||||
peer_lookup(&conn, &key).await?
|
||||
}
|
||||
};
|
||||
|
||||
let outcome = match lookup {
|
||||
@@ -568,11 +580,20 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
|
||||
.join("target")
|
||||
.join(target_subdir_for(&resolved.profile));
|
||||
tracing::info!("fingerprint {}", fp);
|
||||
let ref_key = ref_key_for(&args.peer.namespace, &fp);
|
||||
if args.peer.namespace.as_deref().unwrap_or("").is_empty() {
|
||||
tracing::info!("ref-key: fingerprint (no namespace)");
|
||||
} else {
|
||||
tracing::info!(
|
||||
namespace = %args.peer.namespace.as_deref().unwrap_or(""),
|
||||
"ref-key: namespaced fingerprint"
|
||||
);
|
||||
}
|
||||
|
||||
let (client, conn) = connect_peer(&resolved).await?;
|
||||
|
||||
let mut outcome = CacheOutcome::Miss;
|
||||
match peer_lookup(&conn, &fp).await? {
|
||||
match peer_lookup(&conn, &ref_key).await? {
|
||||
Some((blob_id, stat)) => {
|
||||
tracing::info!(
|
||||
"cache HIT — blob {} ({} bytes, parallel={})",
|
||||
@@ -675,7 +696,7 @@ async fn cmd_build(args: BuildArgs) -> Result<()> {
|
||||
// its blob is byte-identical (content-addressed).
|
||||
let stamped = build_stamped_ref(&blob_id);
|
||||
let merged =
|
||||
call_put_ref_versioned(&conn, fp.as_bytes(), &stamped).await?;
|
||||
call_put_ref_versioned(&conn, &ref_key, &stamped).await?;
|
||||
if merged {
|
||||
tracing::info!(
|
||||
"uploaded blob {} + set stamped ref (clock={})",
|
||||
@@ -769,16 +790,39 @@ async fn cmd_pin(args: PinArgs) -> Result<()> {
|
||||
}
|
||||
};
|
||||
|
||||
// 2. Publish the tag → BlobId mapping.
|
||||
call_put_tag(&conn, &args.name, blob_id.as_bytes()).await?;
|
||||
// 2. Publish the tag → BlobId mapping. Phase 3c: use the stamped
|
||||
// variant so concurrent `pin` calls race deterministically —
|
||||
// higher (clock, node) wins, loser sees AlreadyExists.
|
||||
let stamped_value = crate::cluster::tags::StampedTagValue {
|
||||
value: *blob_id.as_bytes(),
|
||||
clock: std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0),
|
||||
node: match hostname_string() {
|
||||
Some(h) => crate::cluster::refs::node_stamp_for(&h),
|
||||
None => [0u8; 8],
|
||||
},
|
||||
};
|
||||
let merged =
|
||||
call_put_tag_versioned(&conn, &args.name, &stamped_value).await?;
|
||||
if !merged {
|
||||
tracing::info!(
|
||||
"another writer already published a dominant tag for {}; still ok",
|
||||
args.name
|
||||
);
|
||||
}
|
||||
|
||||
// Field finding 2026-07-12: also stash the fingerprint under a
|
||||
// companion tag `<name>.fingerprint`. Downstream nodes fed by
|
||||
// prewarm need to know the fingerprint to publish their own
|
||||
// ref (`PutRef(fingerprint → blob)`) — otherwise a `build` on
|
||||
// the same source misses even though the blob is present.
|
||||
// Companion tag stashes the fingerprint bytes under
|
||||
// `<name>.fingerprint` so prewarm downstream can PutRef locally.
|
||||
let companion = fingerprint_companion_tag(&args.name);
|
||||
call_put_tag(&conn, &companion, fp.as_bytes()).await?;
|
||||
let companion_stamped = crate::cluster::tags::StampedTagValue {
|
||||
value: *fp.as_bytes(),
|
||||
clock: stamped_value.clock,
|
||||
node: stamped_value.node,
|
||||
};
|
||||
let _ =
|
||||
call_put_tag_versioned(&conn, &companion, &companion_stamped).await?;
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
@@ -1180,6 +1224,20 @@ fn build_stamped_ref(
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 3e (2026-07-13): derive the ref-key to use for a lookup.
|
||||
/// Namespaced when the caller opted in via `--namespace` (or the
|
||||
/// layered config), otherwise the raw fingerprint bytes so the
|
||||
/// pre-3e default behavior is preserved.
|
||||
fn ref_key_for(
|
||||
namespace: &Option<String>,
|
||||
fp: &Fingerprint,
|
||||
) -> crate::cluster::refs::RefKey {
|
||||
match namespace.as_deref().filter(|s| !s.is_empty()) {
|
||||
Some(ns) => crate::cluster::refs::namespaced_ref_key(ns, fp.as_bytes()),
|
||||
None => *fp.as_bytes(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Cheap `hostname` probe. Reads `/etc/hostname` on Linux, falls
|
||||
/// back to `HOSTNAME`/`COMPUTERNAME` env vars. `None` on any read
|
||||
/// failure — callers treat that as "no node identity available."
|
||||
|
||||
@@ -99,6 +99,30 @@ pub fn node_stamp_for(name: &str) -> NodeStamp {
|
||||
out
|
||||
}
|
||||
|
||||
/// Phase 3e (2026-07-13): derive a namespaced ref key from a
|
||||
/// `(namespace, fingerprint)` pair. Different namespaces produce
|
||||
/// different 32-byte keys for the same fingerprint — the primitive
|
||||
/// that lets a fleet segregate cache lookups by org/repo/branch/kind
|
||||
/// without touching the underlying content-addressed blob store.
|
||||
///
|
||||
/// Domain: `blake3("clawstor.ns.v1\0" || namespace || "\0" || fp)`.
|
||||
/// The version tag and the NUL delimiter make it collision-resistant
|
||||
/// against a future rewrite of this rule.
|
||||
///
|
||||
/// Empty namespace → the key is a deterministic function of the
|
||||
/// fingerprint alone, which callers can use for cluster-wide
|
||||
/// (default) lookups. Non-empty namespace → siloed lookups.
|
||||
pub fn namespaced_ref_key(namespace: &str, fingerprint: &[u8; 32]) -> RefKey {
|
||||
let mut h = blake3::Hasher::new();
|
||||
h.update(b"clawstor.ns.v1\0");
|
||||
h.update(namespace.as_bytes());
|
||||
h.update(b"\0");
|
||||
h.update(fingerprint);
|
||||
let mut out = [0u8; 32];
|
||||
out.copy_from_slice(&h.finalize().as_bytes()[..32]);
|
||||
out
|
||||
}
|
||||
|
||||
/// Outcome of a stamped put. Callers can distinguish "we accepted
|
||||
/// your write" from "someone else already had a newer/equal write".
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@@ -420,6 +444,28 @@ mod tests {
|
||||
assert!(!c.dominates(&c));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn namespaced_ref_key_is_stable_and_ns_scoped() {
|
||||
// Phase 3e: the same (namespace, fingerprint) always maps to
|
||||
// the same 32-byte key; different namespaces produce distinct
|
||||
// keys for the same fingerprint; the empty namespace is a
|
||||
// legitimate default lookup path.
|
||||
let fp = [42u8; 32];
|
||||
let a1 = namespaced_ref_key("clawverse/main", &fp);
|
||||
let a2 = namespaced_ref_key("clawverse/main", &fp);
|
||||
assert_eq!(a1, a2, "stable for same inputs");
|
||||
|
||||
let b = namespaced_ref_key("clawverse/pr-42", &fp);
|
||||
assert_ne!(a1, b, "different namespaces silo the key");
|
||||
|
||||
let empty = namespaced_ref_key("", &fp);
|
||||
assert_ne!(a1, empty, "empty namespace ≠ named namespace");
|
||||
// Empty is a valid default and must not accidentally match
|
||||
// the raw fingerprint (bypass would break the domain
|
||||
// separation).
|
||||
assert_ne!(empty, fp, "empty namespace still hashes the fp");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn node_stamp_is_stable_for_same_name() {
|
||||
let a = node_stamp_for("tank");
|
||||
|
||||
@@ -158,6 +158,19 @@ pub enum Method {
|
||||
/// MUST NOT forward on miss. Used by daemons doing ref-forwarding
|
||||
/// so they never loop.
|
||||
GetRefVersionedLocal = 0x17,
|
||||
/// Phase 3c (2026-07-13): stamped/versioned PutTag. Merges under
|
||||
/// the same `(clock, node)` total order as
|
||||
/// [`Method::PutRefVersioned`].
|
||||
/// `payload`: `key_len:u16 (LE) || key_bytes || stamped_value:48`.
|
||||
/// Reply: single-byte status —
|
||||
/// `STREAM_STATUS_OK` = accepted (Merged);
|
||||
/// [`ErrorCode::AlreadyExists`] = rejected (older/equal).
|
||||
PutTagVersioned = 0x18,
|
||||
/// Phase 3c: fetch a stamped tag value.
|
||||
/// `payload`: raw tag key bytes (variable length, up to 4 KiB).
|
||||
/// Reply: 48 bytes on hit, single-byte
|
||||
/// [`ErrorCode::NotFound`] on miss.
|
||||
GetTagVersioned = 0x19,
|
||||
}
|
||||
|
||||
impl Method {
|
||||
@@ -188,6 +201,8 @@ impl Method {
|
||||
0x15 => Some(Method::PutRefVersioned),
|
||||
0x16 => Some(Method::GetRefVersioned),
|
||||
0x17 => Some(Method::GetRefVersionedLocal),
|
||||
0x18 => Some(Method::PutTagVersioned),
|
||||
0x19 => Some(Method::GetTagVersioned),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -739,6 +754,50 @@ impl RpcRouter {
|
||||
}
|
||||
}
|
||||
}
|
||||
Method::PutTagVersioned => {
|
||||
let store = match &self.tag_store {
|
||||
Some(s) => s,
|
||||
None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
|
||||
};
|
||||
// Wire: key_len:u16 (LE) || key_bytes || stamped_value:48.
|
||||
let (key, stamped) =
|
||||
match crate::cluster::tags::decode_stamped_record(payload) {
|
||||
Ok(kv) => kv,
|
||||
Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
|
||||
};
|
||||
match store.put_stamped(&key, stamped).await {
|
||||
Ok(crate::cluster::tags::TagPutOutcome::Merged) => {
|
||||
Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
|
||||
}
|
||||
Ok(crate::cluster::tags::TagPutOutcome::Rejected { .. }) => {
|
||||
Ok(HandlerOutcome::Error(ErrorCode::AlreadyExists))
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "PutTagVersioned rejected");
|
||||
Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
|
||||
}
|
||||
}
|
||||
}
|
||||
Method::GetTagVersioned => {
|
||||
let store = match &self.tag_store {
|
||||
Some(s) => s,
|
||||
None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
|
||||
};
|
||||
let key = match std::str::from_utf8(payload) {
|
||||
Ok(s) if !s.is_empty() => s,
|
||||
_ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
|
||||
};
|
||||
match store.get_stamped(key).await? {
|
||||
Some(s) => {
|
||||
self.metrics.record_get_tag_hit();
|
||||
Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()))
|
||||
}
|
||||
None => {
|
||||
self.metrics.record_get_tag_miss();
|
||||
Ok(HandlerOutcome::Error(ErrorCode::NotFound))
|
||||
}
|
||||
}
|
||||
}
|
||||
Method::DeleteTag => {
|
||||
let store = match &self.tag_store {
|
||||
Some(s) => s,
|
||||
|
||||
@@ -546,6 +546,58 @@ async fn call_get_ref_versioned_inner(
|
||||
))
|
||||
}
|
||||
|
||||
// ── Phase 3c: stamped-tag client helpers ─────────────────────────────
|
||||
|
||||
/// Phase 3c (2026-07-13): submit a stamped (CRDT-merge) PutTag.
|
||||
///
|
||||
/// Returns `Ok(true)` when the peer merged the write, `Ok(false)`
|
||||
/// when the peer rejected it because an equal-or-newer version
|
||||
/// already exists (`AlreadyExists`). Any other reply is an error.
|
||||
pub async fn call_put_tag_versioned(
|
||||
conn: &Connection,
|
||||
key: &str,
|
||||
incoming: &crate::cluster::tags::StampedTagValue,
|
||||
) -> Result<bool> {
|
||||
let payload = crate::cluster::tags::encode_stamped_record(key, incoming);
|
||||
let reply = rpc_call(conn, Method::PutTagVersioned, &payload).await?;
|
||||
if reply.len() != 1 {
|
||||
bail!(
|
||||
"expected single-byte PutTagVersioned reply, got {} bytes",
|
||||
reply.len()
|
||||
);
|
||||
}
|
||||
match reply[0] {
|
||||
STREAM_STATUS_OK => Ok(true),
|
||||
code => match decode_error(code) {
|
||||
Some(ErrorCode::AlreadyExists) => Ok(false),
|
||||
Some(err) => bail!("peer rejected PutTagVersioned: {}", err.describe()),
|
||||
None => bail!(
|
||||
"peer replied with unknown byte 0x{:02x} for PutTagVersioned",
|
||||
code
|
||||
),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 3c: fetch a stamped tag value.
|
||||
pub async fn call_get_tag_versioned(
|
||||
conn: &Connection,
|
||||
key: &str,
|
||||
) -> Result<Option<crate::cluster::tags::StampedTagValue>> {
|
||||
let reply = rpc_call(conn, Method::GetTagVersioned, 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 => {}
|
||||
}
|
||||
}
|
||||
Ok(Some(
|
||||
crate::cluster::tags::StampedTagValue::from_bytes(&reply)
|
||||
.context("decoding stamped tag reply")?,
|
||||
))
|
||||
}
|
||||
|
||||
// ── Phase 5g: cache metrics client helper ────────────────────────────
|
||||
|
||||
/// Fetch the peer's current cache-metrics snapshot.
|
||||
|
||||
@@ -1364,3 +1364,74 @@ async fn end_to_end_put_ref_versioned_merges_and_rejects() {
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
acc.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn end_to_end_put_tag_versioned_merges_and_rejects() {
|
||||
// Phase 3c end-to-end: two writers publish stamped tags for the
|
||||
// same key. Higher (clock, node) wins.
|
||||
use crate::cluster::refs::node_stamp_for;
|
||||
use crate::cluster::tags::StampedTagValue;
|
||||
|
||||
let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap();
|
||||
let (_tmp_a, router_a) = router_with_full_stack("a", next_port()).await;
|
||||
|
||||
let server_a = QuicServer::bind(loopback(0), id_a).unwrap();
|
||||
let addr = server_a.local_addr().unwrap();
|
||||
let ra = router_a.clone();
|
||||
let acc = tokio::spawn(async move {
|
||||
while let Some(Ok(conn)) = server_a.accept().await {
|
||||
let r = ra.clone();
|
||||
tokio::spawn(async move {
|
||||
let _ = serve_connection(conn, r).await;
|
||||
});
|
||||
}
|
||||
});
|
||||
let client = QuicClient::new(loopback(0), id_b).unwrap();
|
||||
let conn = client.connect(addr, "a").await.unwrap();
|
||||
|
||||
let key = "clawverse:main:latest";
|
||||
let node_x = node_stamp_for("runner-x");
|
||||
let node_y = node_stamp_for("runner-y");
|
||||
|
||||
assert_eq!(call_get_tag_versioned(&conn, key).await.unwrap(), None);
|
||||
|
||||
let first = StampedTagValue {
|
||||
value: [0x11; 32],
|
||||
clock: 10,
|
||||
node: node_x,
|
||||
};
|
||||
assert!(call_put_tag_versioned(&conn, key, &first).await.unwrap());
|
||||
assert_eq!(
|
||||
call_get_tag_versioned(&conn, key).await.unwrap(),
|
||||
Some(first)
|
||||
);
|
||||
|
||||
// Older clock → rejected.
|
||||
let older = StampedTagValue {
|
||||
value: [0x22; 32],
|
||||
clock: 9,
|
||||
node: node_y,
|
||||
};
|
||||
assert!(!call_put_tag_versioned(&conn, key, &older).await.unwrap());
|
||||
assert_eq!(
|
||||
call_get_tag_versioned(&conn, key).await.unwrap(),
|
||||
Some(first)
|
||||
);
|
||||
|
||||
// Higher clock always merges.
|
||||
let latest = StampedTagValue {
|
||||
value: [0x33; 32],
|
||||
clock: 100,
|
||||
node: node_x,
|
||||
};
|
||||
assert!(call_put_tag_versioned(&conn, key, &latest).await.unwrap());
|
||||
assert_eq!(
|
||||
call_get_tag_versioned(&conn, key).await.unwrap(),
|
||||
Some(latest)
|
||||
);
|
||||
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
client.shutdown().await;
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
acc.abort();
|
||||
}
|
||||
|
||||
@@ -31,6 +31,100 @@ use tokio::io::AsyncWriteExt;
|
||||
/// almost certainly a bug on the caller side.
|
||||
pub const MAX_TAG_KEY_BYTES: usize = 4096;
|
||||
|
||||
/// Phase 3c (2026-07-13): the tag equivalent of
|
||||
/// [`crate::cluster::refs::StampedRef`]. Same `(clock, node)` total
|
||||
/// order, same idempotency semantics on the merge path.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct StampedTagValue {
|
||||
pub value: [u8; 32],
|
||||
pub clock: u64,
|
||||
pub node: crate::cluster::refs::NodeStamp,
|
||||
}
|
||||
|
||||
impl StampedTagValue {
|
||||
/// 48-byte on-wire representation: `value:32 || clock:u64 LE || node:8`.
|
||||
pub const ENCODED_LEN: usize = 48;
|
||||
|
||||
pub fn to_bytes(&self) -> [u8; Self::ENCODED_LEN] {
|
||||
let mut out = [0u8; Self::ENCODED_LEN];
|
||||
out[..32].copy_from_slice(&self.value);
|
||||
out[32..40].copy_from_slice(&self.clock.to_le_bytes());
|
||||
out[40..48].copy_from_slice(&self.node);
|
||||
out
|
||||
}
|
||||
|
||||
pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
|
||||
if bytes.len() != Self::ENCODED_LEN {
|
||||
bail!(
|
||||
"stamped tag value wrong length {} (expected {})",
|
||||
bytes.len(),
|
||||
Self::ENCODED_LEN
|
||||
);
|
||||
}
|
||||
let mut value = [0u8; 32];
|
||||
value.copy_from_slice(&bytes[..32]);
|
||||
let mut clock_bytes = [0u8; 8];
|
||||
clock_bytes.copy_from_slice(&bytes[32..40]);
|
||||
let mut node = [0u8; 8];
|
||||
node.copy_from_slice(&bytes[40..48]);
|
||||
Ok(Self {
|
||||
value,
|
||||
clock: u64::from_le_bytes(clock_bytes),
|
||||
node,
|
||||
})
|
||||
}
|
||||
|
||||
/// Same total-order semantics as [`crate::cluster::refs::StampedRef::dominates`].
|
||||
pub fn stamp_dominates(&self, other: &Self) -> bool {
|
||||
(self.clock, self.node) > (other.clock, other.node)
|
||||
}
|
||||
}
|
||||
|
||||
/// Merge outcome for [`TagStore::put_stamped`].
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum TagPutOutcome {
|
||||
Merged,
|
||||
Rejected { current: StampedTagValue },
|
||||
}
|
||||
|
||||
/// Encode a stamped tag record as
|
||||
/// `key_len:u16 (LE) || key_bytes || stamped_value:48`.
|
||||
pub fn encode_stamped_record(key: &str, stamped: &StampedTagValue) -> Vec<u8> {
|
||||
let key_bytes = key.as_bytes();
|
||||
let mut out = Vec::with_capacity(2 + key_bytes.len() + StampedTagValue::ENCODED_LEN);
|
||||
out.extend_from_slice(&(key_bytes.len() as u16).to_le_bytes());
|
||||
out.extend_from_slice(key_bytes);
|
||||
out.extend_from_slice(&stamped.to_bytes());
|
||||
out
|
||||
}
|
||||
|
||||
/// Reverse of [`encode_stamped_record`].
|
||||
pub fn decode_stamped_record(bytes: &[u8]) -> Result<(String, StampedTagValue)> {
|
||||
if bytes.len() < 2 {
|
||||
bail!(
|
||||
"stamped tag record too short for length prefix ({} bytes)",
|
||||
bytes.len()
|
||||
);
|
||||
}
|
||||
let key_len = u16::from_le_bytes([bytes[0], bytes[1]]) as usize;
|
||||
let expected = 2 + key_len + StampedTagValue::ENCODED_LEN;
|
||||
if bytes.len() != expected {
|
||||
bail!(
|
||||
"stamped tag record length {} does not match declared shape \
|
||||
(key_len={}, expected total {})",
|
||||
bytes.len(),
|
||||
key_len,
|
||||
expected
|
||||
);
|
||||
}
|
||||
let key_bytes = &bytes[2..2 + key_len];
|
||||
let key = std::str::from_utf8(key_bytes)
|
||||
.context("stamped tag key is not valid UTF-8")?
|
||||
.to_string();
|
||||
let stamped = StampedTagValue::from_bytes(&bytes[2 + key_len..])?;
|
||||
Ok((key, stamped))
|
||||
}
|
||||
|
||||
/// A tag entry as returned by `list`.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
pub struct TagEntry {
|
||||
@@ -182,6 +276,68 @@ impl TagStore {
|
||||
.join(format!("{hash_hex}.tag"))
|
||||
}
|
||||
|
||||
/// Phase 3c (2026-07-13): Lamport-stamped tag lookup. Reads from
|
||||
/// `tags-v2/` — a separate namespace from raw `tags/` so the two
|
||||
/// coexist during the cutover. Returns `None` when no stamped
|
||||
/// value exists for this key.
|
||||
pub async fn get_stamped(&self, key: &str) -> Result<Option<StampedTagValue>> {
|
||||
validate_key(key)?;
|
||||
let path = self.stamped_tag_path(key);
|
||||
let bytes = match tokio::fs::read(&path).await {
|
||||
Ok(b) => b,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
|
||||
Err(e) => return Err(anyhow::Error::from(e)),
|
||||
};
|
||||
let (parsed_key, stamped) =
|
||||
decode_stamped_record(&bytes).with_context(|| {
|
||||
format!("decoding stamped tag record at {}", path.display())
|
||||
})?;
|
||||
if parsed_key != key {
|
||||
bail!(
|
||||
"stamped tag record at {} has key {:?} but was requested as {:?}",
|
||||
path.display(),
|
||||
parsed_key,
|
||||
key
|
||||
);
|
||||
}
|
||||
Ok(Some(stamped))
|
||||
}
|
||||
|
||||
/// Phase 3c (2026-07-13): merge an incoming stamped tag. Higher
|
||||
/// `(clock, node)` wins; equal is idempotent (returns
|
||||
/// [`TagPutOutcome::Rejected`] with current).
|
||||
///
|
||||
/// Never lowers the on-disk value.
|
||||
pub async fn put_stamped(
|
||||
&self,
|
||||
key: &str,
|
||||
incoming: StampedTagValue,
|
||||
) -> Result<TagPutOutcome> {
|
||||
validate_key(key)?;
|
||||
let path = self.stamped_tag_path(key);
|
||||
if let Some(current) = self.get_stamped(key).await? {
|
||||
if !incoming.stamp_dominates(¤t) {
|
||||
return Ok(TagPutOutcome::Rejected { current });
|
||||
}
|
||||
}
|
||||
if let Some(parent) = path.parent() {
|
||||
tokio::fs::create_dir_all(parent).await.with_context(|| {
|
||||
format!("creating stamped-tag bucket {}", parent.display())
|
||||
})?;
|
||||
}
|
||||
let bytes = encode_stamped_record(key, &incoming);
|
||||
self.atomic_write(&path, &bytes).await?;
|
||||
Ok(TagPutOutcome::Merged)
|
||||
}
|
||||
|
||||
fn stamped_tag_path(&self, key: &str) -> PathBuf {
|
||||
let hash_hex = hex32(blake3::hash(key.as_bytes()).as_bytes());
|
||||
self.root
|
||||
.join("tags-v2")
|
||||
.join(&hash_hex[..2])
|
||||
.join(format!("{hash_hex}.svtag"))
|
||||
}
|
||||
|
||||
async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
|
||||
let tmp_dir = self.root.join(".tmp");
|
||||
let tmp_name = format!(
|
||||
@@ -430,4 +586,87 @@ mod tests {
|
||||
};
|
||||
assert!(entry.decode_value().is_err());
|
||||
}
|
||||
|
||||
// ── Phase 3c: stamped tags ───────────────────────────────────────
|
||||
|
||||
fn stamped_tag(value: u8, clock: u64, node: u8) -> StampedTagValue {
|
||||
StampedTagValue {
|
||||
value: [value; 32],
|
||||
clock,
|
||||
node: [node; 8],
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stamped_tag_encode_decode_round_trips() {
|
||||
let s = stamped_tag(9, 7, 4);
|
||||
let back = StampedTagValue::from_bytes(&s.to_bytes()).unwrap();
|
||||
assert_eq!(back, s);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stamped_record_encode_decode_round_trips() {
|
||||
let s = stamped_tag(0xEE, 42, 3);
|
||||
let bytes = encode_stamped_record("clawverse:main:latest", &s);
|
||||
let (k, back) = decode_stamped_record(&bytes).unwrap();
|
||||
assert_eq!(k, "clawverse:main:latest");
|
||||
assert_eq!(back, s);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stamped_record_rejects_truncated_input() {
|
||||
assert!(decode_stamped_record(&[]).is_err());
|
||||
assert!(decode_stamped_record(&[1u8]).is_err());
|
||||
// Length prefix says 4 but total is 2+4+48=54 needed.
|
||||
let bad = vec![4, 0, b'a', b'b'];
|
||||
assert!(decode_stamped_record(&bad).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stamped_tag_put_get_round_trip() {
|
||||
let (_tmp, store) = open();
|
||||
let k = "clawverse:main:latest";
|
||||
assert_eq!(store.get_stamped(k).await.unwrap(), None);
|
||||
let s = stamped_tag(1, 1, 1);
|
||||
assert_eq!(
|
||||
store.put_stamped(k, s).await.unwrap(),
|
||||
TagPutOutcome::Merged
|
||||
);
|
||||
assert_eq!(store.get_stamped(k).await.unwrap(), Some(s));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stamped_tag_merges_dominant_and_rejects_lower() {
|
||||
let (_tmp, store) = open();
|
||||
let k = "clawverse:main:latest";
|
||||
let base = stamped_tag(1, 10, 5);
|
||||
store.put_stamped(k, base).await.unwrap();
|
||||
|
||||
let older = stamped_tag(2, 9, 9);
|
||||
assert!(matches!(
|
||||
store.put_stamped(k, older).await.unwrap(),
|
||||
TagPutOutcome::Rejected { current } if current == base
|
||||
));
|
||||
assert_eq!(store.get_stamped(k).await.unwrap(), Some(base));
|
||||
|
||||
let higher = stamped_tag(3, 11, 0);
|
||||
assert_eq!(
|
||||
store.put_stamped(k, higher).await.unwrap(),
|
||||
TagPutOutcome::Merged
|
||||
);
|
||||
assert_eq!(store.get_stamped(k).await.unwrap(), Some(higher));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stamped_and_unstamped_stores_are_independent() {
|
||||
// put() writes to tags/, put_stamped() writes to tags-v2/.
|
||||
// No clobber between the two.
|
||||
let (_tmp, store) = open();
|
||||
let k = "clawverse:main:latest";
|
||||
store.put(k, &[0xAA; 32]).await.unwrap();
|
||||
let s = stamped_tag(0xBB, 1, 1);
|
||||
store.put_stamped(k, s).await.unwrap();
|
||||
assert_eq!(store.get(k).await.unwrap(), Some([0xAA; 32]));
|
||||
assert_eq!(store.get_stamped(k).await.unwrap(), Some(s));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user