netget/client/redis/
resp.rs

1//! RESP for the Redis client: splitting the model's command line into arguments, and reading
2//! one complete server reply off the wire.
3//!
4//! # Why a reply reader and not a line reader
5//!
6//! RESP is length-prefixed, not line-oriented. A bulk string is a `$<len>` header line followed
7//! by `<len>` bytes that may themselves contain CRLF, and an array is a `*<n>` header followed by
8//! `n` further replies. Reading line by line turned `GET k` → `$5\r\nhello\r\n` into **two**
9//! events — `"$5"`, then `"hello"` — so the model was asked what to do about a length header,
10//! and any value containing a newline arrived in pieces. Every reply here is one event.
11//!
12//! # Bounds
13//!
14//! The reader is iterative, so nesting cannot overflow *its* stack — but the `serde_json::Value`
15//! it builds is recursive, and serialising or dropping one nested ~10 000 deep (four bytes a
16//! level: `*1\r\n`) would. So depth is bounded at [`MAX_REPLY_DEPTH`]; real replies (`XREAD`,
17//! `CLUSTER SLOTS`, `COMMAND DOCS`) stay under ten. Every length the server *declares* is
18//! checked against [`MAX_REPLY_BYTES`] / [`MAX_AGGREGATE_LEN`] **before** anything is allocated
19//! for it, and the whole reply shares one byte budget, so no header can make the client buffer
20//! more than [`MAX_REPLY_BYTES`]. A reply that breaks a bound, or is not RESP, is an error: the
21//! connection's framing is then unknowable, and the caller closes it.
22
23use anyhow::{anyhow, bail, Result};
24use serde_json::{Map, Value};
25use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt};
26
27/// Deepest aggregate nesting accepted in one reply.
28pub const MAX_REPLY_DEPTH: usize = 32;
29/// Most bytes one reply may occupy on the wire, headers included.
30pub const MAX_REPLY_BYTES: usize = 16 * 1024 * 1024;
31/// Most elements one aggregate header may declare.
32pub const MAX_AGGREGATE_LEN: usize = 1 << 20;
33/// Longest header or simple-string line accepted.
34pub const MAX_LINE_LEN: usize = 64 * 1024;
35
36/// One reply, as the model sees it.
37#[derive(Debug, Clone, PartialEq)]
38pub struct Reply {
39    /// `simple_string`, `error`, `integer`, `bulk_string`, `null`, `array`, `map`, `set`,
40    /// `push`, `boolean`, `double`, `big_number` or `verbatim_string`.
41    pub reply_type: &'static str,
42    /// The reply as JSON: strings, integers, `null`, arrays, objects, and `{"error": "..."}` for
43    /// an error (nested errors inside an aggregate take the same shape).
44    pub value: Value,
45    /// A one-line rendering in `redis-cli`'s style: `OK`, `(integer) 3`, `"v"`, `(nil)`,
46    /// `(error) ERR …`; aggregates as JSON.
47    pub response: String,
48}
49
50impl Reply {
51    /// The event data for `redis_response_received`.
52    pub fn event_data(&self) -> Value {
53        serde_json::json!({
54            "response": self.response,
55            "reply_type": self.reply_type,
56            "value": self.value,
57        })
58    }
59}
60
61#[derive(Clone, Copy, PartialEq)]
62enum Kind {
63    Array,
64    Set,
65    Push,
66    Map,
67    /// RESP3 attributes precede the reply they annotate and are not counted as an element of
68    /// any aggregate. The map is read, bounded like everything else, and discarded.
69    Attribute,
70}
71
72struct Frame {
73    kind: Kind,
74    remaining: usize,
75    items: Vec<Value>,
76}
77
78/// What one header line produced.
79enum Parsed {
80    /// A complete value and its reply type.
81    Value(&'static str, Value),
82    /// An aggregate header: a frame has been pushed.
83    Open,
84}
85
86/// Read exactly one reply. `Ok(None)` is a clean EOF before the first byte of a reply.
87pub async fn read_reply<R: AsyncBufRead + Unpin>(reader: &mut R) -> Result<Option<Reply>> {
88    let mut budget = MAX_REPLY_BYTES;
89    let mut stack: Vec<Frame> = Vec::new();
90    // The type of the outermost non-attribute value, fixed when its header is read.
91    let mut top_type: Option<&'static str> = None;
92    let mut started = false;
93
94    loop {
95        let Some(line) = read_line(reader, &mut budget).await? else {
96            if !started {
97                return Ok(None);
98            }
99            bail!("connection closed in the middle of a reply");
100        };
101        started = true;
102
103        let at_top = stack.iter().all(|f| f.kind == Kind::Attribute);
104        let parsed = parse_header(reader, &line, &mut budget, &mut stack).await?;
105
106        let (mut kind, mut value) = match parsed {
107            Parsed::Open => {
108                let opened = stack.last().expect("parse_header pushed a frame");
109                if at_top && top_type.is_none() && opened.kind != Kind::Attribute {
110                    top_type = Some(kind_name(opened.kind));
111                }
112                if opened.remaining > 0 {
113                    continue;
114                }
115                let frame = stack.pop().expect("just checked");
116                (Some(frame.kind), finish(frame))
117            }
118            Parsed::Value(t, v) => {
119                if at_top && top_type.is_none() {
120                    top_type = Some(t);
121                }
122                (None, v)
123            }
124        };
125
126        // Deliver the value upward, folding every aggregate it completes.
127        loop {
128            if kind == Some(Kind::Attribute) {
129                // Discarded, and not an element of its parent.
130                break;
131            }
132            match stack.last_mut() {
133                None => {
134                    let reply_type = top_type.unwrap_or("null");
135                    return Ok(Some(Reply {
136                        reply_type,
137                        response: render(reply_type, &value),
138                        value,
139                    }));
140                }
141                Some(frame) => {
142                    frame.items.push(value);
143                    frame.remaining -= 1;
144                    if frame.remaining > 0 {
145                        break;
146                    }
147                    let frame = stack.pop().expect("just checked");
148                    kind = Some(frame.kind);
149                    value = finish(frame);
150                }
151            }
152        }
153    }
154}
155
156fn kind_name(kind: Kind) -> &'static str {
157    match kind {
158        Kind::Array => "array",
159        Kind::Set => "set",
160        Kind::Push => "push",
161        Kind::Map => "map",
162        Kind::Attribute => "attribute",
163    }
164}
165
166/// Interpret one header line. Strings read their body here; aggregates push a frame.
167async fn parse_header<R: AsyncBufRead + Unpin>(
168    reader: &mut R,
169    line: &[u8],
170    budget: &mut usize,
171    stack: &mut Vec<Frame>,
172) -> Result<Parsed> {
173    let (marker, rest) = line.split_at(1);
174    let rest = std::str::from_utf8(rest)
175        .map_err(|_| anyhow!("non-UTF-8 RESP header"))?
176        .to_string();
177
178    Ok(match marker[0] {
179        b'+' => Parsed::Value("simple_string", Value::String(rest)),
180        b'-' => Parsed::Value("error", serde_json::json!({ "error": rest })),
181        b':' => Parsed::Value("integer", Value::from(parse_int(&rest)?)),
182        b'_' => Parsed::Value("null", Value::Null),
183        b'#' => match rest.as_str() {
184            "t" => Parsed::Value("boolean", Value::Bool(true)),
185            "f" => Parsed::Value("boolean", Value::Bool(false)),
186            other => bail!("bad RESP3 boolean {other:?}"),
187        },
188        b',' => Parsed::Value(
189            "double",
190            rest.parse::<f64>()
191                .ok()
192                .and_then(serde_json::Number::from_f64)
193                .map(Value::Number)
194                .unwrap_or(Value::String(rest)),
195        ),
196        b'(' => Parsed::Value("big_number", Value::String(rest)),
197        b'$' | b'!' | b'=' => {
198            let declared = parse_int(&rest)?;
199            if declared < 0 {
200                return Ok(Parsed::Value("null", Value::Null));
201            }
202            let len = declared as usize;
203            // The declared length is checked before a byte is allocated for it.
204            if len.saturating_add(2) > *budget {
205                bail!(
206                    "reply declares a {len}-byte string, over the {MAX_REPLY_BYTES}-byte reply \
207                     limit"
208                );
209            }
210            *budget -= len + 2;
211            let mut body = vec![0u8; len + 2];
212            reader.read_exact(&mut body).await?;
213            if &body[len..] != b"\r\n" {
214                bail!("string body not terminated by CRLF");
215            }
216            body.truncate(len);
217            let mut text = String::from_utf8_lossy(&body).to_string();
218            match marker[0] {
219                b'!' => Parsed::Value("error", serde_json::json!({ "error": text })),
220                b'=' => {
221                    // Verbatim strings carry a three-letter format prefix (`txt:`).
222                    if text.len() >= 4 && text.as_bytes()[3] == b':' {
223                        text = text[4..].to_string();
224                    }
225                    Parsed::Value("verbatim_string", Value::String(text))
226                }
227                _ => Parsed::Value("bulk_string", Value::String(text)),
228            }
229        }
230        b'*' | b'~' | b'>' | b'%' | b'|' => {
231            let declared = parse_int(&rest)?;
232            if declared < 0 {
233                return Ok(Parsed::Value("null", Value::Null));
234            }
235            let n = declared as usize;
236            if n > MAX_AGGREGATE_LEN {
237                bail!("reply declares {n} elements, over the {MAX_AGGREGATE_LEN} limit");
238            }
239            if stack.len() >= MAX_REPLY_DEPTH {
240                bail!("reply nests deeper than {MAX_REPLY_DEPTH} levels");
241            }
242            let kind = match marker[0] {
243                b'*' => Kind::Array,
244                b'~' => Kind::Set,
245                b'>' => Kind::Push,
246                b'%' => Kind::Map,
247                _ => Kind::Attribute,
248            };
249            let remaining = match kind {
250                Kind::Map | Kind::Attribute => n * 2,
251                _ => n,
252            };
253            stack.push(Frame {
254                kind,
255                remaining,
256                items: Vec::with_capacity(remaining.min(1024)),
257            });
258            Parsed::Open
259        }
260        other => bail!("not a RESP reply: unexpected type byte 0x{other:02x}"),
261    })
262}
263
264fn finish(frame: Frame) -> Value {
265    match frame.kind {
266        Kind::Array | Kind::Set | Kind::Push => Value::Array(frame.items),
267        Kind::Map => {
268            let mut map = Map::new();
269            let mut it = frame.items.into_iter();
270            while let (Some(k), Some(v)) = (it.next(), it.next()) {
271                let key = match k {
272                    Value::String(s) => s,
273                    other => other.to_string(),
274                };
275                map.insert(key, v);
276            }
277            Value::Object(map)
278        }
279        Kind::Attribute => Value::Null,
280    }
281}
282
283fn render(reply_type: &str, value: &Value) -> String {
284    match (reply_type, value) {
285        (_, Value::Null) => "(nil)".to_string(),
286        ("simple_string" | "big_number", Value::String(s)) => s.clone(),
287        ("error", Value::Object(o)) => format!(
288            "(error) {}",
289            o.get("error").and_then(|e| e.as_str()).unwrap_or("")
290        ),
291        ("integer", v) => format!("(integer) {v}"),
292        ("boolean", Value::Bool(b)) => format!("({b})"),
293        ("double", v) => format!("(double) {v}"),
294        (_, Value::String(s)) => Value::String(s.clone()).to_string(),
295        (_, v) => v.to_string(),
296    }
297}
298
299fn parse_int(text: &str) -> Result<i64> {
300    text.trim()
301        .parse::<i64>()
302        .map_err(|_| anyhow!("bad RESP length or integer {text:?}"))
303}
304
305/// One CRLF-terminated line, without the terminator. Bounded by [`MAX_LINE_LEN`] and charged to
306/// the reply's byte budget. `None` on EOF before any byte.
307async fn read_line<R: AsyncBufRead + Unpin>(
308    reader: &mut R,
309    budget: &mut usize,
310) -> Result<Option<Vec<u8>>> {
311    let mut line = Vec::new();
312    loop {
313        let available = reader.fill_buf().await?;
314        if available.is_empty() {
315            if line.is_empty() {
316                return Ok(None);
317            }
318            bail!("connection closed in the middle of a line");
319        }
320        let (take, done) = match available.iter().position(|&b| b == b'\n') {
321            Some(i) => (i + 1, true),
322            None => (available.len(), false),
323        };
324        if line.len() + take > MAX_LINE_LEN {
325            bail!("RESP line longer than {MAX_LINE_LEN} bytes");
326        }
327        if line.len() + take > *budget {
328            bail!("reply exceeds the {MAX_REPLY_BYTES}-byte limit");
329        }
330        line.extend_from_slice(&available[..take]);
331        reader.consume(take);
332        if done {
333            *budget -= line.len();
334            if line.len() < 3 || line[line.len() - 2] != b'\r' {
335                bail!("RESP line not terminated by CRLF");
336            }
337            line.truncate(line.len() - 2);
338            return Ok(Some(line));
339        }
340    }
341}
342
343/// Split a command line the way `redis-cli` does (its `sdssplitargs`): whitespace separates
344/// arguments; `"…"` quotes allow `\n \r \t \b \a \\ \"` and `\xHH`; `'…'` quotes allow `\'`; a
345/// closing quote must be followed by whitespace or the end.
346///
347/// So `SET greeting "hello world"` is three arguments, not four with stray quotes — which is
348/// what the model writes, because it is what `redis-cli` accepts. Unbalanced quotes are an
349/// error rather than a guess.
350pub fn split_command(line: &str) -> Result<Vec<Vec<u8>>> {
351    let bytes = line.as_bytes();
352    let mut args = Vec::new();
353    let mut i = 0;
354    while i < bytes.len() {
355        while i < bytes.len() && bytes[i].is_ascii_whitespace() {
356            i += 1;
357        }
358        if i >= bytes.len() {
359            break;
360        }
361        let mut arg = Vec::new();
362        let mut in_double = false;
363        let mut in_single = false;
364        loop {
365            if in_double {
366                if i >= bytes.len() {
367                    bail!("unterminated double quote in Redis command");
368                }
369                match bytes[i] {
370                    b'\\'
371                        if i + 3 < bytes.len()
372                            && bytes[i + 1] == b'x'
373                            && bytes[i + 2].is_ascii_hexdigit()
374                            && bytes[i + 3].is_ascii_hexdigit() =>
375                    {
376                        let hex = std::str::from_utf8(&bytes[i + 2..i + 4]).unwrap_or("00");
377                        arg.push(u8::from_str_radix(hex, 16).unwrap_or(0));
378                        i += 3;
379                    }
380                    b'\\' if i + 1 < bytes.len() => {
381                        i += 1;
382                        arg.push(match bytes[i] {
383                            b'n' => b'\n',
384                            b'r' => b'\r',
385                            b't' => b'\t',
386                            b'b' => 0x08,
387                            b'a' => 0x07,
388                            other => other,
389                        });
390                    }
391                    b'"' => {
392                        if i + 1 < bytes.len() && !bytes[i + 1].is_ascii_whitespace() {
393                            bail!("closing quote must be followed by a space in Redis command");
394                        }
395                        i += 1;
396                        break;
397                    }
398                    other => arg.push(other),
399                }
400                i += 1;
401            } else if in_single {
402                if i >= bytes.len() {
403                    bail!("unterminated single quote in Redis command");
404                }
405                match bytes[i] {
406                    b'\\' if i + 1 < bytes.len() && bytes[i + 1] == b'\'' => {
407                        arg.push(b'\'');
408                        i += 1;
409                    }
410                    b'\'' => {
411                        if i + 1 < bytes.len() && !bytes[i + 1].is_ascii_whitespace() {
412                            bail!("closing quote must be followed by a space in Redis command");
413                        }
414                        i += 1;
415                        break;
416                    }
417                    other => arg.push(other),
418                }
419                i += 1;
420            } else {
421                if i >= bytes.len() || bytes[i].is_ascii_whitespace() {
422                    break;
423                }
424                match bytes[i] {
425                    b'"' => in_double = true,
426                    b'\'' => in_single = true,
427                    other => arg.push(other),
428                }
429                i += 1;
430            }
431        }
432        args.push(arg);
433    }
434    if args.is_empty() {
435        bail!("empty Redis command");
436    }
437    Ok(args)
438}
439
440/// Encode arguments as a RESP array of bulk strings — the form every Redis server accepts.
441pub fn encode_command(args: &[Vec<u8>]) -> Vec<u8> {
442    let mut out = format!("*{}\r\n", args.len()).into_bytes();
443    for arg in args {
444        out.extend_from_slice(format!("${}\r\n", arg.len()).as_bytes());
445        out.extend_from_slice(arg);
446        out.extend_from_slice(b"\r\n");
447    }
448    out
449}
netget/client/redis/resp.rs

netget/client/redis/
resp.rs
