diff --git a/crates/ember-server/src/grpc.rs b/crates/ember-server/src/grpc.rs index 0c0492a3..b012a212 100644 --- a/crates/ember-server/src/grpc.rs +++ b/crates/ember-server/src/grpc.rs @@ -11,6 +11,7 @@ use std::time::{Duration, Instant}; use bytes::Bytes; use ember_core::{Engine, ShardRequest, ShardResponse, TtlResult, Value}; +use ember_protocol::command::ScoreBound; use subtle::ConstantTimeEq; use tokio_stream::wrappers::ReceiverStream; use tonic::service::interceptor::InterceptedService; @@ -197,6 +198,40 @@ fn parse_expire(seconds: u64, millis: u64) -> Option { } } +/// Parses a Redis-style score bound string into a `ScoreBound`. +/// +/// Supports "-inf", "+inf", exclusive bounds like "(5.0", and inclusive +/// bounds like "5.0". Returns an error status for invalid strings. +#[allow(clippy::result_large_err)] +fn parse_score_bound(s: &str) -> Result { + match s { + "-inf" | "-INF" => Ok(ScoreBound::NegInf), + "+inf" | "+INF" | "inf" | "INF" => Ok(ScoreBound::PosInf), + s if s.starts_with('(') => s[1..].parse::().map(ScoreBound::Exclusive).map_err(|_| { + Status::invalid_argument(format!("invalid score bound: {s}")) + }), + s => s.parse::().map(ScoreBound::Inclusive).map_err(|_| { + Status::invalid_argument(format!("invalid score bound: {s}")) + }), + } +} + +/// Converts a ScoredArray response into a ZRangeResponse. +/// +/// When `with_scores` is false, scores are set to 0.0 — clients should +/// ignore them. This avoids a second round-trip to the engine. +fn scored_array_to_zrange(arr: Vec<(String, f64)>, with_scores: bool) -> ZRangeResponse { + ZRangeResponse { + members: arr + .into_iter() + .map(|(member, score)| ScoreMember { + member, + score: if with_scores { score } else { 0.0 }, + }) + .collect(), + } +} + #[tonic::async_trait] impl EmberCache for EmberService { // ----------------------------------------------------------------------- @@ -2097,171 +2132,1352 @@ impl EmberCache for EmberService { } // ----------------------------------------------------------------------- - // pipeline (bidirectional streaming) + // strings (extended) // ----------------------------------------------------------------------- - type PipelineStream = ReceiverStream>; + async fn get_del( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let key = request.into_inner().key; + validate_key(&key, &self.ctx.limits)?; + let resp = self + .route(&key, ShardRequest::GetDel { key: key.clone() }) + .await?; + self.record_command(start, "GETDEL"); - async fn pipeline( + match resp { + ShardResponse::Value(Some(v)) => Ok(Response::new(GetResponse { + value: Some(value_to_bytes(v)), + })), + ShardResponse::Value(None) => Ok(Response::new(GetResponse { value: None })), + other => Err(unexpected_response(&other)), + } + } + + async fn get_ex( &self, - request: Request>, - ) -> Result, Status> { - let mut stream = request.into_inner(); - let engine = self.engine.clone(); - let ctx = Arc::clone(&self.ctx); - let slow_log = Arc::clone(&self.slow_log); - let pubsub = Arc::clone(&self.pubsub); + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; - let (tx, rx) = tokio::sync::mpsc::channel(256); + // map proto expiry fields to the engine's Option> convention: + // None = leave TTL unchanged + // Some(None) = persist (remove TTL) + // Some(Some(ms)) = set TTL to this many milliseconds + let expire = if req.persist { + Some(None) + } else if req.expire_millis > 0 { + Some(Some(req.expire_millis)) + } else if req.expire_seconds > 0 { + Some(Some(req.expire_seconds * 1_000)) + } else { + None + }; - tokio::spawn(async move { - let svc = EmberService::new(engine, ctx, slow_log, pubsub); - while let Ok(Some(req)) = stream.message().await { - let id = req.id; - let result = handle_pipeline_command(&svc, req).await; - let resp = match result { - Ok(pr) => pr, - Err(status) => PipelineResponse { - id, - result: Some(pipeline_response::Result::Error(ErrorResponse { - message: status.message().to_string(), - kind: ErrorKind::Internal as i32, - })), - }, - }; - if tx.send(Ok(resp)).await.is_err() { - break; - } - } - }); + let resp = self + .route( + &req.key, + ShardRequest::GetEx { + key: req.key.clone(), + expire, + }, + ) + .await?; + self.record_command(start, "GETEX"); - Ok(Response::new(ReceiverStream::new(rx))) + match resp { + ShardResponse::Value(Some(v)) => Ok(Response::new(GetResponse { + value: Some(value_to_bytes(v)), + })), + ShardResponse::Value(None) => Ok(Response::new(GetResponse { value: None })), + other => Err(unexpected_response(&other)), + } } -} -/// Dispatches a single pipeline command to the appropriate RPC handler. -/// Dispatches a single pipeline command to the service method and wraps the -/// response in a PipelineResponse. Each arm follows the same pattern: call -/// the service method, extract the inner response, wrap it in the correct -/// result variant. The macro eliminates the boilerplate of ~57 identical arms. -macro_rules! pipeline_dispatch { - ($svc:expr, $id:expr, $cmd:expr, { - $( $Variant:ident => $method:ident => $Result:ident ),* $(,)? - }) => { - match $cmd { - $( - pipeline_request::Command::$Variant(r) => { - let resp = $svc.$method(Request::new(r)).await?; - Ok(PipelineResponse { - id: $id, - result: Some(pipeline_response::Result::$Result(resp.into_inner())), - }) - } - )* + async fn get_range( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::GetRange { + key: req.key.clone(), + start: req.start, + end: req.end, + }, + ) + .await?; + self.record_command(start, "GETRANGE"); + + match resp { + ShardResponse::Value(Some(v)) => Ok(Response::new(GetResponse { + value: Some(value_to_bytes(v)), + })), + ShardResponse::Value(None) => Ok(Response::new(GetResponse { value: None })), + other => Err(unexpected_response(&other)), } - }; -} + } -async fn handle_pipeline_command( - svc: &EmberService, - req: PipelineRequest, -) -> Result { - let id = req.id; - let cmd = req - .command - .ok_or_else(|| Status::invalid_argument("missing command"))?; + async fn set_range( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + validate_value(&req.value, &self.ctx.limits)?; + if req.offset < 0 { + return Err(Status::invalid_argument("offset must not be negative")); + } + let resp = self + .route( + &req.key, + ShardRequest::SetRange { + key: req.key.clone(), + offset: req.offset as usize, + value: bytes::Bytes::from(req.value), + }, + ) + .await?; + self.record_command(start, "SETRANGE"); - pipeline_dispatch!(svc, id, cmd, { - // string commands - Get => get => Get, - Set => set => Set, - Del => del => Del, - Exists => exists => IntVal, - Incr => incr => IntVal, - IncrBy => incr_by => IntVal, - DecrBy => decr_by => IntVal, - IncrByFloat => incr_by_float => FloatVal, - Append => append => IntVal, - Strlen => strlen => IntVal, + match resp { + ShardResponse::Len(n) => Ok(Response::new(IntResponse { value: n as i64 })), + other => Err(unexpected_response(&other)), + } + } - // ttl / expiry - Expire => expire => BoolVal, - Pexpire => p_expire => BoolVal, - Persist => persist => BoolVal, - Ttl => ttl => Ttl, - Pttl => p_ttl => Ttl, - Type => r#type => Type, + // ----------------------------------------------------------------------- + // keys (extended) + // ----------------------------------------------------------------------- - // list commands - Lpush => l_push => IntVal, - Rpush => r_push => IntVal, - Lpop => l_pop => Get, - Rpop => r_pop => Get, - Lrange => l_range => Array, - Llen => l_len => IntVal, + async fn copy(&self, request: Request) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.source, &self.ctx.limits)?; + validate_key(&req.destination, &self.ctx.limits)?; - // hash commands - Hset => h_set => IntVal, - Hget => h_get => Get, - Hgetall => h_get_all => Hash, - Hdel => h_del => IntVal, - Hexists => h_exists => BoolVal, - Hlen => h_len => IntVal, - HincrBy => h_incr_by => IntVal, - Hkeys => h_keys => Keys, - Hvals => h_vals => Array, - Hmget => hm_get => OptionalArray, + if !self.engine.same_shard(&req.source, &req.destination) { + return Err(Status::failed_precondition( + "ERR source and destination keys must hash to the same shard", + )); + } - // set commands - Sadd => s_add => IntVal, - Srem => s_rem => IntVal, - Smembers => s_members => Keys, - Sismember => s_is_member => BoolVal, - Scard => s_card => IntVal, + let resp = self + .route( + &req.source, + ShardRequest::Copy { + source: req.source.clone(), + destination: req.destination, + replace: req.replace, + }, + ) + .await?; + self.record_command(start, "COPY"); - // sorted set commands - Zadd => z_add => IntVal, - Zrem => z_rem => IntVal, - Zscore => z_score => OptionalFloat, - Zrank => z_rank => OptionalInt, - Zcard => z_card => IntVal, - Zrange => z_range => Zrange, + match resp { + ShardResponse::Bool(v) => Ok(Response::new(BoolResponse { value: v })), + ShardResponse::Err(msg) => Err(Status::not_found(msg)), + ShardResponse::OutOfMemory => Err(Status::resource_exhausted("OOM")), + other => Err(unexpected_response(&other)), + } + } - // vector commands - Vadd => v_add => BoolVal, - VaddBatch => v_add_batch => IntVal, - Vsim => v_sim => Vsim, - Vrem => v_rem => BoolVal, - Vget => v_get => Vget, - Vcard => v_card => IntVal, - Vdim => v_dim => IntVal, - Vinfo => v_info => Vinfo, + async fn random_key( + &self, + _request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let responses = self.broadcast(|| ShardRequest::RandomKey).await?; - // server commands - Ping => ping => Ping, - Echo => echo => Echo, - Decr => decr => IntVal, - Unlink => unlink => Del, - Flushdb => flush_db => Status, - Dbsize => db_size => IntVal, - Bgsave => bg_save => Status, - Bgrewriteaof => bg_rewrite_aof => Status, - Mget => m_get => Mget, - Mset => m_set => Mset, - Keys => keys => Keys, - Rename => rename => Status, - Scan => scan => Scan, + // pick the first non-empty result from any shard + let key = responses.into_iter().find_map(|resp| match resp { + ShardResponse::StringArray(mut v) if !v.is_empty() => Some(v.remove(0)), + _ => None, + }); - // slowlog - SlowlogGet => slow_log_get => SlowlogGet, - SlowlogLen => slow_log_len => IntVal, - SlowlogReset => slow_log_reset => Status, + self.record_command(start, "RANDOMKEY"); + Ok(Response::new(GetResponse { + value: key.map(|k| k.into_bytes()), + })) + } - // pub/sub (unary only — Subscribe is streaming) - Publish => publish => IntVal, - PubsubChannels => pub_sub_channels => Keys, - PubsubNumsub => pub_sub_num_sub => PubsubNumsub, - PubsubNumpat => pub_sub_num_pat => IntVal, + async fn touch(&self, request: Request) -> Result, Status> { + let start = Instant::now(); + let keys = request.into_inner().keys; + for k in &keys { + validate_key(k, &self.ctx.limits)?; + } + + let responses = self + .engine + .route_multi(&keys, |k| ShardRequest::Touch { key: k }) + .await + .map_err(|_| Status::unavailable("shard unavailable"))?; + + let mut count = 0i64; + for resp in responses { + if let ShardResponse::Bool(true) = resp { + count += 1; + } + } + self.record_command(start, "TOUCH"); + Ok(Response::new(IntResponse { value: count })) + } + + // ----------------------------------------------------------------------- + // lists (extended) + // ----------------------------------------------------------------------- + + async fn l_index( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::LIndex { + key: req.key.clone(), + index: req.index, + }, + ) + .await?; + self.record_command(start, "LINDEX"); + + match resp { + ShardResponse::Value(Some(v)) => Ok(Response::new(GetResponse { + value: Some(value_to_bytes(v)), + })), + ShardResponse::Value(None) => Ok(Response::new(GetResponse { value: None })), + other => Err(unexpected_response(&other)), + } + } + + async fn l_set( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + validate_value(&req.value, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::LSet { + key: req.key.clone(), + index: req.index, + value: bytes::Bytes::from(req.value), + }, + ) + .await?; + self.record_command(start, "LSET"); + + match resp { + ShardResponse::Ok => Ok(Response::new(StatusResponse { + status: "OK".to_string(), + })), + ShardResponse::Err(msg) => Err(Status::failed_precondition(msg)), + other => Err(unexpected_response(&other)), + } + } + + async fn l_trim( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::LTrim { + key: req.key.clone(), + start: req.start, + stop: req.stop, + }, + ) + .await?; + self.record_command(start, "LTRIM"); + + match resp { + ShardResponse::Ok => Ok(Response::new(StatusResponse { + status: "OK".to_string(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn l_insert( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::LInsert { + key: req.key.clone(), + before: req.before, + pivot: bytes::Bytes::from(req.pivot), + value: bytes::Bytes::from(req.value), + }, + ) + .await?; + self.record_command(start, "LINSERT"); + + match resp { + ShardResponse::Integer(n) => Ok(Response::new(IntResponse { value: n })), + ShardResponse::OutOfMemory => Err(Status::resource_exhausted("OOM")), + other => Err(unexpected_response(&other)), + } + } + + async fn l_rem( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::LRem { + key: req.key.clone(), + count: req.count, + value: bytes::Bytes::from(req.value), + }, + ) + .await?; + self.record_command(start, "LREM"); + + match resp { + ShardResponse::Len(n) => Ok(Response::new(IntResponse { value: n as i64 })), + other => Err(unexpected_response(&other)), + } + } + + async fn l_pos( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + // if count is absent or 0, find first occurrence (count=0 means "all" in + // the engine, but we only return one result via OptionalIntResponse). + let count = req.count.unwrap_or(0) as usize; + let resp = self + .route( + &req.key, + ShardRequest::LPos { + key: req.key.clone(), + element: bytes::Bytes::from(req.value), + rank: 0, + count, + maxlen: 0, + }, + ) + .await?; + self.record_command(start, "LPOS"); + + match resp { + ShardResponse::IntegerArray(positions) => Ok(Response::new(OptionalIntResponse { + value: positions.into_iter().next(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn l_move( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.source, &self.ctx.limits)?; + validate_key(&req.destination, &self.ctx.limits)?; + + if !self.engine.same_shard(&req.source, &req.destination) { + return Err(Status::failed_precondition( + "ERR source and destination keys must hash to the same shard", + )); + } + + let resp = self + .route( + &req.source, + ShardRequest::LMove { + source: req.source.clone(), + destination: req.destination, + src_left: req.src_left, + dst_left: req.dst_left, + }, + ) + .await?; + self.record_command(start, "LMOVE"); + + match resp { + ShardResponse::Value(Some(v)) => Ok(Response::new(GetResponse { + value: Some(value_to_bytes(v)), + })), + ShardResponse::Value(None) => Ok(Response::new(GetResponse { value: None })), + ShardResponse::OutOfMemory => Err(Status::resource_exhausted("OOM")), + other => Err(unexpected_response(&other)), + } + } + + // ----------------------------------------------------------------------- + // sets (extended) + // ----------------------------------------------------------------------- + + async fn s_union( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let keys = request.into_inner().keys; + if keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &keys { + validate_key(k, &self.ctx.limits)?; + } + let resp = self + .route(&keys[0].clone(), ShardRequest::SUnion { keys }) + .await?; + self.record_command(start, "SUNION"); + + match resp { + ShardResponse::StringArray(members) => Ok(Response::new(KeysResponse { keys: members })), + other => Err(unexpected_response(&other)), + } + } + + async fn s_inter( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let keys = request.into_inner().keys; + if keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &keys { + validate_key(k, &self.ctx.limits)?; + } + let resp = self + .route(&keys[0].clone(), ShardRequest::SInter { keys }) + .await?; + self.record_command(start, "SINTER"); + + match resp { + ShardResponse::StringArray(members) => Ok(Response::new(KeysResponse { keys: members })), + other => Err(unexpected_response(&other)), + } + } + + async fn s_diff( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let keys = request.into_inner().keys; + if keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &keys { + validate_key(k, &self.ctx.limits)?; + } + let resp = self + .route(&keys[0].clone(), ShardRequest::SDiff { keys }) + .await?; + self.record_command(start, "SDIFF"); + + match resp { + ShardResponse::StringArray(members) => Ok(Response::new(KeysResponse { keys: members })), + other => Err(unexpected_response(&other)), + } + } + + async fn s_union_store( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.destination, &self.ctx.limits)?; + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let dest = req.destination.clone(); + let resp = self + .route( + &dest, + ShardRequest::SUnionStore { + dest: req.destination, + keys: req.keys, + }, + ) + .await?; + self.record_command(start, "SUNIONSTORE"); + + match resp { + ShardResponse::SetStoreResult { count, .. } => { + Ok(Response::new(IntResponse { value: count as i64 })) + } + other => Err(unexpected_response(&other)), + } + } + + async fn s_inter_store( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.destination, &self.ctx.limits)?; + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let dest = req.destination.clone(); + let resp = self + .route( + &dest, + ShardRequest::SInterStore { + dest: req.destination, + keys: req.keys, + }, + ) + .await?; + self.record_command(start, "SINTERSTORE"); + + match resp { + ShardResponse::SetStoreResult { count, .. } => { + Ok(Response::new(IntResponse { value: count as i64 })) + } + other => Err(unexpected_response(&other)), + } + } + + async fn s_diff_store( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.destination, &self.ctx.limits)?; + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let dest = req.destination.clone(); + let resp = self + .route( + &dest, + ShardRequest::SDiffStore { + dest: req.destination, + keys: req.keys, + }, + ) + .await?; + self.record_command(start, "SDIFFSTORE"); + + match resp { + ShardResponse::SetStoreResult { count, .. } => { + Ok(Response::new(IntResponse { value: count as i64 })) + } + other => Err(unexpected_response(&other)), + } + } + + async fn s_rand_member( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::SRandMember { + key: req.key.clone(), + count: req.count as i64, + }, + ) + .await?; + self.record_command(start, "SRANDMEMBER"); + + match resp { + ShardResponse::StringArray(members) => Ok(Response::new(ArrayResponse { + values: members.into_iter().map(|s| s.into_bytes()).collect(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn s_pop( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::SPop { + key: req.key.clone(), + count: req.count as usize, + }, + ) + .await?; + self.record_command(start, "SPOP"); + + match resp { + ShardResponse::StringArray(members) => Ok(Response::new(ArrayResponse { + values: members.into_iter().map(|s| s.into_bytes()).collect(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn s_mis_member( + &self, + request: Request, + ) -> Result, Status> { + + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::SMisMember { + key: req.key.clone(), + members: req.members, + }, + ) + .await?; + self.record_command(start, "SMISMEMBER"); + + match resp { + ShardResponse::BoolArray(results) => { + Ok(Response::new(BoolArrayResponse { values: results })) + } + other => Err(unexpected_response(&other)), + } + } + + // ----------------------------------------------------------------------- + // hashes (extended) + // ----------------------------------------------------------------------- + + async fn h_scan( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let count = if req.count == 0 { 10 } else { req.count as usize }; + let resp = self + .route( + &req.key, + ShardRequest::HScan { + key: req.key.clone(), + cursor: req.cursor, + count, + pattern: req.pattern, + }, + ) + .await?; + self.record_command(start, "HSCAN"); + + match resp { + ShardResponse::CollectionScan { cursor, items } => { + // items are interleaved: [field, value, field, value, ...] + let fields = items + .chunks(2) + .filter_map(|pair| { + if pair.len() == 2 { + Some(FieldValue { + field: String::from_utf8_lossy(&pair[0]).into_owned(), + value: pair[1].to_vec(), + }) + } else { + None + } + }) + .collect(); + Ok(Response::new(HScanResponse { cursor, fields })) + } + other => Err(unexpected_response(&other)), + } + } + + // ----------------------------------------------------------------------- + // sorted sets (extended) + // ----------------------------------------------------------------------- + + async fn z_rev_rank( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::ZRevRank { + key: req.key.clone(), + member: req.member, + }, + ) + .await?; + self.record_command(start, "ZREVRANK"); + + match resp { + ShardResponse::Rank(r) => Ok(Response::new(OptionalIntResponse { + value: r.map(|n| n as i64), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn z_rev_range( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let with_scores = req.with_scores; + let resp = self + .route( + &req.key, + ShardRequest::ZRevRange { + key: req.key.clone(), + start: req.start, + stop: req.stop, + with_scores, + }, + ) + .await?; + self.record_command(start, "ZREVRANGE"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_count( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let min = parse_score_bound(&req.min)?; + let max = parse_score_bound(&req.max)?; + let resp = self + .route( + &req.key, + ShardRequest::ZCount { + key: req.key.clone(), + min, + max, + }, + ) + .await?; + self.record_command(start, "ZCOUNT"); + + match resp { + ShardResponse::Len(n) => Ok(Response::new(IntResponse { value: n as i64 })), + other => Err(unexpected_response(&other)), + } + } + + async fn z_incr_by( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::ZIncrBy { + key: req.key.clone(), + increment: req.delta, + member: req.member, + }, + ) + .await?; + self.record_command(start, "ZINCRBY"); + + match resp { + ShardResponse::ZIncrByResult { new_score, .. } => Ok(Response::new(FloatResponse { + value: new_score.to_string(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn z_range_by_score( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let min = parse_score_bound(&req.min)?; + let max = parse_score_bound(&req.max)?; + let with_scores = req.with_scores; + let offset = req.offset.unwrap_or(0).max(0) as usize; + let count = req.count.map(|c| c.max(0) as usize); + let resp = self + .route( + &req.key, + ShardRequest::ZRangeByScore { + key: req.key.clone(), + min, + max, + offset, + count, + }, + ) + .await?; + self.record_command(start, "ZRANGEBYSCORE"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_rev_range_by_score( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + // note: for ZREVRANGEBYSCORE, max and min are swapped in the proto + let max = parse_score_bound(&req.max)?; + let min = parse_score_bound(&req.min)?; + let with_scores = req.with_scores; + let offset = req.offset.unwrap_or(0).max(0) as usize; + let count = req.count.map(|c| c.max(0) as usize); + let resp = self + .route( + &req.key, + ShardRequest::ZRevRangeByScore { + key: req.key.clone(), + min, + max, + offset, + count, + }, + ) + .await?; + self.record_command(start, "ZREVRANGEBYSCORE"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_pop_min( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::ZPopMin { + key: req.key.clone(), + count: req.count as usize, + }, + ) + .await?; + self.record_command(start, "ZPOPMIN"); + + match resp { + ShardResponse::ZPopResult(pairs) => Ok(Response::new(ZRangeResponse { + members: pairs + .into_iter() + .map(|(member, score)| ScoreMember { member, score }) + .collect(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn z_pop_max( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let resp = self + .route( + &req.key, + ShardRequest::ZPopMax { + key: req.key.clone(), + count: req.count as usize, + }, + ) + .await?; + self.record_command(start, "ZPOPMAX"); + + match resp { + ShardResponse::ZPopResult(pairs) => Ok(Response::new(ZRangeResponse { + members: pairs + .into_iter() + .map(|(member, score)| ScoreMember { member, score }) + .collect(), + })), + other => Err(unexpected_response(&other)), + } + } + + async fn z_diff( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + if req.keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let with_scores = req.with_scores; + let first_key = req.keys[0].clone(); + let resp = self + .route(&first_key, ShardRequest::ZDiff { keys: req.keys }) + .await?; + self.record_command(start, "ZDIFF"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_inter( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + if req.keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let with_scores = req.with_scores; + let first_key = req.keys[0].clone(); + let resp = self + .route(&first_key, ShardRequest::ZInter { keys: req.keys }) + .await?; + self.record_command(start, "ZINTER"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_union( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + if req.keys.is_empty() { + return Err(Status::invalid_argument("at least one key required")); + } + for k in &req.keys { + validate_key(k, &self.ctx.limits)?; + } + let with_scores = req.with_scores; + let first_key = req.keys[0].clone(); + let resp = self + .route(&first_key, ShardRequest::ZUnion { keys: req.keys }) + .await?; + self.record_command(start, "ZUNION"); + + match resp { + ShardResponse::ScoredArray(arr) => { + Ok(Response::new(scored_array_to_zrange(arr, with_scores))) + } + other => Err(unexpected_response(&other)), + } + } + + async fn z_scan( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let count = if req.count == 0 { 10 } else { req.count as usize }; + let resp = self + .route( + &req.key, + ShardRequest::ZScan { + key: req.key.clone(), + cursor: req.cursor, + count, + pattern: req.pattern, + }, + ) + .await?; + self.record_command(start, "ZSCAN"); + + match resp { + ShardResponse::CollectionScan { cursor, items } => { + // items are interleaved: [member, score_str, member, score_str, ...] + let members = items + .chunks(2) + .filter_map(|pair| { + if pair.len() == 2 { + let member = String::from_utf8_lossy(&pair[0]).into_owned(); + let score = String::from_utf8_lossy(&pair[1]) + .parse::() + .unwrap_or(0.0); + Some(ScoreMember { member, score }) + } else { + None + } + }) + .collect(); + Ok(Response::new(ZScanResponse { cursor, members })) + } + other => Err(unexpected_response(&other)), + } + } + + // ----------------------------------------------------------------------- + // scans + // ----------------------------------------------------------------------- + + async fn s_scan( + &self, + request: Request, + ) -> Result, Status> { + let start = Instant::now(); + let req = request.into_inner(); + validate_key(&req.key, &self.ctx.limits)?; + let count = if req.count == 0 { 10 } else { req.count as usize }; + let resp = self + .route( + &req.key, + ShardRequest::SScan { + key: req.key.clone(), + cursor: req.cursor, + count, + pattern: req.pattern, + }, + ) + .await?; + self.record_command(start, "SSCAN"); + + match resp { + ShardResponse::CollectionScan { cursor, items } => { + let members = items + .into_iter() + .map(|b| String::from_utf8_lossy(&b).into_owned()) + .collect(); + Ok(Response::new(SScanResponse { cursor, members })) + } + other => Err(unexpected_response(&other)), + } + } + + // ----------------------------------------------------------------------- + // server (extended) + // ----------------------------------------------------------------------- + + async fn time(&self, _request: Request) -> Result, Status> { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + Ok(Response::new(TimeResponse { + seconds: now.as_secs() as i64, + microseconds: now.subsec_micros() as i64, + })) + } + + async fn last_save( + &self, + _request: Request, + ) -> Result, Status> { + use std::sync::atomic::Ordering; + Ok(Response::new(IntResponse { + value: self.ctx.last_save_timestamp.load(Ordering::Relaxed) as i64, + })) + } + + // ----------------------------------------------------------------------- + // pipeline (bidirectional streaming) + // ----------------------------------------------------------------------- + + type PipelineStream = ReceiverStream>; + + async fn pipeline( + &self, + request: Request>, + ) -> Result, Status> { + let mut stream = request.into_inner(); + let engine = self.engine.clone(); + let ctx = Arc::clone(&self.ctx); + let slow_log = Arc::clone(&self.slow_log); + let pubsub = Arc::clone(&self.pubsub); + + let (tx, rx) = tokio::sync::mpsc::channel(256); + + tokio::spawn(async move { + let svc = EmberService::new(engine, ctx, slow_log, pubsub); + while let Ok(Some(req)) = stream.message().await { + let id = req.id; + let result = handle_pipeline_command(&svc, req).await; + let resp = match result { + Ok(pr) => pr, + Err(status) => PipelineResponse { + id, + result: Some(pipeline_response::Result::Error(ErrorResponse { + message: status.message().to_string(), + kind: ErrorKind::Internal as i32, + })), + }, + }; + if tx.send(Ok(resp)).await.is_err() { + break; + } + } + }); + + Ok(Response::new(ReceiverStream::new(rx))) + } +} + +/// Dispatches a single pipeline command to the appropriate RPC handler. +/// Dispatches a single pipeline command to the service method and wraps the +/// response in a PipelineResponse. Each arm follows the same pattern: call +/// the service method, extract the inner response, wrap it in the correct +/// result variant. The macro eliminates the boilerplate of ~57 identical arms. +macro_rules! pipeline_dispatch { + ($svc:expr, $id:expr, $cmd:expr, { + $( $Variant:ident => $method:ident => $Result:ident ),* $(,)? + }) => { + match $cmd { + $( + pipeline_request::Command::$Variant(r) => { + let resp = $svc.$method(Request::new(r)).await?; + Ok(PipelineResponse { + id: $id, + result: Some(pipeline_response::Result::$Result(resp.into_inner())), + }) + } + )* + } + }; +} + +async fn handle_pipeline_command( + svc: &EmberService, + req: PipelineRequest, +) -> Result { + let id = req.id; + let cmd = req + .command + .ok_or_else(|| Status::invalid_argument("missing command"))?; + + pipeline_dispatch!(svc, id, cmd, { + // string commands + Get => get => Get, + Set => set => Set, + Del => del => Del, + Exists => exists => IntVal, + Incr => incr => IntVal, + IncrBy => incr_by => IntVal, + DecrBy => decr_by => IntVal, + IncrByFloat => incr_by_float => FloatVal, + Append => append => IntVal, + Strlen => strlen => IntVal, + + // ttl / expiry + Expire => expire => BoolVal, + Pexpire => p_expire => BoolVal, + Persist => persist => BoolVal, + Ttl => ttl => Ttl, + Pttl => p_ttl => Ttl, + Type => r#type => Type, + + // list commands + Lpush => l_push => IntVal, + Rpush => r_push => IntVal, + Lpop => l_pop => Get, + Rpop => r_pop => Get, + Lrange => l_range => Array, + Llen => l_len => IntVal, + + // hash commands + Hset => h_set => IntVal, + Hget => h_get => Get, + Hgetall => h_get_all => Hash, + Hdel => h_del => IntVal, + Hexists => h_exists => BoolVal, + Hlen => h_len => IntVal, + HincrBy => h_incr_by => IntVal, + Hkeys => h_keys => Keys, + Hvals => h_vals => Array, + Hmget => hm_get => OptionalArray, + + // set commands + Sadd => s_add => IntVal, + Srem => s_rem => IntVal, + Smembers => s_members => Keys, + Sismember => s_is_member => BoolVal, + Scard => s_card => IntVal, + + // sorted set commands + Zadd => z_add => IntVal, + Zrem => z_rem => IntVal, + Zscore => z_score => OptionalFloat, + Zrank => z_rank => OptionalInt, + Zcard => z_card => IntVal, + Zrange => z_range => Zrange, + + // vector commands + Vadd => v_add => BoolVal, + VaddBatch => v_add_batch => IntVal, + Vsim => v_sim => Vsim, + Vrem => v_rem => BoolVal, + Vget => v_get => Vget, + Vcard => v_card => IntVal, + Vdim => v_dim => IntVal, + Vinfo => v_info => Vinfo, + + // server commands + Ping => ping => Ping, + Echo => echo => Echo, + Decr => decr => IntVal, + Unlink => unlink => Del, + Flushdb => flush_db => Status, + Dbsize => db_size => IntVal, + Bgsave => bg_save => Status, + Bgrewriteaof => bg_rewrite_aof => Status, + Mget => m_get => Mget, + Mset => m_set => Mset, + Keys => keys => Keys, + Rename => rename => Status, + Scan => scan => Scan, + + // slowlog + SlowlogGet => slow_log_get => SlowlogGet, + SlowlogLen => slow_log_len => IntVal, + SlowlogReset => slow_log_reset => Status, + + // pub/sub (unary only — Subscribe is streaming) + Publish => publish => IntVal, + PubsubChannels => pub_sub_channels => Keys, + PubsubNumsub => pub_sub_num_sub => PubsubNumsub, + PubsubNumpat => pub_sub_num_pat => IntVal, + + // extended strings + GetDel => get_del => Get, + GetEx => get_ex => Get, + GetRange => get_range => Get, + SetRange => set_range => IntVal, + + // extended keys + Copy => copy => BoolVal, + RandomKey => random_key => Get, + Touch => touch => IntVal, + + // extended lists + Lindex => l_index => Get, + Lset => l_set => Status, + Ltrim => l_trim => Status, + Linsert => l_insert => IntVal, + Lrem => l_rem => IntVal, + Lpos => l_pos => OptionalInt, + Lmove => l_move => Get, + + // extended sets + Sunion => s_union => Keys, + Sinter => s_inter => Keys, + Sdiff => s_diff => Keys, + SunionStore => s_union_store => IntVal, + SinterStore => s_inter_store => IntVal, + SdiffStore => s_diff_store => IntVal, + SrandMember => s_rand_member => Array, + Spop => s_pop => Array, + Smismember => s_mis_member => BoolArray, + + // extended hashes + Hscan => h_scan => Hscan, + + // extended sorted sets + ZrevRank => z_rev_rank => OptionalInt, + ZrevRange => z_rev_range => Zrange, + Zcount => z_count => IntVal, + Zincrby => z_incr_by => FloatVal, + ZrangeByScore => z_range_by_score => Zrange, + ZrevRangeByScore => z_rev_range_by_score => Zrange, + Zpopmin => z_pop_min => Zrange, + Zpopmax => z_pop_max => Zrange, + Zdiff => z_diff => Zrange, + Zinter => z_inter => Zrange, + Zunion => z_union => Zrange, + Zscan => z_scan => Zscan, + + // scans + Sscan => s_scan => Sscan, + + // extended server + Time => time => TimeResp, + LastSave => last_save => IntVal, }) } diff --git a/proto/ember/v1/ember.proto b/proto/ember/v1/ember.proto index 35981d98..cd0a7a35 100644 --- a/proto/ember/v1/ember.proto +++ b/proto/ember/v1/ember.proto @@ -111,6 +111,69 @@ service EmberCache { rpc PubSubNumSub(PubSubNumSubRequest) returns (PubSubNumSubResponse); rpc PubSubNumPat(PubSubNumPatRequest) returns (IntResponse); + // --- strings (extended) --- + + rpc GetDel(GetDelRequest) returns (GetResponse); + rpc GetEx(GetExRequest) returns (GetResponse); + rpc GetRange(GetRangeRequest) returns (GetResponse); + rpc SetRange(SetRangeRequest) returns (IntResponse); + + // --- keys (extended) --- + + rpc Copy(CopyRequest) returns (BoolResponse); + rpc RandomKey(RandomKeyRequest) returns (GetResponse); + rpc Touch(TouchRequest) returns (IntResponse); + + // --- lists (extended) --- + + rpc LIndex(LIndexRequest) returns (GetResponse); + rpc LSet(LSetRequest) returns (StatusResponse); + rpc LTrim(LTrimRequest) returns (StatusResponse); + rpc LInsert(LInsertRequest) returns (IntResponse); + rpc LRem(LRemRequest) returns (IntResponse); + rpc LPos(LPosRequest) returns (OptionalIntResponse); + rpc LMove(LMoveRequest) returns (GetResponse); + + // --- sets (extended) --- + + rpc SUnion(SUnionRequest) returns (KeysResponse); + rpc SInter(SInterRequest) returns (KeysResponse); + rpc SDiff(SDiffRequest) returns (KeysResponse); + rpc SUnionStore(SUnionStoreRequest) returns (IntResponse); + rpc SInterStore(SInterStoreRequest) returns (IntResponse); + rpc SDiffStore(SDiffStoreRequest) returns (IntResponse); + rpc SRandMember(SRandMemberRequest) returns (ArrayResponse); + rpc SPop(SPopRequest) returns (ArrayResponse); + rpc SMisMember(SMisMemberRequest) returns (BoolArrayResponse); + + // --- hashes (extended) --- + + rpc HScan(HScanRequest) returns (HScanResponse); + + // --- sorted sets (extended) --- + + rpc ZRevRank(ZRevRankRequest) returns (OptionalIntResponse); + rpc ZRevRange(ZRevRangeRequest) returns (ZRangeResponse); + rpc ZCount(ZCountRequest) returns (IntResponse); + rpc ZIncrBy(ZIncrByRequest) returns (FloatResponse); + rpc ZRangeByScore(ZRangeByScoreRequest) returns (ZRangeResponse); + rpc ZRevRangeByScore(ZRevRangeByScoreRequest) returns (ZRangeResponse); + rpc ZPopMin(ZPopMinRequest) returns (ZRangeResponse); + rpc ZPopMax(ZPopMaxRequest) returns (ZRangeResponse); + rpc ZDiff(ZDiffRequest) returns (ZRangeResponse); + rpc ZInter(ZInterRequest) returns (ZRangeResponse); + rpc ZUnion(ZUnionRequest) returns (ZRangeResponse); + rpc ZScan(ZScanRequest) returns (ZScanResponse); + + // --- scans --- + + rpc SScan(SScanRequest) returns (SScanResponse); + + // --- server (extended) --- + + rpc Time(TimeRequest) returns (TimeResponse); + rpc LastSave(LastSaveRequest) returns (IntResponse); + // --- streaming --- // bidirectional streaming for batch operations, matching RESP3 pipelining. @@ -137,6 +200,10 @@ message StatusResponse { string status = 1; } +message BoolArrayResponse { + repeated bool values = 1; +} + // --------------------------------------------------------------------------- // strings // --------------------------------------------------------------------------- @@ -227,6 +294,32 @@ message StrlenRequest { string key = 1; } +message GetDelRequest { + string key = 1; +} + +message GetExRequest { + string key = 1; + // set expiry in seconds (ignored if expire_millis > 0). + uint64 expire_seconds = 2; + // set expiry in milliseconds (takes precedence over expire_seconds). + uint64 expire_millis = 3; + // remove the existing expiry, making the key persistent. + bool persist = 4; +} + +message GetRangeRequest { + string key = 1; + int64 start = 2; + int64 end = 3; +} + +message SetRangeRequest { + string key = 1; + int64 offset = 2; + bytes value = 3; +} + // --------------------------------------------------------------------------- // keys // --------------------------------------------------------------------------- @@ -294,6 +387,19 @@ message ScanResponse { repeated string keys = 2; } +message CopyRequest { + string source = 1; + string destination = 2; + // overwrite the destination key if it already exists. + bool replace = 3; +} + +message RandomKeyRequest {} + +message TouchRequest { + repeated string keys = 1; +} + // --------------------------------------------------------------------------- // lists // --------------------------------------------------------------------------- @@ -330,6 +436,54 @@ message LLenRequest { string key = 1; } +message LIndexRequest { + string key = 1; + int64 index = 2; +} + +message LSetRequest { + string key = 1; + int64 index = 2; + bytes value = 3; +} + +message LTrimRequest { + string key = 1; + int64 start = 2; + int64 stop = 3; +} + +message LInsertRequest { + string key = 1; + // insert before (true) or after (false) the pivot. + bool before = 2; + bytes pivot = 3; + bytes value = 4; +} + +message LRemRequest { + string key = 1; + // count > 0: remove from head; count < 0: remove from tail; 0: remove all. + int64 count = 2; + bytes value = 3; +} + +message LPosRequest { + string key = 1; + bytes value = 2; + // maximum number of positions to return. absent or 0 returns the first match. + optional uint32 count = 3; +} + +message LMoveRequest { + string source = 1; + string destination = 2; + // pop from the left (true) or right (false) of source. + bool src_left = 3; + // push to the left (true) or right (false) of destination. + bool dst_left = 4; +} + // --------------------------------------------------------------------------- // hashes // --------------------------------------------------------------------------- @@ -394,6 +548,19 @@ message OptionalArrayResponse { repeated OptionalValue values = 1; } +message HScanRequest { + string key = 1; + uint64 cursor = 2; + optional string pattern = 3; + // hint for how many fields to return per call. server may return more or fewer. + uint32 count = 4; +} + +message HScanResponse { + uint64 cursor = 1; + repeated FieldValue fields = 2; +} + // --------------------------------------------------------------------------- // sets // --------------------------------------------------------------------------- @@ -421,6 +588,61 @@ message SCardRequest { string key = 1; } +message SUnionRequest { + repeated string keys = 1; +} + +message SInterRequest { + repeated string keys = 1; +} + +message SDiffRequest { + repeated string keys = 1; +} + +message SUnionStoreRequest { + string destination = 1; + repeated string keys = 2; +} + +message SInterStoreRequest { + string destination = 1; + repeated string keys = 2; +} + +message SDiffStoreRequest { + string destination = 1; + repeated string keys = 2; +} + +message SRandMemberRequest { + string key = 1; + // positive: return that many unique members; negative: allow repeats. + int32 count = 2; +} + +message SPopRequest { + string key = 1; + uint32 count = 2; +} + +message SMisMemberRequest { + string key = 1; + repeated string members = 2; +} + +message SScanRequest { + string key = 1; + uint64 cursor = 2; + optional string pattern = 3; + uint32 count = 4; +} + +message SScanResponse { + uint64 cursor = 1; + repeated string members = 2; +} + // --------------------------------------------------------------------------- // sorted sets // --------------------------------------------------------------------------- @@ -479,6 +701,87 @@ message ZRangeResponse { repeated ScoreMember members = 1; } +message ZRevRankRequest { + string key = 1; + string member = 2; +} + +message ZRevRangeRequest { + string key = 1; + int64 start = 2; + int64 stop = 3; + bool with_scores = 4; +} + +message ZCountRequest { + string key = 1; + // min/max use redis score range syntax: "-inf", "+inf", "(5", "5". + string min = 2; + string max = 3; +} + +message ZIncrByRequest { + string key = 1; + double delta = 2; + string member = 3; +} + +message ZRangeByScoreRequest { + string key = 1; + string min = 2; + string max = 3; + optional int64 offset = 4; + optional int64 count = 5; + bool with_scores = 6; +} + +message ZRevRangeByScoreRequest { + string key = 1; + // note: max comes before min for reverse range queries. + string max = 2; + string min = 3; + optional int64 offset = 4; + optional int64 count = 5; + bool with_scores = 6; +} + +message ZPopMinRequest { + string key = 1; + uint32 count = 2; +} + +message ZPopMaxRequest { + string key = 1; + uint32 count = 2; +} + +message ZDiffRequest { + repeated string keys = 1; + bool with_scores = 2; +} + +message ZInterRequest { + repeated string keys = 1; + bool with_scores = 2; +} + +message ZUnionRequest { + repeated string keys = 1; + bool with_scores = 2; +} + +message ZScanRequest { + string key = 1; + uint64 cursor = 2; + optional string pattern = 3; + uint32 count = 4; +} + +message ZScanResponse { + uint64 cursor = 1; + repeated ScoreMember members = 2; +} + // --------------------------------------------------------------------------- // vectors // --------------------------------------------------------------------------- @@ -614,6 +917,15 @@ message BgSaveRequest {} message BgRewriteAofRequest {} +message TimeRequest {} + +message TimeResponse { + int64 seconds = 1; + int64 microseconds = 2; +} + +message LastSaveRequest {} + // --------------------------------------------------------------------------- // slowlog // --------------------------------------------------------------------------- @@ -757,6 +1069,61 @@ message PipelineRequest { PubSubNumSubRequest pubsub_numsub = 70; PubSubNumPatRequest pubsub_numpat = 71; VAddBatchRequest vadd_batch = 72; + + // extended strings + GetDelRequest get_del = 73; + GetExRequest get_ex = 74; + GetRangeRequest get_range = 75; + SetRangeRequest set_range = 76; + + // extended keys + CopyRequest copy = 77; + RandomKeyRequest random_key = 78; + TouchRequest touch = 79; + + // extended lists + LIndexRequest lindex = 80; + LSetRequest lset = 81; + LTrimRequest ltrim = 82; + LInsertRequest linsert = 83; + LRemRequest lrem = 84; + LPosRequest lpos = 85; + LMoveRequest lmove = 86; + + // extended sets + SUnionRequest sunion = 87; + SInterRequest sinter = 88; + SDiffRequest sdiff = 89; + SUnionStoreRequest sunion_store = 90; + SInterStoreRequest sinter_store = 91; + SDiffStoreRequest sdiff_store = 92; + SRandMemberRequest srand_member = 93; + SPopRequest spop = 94; + SMisMemberRequest smismember = 95; + + // extended hashes + HScanRequest hscan = 96; + + // extended sorted sets + ZRevRankRequest zrev_rank = 97; + ZRevRangeRequest zrev_range = 98; + ZCountRequest zcount = 99; + ZIncrByRequest zincrby = 100; + ZRangeByScoreRequest zrange_by_score = 101; + ZRevRangeByScoreRequest zrev_range_by_score = 102; + ZPopMinRequest zpopmin = 103; + ZPopMaxRequest zpopmax = 104; + ZDiffRequest zdiff = 105; + ZInterRequest zinter = 106; + ZUnionRequest zunion = 107; + ZScanRequest zscan = 108; + + // scans + SScanRequest sscan = 109; + + // extended server + TimeRequest time = 110; + LastSaveRequest last_save = 111; } } @@ -791,6 +1158,11 @@ message PipelineResponse { EchoResponse echo = 27; SlowLogGetResponse slowlog_get = 28; PubSubNumSubResponse pubsub_numsub = 29; + BoolArrayResponse bool_array = 30; + HScanResponse hscan = 31; + ZScanResponse zscan = 32; + SScanResponse sscan = 33; + TimeResponse time_resp = 34; } }