Phase 4b follow-on: pin --ttl RPC + CLI #47
@@ -48,8 +48,9 @@ use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
|
|||||||
use crate::cluster::rpc::{
|
use crate::cluster::rpc::{
|
||||||
call_blob_get_parallel, call_blob_get_stream, call_blob_put_stream, call_blob_stat,
|
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_delete_tag, call_get_metrics, call_get_ref_versioned, call_peer_status,
|
||||||
call_put_ref_versioned, call_put_tag_versioned, prewarm_missing_chunks_between_parallel,
|
call_put_ref_versioned, call_put_tag_versioned, call_set_tag_expiry,
|
||||||
call_get_ref, call_get_tag, call_list_tags, call_put_ref, call_put_tag,
|
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};
|
use crate::cluster::transport::{NodeIdentity, QuicClient};
|
||||||
|
|
||||||
@@ -149,6 +150,13 @@ struct PinArgs {
|
|||||||
/// Human-readable tag to publish (e.g. `clawverse:main:latest`).
|
/// Human-readable tag to publish (e.g. `clawverse:main:latest`).
|
||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
name: String,
|
name: String,
|
||||||
|
/// Optional TTL for the pin. Accepts humantime durations
|
||||||
|
/// (`30d`, `1h`, `168h`, `2w`); when set, the tag expires at
|
||||||
|
/// `now + ttl` (wall clock). Omit for a permanent pin. Pass
|
||||||
|
/// `0` or `clear` to remove any prior TTL sidecar from an
|
||||||
|
/// existing pin without changing the value.
|
||||||
|
#[arg(long)]
|
||||||
|
ttl: Option<String>,
|
||||||
#[command(flatten)]
|
#[command(flatten)]
|
||||||
peer: PeerArgs,
|
peer: PeerArgs,
|
||||||
}
|
}
|
||||||
@@ -828,6 +836,20 @@ async fn cmd_pin(args: PinArgs) -> Result<()> {
|
|||||||
let _ =
|
let _ =
|
||||||
call_put_tag_versioned(&conn, &companion, &companion_stamped).await?;
|
call_put_tag_versioned(&conn, &companion, &companion_stamped).await?;
|
||||||
|
|
||||||
|
// Phase 4b follow-on (2026-07-13): TTL sidecar. Applied to both
|
||||||
|
// the primary tag and its `.fingerprint` companion so eviction
|
||||||
|
// treats them as one lifetime — otherwise prewarm could resurrect
|
||||||
|
// a stale companion after the primary expired.
|
||||||
|
let ttl_display = match args.ttl.as_deref() {
|
||||||
|
None => None,
|
||||||
|
Some(raw) => {
|
||||||
|
let expires_at = parse_ttl_to_absolute(raw)?;
|
||||||
|
call_set_tag_expiry(&conn, &args.name, expires_at).await?;
|
||||||
|
call_set_tag_expiry(&conn, &companion, expires_at).await?;
|
||||||
|
Some((raw.to_string(), expires_at))
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||||
client.shutdown().await;
|
client.shutdown().await;
|
||||||
|
|
||||||
@@ -835,9 +857,152 @@ async fn cmd_pin(args: PinArgs) -> Result<()> {
|
|||||||
println!("companion: {}", companion);
|
println!("companion: {}", companion);
|
||||||
println!("fingerprint: {}", fp);
|
println!("fingerprint: {}", fp);
|
||||||
println!("blob: {}", blob_id);
|
println!("blob: {}", blob_id);
|
||||||
|
match ttl_display {
|
||||||
|
Some((_, 0)) => println!("ttl: cleared"),
|
||||||
|
Some((raw, expires_at)) => {
|
||||||
|
println!("ttl: {raw} (expires_at unix={expires_at})")
|
||||||
|
}
|
||||||
|
None => {}
|
||||||
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Phase 4b follow-on: parse a human TTL string into an absolute
|
||||||
|
/// unix expiry. `"0"` / `"clear"` / `"none"` → 0 (clear-sidecar
|
||||||
|
/// sentinel). Otherwise the string is interpreted as a duration
|
||||||
|
/// added to the current wall clock.
|
||||||
|
fn parse_ttl_to_absolute(raw: &str) -> Result<u64> {
|
||||||
|
let trimmed = raw.trim();
|
||||||
|
if trimmed.is_empty() {
|
||||||
|
anyhow::bail!("--ttl is empty; omit the flag for a permanent pin");
|
||||||
|
}
|
||||||
|
if matches!(trimmed, "0" | "clear" | "none") {
|
||||||
|
return Ok(0);
|
||||||
|
}
|
||||||
|
let dur = parse_human_duration(trimmed)?;
|
||||||
|
let now = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.context("system clock is before 1970")?
|
||||||
|
.as_secs();
|
||||||
|
Ok(now.saturating_add(dur.as_secs()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Phase 4b follow-on: minimal humantime-style duration parser.
|
||||||
|
///
|
||||||
|
/// Accepts sequences of `<number><unit>` pairs, e.g. `1h30m`,
|
||||||
|
/// `2w`, `168h`. Units: `s`, `m` (minute), `h`, `d`, `w`. Case
|
||||||
|
/// insensitive on the unit letter. Kept in-tree so we don't pull
|
||||||
|
/// in a new dependency for a single CLI flag.
|
||||||
|
fn parse_human_duration(input: &str) -> Result<std::time::Duration> {
|
||||||
|
let bytes = input.as_bytes();
|
||||||
|
let mut i = 0usize;
|
||||||
|
let mut total_secs = 0u64;
|
||||||
|
let mut saw_any = false;
|
||||||
|
while i < bytes.len() {
|
||||||
|
// Skip whitespace between pairs.
|
||||||
|
while i < bytes.len() && bytes[i].is_ascii_whitespace() {
|
||||||
|
i += 1;
|
||||||
|
}
|
||||||
|
if i >= bytes.len() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let num_start = i;
|
||||||
|
while i < bytes.len() && bytes[i].is_ascii_digit() {
|
||||||
|
i += 1;
|
||||||
|
}
|
||||||
|
if i == num_start {
|
||||||
|
anyhow::bail!("--ttl {input:?}: expected digit at byte offset {i}");
|
||||||
|
}
|
||||||
|
let n: u64 = input[num_start..i]
|
||||||
|
.parse()
|
||||||
|
.with_context(|| format!("--ttl {input:?}: number {}", &input[num_start..i]))?;
|
||||||
|
if i >= bytes.len() {
|
||||||
|
anyhow::bail!("--ttl {input:?}: number {n} missing unit suffix (s/m/h/d/w)");
|
||||||
|
}
|
||||||
|
let unit = bytes[i].to_ascii_lowercase();
|
||||||
|
i += 1;
|
||||||
|
let mult: u64 = match unit {
|
||||||
|
b's' => 1,
|
||||||
|
b'm' => 60,
|
||||||
|
b'h' => 3600,
|
||||||
|
b'd' => 86_400,
|
||||||
|
b'w' => 7 * 86_400,
|
||||||
|
other => anyhow::bail!(
|
||||||
|
"--ttl {input:?}: unknown unit {:?}; use s/m/h/d/w",
|
||||||
|
other as char
|
||||||
|
),
|
||||||
|
};
|
||||||
|
total_secs = total_secs
|
||||||
|
.checked_add(n.checked_mul(mult).with_context(|| {
|
||||||
|
format!("--ttl {input:?}: overflow multiplying {n} * {mult}")
|
||||||
|
})?)
|
||||||
|
.with_context(|| format!("--ttl {input:?}: total overflow"))?;
|
||||||
|
saw_any = true;
|
||||||
|
}
|
||||||
|
if !saw_any {
|
||||||
|
anyhow::bail!("--ttl {input:?}: no duration components parsed");
|
||||||
|
}
|
||||||
|
Ok(std::time::Duration::from_secs(total_secs))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod ttl_parser_tests {
|
||||||
|
use super::parse_human_duration;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parses_single_unit_forms() {
|
||||||
|
assert_eq!(parse_human_duration("30s").unwrap(), Duration::from_secs(30));
|
||||||
|
assert_eq!(parse_human_duration("5m").unwrap(), Duration::from_secs(300));
|
||||||
|
assert_eq!(parse_human_duration("2h").unwrap(), Duration::from_secs(7200));
|
||||||
|
assert_eq!(parse_human_duration("1d").unwrap(), Duration::from_secs(86_400));
|
||||||
|
assert_eq!(parse_human_duration("1w").unwrap(), Duration::from_secs(604_800));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parses_compound_forms() {
|
||||||
|
assert_eq!(
|
||||||
|
parse_human_duration("1h30m").unwrap(),
|
||||||
|
Duration::from_secs(3600 + 1800)
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
parse_human_duration("2d12h").unwrap(),
|
||||||
|
Duration::from_secs(2 * 86_400 + 12 * 3600)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn accepts_case_insensitive_units() {
|
||||||
|
assert_eq!(parse_human_duration("5H").unwrap(), Duration::from_secs(5 * 3600));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn rejects_bad_input() {
|
||||||
|
assert!(parse_human_duration("").is_err());
|
||||||
|
assert!(parse_human_duration("abc").is_err());
|
||||||
|
assert!(parse_human_duration("10").is_err()); // missing unit
|
||||||
|
assert!(parse_human_duration("10x").is_err()); // unknown unit
|
||||||
|
assert!(parse_human_duration("h10").is_err()); // wrong order
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn absolute_zero_and_clear_map_to_sentinel() {
|
||||||
|
assert_eq!(super::parse_ttl_to_absolute("0").unwrap(), 0);
|
||||||
|
assert_eq!(super::parse_ttl_to_absolute("clear").unwrap(), 0);
|
||||||
|
assert_eq!(super::parse_ttl_to_absolute("none").unwrap(), 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn absolute_ttl_lands_in_the_future() {
|
||||||
|
let now = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_secs();
|
||||||
|
let exp = super::parse_ttl_to_absolute("1h").unwrap();
|
||||||
|
assert!(exp >= now + 3599 && exp <= now + 3601 + 5);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Field finding 2026-07-12: companion tag suffix used by `pin` to
|
/// Field finding 2026-07-12: companion tag suffix used by `pin` to
|
||||||
/// stash the fingerprint alongside the blob-id mapping. Prewarm reads
|
/// stash the fingerprint alongside the blob-id mapping. Prewarm reads
|
||||||
/// it to know what ref to publish downstream.
|
/// it to know what ref to publish downstream.
|
||||||
|
|||||||
@@ -171,6 +171,18 @@ pub enum Method {
|
|||||||
/// Reply: 48 bytes on hit, single-byte
|
/// Reply: 48 bytes on hit, single-byte
|
||||||
/// [`ErrorCode::NotFound`] on miss.
|
/// [`ErrorCode::NotFound`] on miss.
|
||||||
GetTagVersioned = 0x19,
|
GetTagVersioned = 0x19,
|
||||||
|
/// Phase 4b follow-on (2026-07-13): attach a TTL to a stamped
|
||||||
|
/// tag. Sidecar semantics live in [`TagStore::set_stamped_expiry`]
|
||||||
|
/// — `expires_at_unix == 0` clears any prior sidecar; otherwise
|
||||||
|
/// the value is absolute wall-clock seconds.
|
||||||
|
/// `payload`: `key_len:u16 (LE) || key_bytes || expires_at:u64 (LE)`.
|
||||||
|
/// Reply: single-byte `STREAM_STATUS_OK`.
|
||||||
|
SetTagExpiry = 0x1a,
|
||||||
|
/// Phase 4b follow-on: read the TTL sidecar for a stamped tag.
|
||||||
|
/// `payload`: raw tag key bytes.
|
||||||
|
/// Reply: 8 bytes (`u64` LE) on hit, single-byte
|
||||||
|
/// [`ErrorCode::NotFound`] when no sidecar is present.
|
||||||
|
GetTagExpiry = 0x1b,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Method {
|
impl Method {
|
||||||
@@ -203,6 +215,8 @@ impl Method {
|
|||||||
0x17 => Some(Method::GetRefVersionedLocal),
|
0x17 => Some(Method::GetRefVersionedLocal),
|
||||||
0x18 => Some(Method::PutTagVersioned),
|
0x18 => Some(Method::PutTagVersioned),
|
||||||
0x19 => Some(Method::GetTagVersioned),
|
0x19 => Some(Method::GetTagVersioned),
|
||||||
|
0x1a => Some(Method::SetTagExpiry),
|
||||||
|
0x1b => Some(Method::GetTagExpiry),
|
||||||
_ => None,
|
_ => None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -813,6 +827,38 @@ impl RpcRouter {
|
|||||||
Ok(HandlerOutcome::Error(ErrorCode::NotFound))
|
Ok(HandlerOutcome::Error(ErrorCode::NotFound))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Method::SetTagExpiry => {
|
||||||
|
let store = match &self.tag_store {
|
||||||
|
Some(s) => s,
|
||||||
|
None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
|
||||||
|
};
|
||||||
|
let (key, expires_at) =
|
||||||
|
match crate::cluster::tags::decode_expiry_record(payload) {
|
||||||
|
Ok(kv) => kv,
|
||||||
|
Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
|
||||||
|
};
|
||||||
|
match store.set_stamped_expiry(&key, expires_at).await {
|
||||||
|
Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(error = %e, "SetTagExpiry failed");
|
||||||
|
Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Method::GetTagExpiry => {
|
||||||
|
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_expiry(key).await? {
|
||||||
|
Some(exp) => Ok(HandlerOutcome::Reply(exp.to_le_bytes().to_vec())),
|
||||||
|
None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
|
||||||
|
}
|
||||||
|
}
|
||||||
Method::ListTags => {
|
Method::ListTags => {
|
||||||
let store = match &self.tag_store {
|
let store = match &self.tag_store {
|
||||||
Some(s) => s,
|
Some(s) => s,
|
||||||
@@ -1224,3 +1270,7 @@ mod tests_phase5;
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
#[path = "rpc/tests_forwarding.rs"]
|
#[path = "rpc/tests_forwarding.rs"]
|
||||||
mod tests_forwarding;
|
mod tests_forwarding;
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
#[path = "rpc/tests_phase4b_ttl.rs"]
|
||||||
|
mod tests_phase4b_ttl;
|
||||||
|
|||||||
@@ -598,6 +598,65 @@ pub async fn call_get_tag_versioned(
|
|||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Phase 4b follow-on: TTL client helpers ───────────────────────────
|
||||||
|
|
||||||
|
/// Phase 4b follow-on (2026-07-13): attach a TTL sidecar to a stamped
|
||||||
|
/// tag on a peer. `expires_at_unix == 0` clears any prior sidecar.
|
||||||
|
///
|
||||||
|
/// The peer accepts writes even when the stamped tag isn't present
|
||||||
|
/// yet — the sidecar sticks around and takes effect once the tag
|
||||||
|
/// lands (`TagStore::set_stamped_expiry` semantics).
|
||||||
|
pub async fn call_set_tag_expiry(
|
||||||
|
conn: &Connection,
|
||||||
|
key: &str,
|
||||||
|
expires_at_unix: u64,
|
||||||
|
) -> Result<()> {
|
||||||
|
let payload = crate::cluster::tags::encode_expiry_record(key, expires_at_unix);
|
||||||
|
let reply = rpc_call(conn, Method::SetTagExpiry, &payload).await?;
|
||||||
|
if reply.len() != 1 {
|
||||||
|
bail!(
|
||||||
|
"expected single-byte SetTagExpiry reply, got {} bytes",
|
||||||
|
reply.len()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
match reply[0] {
|
||||||
|
STREAM_STATUS_OK => Ok(()),
|
||||||
|
code => match decode_error(code) {
|
||||||
|
Some(err) => bail!("peer rejected SetTagExpiry: {}", err.describe()),
|
||||||
|
None => bail!(
|
||||||
|
"peer replied with unknown byte 0x{:02x} for SetTagExpiry",
|
||||||
|
code
|
||||||
|
),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Phase 4b follow-on: fetch the TTL sidecar for a stamped tag.
|
||||||
|
/// Returns `Ok(None)` when no sidecar is present (never expires or
|
||||||
|
/// no such tag).
|
||||||
|
pub async fn call_get_tag_expiry(
|
||||||
|
conn: &Connection,
|
||||||
|
key: &str,
|
||||||
|
) -> Result<Option<u64>> {
|
||||||
|
let reply = rpc_call(conn, Method::GetTagExpiry, 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() != 8 {
|
||||||
|
bail!(
|
||||||
|
"expected 8-byte GetTagExpiry reply, got {} bytes",
|
||||||
|
reply.len()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Ok(Some(u64::from_le_bytes(
|
||||||
|
reply.as_slice().try_into().expect("checked length"),
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
// ── Phase 5g: cache metrics client helper ────────────────────────────
|
// ── Phase 5g: cache metrics client helper ────────────────────────────
|
||||||
|
|
||||||
/// Fetch the peer's current cache-metrics snapshot.
|
/// Fetch the peer's current cache-metrics snapshot.
|
||||||
|
|||||||
@@ -0,0 +1,177 @@
|
|||||||
|
//! Phase 4b follow-on (2026-07-13): SetTagExpiry / GetTagExpiry RPC
|
||||||
|
//! end-to-end. Kept in its own file to stay under the 1300-line
|
||||||
|
//! ceiling and to keep TTL wiring in one place.
|
||||||
|
|
||||||
|
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;
|
||||||
|
|
||||||
|
static NEXT_PORT: AtomicU16 = AtomicU16::new(47001);
|
||||||
|
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_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 ttl_method_bytes_stable() {
|
||||||
|
assert_eq!(Method::SetTagExpiry.as_byte(), 0x1a);
|
||||||
|
assert_eq!(Method::GetTagExpiry.as_byte(), 0x1b);
|
||||||
|
for m in [Method::SetTagExpiry, Method::GetTagExpiry] {
|
||||||
|
assert_eq!(Method::from_byte(m.as_byte()), Some(m));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ttl_rpcs_return_not_configured_without_store() {
|
||||||
|
let gossip = bootstrap_gossip("solo", next_port()).await;
|
||||||
|
let router = RpcRouter::new(gossip, "solo".into(), "z".into());
|
||||||
|
// SetTagExpiry needs a valid record; GetTagExpiry needs a key.
|
||||||
|
let payload = crate::cluster::tags::encode_expiry_record("k", 42);
|
||||||
|
let mut set_req = vec![Method::SetTagExpiry.as_byte()];
|
||||||
|
set_req.extend_from_slice(&payload);
|
||||||
|
assert_eq!(
|
||||||
|
dispatch(&router, &set_req).await,
|
||||||
|
vec![ErrorCode::NotConfigured.as_byte()]
|
||||||
|
);
|
||||||
|
|
||||||
|
let mut get_req = vec![Method::GetTagExpiry.as_byte()];
|
||||||
|
get_req.extend_from_slice(b"k");
|
||||||
|
assert_eq!(
|
||||||
|
dispatch(&router, &get_req).await,
|
||||||
|
vec![ErrorCode::NotConfigured.as_byte()]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn end_to_end_set_and_get_tag_expiry() {
|
||||||
|
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";
|
||||||
|
|
||||||
|
// No sidecar yet.
|
||||||
|
assert_eq!(call_get_tag_expiry(&conn, key).await.unwrap(), None);
|
||||||
|
|
||||||
|
// Set an absolute expiry — no requirement that the stamped tag
|
||||||
|
// already exist (matches TagStore::set_stamped_expiry semantics).
|
||||||
|
call_set_tag_expiry(&conn, key, 1_800_000_000).await.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
call_get_tag_expiry(&conn, key).await.unwrap(),
|
||||||
|
Some(1_800_000_000)
|
||||||
|
);
|
||||||
|
|
||||||
|
// Overwrite with a later value.
|
||||||
|
call_set_tag_expiry(&conn, key, 1_900_000_000).await.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
call_get_tag_expiry(&conn, key).await.unwrap(),
|
||||||
|
Some(1_900_000_000)
|
||||||
|
);
|
||||||
|
|
||||||
|
// Clear (expires_at == 0).
|
||||||
|
call_set_tag_expiry(&conn, key, 0).await.unwrap();
|
||||||
|
assert_eq!(call_get_tag_expiry(&conn, key).await.unwrap(), None);
|
||||||
|
|
||||||
|
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||||
|
client.shutdown().await;
|
||||||
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||||
|
acc.abort();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ttl_survives_process_boundary_via_pin_flow() {
|
||||||
|
// Real pin flow: PutTagVersioned, then SetTagExpiry, then read
|
||||||
|
// both back through the same connection. Exercises the exact
|
||||||
|
// sequence the pin --ttl CLI will emit.
|
||||||
|
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:pr-42";
|
||||||
|
let stamped = StampedTagValue {
|
||||||
|
value: [0xAB; 32],
|
||||||
|
clock: 42,
|
||||||
|
node: node_stamp_for("runner-x"),
|
||||||
|
};
|
||||||
|
assert!(call_put_tag_versioned(&conn, key, &stamped).await.unwrap());
|
||||||
|
|
||||||
|
let expires_at = 2_000_000_000u64;
|
||||||
|
call_set_tag_expiry(&conn, key, expires_at).await.unwrap();
|
||||||
|
|
||||||
|
// Both surfaces round-trip.
|
||||||
|
assert_eq!(
|
||||||
|
call_get_tag_versioned(&conn, key).await.unwrap(),
|
||||||
|
Some(stamped)
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
call_get_tag_expiry(&conn, key).await.unwrap(),
|
||||||
|
Some(expires_at)
|
||||||
|
);
|
||||||
|
|
||||||
|
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||||
|
client.shutdown().await;
|
||||||
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||||
|
acc.abort();
|
||||||
|
}
|
||||||
@@ -125,6 +125,47 @@ pub fn decode_stamped_record(bytes: &[u8]) -> Result<(String, StampedTagValue)>
|
|||||||
Ok((key, stamped))
|
Ok((key, stamped))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Phase 4b follow-on (2026-07-13): wire encoding for
|
||||||
|
/// `SetTagExpiry` RPC — `key_len:u16 (LE) || key_bytes ||
|
||||||
|
/// expires_at:u64 (LE)`. `expires_at` is absolute wall-clock
|
||||||
|
/// seconds; `0` means "clear any existing sidecar".
|
||||||
|
pub fn encode_expiry_record(key: &str, expires_at_unix: u64) -> Vec<u8> {
|
||||||
|
let key_bytes = key.as_bytes();
|
||||||
|
let mut out = Vec::with_capacity(2 + key_bytes.len() + 8);
|
||||||
|
out.extend_from_slice(&(key_bytes.len() as u16).to_le_bytes());
|
||||||
|
out.extend_from_slice(key_bytes);
|
||||||
|
out.extend_from_slice(&expires_at_unix.to_le_bytes());
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Reverse of [`encode_expiry_record`].
|
||||||
|
pub fn decode_expiry_record(bytes: &[u8]) -> Result<(String, u64)> {
|
||||||
|
if bytes.len() < 2 {
|
||||||
|
bail!(
|
||||||
|
"expiry 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 + 8;
|
||||||
|
if bytes.len() != expected {
|
||||||
|
bail!(
|
||||||
|
"expiry 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("expiry record key is not valid UTF-8")?
|
||||||
|
.to_string();
|
||||||
|
let expires_at =
|
||||||
|
u64::from_le_bytes(bytes[2 + key_len..].try_into().expect("checked length"));
|
||||||
|
Ok((key, expires_at))
|
||||||
|
}
|
||||||
|
|
||||||
/// A tag entry as returned by `list`.
|
/// A tag entry as returned by `list`.
|
||||||
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||||
pub struct TagEntry {
|
pub struct TagEntry {
|
||||||
@@ -802,6 +843,29 @@ mod tests {
|
|||||||
assert!(decode_stamped_record(&bad).is_err());
|
assert!(decode_stamped_record(&bad).is_err());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn expiry_record_encode_decode_round_trips() {
|
||||||
|
let bytes = encode_expiry_record("clawverse:main:latest", 1_800_000_000);
|
||||||
|
let (k, exp) = decode_expiry_record(&bytes).unwrap();
|
||||||
|
assert_eq!(k, "clawverse:main:latest");
|
||||||
|
assert_eq!(exp, 1_800_000_000);
|
||||||
|
|
||||||
|
// Zero (clear-expiry sentinel) round-trips too.
|
||||||
|
let bytes = encode_expiry_record("k", 0);
|
||||||
|
let (k, exp) = decode_expiry_record(&bytes).unwrap();
|
||||||
|
assert_eq!(k, "k");
|
||||||
|
assert_eq!(exp, 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn expiry_record_rejects_malformed_input() {
|
||||||
|
assert!(decode_expiry_record(&[]).is_err());
|
||||||
|
assert!(decode_expiry_record(&[1u8]).is_err());
|
||||||
|
// key_len=4 → need 2+4+8=14 bytes; supply 6.
|
||||||
|
let bad = vec![4, 0, b'a', b'b', b'c', b'd'];
|
||||||
|
assert!(decode_expiry_record(&bad).is_err());
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn stamped_tag_put_get_round_trip() {
|
async fn stamped_tag_put_get_round_trip() {
|
||||||
let (_tmp, store) = open();
|
let (_tmp, store) = open();
|
||||||
|
|||||||
Reference in New Issue
Block a user