use std::io::{Read, Write};
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, Mutex};
static START_GATE: Mutex<()> = Mutex::new(());
fn req(parts: &[&[u8]]) -> Vec<u8> {
let mut v = format!("*{}\r\n", parts.len()).into_bytes();
for p in parts {
v.extend_from_slice(format!("${}\r\n", p.len()).as_bytes());
v.extend_from_slice(p);
v.extend_from_slice(b"\r\n");
}
v
}
fn cmd(s: &mut std::net::TcpStream, parts: &[&[u8]]) -> Vec<u8> {
s.write_all(&req(parts)).unwrap();
std::thread::sleep(std::time::Duration::from_millis(30));
let mut buf = [0u8; 65536];
let n = s.read(&mut buf).unwrap();
buf[..n].to_vec()
}
struct Server {
port: u16,
dir: std::path::PathBuf,
stop: Arc<AtomicBool>,
handle: Option<std::thread::JoinHandle<()>>,
}
impl Server {
fn start(tier_budget: Option<u64>) -> Self {
let _gate = START_GATE.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
let dir = std::env::temp_dir().join(format!(
"kevy-tablee2e-{}",
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()
));
std::fs::create_dir_all(&dir).unwrap();
let stop = Arc::new(AtomicBool::new(false));
let stop_thread = stop.clone();
let dir_thread = dir.clone();
let handle = std::thread::spawn(move || {
let rt = kevy_rt::Runtime::builder(kevy::KevyCommands::sharded(8))
.bind([127, 0, 0, 1], port)
.shards(8)
.with_data_dir(dir_thread)
.with_tier_budget(tier_budget);
rt.run(stop_thread).unwrap();
});
kevy_testnet::assert_listening(port, "the server under test");
Self { port, dir, stop, handle: Some(handle) }
}
fn connect(&self) -> std::net::TcpStream {
let s = std::net::TcpStream::connect(("127.0.0.1", self.port)).unwrap();
s.set_read_timeout(Some(std::time::Duration::from_secs(8))).unwrap();
s
}
}
impl Drop for Server {
fn drop(&mut self) {
self.stop.store(true, std::sync::atomic::Ordering::SeqCst);
let _ = std::net::TcpStream::connect(("127.0.0.1", self.port));
if let Some(h) = self.handle.take() {
let _ = h.join();
}
let _ = std::fs::remove_dir_all(&self.dir);
}
}
fn bulks(reply: &[u8]) -> Vec<Vec<u8>> {
let mut out = Vec::new();
let mut i = 0;
while i < reply.len() {
if reply[i] == b'$' {
let nl = reply[i..].windows(2).position(|w| w == b"\r\n").unwrap() + i;
if let Ok(n) = std::str::from_utf8(&reply[i + 1..nl]).unwrap().parse::<i64>()
&& n >= 0
{
let start = nl + 2;
out.push(reply[start..start + n as usize].to_vec());
i = start + n as usize + 2;
continue;
}
}
i += 1;
}
out
}
fn page_keys(reply: &[u8]) -> Vec<Vec<u8>> {
let b = bulks(reply);
b[1..].chunks(2).map(|p| p[0].clone()).collect()
}
fn labeled(fields: &[Vec<u8>], label: &[u8]) -> Vec<Vec<u8>> {
fields
.windows(2)
.filter(|w| w[0] == label)
.map(|w| w[1].clone())
.collect()
}
fn keys_eq(reply: &[u8], want: &[&str]) {
let got: Vec<String> =
page_keys(reply).iter().map(|k| String::from_utf8_lossy(k).into_owned()).collect();
assert_eq!(got, want.to_vec(), "reply: {}", String::from_utf8_lossy(reply));
}
fn info_gauge(s: &mut std::net::TcpStream, name: &str) -> u64 {
let reply = String::from_utf8_lossy(&cmd(s, &[b"INFO"])).into_owned();
reply
.lines()
.find_map(|l| l.strip_prefix(&format!("{name}:")))
.and_then(|v| v.trim().parse().ok())
.unwrap_or_else(|| panic!("INFO has no {name} gauge:\n{reply}"))
}
fn wait_cold_stable(s: &mut std::net::TcpStream) -> u64 {
let mut last = 0u64;
for _ in 0..100 {
std::thread::sleep(std::time::Duration::from_millis(150));
let now = info_gauge(s, "cold_keys");
if now > 0 && now == last {
return now;
}
last = now;
}
panic!("cold_keys never stabilized (last = {last})");
}
const DECLARE_USER: &[&[u8]] = &[
b"TABLE.DECLARE", b"user", b"PREFIX", b"u:", b"PK", b"id",
b"COLUMN", b"id", b"str", b"COLUMN", b"name", b"str", b"COLUMN", b"age", b"i64",
b"COLUMN", b"dept", b"str", b"COLUMN", b"email", b"str", b"COLUMN", b"deleted", b"i64",
b"INDEX", b"age", b"RANGE", b"VALUES", b"dept", b"name", b"deleted",
b"INDEX", b"dept", b"RANGE",
b"INDEX", b"email", b"UNIQUE",
b"ORDERPATH", b"by_dept_age", b"ON", b"dept", b"THEN", b"age", b"DESC",
];
const ROWS: &[(&str, &str, &str, &str, &str)] = &[
("u:1", "alice", "30", "eng", "a@x"),
("u:2", "bob", "45", "eng", "b@x"),
("u:3", "carol", "25", "ops", "c@x"),
("u:4", "dave", "38", "eng", "d@x"),
("u:5", "erin", "52", "ops", "e@x"),
("u:6", "frank", "61", "", "f@x"),
];
fn write_rows(s: &mut std::net::TcpStream) {
for (key, name, age, dept, email) in ROWS {
let mut argv: Vec<&[u8]> = vec![
b"HSET", key.as_bytes(), b"id", &key.as_bytes()[2..], b"name", name.as_bytes(),
b"age", age.as_bytes(), b"email", email.as_bytes(), b"deleted", b"0",
];
if !dept.is_empty() {
argv.push(b"dept");
argv.push(dept.as_bytes());
}
let reply = cmd(s, &argv);
assert!(reply.starts_with(b":"), "HSET {key}: {}", String::from_utf8_lossy(&reply));
}
}
fn wait_ready(s: &mut std::net::TcpStream, probe: &[&[u8]]) -> Vec<u8> {
for _ in 0..200 {
let reply = cmd(s, probe);
if !reply.starts_with(b"-INDEXBUILDING") {
return reply;
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
panic!("index never left BUILDING for {probe:?}");
}
fn setup_user(c: &mut std::net::TcpStream) {
assert_eq!(cmd(c, DECLARE_USER), b"+OK\r\n", "TABLE.DECLARE");
write_rows(c);
wait_ready(c, &[b"IDX.QUERY", b"user.age", b"EQ", b"0"]);
wait_ready(c, &[b"IDX.QUERY", b"user.dept", b"EQ", b"x"]);
wait_ready(c, &[b"IDX.QUERY", b"user.email", b"EQ", b"x"]);
wait_ready(c, &[b"IDX.QUERY", b"user.by_dept_age", b"WHERE", b"dept", b"EQ", b"x"]);
}
#[test]
fn r1_declare_round_trip_and_compiled_paths() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
let list = bulks(&cmd(&mut c, &[b"TABLE.LIST"]));
assert_eq!(labeled(&list, b"name"), vec![b"user".to_vec()]);
assert_eq!(labeled(&list, b"prefix"), vec![b"u:".to_vec()]);
assert_eq!(labeled(&list, b"pk"), vec![b"id".to_vec()]);
assert_eq!(labeled(&list, b"columns"), vec![b"6".to_vec()]);
assert_eq!(labeled(&list, b"indexes"), vec![b"3".to_vec()]);
assert_eq!(labeled(&list, b"orderpaths"), vec![b"1".to_vec()]);
let idx = bulks(&cmd(&mut c, &[b"IDX.LIST"]));
let names = labeled(&idx, b"name");
for want in ["user.age", "user.dept", "user.email", "user.by_dept_age"] {
assert!(names.iter().any(|n| n == want.as_bytes()), "IDX.LIST missing {want}");
}
assert_eq!(cmd(&mut c, &[b"TABLE.DROP", b"user"]), b":1\r\n");
let idx = bulks(&cmd(&mut c, &[b"IDX.LIST"]));
assert!(labeled(&idx, b"name").is_empty(), "compiled indexes must drop with the table");
}
#[test]
fn r2_r3_r4_point_range_compose_and_residual_filter() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
keys_eq(&cmd(&mut c, &[b"IDX.QUERY", b"user.age", b"RANGE", b"25", b"45"]),
&["u:3", "u:1", "u:4", "u:2"]);
keys_eq(&cmd(&mut c, &[b"IDX.QUERY", b"user.dept", b"EQ", b"ops"]), &["u:3", "u:5"]);
keys_eq(&cmd(&mut c, &[b"IDX.QUERY", b"user.email", b"EQ", b"d@x"]), &["u:4"]);
let and = cmd(&mut c, &[
b"IDX.QUERY", b"COMPOSE", b"AND", b"user.age", b"RANGE", b"30", b"50",
b"user.dept", b"EQ", b"eng",
]);
let and_keys: Vec<Vec<u8>> = bulks(&and)[1..].to_vec();
assert_eq!(and_keys, vec![b"u:1".to_vec(), b"u:2".to_vec(), b"u:4".to_vec()], "AND, key order");
let or = cmd(&mut c, &[
b"IDX.QUERY", b"COMPOSE", b"OR", b"user.age", b"RANGE", b"60", b"100",
b"user.dept", b"EQ", b"ops",
]);
let or_keys: Vec<Vec<u8>> = bulks(&or)[1..].to_vec();
assert_eq!(or_keys, vec![b"u:3".to_vec(), b"u:5".to_vec(), b"u:6".to_vec()]);
keys_eq(&cmd(&mut c, &[
b"IDX.QUERY", b"user.age", b"RANGE", b"0", b"100", b"FILTER", b"dept", b"EQ", b"eng",
]), &["u:1", "u:4", "u:2"]);
}
#[test]
fn r5_order_by_single_and_composite_orderpath() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
keys_eq(&cmd(&mut c, &[
b"IDX.QUERY", b"user.age", b"RANGE", b"0", b"100", b"SORT", b"name", b"ASC",
]), &["u:1", "u:2", "u:3", "u:4", "u:5", "u:6"]);
keys_eq(&cmd(&mut c, &[
b"IDX.QUERY", b"user.by_dept_age", b"WHERE", b"dept", b"EQ", b"eng",
]), &["u:2", "u:4", "u:1"]);
keys_eq(&cmd(&mut c, &[
b"IDX.QUERY", b"user.by_dept_age", b"WHERE", b"dept", b"EQ", b"eng",
b"RANGE", b"age", b"31", b"46",
]), &["u:2", "u:4"]);
keys_eq(&cmd(&mut c, &[b"IDX.QUERY", b"user.age", b"EQ", b"61"]), &["u:6"]);
assert_eq!(
cmd(&mut c, &[b"IDX.COUNT", b"user.by_dept_age", b"WHERE", b"dept", b"EQ", b"eng"]),
b":3\r\n"
);
assert_eq!(
cmd(&mut c, &[b"IDX.COUNT", b"user.by_dept_age", b"WHERE", b"dept", b"EQ", b"ops"]),
b":2\r\n"
);
}
#[test]
fn r6_limit_offset_two_pages_do_not_overlap() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
let p1 = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"user.age", b"RANGE", b"0", b"100", b"LIMIT", b"3",
]));
let p2 = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"user.age", b"RANGE", b"0", b"100", b"LIMIT", b"3", b"OFFSET", b"3",
]));
assert_eq!(p1.len(), 3);
assert_eq!(p2.len(), 3);
assert!(p1.iter().all(|k| !p2.contains(k)), "pages must not overlap: {p1:?} vs {p2:?}");
let mut union: Vec<Vec<u8>> = p1.into_iter().chain(p2).collect();
union.sort();
assert_eq!(union.len(), 6, "the two pages must cover the whole range");
}
#[test]
fn r7_count_and_the_agg_kind_is_a_named_refusal() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
assert_eq!(cmd(&mut c, &[b"IDX.COUNT", b"user.age", b"RANGE", b"0", b"100"]), b":6\r\n");
assert_eq!(cmd(&mut c, &[b"IDX.COUNT", b"user.dept", b"EQ", b"eng"]), b":3\r\n");
assert_eq!(
cmd(&mut c, &[
b"TABLE.DECLARE", b"t2", b"PREFIX", b"t2:", b"PK", b"id",
b"COLUMN", b"id", b"str", b"COLUMN", b"n", b"i64", b"INDEX", b"n", b"agg",
]),
b"-ERR INDEX kind must be range|unique\r\n"
);
}
#[test]
fn r8_via_lookup_over_a_compiled_index_unchanged() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
assert_eq!(
cmd(&mut c, &[
b"VIEW.CREATE", b"engers", b"QUERY", b"user.dept", b"EQ", b"eng",
b"ORDER", b"BY", b"user.age", b"VIA", b"{key}",
]),
b"+OK\r\n"
);
let reply = cmd(&mut c, &[b"VIEW.QUERY", b"engers", b"FIELDS", b"name"]);
let b = bulks(&reply);
for name in [b"alice".as_slice(), b"bob", b"dave"] {
assert!(b.iter().any(|v| v == name), "VIA hydration missing {}: {}",
String::from_utf8_lossy(name), String::from_utf8_lossy(&reply));
}
}
#[test]
fn r10_optimistic_lock_via_watch_multi_exec() {
let srv = Server::start(None);
let mut c1 = srv.connect();
let mut c2 = srv.connect();
setup_user(&mut c1);
assert_eq!(cmd(&mut c1, &[b"WATCH", b"u:1"]), b"+OK\r\n");
assert!(cmd(&mut c1, &[b"HGET", b"u:1", b"age"]).starts_with(b"$2\r\n30"));
assert_eq!(cmd(&mut c1, &[b"MULTI"]), b"+OK\r\n");
assert_eq!(cmd(&mut c1, &[b"HSET", b"u:1", b"age", b"31"]), b"+QUEUED\r\n");
let exec = cmd(&mut c1, &[b"EXEC"]);
assert!(exec.starts_with(b"*1\r\n"), "clean EXEC applies: {}", String::from_utf8_lossy(&exec));
assert_eq!(cmd(&mut c1, &[b"WATCH", b"u:2"]), b"+OK\r\n");
assert!(cmd(&mut c2, &[b"HSET", b"u:2", b"age", b"46"]).starts_with(b":"));
assert_eq!(cmd(&mut c1, &[b"MULTI"]), b"+OK\r\n");
assert_eq!(cmd(&mut c1, &[b"HSET", b"u:2", b"age", b"99"]), b"+QUEUED\r\n");
assert_eq!(cmd(&mut c1, &[b"EXEC"]), b"*-1\r\n", "conflicted EXEC must abort");
wait_ready(&mut c1, &[b"IDX.QUERY", b"user.age", b"EQ", b"46"]);
keys_eq(&cmd(&mut c1, &[b"IDX.QUERY", b"user.age", b"EQ", b"46"]), &["u:2"]);
}
#[test]
fn r11_uniqueness_sequence_and_soft_delete_recipes() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
assert!(cmd(&mut c, &[b"HSET", b"u:2", b"email", b"a@x"]).starts_with(b":"));
keys_eq(&cmd(&mut c, &[b"IDX.QUERY", b"user.email", b"EQ", b"a@x"]), &["u:1", "u:2"]);
for i in 7..14u32 {
let key = format!("u:{i}");
let reply = cmd(&mut c, &[b"HSET", key.as_bytes(), b"id", format!("{i}").as_bytes(), b"email", b"a@x"]);
assert!(reply.starts_with(b":"), "HSET {key}");
}
let verify = bulks(&cmd(&mut c, &[b"IDX.VERIFY", b"user.email"]));
let dup: u64 = String::from_utf8_lossy(&labeled(&verify, b"duplicates")[0]).parse().unwrap();
assert!(dup >= 1, "the fence must report a same-shard duplicate, got {dup}");
assert_eq!(cmd(&mut c, &[b"INCR", b"user:seq"]), b":1\r\n");
assert_eq!(cmd(&mut c, &[b"INCR", b"user:seq"]), b":2\r\n");
assert!(cmd(&mut c, &[b"HSET", b"u:5", b"deleted", b"1"]).starts_with(b":"));
keys_eq(&cmd(&mut c, &[
b"IDX.QUERY", b"user.age", b"RANGE", b"0", b"100", b"FILTER", b"deleted", b"EQ", b"0",
]), &["u:3", "u:1", "u:4", "u:2", "u:6"]);
}
#[test]
fn c2_table_verify_clean_after_declare_and_writes() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
let verify = bulks(&cmd(&mut c, &[b"TABLE.VERIFY", b"user"]));
let names = labeled(&verify, b"index");
assert_eq!(names.len(), 4, "one element per compiled index");
for d in labeled(&verify, b"drift") {
assert_eq!(d, b"0".to_vec(), "a hook-maintained compiled index must not drift");
}
assert_eq!(labeled(&verify, b"spotcheck_rows"), vec![b"6".to_vec()]);
assert_eq!(labeled(&verify, b"spotcheck_type_mismatches"), vec![b"0".to_vec()]);
assert_eq!(
labeled(&verify, b"entries"),
vec![b"6".to_vec(), b"5".to_vec(), b"6".to_vec(), b"5".to_vec()]
);
assert!(cmd(&mut c, &[b"HSET", b"u:9", b"id", b"9", b"age", b"not-a-number"]).starts_with(b":"));
let verify = bulks(&cmd(&mut c, &[b"TABLE.VERIFY", b"user"]));
assert_eq!(labeled(&verify, b"spotcheck_type_mismatches"), vec![b"1".to_vec()]);
assert_eq!(labeled(&verify, b"rows"), vec![b"7".to_vec(); 4], "every prefix row walked");
assert_eq!(
labeled(&verify, b"coerce_failures"),
vec![b"1".to_vec(), b"0".to_vec(), b"0".to_vec(), b"0".to_vec()],
"u:9's age is present-but-not-i64 — on the age index only"
);
assert_eq!(
labeled(&verify, b"absent"),
vec![b"0".to_vec(), b"2".to_vec(), b"1".to_vec(), b"2".to_vec()],
"u:6 and u:9 lack dept, u:9 lacks email — counted as NULL, not coercion"
);
assert_eq!(labeled(&verify, b"missing"), vec![b"0".to_vec(); 4], "no forgotten writer");
}
#[test]
fn r9_c3_the_refusal_surface_errors_by_name() {
let srv = Server::start(None);
let mut c = srv.connect();
setup_user(&mut c);
assert!(cmd(&mut c, &[b"INSERT", b"INTO", b"user"]).starts_with(b"-ERR unknown command"));
assert!(cmd(&mut c, &[b"JOIN", b"user", b"orders"]).starts_with(b"-ERR unknown command"));
assert!(cmd(&mut c, &[b"HAVING", b"count"]).starts_with(b"-ERR unknown command"));
assert!(cmd(&mut c, &[b"SELECT", b"*", b"FROM", b"user"]).starts_with(b"-ERR"));
assert_eq!(
cmd(&mut c, &[b"TABLE.DECLARE", b"t3", b"PREFIX", b"t3:", b"PK", b"id",
b"COLUMN", b"id", b"str", b"INDEX", b"ghost", b"RANGE"]),
b"-ERR INDEX names unknown column 'ghost'\r\n"
);
assert_eq!(
cmd(&mut c, &[b"TABLE.DECLARE", b"t3", b"PREFIX", b"t3:", b"PK", b"id",
b"COLUMN", b"id", b"uuid"]),
b"-ERR COLUMN type must be i64|f64|str\r\n"
);
let r = cmd(&mut c, &[b"IDX.QUERY", b"user.age", b"WHERE", b"dept", b"EQ", b"eng"]);
assert!(
r.starts_with(b"-ERR IDX.QUERY 'user.age': WHERE requires a composite index"),
"{}",
String::from_utf8_lossy(&r)
);
let r = cmd(&mut c, &[b"IDX.QUERY", b"user.by_dept_age", b"WHERE", b"age", b"EQ", b"30"]);
assert!(
r.starts_with(b"-ERR IDX.QUERY 'user.by_dept_age': WHERE columns must be a leading prefix"),
"{}",
String::from_utf8_lossy(&r)
);
}
#[test]
fn c6_index_only_queries_touch_zero_cold_rows() {
let srv = Server::start(Some(256 * 1024));
let mut c = srv.connect();
let pad = vec![b'p'; 8192];
for i in 0..60u32 {
let key = format!("row:{i:02}");
let n = i.to_string();
let cval = if i % 2 == 0 { "even" } else { "odd" };
let reply = cmd(&mut c, &[
b"HSET", key.as_bytes(), b"id", n.as_bytes(), b"n", n.as_bytes(),
b"c", cval.as_bytes(), b"pad", &pad,
]);
assert!(reply.starts_with(b":"), "HSET {key}");
}
let cold_keys = wait_cold_stable(&mut c);
assert!(cold_keys > 8, "need most rows cold, got {cold_keys}");
assert_eq!(
cmd(&mut c, &[
b"TABLE.DECLARE", b"bench", b"PREFIX", b"row:", b"PK", b"id",
b"COLUMN", b"id", b"str", b"COLUMN", b"n", b"i64", b"COLUMN", b"c", b"str",
b"INDEX", b"n", b"RANGE", b"VALUES", b"c",
]),
b"+OK\r\n"
);
wait_ready(&mut c, &[b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"0"]);
std::thread::sleep(std::time::Duration::from_millis(300)); assert_eq!(info_gauge(&mut c, "promotions_total"), 0, "backfill must not promote");
let pre = info_gauge(&mut c, "peek_preads_total");
let filtered = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"100", b"FILTER", b"c", b"EQ", b"even",
b"LIMIT", b"100",
]));
assert_eq!(filtered.len(), 30);
let sorted = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"100", b"SORT", b"c", b"ASC", b"LIMIT", b"10",
]));
assert_eq!(sorted.len(), 10);
assert_eq!(cmd(&mut c, &[b"IDX.COUNT", b"bench.n", b"RANGE", b"0", b"100"]), b":60\r\n");
std::thread::sleep(std::time::Duration::from_millis(300));
assert_eq!(
info_gauge(&mut c, "peek_preads_total") - pre,
0,
"index-only queries must touch ZERO rows (the D2 shape)"
);
assert_eq!(info_gauge(&mut c, "promotions_total"), 0);
let pre = info_gauge(&mut c, "peek_preads_total");
let page = cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"9", b"LIMIT", b"10", b"FIELDS", b"c", b"n",
]);
let returned = bulks(&page).iter().filter(|b| b.starts_with(b"row:")).count() as u64;
assert_eq!(returned, 10);
std::thread::sleep(std::time::Duration::from_millis(300));
let preads = info_gauge(&mut c, "peek_preads_total") - pre;
assert!(preads <= returned, "one read per ROW at most: {preads} for {returned} rows");
assert!(preads > 0, "a mostly-cold page must have paid some reads");
assert_eq!(info_gauge(&mut c, "promotions_total"), 0, "hydration is not an access signal");
}
#[test]
fn d2_fully_cold_table_index_only_queries_pay_zero_cold_reads() {
let srv = Server::start(Some(256 * 1024));
let mut c = srv.connect();
let pad = vec![b'p'; 8192];
for i in 0..60u32 {
let key = format!("row:{i:02}");
let n = i.to_string();
let cval = if i % 2 == 0 { "even" } else { "odd" };
let reply = cmd(&mut c, &[
b"HSET", key.as_bytes(), b"id", n.as_bytes(), b"n", n.as_bytes(),
b"c", cval.as_bytes(), b"pad", &pad,
]);
assert!(reply.starts_with(b":"), "HSET {key}");
}
assert_eq!(
cmd(&mut c, &[
b"TABLE.DECLARE", b"bench", b"PREFIX", b"row:", b"PK", b"id",
b"COLUMN", b"id", b"str", b"COLUMN", b"n", b"i64", b"COLUMN", b"c", b"str",
b"INDEX", b"n", b"RANGE", b"VALUES", b"c",
]),
b"+OK\r\n"
);
wait_ready(&mut c, &[b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"0"]);
assert_eq!(cmd(&mut c, &[b"CONFIG", b"SET", b"tiering-budget", b"1"]), b"+OK\r\n");
let mut cold = 0;
for _ in 0..200 {
std::thread::sleep(std::time::Duration::from_millis(50));
cold = info_gauge(&mut c, "cold_keys");
if cold >= 60 {
break;
}
}
assert!(cold >= 60, "every row must be cold, got {cold}");
let pre_preads = info_gauge(&mut c, "peek_preads_total");
let pre_promo = info_gauge(&mut c, "promotions_total");
let filtered = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"100", b"FILTER", b"c", b"EQ", b"even",
b"LIMIT", b"100",
]));
assert_eq!(filtered.len(), 30);
let sorted = page_keys(&cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"100", b"SORT", b"c", b"DESC", b"LIMIT", b"10",
]));
assert_eq!(sorted.len(), 10);
assert_eq!(cmd(&mut c, &[b"IDX.COUNT", b"bench.n", b"RANGE", b"0", b"100"]), b":60\r\n");
std::thread::sleep(std::time::Duration::from_millis(300));
assert_eq!(
info_gauge(&mut c, "peek_preads_total") - pre_preads,
0,
"index-only queries on a FULLY-cold table must touch zero rows"
);
assert_eq!(info_gauge(&mut c, "promotions_total") - pre_promo, 0);
let pre = info_gauge(&mut c, "peek_preads_total");
let page = cmd(&mut c, &[
b"IDX.QUERY", b"bench.n", b"RANGE", b"0", b"9", b"LIMIT", b"10", b"FIELDS", b"c", b"n",
]);
let returned = bulks(&page).iter().filter(|b| b.starts_with(b"row:")).count() as u64;
assert_eq!(returned, 10);
std::thread::sleep(std::time::Duration::from_millis(300));
assert_eq!(
info_gauge(&mut c, "peek_preads_total") - pre,
returned,
"fully-cold: exactly one read per returned row"
);
assert_eq!(info_gauge(&mut c, "promotions_total") - pre_promo, 0);
}
#[test]
fn ensure_and_replace_are_the_boot_verbs() {
let srv = Server::start(None);
let mut c = srv.connect();
let ensure_user: Vec<&[u8]> = {
let mut v = DECLARE_USER.to_vec();
v[0] = b"TABLE.ENSURE";
v
};
assert_eq!(cmd(&mut c, &ensure_user), b"+OK\r\n", "first boot declares");
assert_eq!(cmd(&mut c, &ensure_user), b"+UNCHANGED\r\n", "every later boot is a no-op");
let mut changed = DECLARE_USER.to_vec();
changed[0] = b"TABLE.ENSURE";
changed.extend_from_slice(&[b"COLUMN", b"extra", b"i64"]);
let r = cmd(&mut c, &changed);
let s = String::from_utf8_lossy(&r);
assert!(s.starts_with("-ERR"), "changed spec must refuse: {s}");
assert!(s.contains("COLUMNS"), "the refusal names what changed: {s}");
let bad: &[&[u8]] = &[
b"TABLE.REPLACE", b"user", b"PREFIX", b"u:", b"PK", b"id",
b"COLUMN", b"id", b"str",
b"ORDERPATH", b"by_ghost", b"ON", b"ghost",
];
let r = cmd(&mut c, bad);
assert!(r.starts_with(b"-ERR"), "bad replacement refuses");
let list = cmd(&mut c, &[b"TABLE.LIST"]);
assert!(
String::from_utf8_lossy(&list).contains("user"),
"the old table must still stand after a refused REPLACE"
);
let mut good = DECLARE_USER.to_vec();
good[0] = b"TABLE.REPLACE";
good.extend_from_slice(&[b"COLUMN", b"extra", b"i64"]);
assert_eq!(cmd(&mut c, &good), b"+OK\r\n", "explicit rebuild");
}