use std::{
fmt::Write as _,
str,
time::{SystemTime, UNIX_EPOCH},
};
use super::{
cmd_strings as cs,
cmd_strings::{
abort_with_error_message, abort_with_wrong_number_of_arguments, write_error_raw,
write_map_len_resp2, write_raw,
},
parser::{
resp_ext::{RespSliceExt, RespVecExt},
session_parse_state::{strict_f64, strict_i32, strict_i64},
},
resp_server_session::RespServerSession,
ttl_sync::{
del_ttl_sync, now_unix_ms, probe_alive, put_ttl_sync, read_adjudicated_sync, ttl_of_sync,
},
};
pub(crate) const MAX_STRING_PAYLOAD_BYTES: usize = 512 * 1024 * 1024;
const ERR_NOT_INTEGER: &str = cs::RESP_ERR_GENERIC_VALUE_IS_NOT_INTEGER;
const ERR_OFFSET_OUT_OF_RANGE: &str = cs::RESP_ERR_GENERIC_OFFSETOUTOFRANGE;
const ERR_STRING_EXCEEDS_MAX: &str = "ERR string exceeds maximum allowed size (proto-max-bulk-len)";
const CAN_AUTHENTICATE: bool = false;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SetCmd {
Set,
SetExNx,
SetExXx,
SetKeepTtl,
SetKeepTtlXx,
}
impl SetCmd {
#[inline]
const fn is_keep_ttl(self) -> bool {
matches!(self, Self::SetKeepTtl | Self::SetKeepTtlXx)
}
#[inline]
const fn is_xx(self) -> bool {
matches!(self, Self::SetExXx | Self::SetKeepTtlXx)
}
#[inline]
const fn is_nx(self) -> bool {
matches!(self, Self::SetExNx)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IncrCmd {
Incr,
Decr,
IncrBy,
DecrBy,
}
impl IncrCmd {
const fn as_str(self) -> &'static str {
match self {
Self::Incr => "INCR",
Self::Decr => "DECR",
Self::IncrBy => "INCRBY",
Self::DecrBy => "DECRBY",
}
}
const fn sign(self) -> i64 {
match self {
Self::Decr | Self::DecrBy => -1,
_ => 1,
}
}
const fn has_by(self) -> bool {
matches!(self, Self::IncrBy | Self::DecrBy)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ObjectSubCmd {
Encoding,
Freq,
Idletime,
Refcount,
}
impl ObjectSubCmd {
const fn as_str(self) -> &'static str {
match self {
Self::Encoding => "object|encoding",
Self::Freq => "object|freq",
Self::Idletime => "object|idletime",
Self::Refcount => "object|refcount",
}
}
}
enum GetexExpiry {
None,
Persist,
At(u64),
}
pub struct SetOptions<'a> {
pub key: &'a [u8],
pub val: &'a [u8],
pub expiry: i64,
pub exp_high_precision: bool,
cmd: SetCmd,
pub get_value: bool,
}
fn expiry_ms_from_now(expiry: i64, high_precision: bool) -> u64 {
let now = now_unix_ms() as i64;
let span = if high_precision {
expiry
} else {
expiry.saturating_mul(1000)
};
now.saturating_add(span).max(0) as u64
}
fn apply_set_with_expiry<'a, D: wdev::Device>(
store: &wkv::BatchStoreSession<'a, D>,
key: &[u8],
val: &[u8],
expiry: i64,
high_precision: bool,
keep_ttl: Option<Option<u64>>,
output: &mut Vec<u8>,
) -> Result<bool, ()> {
match store.try_upsert_sync(key, val) {
Ok(Ok(_)) => {}
Ok(Err(_)) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Err(());
}
}
let apply_ttl = |ms: u64| put_ttl_sync(store, key, ms).map_err(|_| ());
if let Some(old_ttl) = keep_ttl {
return match old_ttl {
Some(ms) => apply_ttl(ms),
None => Ok(true),
};
}
if expiry != 0 {
let expire_at_ms = expiry_ms_from_now(expiry, high_precision);
return apply_ttl(expire_at_ms);
}
Ok(true)
}
pub fn try_get_client_name(raw: &[u8]) -> Option<&str> {
let name = str::from_utf8(raw).ok()?;
if name.is_empty() {
return Some(name);
}
name
.bytes()
.all(|c| (33..=126).contains(&c))
.then_some(name)
}
fn format_error_option(cmd: &str, option: &str) -> String {
format!("ERR Syntax error in {cmd} option '{option}'")
}
fn try_parse_double(raw: &[u8]) -> Option<f64> {
strict_f64(raw, true)
}
impl RespServerSession {
pub fn network_get<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.is_empty() {
return Ok(false);
}
let key = parse_state[0];
match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(Some(val))) => {
output.write_resp_bulk_string(&val);
}
Ok(Some(None)) => {
output.extend_from_slice(b"$-1\r\n");
}
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_getex<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.is_empty() || parse_state.len() > 3 {
abort_with_wrong_number_of_arguments(output, "GETEX");
return Ok(true);
}
let key = parse_state[0];
let expiry = if parse_state.len() > 1 {
let option = parse_state[1];
if option.eq_ignore_ascii_case(b"PERSIST") {
GetexExpiry::Persist
} else {
let Some(expire_time) = parse_state.get(2).copied().and_then(strict_i64) else {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_VALUE_IS_OUT_OF_RANGE);
return Ok(true);
};
if expire_time <= 0 {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_VALUE_IS_OUT_OF_RANGE);
return Ok(true);
}
let now = now_unix_ms() as i64;
if option.eq_ignore_ascii_case(b"EX") {
GetexExpiry::At((now + expire_time.saturating_mul(1000)).max(0) as u64)
} else if option.eq_ignore_ascii_case(b"PX") {
GetexExpiry::At((now + expire_time).max(0) as u64)
} else if option.eq_ignore_ascii_case(b"EXAT") {
GetexExpiry::At(expire_time.saturating_mul(1000).max(0) as u64)
} else if option.eq_ignore_ascii_case(b"PXAT") {
GetexExpiry::At(expire_time.max(0) as u64)
} else {
abort_with_error_message(
output,
&cs::GENERIC_ERR_UNSUPPORTED_OPTION.replace("{0}", option.as_str_safe()),
);
return Ok(true);
}
}
} else {
GetexExpiry::None
};
match read_adjudicated_sync(store, key, |v| v.to_vec()) {
Ok(Some(Some(val))) => {
let applied = match &expiry {
GetexExpiry::None => Ok(true),
GetexExpiry::Persist => del_ttl_sync(store, key),
GetexExpiry::At(ms) => put_ttl_sync(store, key, *ms),
};
match applied {
Ok(true) => output.write_resp_bulk_string(&val),
Ok(false) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
}
Ok(Some(None)) => {
output.extend_from_slice(b"$-1\r\n");
}
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_get_async<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
self.network_get(parse_state, store, output)
}
pub fn network_get_sg<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
self.network_get(parse_state, store, output)
}
pub fn network_set<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() < 2 {
abort_with_wrong_number_of_arguments(output, "SET");
return Ok(true);
}
if parse_state.len() > 2 {
return self.network_setexnx(parse_state, store, output);
}
let key = parse_state[0];
let value = parse_state[1];
match store.try_upsert_sync(key, value) {
Ok(Ok(_)) => output.write_resp_simple_string("OK"),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_getset<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 2 {
abort_with_wrong_number_of_arguments(output, "GETSET");
return Ok(true);
}
let opts = SetOptions {
key: parse_state[0],
val: parse_state[1],
expiry: 0,
exp_high_precision: false,
cmd: SetCmd::Set,
get_value: true,
};
self.network_set__conditional(&opts, store, output)
}
pub fn network_set_range<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 3 {
abort_with_wrong_number_of_arguments(output, "SETRANGE");
return Ok(true);
}
let key = parse_state[0];
let Some(offset) = strict_i32(parse_state[1]) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
let offset: i64 = i64::from(offset);
let val = parse_state[2];
if offset < 0 {
abort_with_error_message(output, ERR_OFFSET_OUT_OF_RANGE);
return Ok(true);
}
if offset as u64 + val.len() as u64 > MAX_STRING_PAYLOAD_BYTES as u64 {
abort_with_error_message(output, ERR_STRING_EXCEEDS_MAX);
return Ok(true);
}
let offset = offset as usize;
match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(Some(mut existing))) => {
if offset + val.len() > existing.len() {
existing.resize(offset + val.len(), 0);
}
existing[offset..offset + val.len()].copy_from_slice(val);
match store.try_upsert_sync(key, &existing) {
Ok(Ok(_)) => output.write_resp_int(existing.len() as i64),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
}
Ok(Some(None)) => {
let mut new_val = vec![0; offset + val.len()];
new_val[offset..offset + val.len()].copy_from_slice(val);
match store.try_upsert_sync(key, &new_val) {
Ok(Ok(_)) => output.write_resp_int(new_val.len() as i64),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
}
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_get_range<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 3 {
abort_with_wrong_number_of_arguments(output, "GETRANGE");
return Ok(true);
}
let Some(mut start) = strict_i32(parse_state[1]).map(i64::from) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
let Some(mut end) = strict_i32(parse_state[2]).map(i64::from) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
let key = parse_state[0];
match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(Some(val))) => {
let len = val.len() as i64;
if start < 0 {
start = start.saturating_add(len);
}
if end < 0 {
end = end.saturating_add(len);
}
if start < 0 {
start = 0;
}
if end >= len {
end = len - 1;
}
if start > end || start >= len {
output.write_resp_bulk_string(b"");
} else {
output.write_resp_bulk_string(&val[(start as usize)..=(end as usize)]);
}
}
Ok(Some(None)) => output.write_resp_bulk_string(b""),
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_setex<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
self.network_setex_impl(false, "SETEX", parse_state, store, output)
}
pub fn network_psetex<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
self.network_setex_impl(true, "PSETEX", parse_state, store, output)
}
fn network_setex_impl<'a, D: wdev::Device>(
&mut self,
high_precision: bool,
cmd_name: &str,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 3 {
abort_with_wrong_number_of_arguments(output, cmd_name);
return Ok(true);
}
let key = parse_state[0];
let Some(expiry) = strict_i32(parse_state[1]) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
if expiry <= 0 {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_INVALIDEXP_IN_SET);
return Ok(true);
}
let val = parse_state[2];
match store.try_upsert_sync(key, val) {
Ok(Ok(_)) => {}
Ok(Err(_)) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
}
let expire_at_ms = expiry_ms_from_now(i64::from(expiry), high_precision);
match put_ttl_sync(store, key, expire_at_ms) {
Ok(true) => output.write_resp_simple_string("OK"),
Ok(false) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_setnx<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 2 {
abort_with_wrong_number_of_arguments(output, "SETNX");
return Ok(true);
}
let key = parse_state[0];
let val = parse_state[1];
match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(Some(_))) => output.write_resp_int(0),
Ok(Some(None)) => match store.try_upsert_sync(key, val) {
Ok(Ok(_)) => output.write_resp_int(1),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
},
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_setexnx<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let Some(opts) = parse_set_options(parse_state, output) else {
return Ok(true);
};
if opts.cmd == SetCmd::Set && !opts.get_value {
return self.network_set_ex(&opts, store, output);
}
self.network_set__conditional(&opts, store, output)
}
pub fn network_set_ex<'a, D: wdev::Device>(
&mut self,
opts: &SetOptions,
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
match apply_set_with_expiry(
store,
opts.key,
opts.val,
opts.expiry,
opts.exp_high_precision,
None,
output,
) {
Ok(true) => {
output.write_resp_simple_string("OK");
Ok(true)
}
Ok(false) => Ok(false),
Err(()) => Ok(true),
}
}
pub fn network_set__conditional<'a, D: wdev::Device>(
&mut self,
opts: &SetOptions,
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let SetOptions {
key,
val,
expiry,
exp_high_precision: high_precision,
cmd,
get_value,
..
} = *opts;
if !get_value {
let (must_exist, must_absent) = (cmd.is_xx(), cmd.is_nx());
let old_ttl = if cmd.is_keep_ttl() {
match ttl_of_sync(store, key) {
Ok(Some(ttl)) => Some(ttl),
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
}
} else {
None
};
let exists = if cmd.is_keep_ttl() {
true
} else {
match probe_alive(store, key) {
Ok(Some(alive)) => alive,
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
}
};
if !cmd.is_keep_ttl() && ((must_exist && !exists) || (must_absent && exists)) {
output.extend_from_slice(b"$-1\r\n");
return Ok(true);
}
match apply_set_with_expiry(store, key, val, expiry, high_precision, old_ttl, output) {
Ok(true) => output.write_resp_simple_string("OK"),
Ok(false) => return Ok(false),
Err(()) => {}
}
return Ok(true);
}
let old = match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(found)) => found,
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
};
let should_set = if cmd.is_keep_ttl() {
!cmd.is_xx() || old.is_some()
} else if cmd.is_nx() {
old.is_none()
} else if cmd.is_xx() {
old.is_some()
} else {
true
};
let old_ttl = if should_set && cmd.is_keep_ttl() {
match ttl_of_sync(store, key) {
Ok(Some(ttl)) => Some(ttl),
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
}
} else {
None
};
if should_set {
match apply_set_with_expiry(store, key, val, expiry, high_precision, old_ttl, output) {
Ok(true) => {}
Ok(false) => return Ok(false),
Err(()) => return Ok(true),
}
}
match old {
Some(old) => output.write_resp_bulk_string(&old),
None => output.extend_from_slice(b"$-1\r\n"),
}
Ok(true)
}
pub fn network_increment<'a, D: wdev::Device>(
&mut self,
cmd: IncrCmd,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let min_args = if cmd.has_by() { 2 } else { 1 };
if parse_state.len() < min_args {
abort_with_wrong_number_of_arguments(output, cmd.as_str());
return Ok(true);
}
let key = parse_state[0];
let mut delta = cmd.sign();
if cmd.has_by() {
let Some(by) = strict_i64(parse_state[1]) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
delta = delta.saturating_mul(by);
}
let cur = match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(found)) => found,
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
};
let val = match cur {
Some(bytes) => match bytes.as_str_safe().parse::<i64>() {
Ok(v) => v,
Err(_) => {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
}
},
None => 0,
};
let Some(next) = val.checked_add(delta) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
let mut buf = itoa::Buffer::new();
match store.try_upsert_sync(key, buf.format(next).as_bytes()) {
Ok(Ok(_)) => output.write_resp_int(next),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_increment_by_float<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() < 2 {
abort_with_wrong_number_of_arguments(output, "INCRBYFLOAT");
return Ok(true);
}
let key = parse_state[0];
let Some(incr_by) = try_parse_double(parse_state[1]) else {
abort_with_error_message(output, cs::RESP_ERR_NOT_VALID_FLOAT);
return Ok(true);
};
if incr_by.is_infinite() {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_NAN_INFINITY_INCR);
return Ok(true);
}
let cur = match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(found)) => found,
Ok(None) => return Ok(false),
Err(_) => {
output.write_resp_error("generic error");
return Ok(true);
}
};
let val = match cur {
Some(bytes) => match try_parse_double(&bytes) {
Some(v) => v,
None => {
abort_with_error_message(output, cs::RESP_ERR_NOT_VALID_FLOAT);
return Ok(true);
}
},
None => 0.0,
};
let next = val + incr_by;
if next.is_nan() || next.is_infinite() {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_NAN_INFINITY_INCR);
return Ok(true);
}
let mut formatted = String::new();
let _ = write!(&mut formatted, "{next}");
match store.try_upsert_sync(key, formatted.as_bytes()) {
Ok(Ok(_)) => output.write_resp_bulk_string(formatted.as_bytes()),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_append<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 2 {
abort_with_wrong_number_of_arguments(output, "APPEND");
return Ok(true);
}
let key = parse_state[0];
let val = parse_state[1];
match store.try_read_sync(key, |v| v.to_vec()) {
Ok(Some(Some(mut existing))) => {
existing.extend_from_slice(val);
match store.try_upsert_sync(key, &existing) {
Ok(Ok(_)) => output.write_resp_int(existing.len() as i64),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
}
Ok(Some(None)) => match store.try_upsert_sync(key, val) {
Ok(Ok(_)) => output.write_resp_int(val.len() as i64),
Ok(Err(_)) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
},
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_ping(
&mut self,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() > 1 {
abort_with_wrong_number_of_arguments(output, "PING");
return Ok(true);
}
if let Some(msg) = parse_state.first() {
output.write_resp_bulk_string(msg);
} else {
output.extend_from_slice(cs::RESP_PONG);
}
Ok(true)
}
pub fn network_asking<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
write_raw(output, cs::RESP_OK);
Ok(true)
}
pub fn network_quit(
&mut self,
_parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
output.write_resp_simple_string("OK");
Ok(true)
}
pub fn network_flushdb(
&mut self,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() > 2 {
abort_with_wrong_number_of_arguments(output, "FLUSHDB");
return Ok(true);
}
self.flush_db("FLUSHDB", parse_state, output)
}
pub fn network_flushall(
&mut self,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() > 3 {
abort_with_wrong_number_of_arguments(output, "FLUSHALL");
return Ok(true);
}
self.flush_db("FLUSHALL", parse_state, output)
}
pub fn network_readonly<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
write_raw(output, cs::RESP_OK);
Ok(true)
}
pub fn network_readwrite<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
write_raw(output, cs::RESP_OK);
Ok(true)
}
pub fn network_strlen<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 1 {
abort_with_wrong_number_of_arguments(output, "STRLEN");
return Ok(true);
}
let key = parse_state[0];
match store.try_read_sync(key, |v| v.len()) {
Ok(Some(Some(len))) => output.write_resp_int(len as i64),
Ok(Some(None)) | Ok(None) => output.write_resp_int(0),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn write_command_response<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
write_raw(output, cs::RESP_EMPTYLIST);
Ok(true)
}
pub fn network_command<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if !parse_state.is_empty() {
let error_msg = cs::GENERIC_ERR_UNKNOWN_SUB_COMMAND
.replace("{0}", parse_state[0].as_str_safe())
.replace("{1}", "COMMAND");
write_error_raw(output, &error_msg);
} else {
self.write_command_response(parse_state, store, output)?;
}
Ok(true)
}
pub fn network_command_count<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if !parse_state.is_empty() {
abort_with_wrong_number_of_arguments(output, "COMMAND|COUNT");
return Ok(true);
}
output.write_resp_int(0);
Ok(true)
}
pub fn network_command_docs<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let _ = parse_state.len();
write_map_len_resp2(output, 0);
Ok(true)
}
pub fn network_command_info<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let count = parse_state.len();
if count == 0 {
return self.write_command_response(parse_state, store, output);
}
output.write_resp_array_len(count);
for _ in 0..count {
output.extend_from_slice(b"$-1\r\n");
}
Ok(true)
}
pub fn network_command_getkeys<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.is_empty() {
abort_with_wrong_number_of_arguments(output, "COMMAND|GETKEYS");
return Ok(true);
}
abort_with_error_message(output, cs::RESP_INVALID_COMMAND_SPECIFIED);
Ok(true)
}
pub fn network_command_getkeysandflags<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.is_empty() {
abort_with_wrong_number_of_arguments(output, "COMMAND|GETKEYSANDFLAGS");
return Ok(true);
}
abort_with_error_message(output, cs::RESP_INVALID_COMMAND_SPECIFIED);
Ok(true)
}
pub fn network_echo(
&mut self,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 1 {
abort_with_wrong_number_of_arguments(output, "ECHO");
return Ok(true);
}
let msg = parse_state[0];
output.write_resp_bulk_string(msg);
Ok(true)
}
pub fn network_hello<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let count = parse_state.len();
if count > 6 {
abort_with_wrong_number_of_arguments(output, "HELLO");
return Ok(true);
}
let mut tmp_resp_protocol_version: Option<u8> = None;
let mut auth_username: &[u8] = &[];
let mut tmp_client_name: Option<&str> = None;
let mut error_msg: Option<String> = None;
if count > 0 {
let mut token_idx = 0usize;
let Some(local_resp_protocol_version) = strict_i32(parse_state[token_idx]) else {
abort_with_error_message(output, cs::RESP_ERR_PROTOCOL_VALUE_IS_NOT_INTEGER);
return Ok(true);
};
token_idx += 1;
if !(2..=3).contains(&local_resp_protocol_version) {
abort_with_error_message(output, cs::RESP_ERR_UNSUPPORTED_PROTOCOL_VERSION);
return Ok(true);
}
tmp_resp_protocol_version = Some(local_resp_protocol_version as u8);
while token_idx < count {
let param = parse_state[token_idx];
token_idx += 1;
if param.eq_ignore_ascii_case(b"AUTH") {
if count - token_idx < 2 {
error_msg = Some(format_error_option("HELLO", "AUTH"));
break;
}
auth_username = parse_state[token_idx];
token_idx += 2;
} else if param.eq_ignore_ascii_case(b"SETNAME") {
if count - token_idx < 1 {
error_msg = Some(format_error_option("HELLO", "SETNAME"));
break;
}
let Some(name) = try_get_client_name(parse_state[token_idx]) else {
abort_with_error_message(output, cs::RESP_ERR_INVALID_CLIENT_NAME);
return Ok(true);
};
token_idx += 1;
tmp_client_name = Some(name);
} else {
error_msg = Some(format_error_option("HELLO", param.as_str_safe()));
break;
}
}
}
if let Some(error_msg) = error_msg {
write_error_raw(output, &error_msg);
return Ok(true);
}
self.process_hello_command(
tmp_resp_protocol_version,
auth_username,
tmp_client_name,
store,
output,
)
}
pub fn network_time(
&mut self,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if !parse_state.is_empty() {
abort_with_wrong_number_of_arguments(output, "TIME");
return Ok(true);
}
let Ok(elapsed) = SystemTime::now().duration_since(UNIX_EPOCH) else {
output.write_resp_error("generic error");
return Ok(true);
};
let seconds = elapsed.as_secs();
let micros = elapsed.subsec_micros();
let frame = format!(
"*2\r\n${}\r\n{seconds}\r\n$6\r\n{micros:06}\r\n",
digits_len(seconds)
);
write_raw(output, frame.as_bytes());
Ok(true)
}
pub fn network_auth<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let count = parse_state.len();
if !(1..=2).contains(&count) {
abort_with_wrong_number_of_arguments(output, "AUTH");
return Ok(true);
}
if CAN_AUTHENTICATE {
write_raw(output, cs::RESP_OK);
} else {
write_error_raw(
output,
"ERR Client sent AUTH, but configured authenticator does not accept passwords",
);
}
Ok(true)
}
pub fn network_memory_usage<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let count = parse_state.len();
if count != 1 && count != 3 {
abort_with_wrong_number_of_arguments(output, "MEMORY|USAGE");
return Ok(true);
}
let key = parse_state[0];
if count == 3 {
if !parse_state[1].eq_ignore_ascii_case(b"SAMPLES") {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
let Some(samples) = strict_i32(parse_state[2]) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return Ok(true);
};
if samples < 0 {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
}
match store.try_read_sync(key, |_| ()) {
Ok(Some(_)) => output.extend_from_slice(b"$-1\r\n"),
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_object<'a, D: wdev::Device>(
&mut self,
sub_cmd: ObjectSubCmd,
parse_state: &[&[u8]],
store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if parse_state.len() != 1 {
abort_with_wrong_number_of_arguments(output, sub_cmd.as_str());
return Ok(true);
}
let key = parse_state[0];
match store.try_read_sync(key, |_| ()) {
Ok(Some(Some(()))) => match sub_cmd {
ObjectSubCmd::Encoding => output.write_resp_bulk_string(b"raw"),
ObjectSubCmd::Refcount => output.write_resp_int(1),
ObjectSubCmd::Idletime => output.write_resp_int(0),
ObjectSubCmd::Freq => {
abort_with_error_message(output, cs::RESP_ERR_OBJECT_FREQ_UNSUPPORTED)
}
},
Ok(Some(None)) => output.extend_from_slice(b"$-1\r\n"),
Ok(None) => return Ok(false),
Err(_) => output.write_resp_error("generic error"),
}
Ok(true)
}
pub fn network_objecthelp<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if !parse_state.is_empty() {
abort_with_wrong_number_of_arguments(output, "object|help");
return Ok(true);
}
const OBJECT_HELP: [&str; 11] = [
"OBJECT <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",
"ENCODING <key>",
"\tReturn the kind of internal representation used in order to store the value associated with a <key>.",
"FREQ <key>",
"\tNot supported in Garnet: always returns an error, as access frequency (LFU) is not tracked.",
"IDLETIME <key>",
"\tReturn the idle time of the <key>. Garnet does not track per-key idle time, so this is always 0.",
"REFCOUNT <key>",
"\tReturn the number of references of the value associated with the <key>. Garnet does not share value objects, so this is always 1.",
"HELP",
"\tPrints this help.",
];
output.write_resp_array_len(OBJECT_HELP.len());
for line in OBJECT_HELP {
output.write_resp_simple_string(line);
}
Ok(true)
}
pub fn network_async<'a, D: wdev::Device>(
&mut self,
parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let resp_protocol_version: u8 = 2;
if resp_protocol_version <= 2 {
abort_with_error_message(output, cs::RESP_ERR_NOT_SUPPORTED_RESP2);
return Ok(true);
}
if parse_state.len() != 1 {
abort_with_wrong_number_of_arguments(output, "ASYNC");
return Ok(true);
}
let param = parse_state[0];
if param.eq_ignore_ascii_case(b"ON") || param.eq_ignore_ascii_case(b"OFF") {
} else if param.eq_ignore_ascii_case(b"BARRIER") {
} else {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
write_raw(output, cs::RESP_OK);
Ok(true)
}
pub fn process_hello_command<'a, D: wdev::Device>(
&mut self,
resp_protocol_version: Option<u8>,
username: &[u8],
client_name: Option<&str>,
_store: &wkv::BatchStoreSession<'a, D>,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
if !username.is_empty() && !CAN_AUTHENTICATE {
write_error_raw(output, cs::RESP_WRONGPASS_INVALID_USERNAME_PASSWORD);
return Ok(true);
}
if let Some(version) = resp_protocol_version {
self.update_resp_protocol_version(version);
}
if let Some(name) = client_name {
self.set_client_name(Some(name));
}
write_map_len_resp2(output, 8);
output.write_resp_bulk_string(b"server");
output.write_resp_bulk_string(b"redis");
output.write_resp_bulk_string(b"version");
output.write_resp_bulk_string(super::resp_server_session::REDIS_PROTOCOL_VERSION.as_bytes());
output.write_resp_bulk_string(b"garnet_version");
output.write_resp_bulk_string(env!("CARGO_PKG_VERSION").as_bytes());
output.write_resp_bulk_string(b"proto");
output.write_resp_int(i64::from(self.resp_protocol_version));
output.write_resp_bulk_string(b"id");
output.write_resp_int(self.id);
output.write_resp_bulk_string(b"mode");
output.write_resp_bulk_string(b"standalone");
output.write_resp_bulk_string(b"role");
output.write_resp_bulk_string(b"master");
output.write_resp_bulk_string(b"modules");
output.extend_from_slice(b"*0\r\n");
Ok(true)
}
pub fn flush_db(
&mut self,
cmd: &str,
parse_state: &[&[u8]],
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
let mut unsafe_truncate_log = false;
let mut async_flush = false;
let mut sync_flush = false;
for token in parse_state {
if token.eq_ignore_ascii_case(b"UNSAFETRUNCATELOG") {
if unsafe_truncate_log {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
unsafe_truncate_log = true;
} else if token.eq_ignore_ascii_case(b"ASYNC") {
if sync_flush || async_flush {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
async_flush = true;
} else if token.eq_ignore_ascii_case(b"SYNC") {
if sync_flush || async_flush {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
sync_flush = true;
} else {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return Ok(true);
}
}
self.execute_flush_db(cmd, unsafe_truncate_log, async_flush, output)
}
pub fn execute_flush_db(
&mut self,
_cmd: &str,
_unsafe_truncate_log: bool,
_async_flush: bool,
output: &mut Vec<u8>,
) -> wresp::Result<bool> {
output.write_resp_error("generic error");
Ok(true)
}
pub fn parse_get_and_key<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
_output: &mut Vec<u8>,
) -> wresp::Result<bool> {
Ok(false)
}
pub fn next_command_maybe_get(&self) -> bool {
false
}
pub fn try_get_simple_command_info<'a, D: wdev::Device>(
&mut self,
_cmd_name: &[u8],
_store: &wkv::BatchStoreSession<'a, D>,
_output: &mut Vec<u8>,
) -> wresp::Result<bool> {
Ok(false)
}
pub fn set_result<'a, D: wdev::Device>(
&mut self,
_parse_state: &[&[u8]],
_store: &wkv::BatchStoreSession<'a, D>,
_output: &mut Vec<u8>,
) -> wresp::Result<bool> {
Ok(true)
}
}
const fn digits_len(mut v: u64) -> usize {
let mut n = 1;
while v >= 10 {
v /= 10;
n += 1;
}
n
}
fn parse_set_options<'p>(parse_state: &[&'p [u8]], output: &mut Vec<u8>) -> Option<SetOptions<'p>> {
let key = parse_state[0];
let val = parse_state[1];
let mut expiry: i64 = 0;
let mut exp_high_precision = false;
let mut exp_keep_ttl = false;
let mut exist_nx = false;
let mut exist_xx = false;
let mut get_value = false;
let mut token_idx = 2usize;
while token_idx < parse_state.len() {
let next_opt = parse_state[token_idx];
token_idx += 1;
let is_expiry_option = next_opt.eq_ignore_ascii_case(b"EX")
|| next_opt.eq_ignore_ascii_case(b"PX")
|| next_opt.eq_ignore_ascii_case(b"KEEPTTL")
|| next_opt.eq_ignore_ascii_case(b"EXAT")
|| next_opt.eq_ignore_ascii_case(b"PXAT");
if is_expiry_option {
if exp_keep_ttl
|| expiry != 0
|| next_opt.eq_ignore_ascii_case(b"EXAT")
|| next_opt.eq_ignore_ascii_case(b"PXAT")
{
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return None;
}
if next_opt.eq_ignore_ascii_case(b"KEEPTTL") {
exp_keep_ttl = true;
continue;
}
let Some(raw) = parse_state.get(token_idx) else {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return None;
};
token_idx += 1;
let Some(v) = strict_i32(raw) else {
abort_with_error_message(output, ERR_NOT_INTEGER);
return None;
};
if v <= 0 {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_INVALIDEXP_IN_SET);
return None;
}
expiry = i64::from(v);
exp_high_precision = next_opt.eq_ignore_ascii_case(b"PX");
continue;
}
if next_opt.eq_ignore_ascii_case(b"NX") {
if exist_nx || exist_xx {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return None;
}
exist_nx = true;
} else if next_opt.eq_ignore_ascii_case(b"XX") {
if exist_nx || exist_xx {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_SYNTAX_ERROR);
return None;
}
exist_xx = true;
} else if next_opt.eq_ignore_ascii_case(b"GET") {
get_value = true;
} else {
abort_with_error_message(output, cs::RESP_ERR_GENERIC_UNK_CMD);
return None;
}
}
if exp_keep_ttl {
expiry = 0;
}
let cmd = if exist_nx {
SetCmd::SetExNx
} else if exist_xx {
if exp_keep_ttl {
SetCmd::SetKeepTtlXx
} else {
SetCmd::SetExXx
}
} else if exp_keep_ttl {
SetCmd::SetKeepTtl
} else {
SetCmd::Set
};
Some(SetOptions {
key,
val,
expiry,
exp_high_precision,
cmd,
get_value,
})
}
#[cfg(test)]
mod tests {
use super::{
super::{batch_harness::with_batch, ttl_sync::ttl_of_sync},
*,
};
#[test]
fn set_get_roundtrip_and_null() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_set(&[b"k", b"v1"], batch, &mut out).unwrap();
assert_eq!(out, b"+OK\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b"$2\r\nv1\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"missing"], batch, &mut out).unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s
.network_set(&[b"k", b"v2", b"GET"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$2\r\nv1\r\n");
});
}
#[test]
fn set_wrong_args_error() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_set(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b"-ERR wrong number of arguments for 'SET' command\r\n");
});
}
#[test]
fn append_and_strlen() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_append(&[b"k", b"ab"], batch, &mut out).unwrap();
assert_eq!(out, b":2\r\n");
let mut out = Vec::new();
let _ = s.network_append(&[b"k", b"cd"], batch, &mut out).unwrap();
assert_eq!(out, b":4\r\n");
let mut out = Vec::new();
let _ = s.network_strlen(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b":4\r\n");
let mut out = Vec::new();
let _ = s.network_strlen(&[b"x"], batch, &mut out).unwrap();
assert_eq!(out, b":0\r\n");
});
}
#[test]
fn setnx_semantics() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_setnx(&[b"k", b"v"], batch, &mut out).unwrap();
assert_eq!(out, b":1\r\n");
let mut out = Vec::new();
let _ = s.network_setnx(&[b"k", b"w"], batch, &mut out).unwrap();
assert_eq!(out, b":0\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b"$1\r\nv\r\n");
});
}
#[test]
fn incr_family_and_overflow() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::Incr, &[b"n"], batch, &mut out)
.unwrap();
assert_eq!(out, b":1\r\n");
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::IncrBy, &[b"n", b"41"], batch, &mut out)
.unwrap();
assert_eq!(out, b":42\r\n");
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::Decr, &[b"n"], batch, &mut out)
.unwrap();
assert_eq!(out, b":41\r\n");
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::DecrBy, &[b"n", b"1"], batch, &mut out)
.unwrap();
assert_eq!(out, b":40\r\n");
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::IncrBy, &[b"n", b"x"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not an integer or out of range.\r\n");
let _ = s
.network_set(&[b"s", b"abc"], batch, &mut Vec::new())
.unwrap();
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::Incr, &[b"s"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not an integer or out of range.\r\n");
let _ = s
.network_set(&[b"m", b"9223372036854775807"], batch, &mut Vec::new())
.unwrap();
let mut out = Vec::new();
let _ = s
.network_increment(IncrCmd::Incr, &[b"m"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not an integer or out of range.\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"m"], batch, &mut out).unwrap();
assert_eq!(out, b"$19\r\n9223372036854775807\r\n");
});
}
#[test]
fn incrbyfloat_formatting() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"10.5"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$4\r\n10.5\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"0.1"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$4\r\n10.6\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"z", b"0.2"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$3\r\n0.2\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"z", b"0.1"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$19\r\n0.30000000000000004\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"nan"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not a valid float\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"inf"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR increment would produce NaN or Infinity\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"1e999"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR increment would produce NaN or Infinity\r\n");
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"f", b"abc"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not a valid float\r\n");
let _ = s
.network_set(&[b"g", b"abc"], batch, &mut Vec::new())
.unwrap();
let mut out = Vec::new();
let _ = s
.network_increment_by_float(&[b"g", b"1.0"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not a valid float\r\n");
});
}
#[test]
fn setex_writes_ttl_record() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_setex(&[b"k", b"100", b"v"], batch, &mut out)
.unwrap();
assert_eq!(out, b"+OK\r\n");
let ttl = ttl_of_sync(batch, b"k").unwrap().unwrap().unwrap();
assert!(ttl > now_unix_ms() + 99_000);
let mut out = Vec::new();
let _ = s
.network_setex(&[b"k", b"0", b"v"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR invalid expire time in 'set' command\r\n");
let mut out = Vec::new();
let _ = s
.network_setex(&[b"k", b"abc", b"v"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not an integer or out of range.\r\n");
let mut out = Vec::new();
let _ = s
.network_psetex(&[b"p", b"500", b"v"], batch, &mut out)
.unwrap();
assert_eq!(out, b"+OK\r\n");
let ttl = ttl_of_sync(batch, b"p").unwrap().unwrap().unwrap();
assert!(ttl > now_unix_ms() + 400);
});
}
#[test]
fn getex_persist_and_ex() {
with_batch(|s, batch| {
let _ = s
.network_set(&[b"k", b"v"], batch, &mut Vec::new())
.unwrap();
let _ = put_ttl_sync(batch, b"k", now_unix_ms() + 60_000).unwrap();
let mut out = Vec::new();
let _ = s
.network_getex(&[b"k", b"PERSIST"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$1\r\nv\r\n");
assert_eq!(ttl_of_sync(batch, b"k").unwrap(), Some(None));
let mut out = Vec::new();
let _ = s
.network_getex(&[b"k", b"EX", b"100"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$1\r\nv\r\n");
assert!(ttl_of_sync(batch, b"k").unwrap().unwrap().unwrap() > now_unix_ms() + 99_000);
let mut out = Vec::new();
let _ = s
.network_getex(&[b"k", b"PX", b"0"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is out of range, must be positive.\r\n");
let mut out = Vec::new();
let _ = s
.network_getex(&[b"k", b"EXAT", b"100"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$1\r\nv\r\n");
let ttl = ttl_of_sync(batch, b"k").unwrap().unwrap().unwrap();
assert!(ttl < now_unix_ms() + 200_000);
let mut out = Vec::new();
let _ = s
.network_getex(&[b"k", b"EXXX", b"100"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR Unsupported option EXXX\r\n");
});
}
#[test]
fn getset_returns_old_value() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_getset(&[b"k", b"v2"], batch, &mut out).unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b"$2\r\nv2\r\n");
let mut out = Vec::new();
let _ = s.network_getset(&[b"k", b"v3"], batch, &mut out).unwrap();
assert_eq!(out, b"$2\r\nv2\r\n");
});
}
#[test]
fn set_nx_xx_get_variants() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v1", b"NX"], batch, &mut out)
.unwrap();
assert_eq!(out, b"+OK\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v2", b"NX"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v3", b"XX", b"GET"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$2\r\nv1\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"missing", b"v", b"XX", b"GET"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"missing"], batch, &mut out).unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"t", b"v", b"EX", b"100"], batch, &mut out)
.unwrap();
assert_eq!(out, b"+OK\r\n");
let ttl = ttl_of_sync(batch, b"t").unwrap().unwrap().unwrap();
assert!(ttl > now_unix_ms() + 99_000);
let _ = put_ttl_sync(batch, b"t", now_unix_ms() + 50_000).unwrap();
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"t", b"v2", b"KEEPTTL"], batch, &mut out)
.unwrap();
assert_eq!(out, b"+OK\r\n");
let ttl = ttl_of_sync(batch, b"t").unwrap().unwrap().unwrap();
assert!(ttl > now_unix_ms() + 49_000 && ttl < now_unix_ms() + 51_000);
});
}
#[test]
fn setexnx_option_errors_match_csharp() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v", b"EXAT", b"100"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v", b"EX"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v", b"NX", b"NX"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s
.network_setexnx(&[b"k", b"v", b"WHATEVER"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR unknown command\r\n");
});
}
#[test]
fn setrange_getrange_bounds() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s
.network_set_range(&[b"k", b"1", b"ab"], batch, &mut out)
.unwrap();
assert_eq!(out, b":3\r\n");
let mut out = Vec::new();
let _ = s
.network_set_range(&[b"k", b"5", b"cd"], batch, &mut out)
.unwrap();
assert_eq!(out, b":7\r\n");
let mut out = Vec::new();
let _ = s.network_get(&[b"k"], batch, &mut out).unwrap();
assert_eq!(out, b"$7\r\n\x00ab\x00\x00cd\r\n");
let mut out = Vec::new();
let _ = s
.network_set_range(&[b"k", b"-1", b"x"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR offset is out of range\r\n");
let mut out = Vec::new();
let _ = s
.network_set_range(&[b"k", b"x", b"x"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR value is not an integer or out of range.\r\n");
let mut out = Vec::new();
let _ = s
.network_get_range(&[b"k", b"-2", b"-1"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$2\r\ncd\r\n");
let mut out = Vec::new();
let _ = s
.network_get_range(&[b"k", b"0", b"999"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$7\r\n\x00ab\x00\x00cd\r\n");
let mut out = Vec::new();
let _ = s
.network_get_range(&[b"k", b"5", b"2"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$0\r\n\r\n");
});
}
#[test]
fn simple_frames_ping_echo_time_quit() {
let mut s = RespServerSession::default();
let mut out = Vec::new();
let _ = s.network_ping(&[], &mut out).unwrap();
assert_eq!(out, b"+PONG\r\n");
let mut out = Vec::new();
let _ = s.network_ping(&[b"hey"], &mut out).unwrap();
assert_eq!(out, b"$3\r\nhey\r\n");
let mut out = Vec::new();
let _ = s.network_ping(&[b"a", b"b"], &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'PING' command\r\n"
);
let mut out = Vec::new();
let _ = s.network_echo(&[b"msg"], &mut out).unwrap();
assert_eq!(out, b"$3\r\nmsg\r\n");
let mut out = Vec::new();
let _ = s.network_echo(&[], &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'ECHO' command\r\n"
);
let mut out = Vec::new();
let _ = s.network_time(&[b"x"], &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'TIME' command\r\n"
);
let mut out = Vec::new();
let _ = s.network_time(&[], &mut out).unwrap();
assert!(out.starts_with(b"*2\r\n$"));
let tail = &out[out.len() - 12..];
assert!(tail.starts_with(b"$6\r\n") && tail.ends_with(b"\r\n"));
let mut out = Vec::new();
let _ = s.network_quit(&[], &mut out).unwrap();
assert_eq!(out, b"+OK\r\n");
}
#[test]
fn auth_default_noauthenticator_error() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_auth(&[b"pass"], batch, &mut out).unwrap();
assert_eq!(
out,
b"-ERR Client sent AUTH, but configured authenticator does not accept passwords\r\n"
);
let mut out = Vec::new();
let _ = s.network_auth(&[], batch, &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'AUTH' command\r\n"
);
let mut out = Vec::new();
let _ = s
.network_auth(&[b"u", b"p", b"x"], batch, &mut out)
.unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'AUTH' command\r\n"
);
});
}
#[test]
fn hello_validation_and_frame() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_hello(&[b"4"], batch, &mut out).unwrap();
assert_eq!(out, b"-ERR Unsupported protocol version\r\n");
let mut out = Vec::new();
let _ = s.network_hello(&[b"abc"], batch, &mut out).unwrap();
assert_eq!(
out,
b"-ERR Protocol version is not an integer or out of range.\r\n"
);
let mut out = Vec::new();
let _ = s
.network_hello(
&[b"2", b"AUTH", b"a", b"b", b"SETNAME", b"c", b"d"],
batch,
&mut out,
)
.unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'HELLO' command\r\n"
);
let mut out = Vec::new();
let _ = s
.network_hello(&[b"2", b"AUTH", b"u"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR Syntax error in HELLO option 'AUTH'\r\n");
let mut out = Vec::new();
let _ = s
.network_hello(&[b"2", b"SETNAME", b"a b"], batch, &mut out)
.unwrap();
assert_eq!(
out,
b"-ERR Client names cannot contain spaces, newlines or special characters.\r\n"
);
let mut out = Vec::new();
let _ = s.network_hello(&[b"2", b"WHAT"], batch, &mut out).unwrap();
assert_eq!(out, b"-ERR Syntax error in HELLO option 'WHAT'\r\n");
let mut out = Vec::new();
let _ = s.network_hello(&[], batch, &mut out).unwrap();
assert!(out.starts_with(b"*16\r\n$6\r\nserver\r\n$5\r\nredis\r\n"));
let mut out = Vec::new();
let _ = s
.process_hello_command(Some(3), b"user", None, batch, &mut out)
.unwrap();
assert_eq!(out, b"-WRONGPASS Invalid username/password combination\r\n");
});
}
#[test]
fn command_family_degradation_frames() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_command(&[], batch, &mut out).unwrap();
assert_eq!(out, b"*0\r\n");
let mut out = Vec::new();
let _ = s.network_command_count(&[], batch, &mut out).unwrap();
assert_eq!(out, b":0\r\n");
let mut out = Vec::new();
let _ = s.network_command_count(&[b"X"], batch, &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'COMMAND|COUNT' command\r\n"
);
let mut out = Vec::new();
let _ = s.network_command(&[b"WHAT"], batch, &mut out).unwrap();
assert_eq!(out, b"-ERR unknown subcommand 'WHAT'. Try COMMAND HELP\r\n");
let mut out = Vec::new();
let _ = s.network_command_docs(&[], batch, &mut out).unwrap();
assert_eq!(out, b"*0\r\n");
let mut out = Vec::new();
let _ = s.network_command_info(&[b"get"], batch, &mut out).unwrap();
assert_eq!(out, b"*1\r\n$-1\r\n");
let mut out = Vec::new();
let _ = s
.network_command_getkeys(&[b"get"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-Invalid command specified\r\n");
let mut out = Vec::new();
let _ = s
.network_command_getkeysandflags(&[], batch, &mut out)
.unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'COMMAND|GETKEYSANDFLAGS' command\r\n"
);
});
}
#[test]
fn object_and_memory_validation() {
with_batch(|s, batch| {
let _ = s
.network_set(&[b"k", b"v"], batch, &mut Vec::new())
.unwrap();
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Encoding, &[b"k"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$3\r\nraw\r\n");
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Refcount, &[b"k"], batch, &mut out)
.unwrap();
assert_eq!(out, b":1\r\n");
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Idletime, &[b"k"], batch, &mut out)
.unwrap();
assert_eq!(out, b":0\r\n");
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Freq, &[b"k"], batch, &mut out)
.unwrap();
assert_eq!(
out,
&b"-ERR OBJECT FREQ is not supported: Garnet does not track access frequency (no LFU maxmemory policy).\x0d\x0a"[..]
);
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Idletime, &[b"nope"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$-1\r\n");
let mut out = Vec::new();
let _ = s
.network_object(ObjectSubCmd::Encoding, &[], batch, &mut out)
.unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'object|encoding' command\r\n"
);
let mut out = Vec::new();
let _ = s
.network_memory_usage(&[b"k", b"SAMPLES", b"-1"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s
.network_memory_usage(&[b"k", b"BAD", b"1"], batch, &mut out)
.unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s
.network_memory_usage(&[b"k", b"SAMPLES", b"5"], batch, &mut out)
.unwrap();
assert_eq!(out, b"$-1\r\n");
});
}
#[test]
fn object_help_frame() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_objecthelp(&[], batch, &mut out).unwrap();
assert!(out.starts_with(b"*11\r\n+OBJECT <subcommand>"));
let mut out = Vec::new();
let _ = s.network_objecthelp(&[b"x"], batch, &mut out).unwrap();
assert_eq!(
out,
b"-ERR wrong number of arguments for 'object|help' command\r\n"
);
});
}
#[test]
fn async_rejected_on_resp2() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_async(&[b"ON"], batch, &mut out).unwrap();
assert_eq!(out, b"-ERR command not supported in RESP2\r\n");
});
}
#[test]
fn flush_options_and_unavailable_flush() {
let mut s = RespServerSession::default();
let mut out = Vec::new();
let _ = s.network_flushdb(&[b"WHAT"], &mut out).unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s.network_flushall(&[b"ASYNC", b"SYNC"], &mut out).unwrap();
assert_eq!(out, b"-ERR syntax error\r\n");
let mut out = Vec::new();
let _ = s.network_flushdb(&[], &mut out).unwrap();
assert_eq!(out, b"-ERR generic error\r\n");
let mut out = Vec::new();
let _ = s.network_flushall(&[], &mut out).unwrap();
assert_eq!(out, b"-ERR generic error\r\n");
}
#[test]
fn readonly_readwrite_asking_frames() {
with_batch(|s, batch| {
let mut out = Vec::new();
let _ = s.network_asking(&[], batch, &mut out).unwrap();
assert_eq!(out, b"+OK\r\n");
let mut out = Vec::new();
let _ = s.network_readonly(&[], batch, &mut out).unwrap();
assert_eq!(out, b"+OK\r\n");
let mut out = Vec::new();
let _ = s.network_readwrite(&[], batch, &mut out).unwrap();
assert_eq!(out, b"+OK\r\n");
});
}
#[test]
fn client_name_validation() {
assert_eq!(try_get_client_name(b"client-1"), Some("client-1"));
assert_eq!(try_get_client_name(b""), Some(""));
assert_eq!(try_get_client_name(b"a b"), None);
assert_eq!(try_get_client_name(b"\xff"), None);
}
#[test]
fn digits_len_counts_decimal_digits() {
assert_eq!(digits_len(0), 1);
assert_eq!(digits_len(9), 1);
assert_eq!(digits_len(10), 2);
assert_eq!(digits_len(1_700_000_000), 10);
}
}