mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-19 16:16:45 +00:00
fix(redis): allow hash operations on Kvrocks legacy encoding
Closes #10948
This commit is contained in:
@@ -3433,7 +3433,10 @@ where
|
||||
local source_ttl = -2
|
||||
local ttl_reply = redis.pcall('HTTL', key, 'FIELDS', 1, old_field)
|
||||
if type(ttl_reply) == 'table' and ttl_reply.err ~= nil then
|
||||
if not optional_expiry_error(ttl_reply) then
|
||||
local detail = string.lower(ttl_reply.err)
|
||||
if not optional_expiry_error(ttl_reply)
|
||||
and detail ~= 'invalid argument: hash field expiration is not supported by legacy hash encoding'
|
||||
and detail ~= 'err invalid argument: hash field expiration is not supported by legacy hash encoding' then
|
||||
return -3
|
||||
end
|
||||
elseif type(ttl_reply) == 'table' and ttl_reply[1] ~= nil then
|
||||
@@ -4499,6 +4502,7 @@ fn is_optional_hash_field_expiry_error(error: &redis::RedisError) -> bool {
|
||||
|| detail.contains("syntax error")
|
||||
|| detail.contains("noperm")
|
||||
|| detail.contains("no permission")
|
||||
|| detail == "invalid argument: hash field expiration is not supported by legacy hash encoding"
|
||||
}
|
||||
|
||||
fn blob_matches_query(blob: &RedisBlob, lowered_query: &str) -> bool {
|
||||
@@ -6858,6 +6862,337 @@ mod tests {
|
||||
assert_eq!(con.command_count("HEXPIRE"), 1);
|
||||
}
|
||||
|
||||
fn legacy_hash_expiry_error() -> redis::RedisError {
|
||||
redis::parse_redis_value(
|
||||
b"-ERR Invalid argument: hash field expiration is not supported by legacy hash encoding\r\n",
|
||||
)
|
||||
.unwrap()
|
||||
.extract_error()
|
||||
.unwrap_err()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_value_keeps_bounded_pages_and_key_ttl() {
|
||||
let pairs = (0..201).map(|index| (format!("f{index}"), format!("v{index}"))).collect();
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Ok(bulk("hash")),
|
||||
Ok(RedisRawValue::Int(73)),
|
||||
Ok(RedisRawValue::Int(201)),
|
||||
Ok(hscan_response_owned("0", pairs)),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
]);
|
||||
|
||||
let value = super::get_value(&mut con, b"legacy-paged-hash").await.unwrap();
|
||||
assert_eq!(value.ttl, 73);
|
||||
let RedisValueData::Hash { items, total, scan_cursor } = value.data else {
|
||||
panic!("expected hash");
|
||||
};
|
||||
assert_eq!(total, 201);
|
||||
assert_eq!(items.len(), 200);
|
||||
assert!(items.iter().all(|item| item.field_ttl.is_none()));
|
||||
assert_eq!(items[199].field, text_blob("f199"));
|
||||
let page =
|
||||
super::load_more_collection(&mut con, b"legacy-paged-hash", "hash", scan_cursor.unwrap(), 200, None, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let RedisCollectionPage::Hash { items, scan_cursor } = page else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(items, vec![RedisHashItem { field: text_blob("f200"), value: text_blob("v200"), field_ttl: None }]);
|
||||
assert_eq!(scan_cursor, None);
|
||||
assert_eq!(con.command_count("HSCAN"), 1);
|
||||
assert_eq!(con.command_count("HTTL"), 2);
|
||||
assert!(con.commands[4].contains("\r\n200\r\n"));
|
||||
assert!(!con.commands[4].contains("\r\nf200\r\n"));
|
||||
assert!(con.responses.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_load_more_does_not_disable_ttl_for_other_hashes() {
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Ok(hscan_response("17", vec![("field", "value")])),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Ok(hscan_response("0", vec![("field", "value")])),
|
||||
Ok(httl_response(1, 42)),
|
||||
]);
|
||||
for (key, cursor, expected_ttl) in [(b"legacy".as_slice(), Some(17), None), (b"modern", None, Some(42))] {
|
||||
let page = super::load_more_collection(&mut con, key, "hash", 0, 1, None, None).await.unwrap();
|
||||
let RedisCollectionPage::Hash { items, scan_cursor } = page else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(scan_cursor, cursor);
|
||||
assert_eq!(
|
||||
items,
|
||||
vec![RedisHashItem { field: text_blob("field"), value: text_blob("value"), field_ttl: expected_ttl }]
|
||||
);
|
||||
}
|
||||
assert_eq!(con.command_count("HTTL"), 2);
|
||||
assert!(con.responses.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_set_does_not_disable_ttl_for_other_hashes() {
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Ok(RedisRawValue::Okay),
|
||||
Ok(httl_response(1, 42)),
|
||||
Ok(RedisRawValue::Okay),
|
||||
Ok(httl_response(1, 1)),
|
||||
]);
|
||||
super::hash_set(&mut con, b"legacy", "field", "value", None).await.unwrap();
|
||||
super::hash_set(&mut con, b"modern", "field", "value", None).await.unwrap();
|
||||
assert_eq!(con.command_count("HTTL"), 2);
|
||||
assert_eq!(con.command_count("HSET"), 2);
|
||||
assert_eq!(con.command_count("HEXPIRE"), 1);
|
||||
assert_eq!(con.command_count("EXPIRE"), 0);
|
||||
assert!(con.commands[4].contains("\r\nmodern\r\n$2\r\n42\r\n"));
|
||||
assert!(con.responses.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hash_ttl_discovery_keeps_unrelated_errors_fatal() {
|
||||
for detail in [
|
||||
"operation not supported",
|
||||
"Invalid argument: hash field expiration is not supported by another encoding",
|
||||
"WRONGTYPE Operation against a key holding the wrong kind of value",
|
||||
"Invalid argument: hash field expiration is not supported by legacy hash encoding: extra failure",
|
||||
] {
|
||||
let error =
|
||||
|| redis::RedisError::from((redis::ErrorKind::ResponseError, "server error", detail.to_string()));
|
||||
let mut con = FakeRedisConnection::with_results(vec![Err(error())]);
|
||||
assert!(super::hash_set(&mut con, b"hash", "field", "value", None).await.unwrap_err().contains(detail));
|
||||
assert_eq!(con.command_count("HSET"), 0);
|
||||
let mut con =
|
||||
FakeRedisConnection::with_results(vec![Ok(hscan_response("0", vec![("f", "v")])), Err(error())]);
|
||||
assert!(super::load_more_collection(&mut con, b"hash", "hash", 0, 1, None, None)
|
||||
.await
|
||||
.unwrap_err()
|
||||
.contains(detail));
|
||||
}
|
||||
let disconnected =
|
||||
|| redis::RedisError::from(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "connection reset"));
|
||||
let mut con = FakeRedisConnection::with_results(vec![Err(disconnected())]);
|
||||
assert!(super::hash_set(&mut con, b"hash", "field", "value", None).await.is_err());
|
||||
assert_eq!(con.command_count("HSET"), 0);
|
||||
let mut con =
|
||||
FakeRedisConnection::with_results(vec![Ok(hscan_response("0", vec![("f", "v")])), Err(disconnected())]);
|
||||
assert!(super::load_more_collection(&mut con, b"hash", "hash", 0, 1, None, None).await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_explicit_expiry_errors_remain_visible() {
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
]);
|
||||
assert!(super::set_hash_field_ttl(&mut con, b"hash", "field", 60).await.is_err());
|
||||
assert!(super::set_hash_field_ttl(&mut con, b"hash", "field", -1).await.is_err());
|
||||
assert!(super::set_hash_field_expire_at(&mut con, b"hash", "field", 1_735_689_600).await.is_err());
|
||||
assert_eq!(con.command_count("HEXPIRE"), 1);
|
||||
assert_eq!(con.command_count("HPERSIST"), 1);
|
||||
assert_eq!(con.command_count("HEXPIREAT"), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hash_set_keeps_explicit_key_ttl_and_missing_field_behavior() {
|
||||
for ttl in [Some(60), Some(0), Some(-1)] {
|
||||
let mut con = FakeRedisConnection::new(vec![RedisRawValue::Okay, RedisRawValue::Okay]);
|
||||
super::hash_set(&mut con, b"hash", "field", "value", ttl).await.unwrap();
|
||||
assert_eq!(con.command_count("HTTL"), 0);
|
||||
assert_eq!(con.command_count("HSET"), 1);
|
||||
assert_eq!(con.command_count("EXPIRE"), usize::from(ttl == Some(60)));
|
||||
}
|
||||
for response in [httl_response(1, -2), httl_response(1, -1), httl_response(0, -2)] {
|
||||
let mut con = FakeRedisConnection::new(vec![response, RedisRawValue::Okay]);
|
||||
super::hash_set(&mut con, b"hash", "field", "value", None).await.unwrap();
|
||||
assert_eq!(con.command_count("HSET"), 1);
|
||||
assert_eq!(con.command_count("HEXPIRE"), 0);
|
||||
}
|
||||
let mut con = FakeRedisConnection::new(vec![hscan_response("0", vec![])]);
|
||||
let page = super::load_more_collection(&mut con, b"hash", "hash", 0, 1, None, None).await.unwrap();
|
||||
assert!(matches!(page, RedisCollectionPage::Hash { items, scan_cursor: None } if items.is_empty()));
|
||||
assert_eq!(con.command_count("HTTL"), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_update_handles_encoding_only_in_discovery() {
|
||||
let mut con = FakeRedisConnection::new(vec![RedisRawValue::Int(2)]);
|
||||
super::hash_field_update(&mut con, b"hash", "old", "new", "value").await.unwrap();
|
||||
assert_eq!(con.commands.len(), 1);
|
||||
assert_eq!(con.command_count("EVAL"), 1);
|
||||
let script = &con.commands[0];
|
||||
let discovery =
|
||||
script.split("local ttl_reply =").nth(1).unwrap().split("local function restore_source").next().unwrap();
|
||||
assert!(discovery.contains("hash field expiration is not supported by legacy hash encoding"));
|
||||
let writes = script.split("local function restore_source").nth(1).unwrap();
|
||||
assert!(!writes.contains("legacy hash encoding"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_pages_keep_byte_and_sparse_scan_limits() {
|
||||
let value = "v".repeat(super::COLLECTION_PAGE_MAX_BYTES / 2);
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Ok(hscan_response("0", vec![("first", &value), ("second", &value)])),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
]);
|
||||
let mut cursor = 0;
|
||||
for field in ["first", "second"] {
|
||||
let page = super::load_more_collection(&mut con, b"legacy-byte-limit", "hash", cursor, 200, None, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let RedisCollectionPage::Hash { items, scan_cursor } = page else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(items.len(), 1);
|
||||
assert_eq!(items[0].field, text_blob(field));
|
||||
assert_eq!(items[0].value, text_blob(&value));
|
||||
assert!(items[0].field_ttl.is_none());
|
||||
assert!(super::hash_item_page_bytes(&items[0]) <= super::COLLECTION_PAGE_MAX_BYTES);
|
||||
cursor = scan_cursor.unwrap_or(0);
|
||||
}
|
||||
assert_eq!(cursor, 0);
|
||||
assert_eq!(con.command_count("HSCAN"), 1);
|
||||
assert_eq!(con.command_count("HTTL"), 2);
|
||||
|
||||
let mut responses = (1..=super::COLLECTION_FILTER_SCAN_MAX_ITERATIONS)
|
||||
.map(|cursor| Ok(hscan_response(&cursor.to_string(), vec![("field", "unmatched")])))
|
||||
.collect::<Vec<_>>();
|
||||
*responses.last_mut().unwrap() = Ok(hscan_response("10", vec![("field", "wanted")]));
|
||||
responses.push(Err(legacy_hash_expiry_error()));
|
||||
let mut con = FakeRedisConnection::with_results(responses);
|
||||
let page =
|
||||
super::load_more_collection(&mut con, b"legacy-sparse", "hash", 0, 20, Some("wanted"), None).await.unwrap();
|
||||
let RedisCollectionPage::Hash { items, scan_cursor } = page else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(items.len(), 1);
|
||||
assert_eq!(items[0].value, text_blob("wanted"));
|
||||
assert!(items[0].field_ttl.is_none());
|
||||
assert_eq!(scan_cursor, Some(10));
|
||||
assert_eq!(con.command_count("HSCAN"), 10);
|
||||
assert_eq!(con.command_count("HTTL"), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hash_ttl_discovery_preserves_rust_permission_fallback() {
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Err(noperm("httl")),
|
||||
Ok(RedisRawValue::Okay),
|
||||
Ok(hscan_response("0", vec![("f", "v")])),
|
||||
Err(noperm("httl")),
|
||||
]);
|
||||
super::hash_set(&mut con, b"hash", "field", "value", None).await.unwrap();
|
||||
let page = super::load_more_collection(&mut con, b"hash", "hash", 0, 1, None, None).await.unwrap();
|
||||
assert!(matches!(page, RedisCollectionPage::Hash { items, .. } if items[0].field_ttl.is_none()));
|
||||
assert_eq!(con.command_count("HTTL"), 2);
|
||||
assert_eq!(con.command_count("HSET"), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_hash_fallback_keeps_write_and_scan_errors_visible() {
|
||||
let mut con = FakeRedisConnection::with_results(vec![Err(legacy_hash_expiry_error()), Err(noperm("hset"))]);
|
||||
assert!(super::hash_set(&mut con, b"hash", "field", "value", None).await.is_err());
|
||||
assert_eq!(con.command_count("HSET"), 1);
|
||||
assert_eq!(con.commands.len(), 2);
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Ok(httl_response(1, 42)),
|
||||
Ok(RedisRawValue::Okay),
|
||||
Err(legacy_hash_expiry_error()),
|
||||
]);
|
||||
assert!(super::hash_set(&mut con, b"hash", "field", "value", None).await.is_err());
|
||||
assert_eq!(con.command_count("HEXPIRE"), 1);
|
||||
let mut con = FakeRedisConnection::with_results(vec![
|
||||
Ok(bulk("hash")),
|
||||
Ok(RedisRawValue::Int(-1)),
|
||||
Ok(RedisRawValue::Int(1)),
|
||||
Err(noperm("hscan")),
|
||||
]);
|
||||
assert!(super::get_value(&mut con, b"hash").await.is_err());
|
||||
assert_eq!(con.command_count("HTTL"), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires disposable Redis via DBX_TEST_REDIS_URL; injects Kvrocks HTTL replies into production Lua"]
|
||||
async fn hash_field_update_lua_legacy_encoding_regression() {
|
||||
let mut trace = FakeRedisConnection::new(vec![RedisRawValue::Int(2)]);
|
||||
super::hash_field_update(&mut trace, b"hash", "old", "new", "after").await.unwrap();
|
||||
let RedisRawValue::Array(args) = redis::parse_redis_value(trace.commands[0].as_bytes()).unwrap() else {
|
||||
panic!("expected EVAL arguments");
|
||||
};
|
||||
let script = String::from_utf8(redis_value_to_bytes(args[1].clone()).unwrap()).unwrap();
|
||||
let probe = "redis.pcall('HTTL', key, 'FIELDS', 1, old_field)";
|
||||
assert_eq!(script.matches(probe).count(), 1);
|
||||
let url = std::env::var("DBX_TEST_REDIS_URL").expect("DBX_TEST_REDIS_URL");
|
||||
let mut con = super::connect(&url, std::time::Duration::from_secs(5)).await.unwrap();
|
||||
let nonce = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos();
|
||||
let key = format!("dbx-10948-{}-{nonce}", std::process::id());
|
||||
for (reply, success) in [
|
||||
("Invalid argument: hash field expiration is not supported by legacy hash encoding", true),
|
||||
("ERR Invalid argument: hash field expiration is not supported by legacy hash encoding", true),
|
||||
("NOPERM no permission to run HTTL", false),
|
||||
("WRONGTYPE Operation against a key holding the wrong kind of value", false),
|
||||
("ERR operation not supported", false),
|
||||
(
|
||||
"ERR Invalid argument: hash field expiration is not supported by legacy hash encoding: extra failure",
|
||||
false,
|
||||
),
|
||||
] {
|
||||
let injected = script.replace(probe, &format!("{{err = '{reply}'}}"));
|
||||
for destination in ["old", "new"] {
|
||||
redis::cmd("DEL").arg(&key).query_async::<()>(&mut con).await.unwrap();
|
||||
redis::cmd("HSET")
|
||||
.arg(&key)
|
||||
.arg("old")
|
||||
.arg("before")
|
||||
.arg("keeper")
|
||||
.arg("keep")
|
||||
.query_async::<()>(&mut con)
|
||||
.await
|
||||
.unwrap();
|
||||
redis::cmd("EXPIRE").arg(&key).arg(60).query_async::<()>(&mut con).await.unwrap();
|
||||
let result: i64 = redis::cmd("EVAL")
|
||||
.arg(&injected)
|
||||
.arg(1)
|
||||
.arg(&key)
|
||||
.arg("old")
|
||||
.arg(destination)
|
||||
.arg("after")
|
||||
.query_async(&mut con)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result,
|
||||
if success {
|
||||
if destination == "old" {
|
||||
1
|
||||
} else {
|
||||
2
|
||||
}
|
||||
} else {
|
||||
-3
|
||||
},
|
||||
"{reply}"
|
||||
);
|
||||
let fields: HashMap<String, String> =
|
||||
redis::cmd("HGETALL").arg(&key).query_async(&mut con).await.unwrap();
|
||||
let expected = if success { (destination, "after") } else { ("old", "before") };
|
||||
assert_eq!(
|
||||
fields,
|
||||
HashMap::from([
|
||||
(expected.0.to_string(), expected.1.to_string()),
|
||||
("keeper".to_string(), "keep".to_string()),
|
||||
])
|
||||
);
|
||||
let ttl: i64 = redis::cmd("TTL").arg(&key).query_async(&mut con).await.unwrap();
|
||||
assert!((1..=60).contains(&ttl));
|
||||
}
|
||||
}
|
||||
redis::cmd("DEL").arg(&key).query_async::<()>(&mut con).await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hash_set_degrades_when_field_ttl_is_rejected_as_unknown_redis_command() {
|
||||
let unsupported = redis::RedisError::from((
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
use base64::Engine;
|
||||
use dbx_driver_redis::{
|
||||
connect, get_value, hash_field_update, hash_set, load_more_collection, set_hash_field_expire_at,
|
||||
set_hash_field_ttl, RedisCollectionPage, RedisHashItem, RedisValueData,
|
||||
};
|
||||
use redis::aio::MultiplexedConnection;
|
||||
use std::collections::HashMap;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
fn fields(items: Vec<RedisHashItem>, expected_ttl: Option<i64>) -> HashMap<String, String> {
|
||||
items
|
||||
.into_iter()
|
||||
.map(|item| {
|
||||
assert_eq!(item.field_ttl, expected_ttl);
|
||||
let decode = |value: &str| {
|
||||
String::from_utf8(base64::engine::general_purpose::STANDARD.decode(value).unwrap()).unwrap()
|
||||
};
|
||||
(decode(&item.field.raw_base64), decode(&item.value.raw_base64))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn snapshot(con: &mut MultiplexedConnection, key: &str) -> HashMap<String, String> {
|
||||
redis::cmd("HGETALL").arg(key).query_async(con).await.unwrap()
|
||||
}
|
||||
|
||||
async fn field_ttl(con: &mut MultiplexedConnection, key: &str, field: &str) -> i64 {
|
||||
let ttls: Vec<i64> = redis::cmd("HTTL").arg(key).arg("FIELDS").arg(1).arg(field).query_async(con).await.unwrap();
|
||||
ttls[0]
|
||||
}
|
||||
|
||||
async fn seed(con: &mut MultiplexedConnection, keys: &[String], kvrocks: bool) -> redis::RedisResult<()> {
|
||||
if kvrocks {
|
||||
redis::cmd("CONFIG").arg("SET").arg("hash-encoding-mode").arg("legacy").query_async::<()>(con).await?;
|
||||
}
|
||||
let mut command = redis::cmd("HSET");
|
||||
command.arg(&keys[0]);
|
||||
for index in 0..401 {
|
||||
command.arg(format!("f{index}")).arg(format!("v{index}"));
|
||||
}
|
||||
command.query_async::<()>(con).await?;
|
||||
if kvrocks {
|
||||
redis::cmd("CONFIG")
|
||||
.arg("SET")
|
||||
.arg("hash-encoding-mode")
|
||||
.arg("field-expiration")
|
||||
.query_async::<()>(con)
|
||||
.await?;
|
||||
}
|
||||
redis::cmd("HSET").arg(&keys[1]).arg("expiring").arg("before").query_async::<()>(con).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn exercise(mut con: MultiplexedConnection, keys: Vec<String>, flavor: String) {
|
||||
let key = &keys[0];
|
||||
let modern = &keys[1];
|
||||
let missing = &keys[2];
|
||||
let wrong_type = &keys[3];
|
||||
let expected_ttl = (flavor == "redis74").then_some(-1);
|
||||
redis::cmd("EXPIRE").arg(key).arg(300).query_async::<()>(&mut con).await.unwrap();
|
||||
let value = get_value(&mut con, key.as_bytes()).await.unwrap();
|
||||
assert!((1..=300).contains(&value.ttl));
|
||||
let RedisValueData::Hash { items, total, mut scan_cursor } = value.data else {
|
||||
panic!("expected hash");
|
||||
};
|
||||
assert_eq!(total, 401);
|
||||
assert!(!items.is_empty() && items.len() <= 200);
|
||||
let mut actual = fields(items, expected_ttl);
|
||||
let mut pages = 0;
|
||||
while let Some(cursor) = scan_cursor {
|
||||
pages += 1;
|
||||
assert!(pages < 100, "pagination failed to finish");
|
||||
let page = load_more_collection(&mut con, key.as_bytes(), "hash", cursor, 200, None, None).await.unwrap();
|
||||
let RedisCollectionPage::Hash { items, scan_cursor: next } = page else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert!(items.len() <= 200);
|
||||
actual.extend(fields(items, expected_ttl));
|
||||
scan_cursor = next;
|
||||
}
|
||||
assert!(pages > 0);
|
||||
assert_eq!(actual, (0..401).map(|index| (format!("f{index}"), format!("v{index}"))).collect());
|
||||
println!("PASS production listing/load-more: 401 fields, bounded pages, key TTL");
|
||||
|
||||
hash_set(&mut con, key.as_bytes(), "added", "value", None).await.unwrap();
|
||||
hash_field_update(&mut con, key.as_bytes(), "f0", "f0", "edited").await.unwrap();
|
||||
assert_eq!(snapshot(&mut con, key).await["f0"], "edited");
|
||||
hash_field_update(&mut con, key.as_bytes(), "f0", "renamed", "moved").await.unwrap();
|
||||
let after = snapshot(&mut con, key).await;
|
||||
assert_eq!(after["added"], "value");
|
||||
assert_eq!(after["renamed"], "moved");
|
||||
assert!(!after.contains_key("f0"));
|
||||
assert!(hash_field_update(&mut con, key.as_bytes(), "f1", "f2", "collision").await.is_err());
|
||||
assert!(hash_field_update(&mut con, key.as_bytes(), "absent", "new", "missing").await.is_err());
|
||||
assert_eq!(snapshot(&mut con, key).await, after);
|
||||
let ttl: i64 = redis::cmd("TTL").arg(key).query_async(&mut con).await.unwrap();
|
||||
assert!((1..=300).contains(&ttl));
|
||||
println!("PASS production add/edit/rename/source deletion/collision/missing field/key TTL");
|
||||
|
||||
if flavor != "redis74" {
|
||||
for result in [
|
||||
set_hash_field_ttl(&mut con, key.as_bytes(), "added", 120).await,
|
||||
set_hash_field_ttl(&mut con, key.as_bytes(), "added", -1).await,
|
||||
set_hash_field_expire_at(&mut con, key.as_bytes(), "added", 4_000_000_000).await,
|
||||
] {
|
||||
let error = result.unwrap_err();
|
||||
if flavor == "kvrocks" {
|
||||
assert!(error.contains("hash field expiration is not supported by legacy hash encoding"), "{error}");
|
||||
} else {
|
||||
assert!(error.to_ascii_lowercase().contains("unknown command"), "{error}");
|
||||
}
|
||||
}
|
||||
assert_eq!(snapshot(&mut con, key).await, after);
|
||||
println!("PASS production explicit HEXPIRE/HPERSIST/HEXPIREAT errors remain visible");
|
||||
}
|
||||
|
||||
if flavor != "redis72" {
|
||||
set_hash_field_ttl(&mut con, modern.as_bytes(), "expiring", 180).await.unwrap();
|
||||
for (old, new, value) in [("expiring", "expiring", "edited"), ("expiring", "renamed", "moved")] {
|
||||
let before = field_ttl(&mut con, modern, old).await;
|
||||
hash_set(&mut con, key.as_bytes(), "added", "legacy still works", None).await.unwrap();
|
||||
hash_set(&mut con, modern.as_bytes(), old, "overwrite", None).await.unwrap();
|
||||
let overwritten = field_ttl(&mut con, modern, old).await;
|
||||
assert!((1..=before).contains(&overwritten));
|
||||
hash_field_update(&mut con, modern.as_bytes(), old, new, value).await.unwrap();
|
||||
let remaining = field_ttl(&mut con, modern, new).await;
|
||||
assert!((1..=overwritten).contains(&remaining));
|
||||
assert_eq!(snapshot(&mut con, modern).await, HashMap::from([(new.to_string(), value.to_string())]));
|
||||
let RedisValueData::Hash { items, .. } = get_value(&mut con, modern.as_bytes()).await.unwrap().data else {
|
||||
panic!("expected hash");
|
||||
};
|
||||
assert_eq!(items.len(), 1);
|
||||
assert!(items[0].field_ttl.is_some_and(|ttl| (1..=remaining).contains(&ttl)));
|
||||
let RedisCollectionPage::Hash { items, .. } =
|
||||
load_more_collection(&mut con, modern.as_bytes(), "hash", 0, 200, None, None).await.unwrap()
|
||||
else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(items.len(), 1);
|
||||
assert!(items[0].field_ttl.is_some_and(|ttl| (1..=remaining).contains(&ttl)));
|
||||
let RedisCollectionPage::Hash { items, .. } =
|
||||
load_more_collection(&mut con, key.as_bytes(), "hash", 0, 1, None, None).await.unwrap()
|
||||
else {
|
||||
panic!("expected hash page");
|
||||
};
|
||||
assert_eq!(items[0].field_ttl, expected_ttl);
|
||||
}
|
||||
set_hash_field_ttl(&mut con, modern.as_bytes(), "renamed", -1).await.unwrap();
|
||||
assert_eq!(field_ttl(&mut con, modern, "renamed").await, -1);
|
||||
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64;
|
||||
set_hash_field_expire_at(&mut con, modern.as_bytes(), "renamed", now + 180).await.unwrap();
|
||||
assert!((1..=180).contains(&field_ttl(&mut con, modern, "renamed").await));
|
||||
println!("PASS production per-key TTL capability coexistence, remaining TTL on overwrite/edit/rename, explicit expiry");
|
||||
}
|
||||
|
||||
assert!(hash_field_update(&mut con, missing.as_bytes(), "absent", "new", "value").await.is_err());
|
||||
let exists: bool = redis::cmd("EXISTS").arg(missing).query_async(&mut con).await.unwrap();
|
||||
assert!(!exists);
|
||||
let page = load_more_collection(&mut con, missing.as_bytes(), "hash", 0, 200, None, None).await.unwrap();
|
||||
assert!(matches!(page, RedisCollectionPage::Hash { items, scan_cursor: None } if items.is_empty()));
|
||||
hash_set(&mut con, missing.as_bytes(), "created", "value", None).await.unwrap();
|
||||
assert!(matches!(
|
||||
get_value(&mut con, missing.as_bytes()).await.unwrap().data,
|
||||
RedisValueData::Hash { total: 1, .. }
|
||||
));
|
||||
hash_set(&mut con, missing.as_bytes(), "created", "value", Some(60)).await.unwrap();
|
||||
let ttl: i64 = redis::cmd("TTL").arg(missing).query_async(&mut con).await.unwrap();
|
||||
assert!((1..=60).contains(&ttl));
|
||||
println!("PASS production missing/empty hash, create, explicit key TTL");
|
||||
|
||||
redis::cmd("SET").arg(wrong_type).arg("string").query_async::<()>(&mut con).await.unwrap();
|
||||
assert!(hash_set(&mut con, wrong_type.as_bytes(), "field", "value", None).await.is_err());
|
||||
assert!(load_more_collection(&mut con, wrong_type.as_bytes(), "hash", 0, 200, None, None).await.is_err());
|
||||
assert!(hash_field_update(&mut con, wrong_type.as_bytes(), "field", "new", "value").await.is_err());
|
||||
let value: String = redis::cmd("GET").arg(wrong_type).query_async(&mut con).await.unwrap();
|
||||
assert_eq!(value, "string");
|
||||
println!("PASS production WRONGTYPE errors preserve data");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires disposable server via DBX_TEST_REDIS_URL and DBX_TEST_REDIS_FLAVOR=kvrocks|redis74|redis72"]
|
||||
async fn production_hash_compatibility() {
|
||||
let url = std::env::var("DBX_TEST_REDIS_URL").expect("DBX_TEST_REDIS_URL");
|
||||
let flavor = std::env::var("DBX_TEST_REDIS_FLAVOR").expect("DBX_TEST_REDIS_FLAVOR");
|
||||
assert!(["kvrocks", "redis74", "redis72"].contains(&flavor.as_str()));
|
||||
let mut con = connect(&url, Duration::from_secs(5)).await.unwrap();
|
||||
let info: String = redis::cmd("INFO").arg("server").query_async(&mut con).await.unwrap();
|
||||
println!(
|
||||
"{}",
|
||||
info.lines()
|
||||
.filter(|line| line.contains("version:") || line.contains("git_sha1:"))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n")
|
||||
);
|
||||
let nonce = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
|
||||
let keys: Vec<_> = (0..4).map(|index| format!("dbx-10948-{}-{nonce}-{index}", std::process::id())).collect();
|
||||
let kvrocks = flavor == "kvrocks";
|
||||
let original: Option<HashMap<String, String>> = if kvrocks {
|
||||
Some(redis::cmd("CONFIG").arg("GET").arg("hash-encoding-mode").query_async(&mut con).await.unwrap())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let seeded = seed(&mut con, &keys, kvrocks).await;
|
||||
if let Some(original) = original {
|
||||
redis::cmd("CONFIG")
|
||||
.arg("SET")
|
||||
.arg("hash-encoding-mode")
|
||||
.arg(&original["hash-encoding-mode"])
|
||||
.query_async::<()>(&mut con)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let outcome =
|
||||
if seeded.is_ok() { Some(tokio::spawn(exercise(con.clone(), keys.clone(), flavor)).await) } else { None };
|
||||
redis::cmd("DEL").arg(&keys).query_async::<()>(&mut con).await.unwrap();
|
||||
seeded.unwrap();
|
||||
outcome.unwrap().unwrap();
|
||||
}
|
||||
Reference in New Issue
Block a user