From cc31b33ade8859feb1fbde3883ee9a5424282195 Mon Sep 17 00:00:00 2001 From: kclaka <33363343+kclaka@users.noreply.github.com> Date: Wed, 16 Sep 2026 07:19:43 -0700 Subject: [PATCH 1/2] Isolate binary RESP keys and validate Redis commands --- README.md | 10 +- crates/proxy-server/src/cache_layer.rs | 44 +++- crates/proxy-server/src/resp/cmd.rs | 279 ++++++++++++++------- crates/proxy-server/src/resp/connection.rs | 17 +- crates/proxy-server/src/resp/mod.rs | 6 +- 5 files changed, 249 insertions(+), 107 deletions(-) diff --git a/README.md b/README.md index e203584..d6aa00c 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ --- -Colander is a drop-in caching reverse proxy that replaces LRU with the [SIEVE](https://cachemon.github.io/SIEVE-website/) eviction algorithm (published at [NSDI '24](https://www.usenix.org/conference/nsdi24/presentation/zhang-yazhuo)). It sits between your clients and your backend, caches HTTP responses, and speaks both **HTTP** and the **Redis wire protocol (RESP2)** — so you can swap out Redis for SIEVE-powered caching with zero code changes. +Colander is a drop-in caching reverse proxy that replaces LRU with the [SIEVE](https://cachemon.github.io/SIEVE-website/) eviction algorithm (published at [NSDI '24](https://www.usenix.org/conference/nsdi24/presentation/zhang-yazhuo)). It sits between your clients and your backend, caches HTTP responses, and speaks both **HTTP** and the **Redis wire protocol (RESP2)** — with a small Redis-compatible command subset for cache workloads. ```bash docker compose up # proxy + backend + load generator + dashboard @@ -223,10 +223,10 @@ redis-cli -p 6379 |---------|--------|-------------| | **PING** | `PING` | Health check. Returns `PONG`. | | **GET** | `GET key` | Retrieve a cached value. Returns bulk string or `(nil)`. | -| **SET** | `SET key value [EX seconds]` | Store a value with optional TTL. Returns `OK`. | +| **SET** | `SET key value [EX seconds | PX milliseconds]` | Store a value with optional positive TTL. Without EX/PX, it persists until eviction or deletion. Unsupported options return an error. | | **DEL** | `DEL key [key ...]` | Delete one or more keys. Returns count of deleted keys. | -| **TTL** | `TTL key` | Seconds remaining before expiry. Returns `-2` if key missing. | -| **EXPIRE** | `EXPIRE key seconds` | Not supported (TTL is set-at-insert). Returns `0`. | +| **TTL** | `TTL key` | Seconds remaining before expiry. Returns `-2` if key missing and `-1` if it has no expiration. | +| **EXPIRE** | `EXPIRE key seconds` | Not supported (TTL is set-at-insert). Returns an explicit error. | | **COMMAND** | `COMMAND` | Client compatibility (redis-cli sends this on connect). Returns `OK`. | ### Example @@ -245,7 +245,7 @@ OK (nil) ``` -> **Shared cache**: The RESP interface shares the same in-memory cache as the HTTP proxy. A `SET` via Redis is visible to HTTP `GET` responses, and vice versa. +> **Isolated keys**: RESP and HTTP use separate key namespaces within the same bounded primary cache. Binary Redis keys cannot overwrite HTTP responses. RESP operations do not update the comparison cache. Values are limited by `max_body_size_bytes`; incoming frames are limited to 8 MiB. --- diff --git a/crates/proxy-server/src/cache_layer.rs b/crates/proxy-server/src/cache_layer.rs index 8b38d32..c71dd2b 100644 --- a/crates/proxy-server/src/cache_layer.rs +++ b/crates/proxy-server/src/cache_layer.rs @@ -9,6 +9,15 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; +fn raw_key(key: &[u8]) -> String { + use std::fmt::Write; + let mut encoded = String::from("\0resp:"); + for byte in key { + write!(encoded, "{byte:02x}").expect("writing to String cannot fail"); + } + encoded +} + /// Runtime mode for the dual-cache system. /// - Demo: updates both caches, serves from primary (fair hit-rate comparison) /// - Bench: updates only primary cache (true latency/throughput) @@ -156,28 +165,39 @@ impl CacheLayer { self.primary.insert(key, value); } - /// Remove a key from the primary cache. Returns true if the key existed. - pub fn remove(&self, key: &str) -> bool { - self.primary.remove(key) + /// RESP keys are binary-safe and isolated from HTTP cache keys. + pub fn get_raw(&self, key: &[u8]) -> Option> { + self.primary.get(&raw_key(key)) } - /// Insert raw bytes (for RESP SET — bypasses HTTP response wrapping). - /// Only inserts into primary (RESP ops don't participate in demo comparison). - pub fn insert_raw(&self, key: String, value: Bytes, ttl: Option) { + pub fn remove_raw(&self, key: &[u8]) -> bool { + let key = raw_key(key); + // Expired entries count as absent, as they do for GET. + self.primary.get(&key).is_some() && self.primary.remove(&key) + } + + /// SET without an expiration is persistent until eviction or deletion. + pub fn insert_raw(&self, key: &[u8], value: Bytes, ttl: Option) { let response = CachedResponse { status: 0, headers: vec![], body: value, inserted_at: Instant::now(), - ttl: ttl.unwrap_or(self.default_ttl()), + ttl: ttl.unwrap_or(Duration::MAX), }; - self.primary.insert(key, response); + self.primary.insert(raw_key(key), response); } - /// Get TTL remaining for a key. Returns None if key missing/expired. - pub fn ttl_remaining(&self, key: &str) -> Option { - let entry = self.primary.get(key)?; - entry.ttl.checked_sub(entry.inserted_at.elapsed()) + pub fn raw_ttl(&self, key: &[u8]) -> i64 { + match self.get_raw(key) { + None => -2, + Some(entry) if entry.ttl == Duration::MAX => -1, + Some(entry) => entry + .ttl + .saturating_sub(entry.inserted_at.elapsed()) + .as_secs() + .min(i64::MAX as u64) as i64, + } } /// Build a CachedResponse from raw HTTP response parts. diff --git a/crates/proxy-server/src/resp/cmd.rs b/crates/proxy-server/src/resp/cmd.rs index fffc4b8..4e200ed 100644 --- a/crates/proxy-server/src/resp/cmd.rs +++ b/crates/proxy-server/src/resp/cmd.rs @@ -3,109 +3,218 @@ use bytes::Bytes; use redis_protocol::resp2::types::BytesFrame; use std::time::Duration; -/// Dispatch a RESP2 frame (expected to be an Array of bulk strings) to the appropriate handler. +/// Validate the complete command before accessing or mutating the cache. pub fn dispatch(frame: &BytesFrame, state: &AppState) -> BytesFrame { - let args = match frame { - BytesFrame::Array(arr) => arr, - _ => return error_frame("ERR expected array"), + let frames = match frame { + BytesFrame::Array(args) if !args.is_empty() => args, + _ => return error_frame("ERR expected a non-empty array"), }; - - if args.is_empty() { - return error_frame("ERR empty command"); - } - - let cmd = match &args[0] { - BytesFrame::BulkString(b) => String::from_utf8_lossy(b).to_uppercase(), - _ => return error_frame("ERR invalid command format"), + let args: Option> = frames + .iter() + .map(|frame| match frame { + BytesFrame::BulkString(bytes) => Some(bytes), + _ => None, + }) + .collect(); + let Some(args) = args else { + return error_frame("ERR command arguments must be bulk strings"); }; - + let command = args[0].to_ascii_uppercase(); let cache = state.cache.load(); - - match cmd.as_str() { - "PING" => BytesFrame::SimpleString("PONG".into()), - "COMMAND" => BytesFrame::SimpleString("OK".into()), - "GET" => { - if args.len() < 2 { - return error_frame("ERR wrong number of arguments for 'GET' command"); - } - let key = bulk_to_string(&args[1]); - let lookup = cache.get(&key); - match lookup.value { - Some(cached) => BytesFrame::BulkString(cached.body.clone()), - None => BytesFrame::Null, - } - } - "SET" => { - if args.len() < 3 { - return error_frame("ERR wrong number of arguments for 'SET' command"); + match command.as_slice() { + b"PING" => match args.len() { + 1 => BytesFrame::SimpleString("PONG".into()), + 2 => BytesFrame::BulkString(args[1].clone()), + _ => error_frame("ERR wrong number of arguments for 'PING' command"), + }, + b"GET" if args.len() == 2 => cache + .get_raw(args[1]) + .map(|entry| BytesFrame::BulkString(entry.body.clone())) + .unwrap_or(BytesFrame::Null), + b"SET" if args.len() >= 3 => { + let ttl = match parse_set_options(&args[3..]) { + Ok(ttl) => ttl, + Err(error) => return error_frame(error), + }; + if args[2].len() > cache.max_body_size { + return error_frame("ERR value exceeds configured maximum body size"); } - let key = bulk_to_string(&args[1]); - let value = bulk_to_bytes(&args[2]); - let ttl = parse_set_options(&args[3..]); - cache.insert_raw(key, value, ttl); + cache.insert_raw(args[1], args[2].clone(), ttl); BytesFrame::SimpleString("OK".into()) } - "DEL" => { - if args.len() < 2 { - return error_frame("ERR wrong number of arguments for 'DEL' command"); - } - let mut count: i64 = 0; - for arg in &args[1..] { - let key = bulk_to_string(arg); - if cache.remove(&key) { - count += 1; - } - } - BytesFrame::Integer(count) - } - "EXPIRE" => { - // TTL is set-at-insert only — EXPIRE is not supported - BytesFrame::Integer(0) - } - "TTL" => { - if args.len() < 2 { - return error_frame("ERR wrong number of arguments for 'TTL' command"); - } - let key = bulk_to_string(&args[1]); - match cache.ttl_remaining(&key) { - Some(remaining) => BytesFrame::Integer(remaining.as_secs() as i64), - None => BytesFrame::Integer(-2), - } + b"DEL" if args.len() >= 2 => { + let count = args[1..].iter().filter(|key| cache.remove_raw(key)).count(); + BytesFrame::Integer(count as i64) } - other => error_frame(&format!("ERR unknown command '{other}'")), + b"TTL" if args.len() == 2 => BytesFrame::Integer(cache.raw_ttl(args[1])), + b"GET" | b"SET" | b"DEL" | b"TTL" => error_frame("ERR wrong number of arguments"), + b"EXPIRE" => error_frame("ERR EXPIRE is not supported; use SET with EX or PX"), + _ => error_frame("ERR unknown or unsupported command"), } } -fn error_frame(msg: &str) -> BytesFrame { - BytesFrame::Error(msg.into()) +fn error_frame(message: &str) -> BytesFrame { + BytesFrame::Error(message.into()) } -fn bulk_to_string(frame: &BytesFrame) -> String { - match frame { - BytesFrame::BulkString(b) => String::from_utf8_lossy(b).into_owned(), - _ => String::new(), +fn parse_set_options(args: &[&Bytes]) -> Result, &'static str> { + if args.is_empty() { + return Ok(None); + } + if args.len() != 2 { + return Err("ERR syntax error"); } + let milliseconds = if args[0].eq_ignore_ascii_case(b"EX") { + false + } else if args[0].eq_ignore_ascii_case(b"PX") { + true + } else { + return Err("ERR unsupported SET option"); + }; + let amount = std::str::from_utf8(args[1]) + .ok() + .filter(|value| !value.is_empty() && value.bytes().all(|c| c.is_ascii_digit())) + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0 && *value <= i64::MAX as u64) + .ok_or("ERR invalid expire time in 'SET' command")?; + Ok(Some(if milliseconds { + Duration::from_millis(amount) + } else { + Duration::from_secs(amount) + })) } -fn bulk_to_bytes(frame: &BytesFrame) -> Bytes { - match frame { - BytesFrame::BulkString(b) => b.clone(), - _ => Bytes::new(), +#[cfg(test)] +mod tests { + use super::*; + use crate::cache_layer::CacheLayer; + use arc_swap::ArcSwap; + use hyper_util::{client::legacy::Client, rt::TokioExecutor}; + use std::sync::Arc; + + fn state() -> AppState { + AppState { + cache: ArcSwap::from(Arc::new(CacheLayer::new( + "sieve", + Some("lru"), + 1024, + Duration::from_secs(60), + 1024, + ))), + client: Client::builder(TokioExecutor::new()).build_http(), + upstream_url: "http://127.0.0.1:1".into(), + } } -} -/// Parse SET options: SET key value [EX seconds] -fn parse_set_options(args: &[BytesFrame]) -> Option { - let mut i = 0; - while i < args.len() { - let opt = bulk_to_string(&args[i]).to_uppercase(); - if opt == "EX" && i + 1 < args.len() { - let secs_str = bulk_to_string(&args[i + 1]); - if let Ok(secs) = secs_str.parse::() { - return Some(Duration::from_secs(secs)); - } + fn command(state: &AppState, args: &[&[u8]]) -> BytesFrame { + dispatch( + &BytesFrame::Array( + args.iter() + .map(|arg| BytesFrame::BulkString(Bytes::copy_from_slice(arg))) + .collect(), + ), + state, + ) + } + + #[tokio::test] + async fn binary_keys_and_http_namespace_do_not_collide() { + let state = state(); + command(&state, &[b"SET", b"\xff", b"first"]); + command(&state, &[b"SET", b"\xfe", b"second"]); + assert_eq!( + command(&state, &[b"GET", b"\xff"]), + BytesFrame::BulkString("first".into()) + ); + command(&state, &[b"SET", b"GET:/", b"injected"]); + assert!(state.cache.load().get("GET:/").value.is_none()); + assert_eq!( + state.cache.load().comparison_stats().unwrap().current_size, + 0 + ); + } + + #[tokio::test] + async fn malformed_set_does_not_overwrite_existing_value() { + let state = state(); + command(&state, &[b"SET", b"key", b"old"]); + for options in [ + vec![b"EX".as_slice()], + vec![b"EX", b"0"], + vec![b"EX", b"-1"], + vec![b"PX", b"no"], + vec![b"NX"], + vec![b"EX", b"1", b"NX"], + ] { + let mut args = vec![b"SET".as_slice(), b"key", b"new"]; + args.extend(options); + assert!(matches!(command(&state, &args), BytesFrame::Error(_))); + assert_eq!( + command(&state, &[b"GET", b"key"]), + BytesFrame::BulkString("old".into()) + ); } - i += 1; } - None + + #[tokio::test] + async fn persistent_ttl_expiration_and_delete_semantics() { + let state = state(); + command(&state, &[b"SET", b"key", b"value"]); + assert_eq!(command(&state, &[b"TTL", b"key"]), BytesFrame::Integer(-1)); + assert_eq!( + command(&state, &[b"TTL", b"missing"]), + BytesFrame::Integer(-2) + ); + state + .cache + .load() + .insert_raw(b"expired", "old".into(), Some(Duration::ZERO)); + assert_eq!( + command(&state, &[b"DEL", b"expired"]), + BytesFrame::Integer(0) + ); + assert_eq!( + command(&state, &[b"DEL", b"key", b"key"]), + BytesFrame::Integer(1) + ); + command(&state, &[b"SET", b"key", b"value", b"PX", b"10000"]); + assert!(matches!( + command(&state, &[b"TTL", b"key"]), + BytesFrame::Integer(0..=10) + )); + assert_eq!(state.cache.load().comparison_stats().unwrap().hits, 0); + assert_eq!(state.cache.load().comparison_stats().unwrap().misses, 0); + } + + #[tokio::test] + async fn rejects_wrong_arity_types_and_oversized_values() { + let state = state(); + assert!(matches!( + command(&state, &[b"GET", b"a", b"b"]), + BytesFrame::Error(_) + )); + assert!(matches!( + command(&state, &[b"SET", b"key", &vec![0; 1025]]), + BytesFrame::Error(_) + )); + assert!(matches!( + dispatch( + &BytesFrame::Array(vec![ + BytesFrame::BulkString("SET".into()), + BytesFrame::Null, + BytesFrame::Null + ]), + &state + ), + BytesFrame::Error(_) + )); + assert_eq!( + command(&state, &[b"PING", b"hello"]), + BytesFrame::BulkString("hello".into()) + ); + assert!(matches!( + command(&state, &[b"EXPIRE", b"key", b"1"]), + BytesFrame::Error(_) + )); + } } diff --git a/crates/proxy-server/src/resp/connection.rs b/crates/proxy-server/src/resp/connection.rs index 8bf6589..b64d9e0 100644 --- a/crates/proxy-server/src/resp/connection.rs +++ b/crates/proxy-server/src/resp/connection.rs @@ -7,15 +7,24 @@ use redis_protocol::resp2::types::BytesFrame; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; +const MAX_FRAME_SIZE: usize = 8 * 1024 * 1024; + /// Handle a single RESP client connection: read frames, dispatch commands, write responses. pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { let mut buf = BytesMut::with_capacity(4096); loop { - // Read data from client - match stream.read_buf(&mut buf).await { + if buf.len() >= MAX_FRAME_SIZE { + let _ = stream + .write_all(b"-ERR frame exceeds 8 MiB limit\r\n") + .await; + return; + } + let mut chunk = [0u8; 8192]; + let available = chunk.len().min(MAX_FRAME_SIZE - buf.len()); + match stream.read(&mut chunk[..available]).await { Ok(0) => break, // EOF - Ok(_) => {} + Ok(read) => buf.extend_from_slice(&chunk[..read]), Err(e) => { tracing::debug!(error = %e, "RESP read error"); break; @@ -24,7 +33,7 @@ pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { // Try to decode complete frames from the buffer loop { - // clone().freeze() gives us &Bytes without copying the data + // The decoder returns owned byte slices for dispatch. let (frame, consumed) = match decode_bytes(&buf.clone().freeze()) { Ok(Some((frame, consumed))) => (frame, consumed), Ok(None) => break, // Need more data diff --git a/crates/proxy-server/src/resp/mod.rs b/crates/proxy-server/src/resp/mod.rs index ce2d9b6..e6ac1ae 100644 --- a/crates/proxy-server/src/resp/mod.rs +++ b/crates/proxy-server/src/resp/mod.rs @@ -29,9 +29,13 @@ pub async fn run_resp_server(addr: &str, state: Arc, shutdown: Cancell match result { Ok((stream, peer)) => { let state = Arc::clone(&state); + let client_shutdown = shutdown.clone(); tokio::spawn(async move { tracing::debug!(peer = %peer, "RESP client connected"); - connection::handle_connection(stream, &state).await; + tokio::select! { + _ = client_shutdown.cancelled() => {}, + _ = connection::handle_connection(stream, &state) => {}, + } tracing::debug!(peer = %peer, "RESP client disconnected"); }); } From 517d6455afc4699a68c788b1472202f18265b53d Mon Sep 17 00:00:00 2001 From: kclaka <33363343+kclaka@users.noreply.github.com> Date: Wed, 16 Sep 2026 07:24:54 -0700 Subject: [PATCH 2/2] Allocate RESP reply buffers and cover fragmented and pipelined transport --- crates/proxy-server/src/resp/connection.rs | 75 ++++++++++++++++++++-- 1 file changed, 68 insertions(+), 7 deletions(-) diff --git a/crates/proxy-server/src/resp/connection.rs b/crates/proxy-server/src/resp/connection.rs index b64d9e0..b24f2c9 100644 --- a/crates/proxy-server/src/resp/connection.rs +++ b/crates/proxy-server/src/resp/connection.rs @@ -2,15 +2,14 @@ use super::cmd; use crate::proxy::AppState; use bytes::BytesMut; use redis_protocol::resp2::decode::decode_bytes; -use redis_protocol::resp2::encode::encode_bytes; +use redis_protocol::resp2::encode::extend_encode; use redis_protocol::resp2::types::BytesFrame; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpStream; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; const MAX_FRAME_SIZE: usize = 8 * 1024 * 1024; /// Handle a single RESP client connection: read frames, dispatch commands, write responses. -pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { +pub async fn handle_connection(mut stream: S, state: &AppState) { let mut buf = BytesMut::with_capacity(4096); loop { @@ -42,7 +41,7 @@ pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { let err_frame = BytesFrame::Error("ERR protocol error".into()); let mut out = BytesMut::new(); // false = don't encode integers as bulk strings (standard RESP2) - if encode_bytes(&mut out, &err_frame, false).is_ok() { + if extend_encode(&mut out, &err_frame, false).is_ok() { let _ = stream.write_all(&out).await; } return; @@ -57,9 +56,9 @@ pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { // Encode and send the response let mut out = BytesMut::new(); - if let Err(e) = encode_bytes(&mut out, &response, false) { + if let Err(e) = extend_encode(&mut out, &response, false) { tracing::debug!(error = %e, "RESP encode error"); - break; + return; } if let Err(e) = stream.write_all(&out).await { tracing::debug!(error = %e, "RESP write error"); @@ -68,3 +67,65 @@ pub async fn handle_connection(mut stream: TcpStream, state: &AppState) { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::cache_layer::CacheLayer; + use arc_swap::ArcSwap; + use hyper_util::{client::legacy::Client, rt::TokioExecutor}; + use std::{sync::Arc, time::Duration}; + + fn connection() -> (tokio::io::DuplexStream, tokio::task::JoinHandle<()>) { + let (client, server) = tokio::io::duplex(4096); + let state = AppState { + cache: ArcSwap::from(Arc::new(CacheLayer::new( + "sieve", + None, + 64, + Duration::from_secs(60), + 1024, + ))), + client: Client::builder(TokioExecutor::new()).build_http(), + upstream_url: "http://127.0.0.1:1".into(), + }; + let task = tokio::spawn(async move { + handle_connection(server, &state).await; + }); + (client, task) + } + + #[tokio::test] + async fn fragmented_and_pipelined_commands_receive_complete_encoded_replies() { + let (mut client, task) = connection(); + client.write_all(b"*1\r\n$4\r\nPI").await.unwrap(); + tokio::task::yield_now().await; + client + .write_all( + b"NG\r\n*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n*2\r\n$3\r\nGET\r\n$1\r\nk\r\n", + ) + .await + .unwrap(); + client.shutdown().await.unwrap(); + let mut output = Vec::new(); + tokio::time::timeout(Duration::from_secs(1), client.read_to_end(&mut output)) + .await + .unwrap() + .unwrap(); + assert_eq!(output, b"+PONG\r\n+OK\r\n$1\r\nv\r\n"); + task.await.unwrap(); + } + + #[tokio::test] + async fn protocol_errors_are_encoded_before_the_connection_closes() { + let (mut client, task) = connection(); + client.write_all(b"?invalid\r\n").await.unwrap(); + let mut output = Vec::new(); + tokio::time::timeout(Duration::from_secs(1), client.read_to_end(&mut output)) + .await + .unwrap() + .unwrap(); + assert_eq!(output, b"-ERR protocol error\r\n"); + task.await.unwrap(); + } +}