clawstor/claw-store/src/cluster/rpc/tests_phase4b_ttl.rs

//! 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();
}