dynomite/proto/redis/
parser.rs

1//! Redis (RESP) wire-protocol parser.
2//!
3//! The parser walks the bytes of a flattened request or response
4//! buffer through a single state machine and updates the message
5//! in place. The state alphabet, transitions, and error paths
6//! mirror `redis_parse_req` and `redis_parse_rsp` in the reference
7//! engine.
8//!
9//! The parser is byte-driven and never allocates outside the
10//! [`Msg`]'s key and argument buffers. It must not panic on any
11//! input.
12
13// The parser truncates ASCII-decimal accumulators into fixed-width
14// counters (`u32` for rlen / rntokens / nkeys, `usize` for cursor
15// positions). The allowance covers these intentional `as` casts;
16// out-of-range numerals surface as parse errors elsewhere in the
17// state machine.
18#![allow(clippy::cast_possible_truncation)]
19#![allow(clippy::cast_possible_wrap)]
20#![allow(clippy::too_many_arguments)]
21#![allow(clippy::match_same_arms)]
22#![allow(clippy::needless_continue)]
23#![allow(clippy::manual_let_else)]
24
25use super::commands::{classify, error_lookup, lookup, CommandClass, RoutingOverride};
26use crate::io::mbuf::MBUF_MAX_SIZE;
27use crate::msg::{ArgPos, KeyPos, Msg, MsgParseResult, MsgRouting, MsgType};
28
29const CR: u8 = b'\r';
30const LF: u8 = b'\n';
31
32/// Optional hash-tag delimiters. When set the parser carves out
33/// the inner range as the routing tag on every parsed key.
34#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
35pub struct HashTag {
36    /// Opening byte of the hash tag.
37    pub open: u8,
38    /// Closing byte of the hash tag.
39    pub close: u8,
40}
41
42#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
43#[repr(u32)]
44enum ReqState {
45    #[default]
46    Start = 0,
47    Narg = 1,
48    NargLf = 2,
49    ReqTypeLen = 3,
50    ReqTypeLenLf = 4,
51    ReqType = 5,
52    ReqTypeLf = 6,
53    KeyLen = 7,
54    KeyLenLf = 8,
55    Key = 9,
56    KeyLf = 10,
57    Arg1Len = 11,
58    Arg1LenLf = 12,
59    Arg1 = 13,
60    Arg1Lf = 14,
61    Arg2Len = 15,
62    Arg2LenLf = 16,
63    Arg2 = 17,
64    Arg2Lf = 18,
65    Arg3Len = 19,
66    Arg3LenLf = 20,
67    Arg3 = 21,
68    Arg3Lf = 22,
69    ArgnLen = 23,
70    ArgnLenLf = 24,
71    Argn = 25,
72    ArgnLf = 26,
73    InlinePing = 27,
74}
75
76impl ReqState {
77    fn from_u32(v: u32) -> Self {
78        match v {
79            1 => Self::Narg,
80            2 => Self::NargLf,
81            3 => Self::ReqTypeLen,
82            4 => Self::ReqTypeLenLf,
83            5 => Self::ReqType,
84            6 => Self::ReqTypeLf,
85            7 => Self::KeyLen,
86            8 => Self::KeyLenLf,
87            9 => Self::Key,
88            10 => Self::KeyLf,
89            11 => Self::Arg1Len,
90            12 => Self::Arg1LenLf,
91            13 => Self::Arg1,
92            14 => Self::Arg1Lf,
93            15 => Self::Arg2Len,
94            16 => Self::Arg2LenLf,
95            17 => Self::Arg2,
96            18 => Self::Arg2Lf,
97            19 => Self::Arg3Len,
98            20 => Self::Arg3LenLf,
99            21 => Self::Arg3,
100            22 => Self::Arg3Lf,
101            23 => Self::ArgnLen,
102            24 => Self::ArgnLenLf,
103            25 => Self::Argn,
104            26 => Self::ArgnLf,
105            27 => Self::InlinePing,
106            _ => Self::Start,
107        }
108    }
109}
110
111#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
112#[repr(u32)]
113enum RspState {
114    #[default]
115    Start = 0,
116    Status = 1,
117    Error = 2,
118    Integer = 3,
119    IntegerStart = 4,
120    Simple = 5,
121    Bulk = 6,
122    BulkLf = 7,
123    BulkArg = 8,
124    BulkArgLf = 9,
125    Multibulk = 10,
126    MultibulkNargLf = 11,
127    MultibulkArgnLen = 12,
128    MultibulkArgnLenLf = 13,
129    MultibulkArgn = 14,
130    MultibulkArgnLf = 15,
131    RuntoCrlf = 16,
132    AlmostDone = 17,
133}
134
135impl RspState {
136    fn from_u32(v: u32) -> Self {
137        match v {
138            1 => Self::Status,
139            2 => Self::Error,
140            3 => Self::Integer,
141            4 => Self::IntegerStart,
142            5 => Self::Simple,
143            6 => Self::Bulk,
144            7 => Self::BulkLf,
145            8 => Self::BulkArg,
146            9 => Self::BulkArgLf,
147            10 => Self::Multibulk,
148            11 => Self::MultibulkNargLf,
149            12 => Self::MultibulkArgnLen,
150            13 => Self::MultibulkArgnLenLf,
151            14 => Self::MultibulkArgn,
152            15 => Self::MultibulkArgnLf,
153            16 => Self::RuntoCrlf,
154            17 => Self::AlmostDone,
155            _ => Self::Start,
156        }
157    }
158}
159
160/// Parse a Redis request from `input` and update `r` in place.
161///
162/// On success `r.ty()` is set to the recognised command, the
163/// parsed keys are appended to [`Msg::keys`], the argument buffer
164/// is left untouched (callers that need argument bytes use
165/// [`redis_parse_req_with_args`]), and the parser cursor advances
166/// just past the trailing LF.
167///
168/// # Examples
169///
170/// ```
171/// use dynomite::msg::{Msg, MsgParseResult, MsgType};
172/// use dynomite::proto::redis::redis_parse_req;
173///
174/// let mut r = Msg::new(0, MsgType::Unknown, true);
175/// let res = redis_parse_req(&mut r, b"*2\r\n$3\r\nGET\r\n$3\r\nfoo\r\n");
176/// assert_eq!(res, MsgParseResult::Ok);
177/// assert_eq!(r.ty(), MsgType::ReqRedisGet);
178/// assert_eq!(r.keys()[0].key(), b"foo");
179/// ```
180pub fn redis_parse_req(r: &mut Msg, input: &[u8]) -> MsgParseResult {
181    redis_parse_req_with_args(r, input, None, true)
182}
183
184/// Variant of [`redis_parse_req`] that records all bulk arguments
185/// (beyond the keys) into [`Msg::args`]. Used by the rewrite path
186/// that needs each argument's bytes.
187///
188/// # Examples
189///
190/// ```
191/// use dynomite::msg::{Msg, MsgParseResult, MsgType};
192/// use dynomite::proto::redis::parser::redis_parse_req_with_args;
193///
194/// let mut r = Msg::new(0, MsgType::Unknown, true);
195/// let bytes = b"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n";
196/// let res = redis_parse_req_with_args(&mut r, bytes, None, true);
197/// assert_eq!(res, MsgParseResult::Ok);
198/// assert_eq!(r.args()[0].bytes(), b"bar");
199/// ```
200#[allow(clippy::too_many_lines)]
201pub fn redis_parse_req_with_args(
202    r: &mut Msg,
203    input: &[u8],
204    hash_tag: Option<HashTag>,
205    record_args: bool,
206) -> MsgParseResult {
207    if !r.is_request() {
208        r.set_parse_result(MsgParseResult::Error);
209        return MsgParseResult::Error;
210    }
211    let mut state = ReqState::from_u32(r.parser_state());
212    let mut p = r.parser_pos();
213    let mut token: Option<usize> = r.parser_token();
214    let mut rlen = r.rlen();
215    let mut rntokens = r.rntokens();
216    let mut ntokens = r.ntokens();
217    let mut nkeys = r.nkeys();
218    let mut ty = r.ty();
219    let mut is_read = r.flags().is_read;
220    let mut quit = r.flags().quit;
221    let mut routing = r.routing();
222
223    while p < input.len() {
224        let ch = input[p];
225        match state {
226            ReqState::Start | ReqState::Narg => {
227                if token.is_none() {
228                    if ch == b'p' || ch == b'P' {
229                        state = ReqState::InlinePing;
230                        p += 1;
231                        continue;
232                    }
233                    if ch != b'*' {
234                        return finish_req_error(
235                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
236                            routing,
237                        );
238                    }
239                    token = Some(p);
240                    rntokens = 0;
241                    state = ReqState::Narg;
242                    p += 1;
243                } else if ch.is_ascii_digit() {
244                    rntokens = rntokens
245                        .saturating_mul(10)
246                        .saturating_add(u32::from(ch - b'0'));
247                    p += 1;
248                } else if ch == CR {
249                    if rntokens == 0 {
250                        return finish_req_error(
251                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
252                            routing,
253                        );
254                    }
255                    ntokens = rntokens;
256                    token = None;
257                    state = ReqState::NargLf;
258                    p += 1;
259                } else {
260                    return finish_req_error(
261                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
262                        routing,
263                    );
264                }
265            }
266            ReqState::InlinePing => {
267                // `pInG` then trailing CRLF (4 more bytes).
268                if input.len() - p >= 5
269                    && input[p].eq_ignore_ascii_case(&b'i')
270                    && input[p + 1].eq_ignore_ascii_case(&b'n')
271                    && input[p + 2].eq_ignore_ascii_case(&b'g')
272                    && input[p + 3] == CR
273                    && input[p + 4] == LF
274                {
275                    ty = MsgType::ReqRedisPing;
276                    is_read = true;
277                    routing = MsgRouting::LocalNodeOnly;
278                    p += 5;
279                    return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
280                }
281                return finish_req_error(
282                    r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit, routing,
283                );
284            }
285            ReqState::NargLf => {
286                if ch == LF {
287                    state = ReqState::ReqTypeLen;
288                    p += 1;
289                } else {
290                    return finish_req_error(
291                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
292                        routing,
293                    );
294                }
295            }
296            ReqState::ReqTypeLen => {
297                if token.is_none() {
298                    if ch != b'$' {
299                        return finish_req_error(
300                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
301                            routing,
302                        );
303                    }
304                    token = Some(p);
305                    rlen = 0;
306                    p += 1;
307                } else if ch.is_ascii_digit() {
308                    rlen = rlen.saturating_mul(10).saturating_add(u32::from(ch - b'0'));
309                    p += 1;
310                } else if ch == CR {
311                    if rlen == 0 || rntokens == 0 {
312                        return finish_req_error(
313                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
314                            routing,
315                        );
316                    }
317                    rntokens -= 1;
318                    token = None;
319                    state = ReqState::ReqTypeLenLf;
320                    p += 1;
321                } else {
322                    return finish_req_error(
323                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
324                        routing,
325                    );
326                }
327            }
328            ReqState::ReqTypeLenLf => {
329                if ch == LF {
330                    state = ReqState::ReqType;
331                    p += 1;
332                } else {
333                    return finish_req_error(
334                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
335                        routing,
336                    );
337                }
338            }
339            ReqState::ReqType => {
340                if token.is_none() {
341                    token = Some(p);
342                }
343                let start = token.expect("token recorded");
344                let needed = start + rlen as usize;
345                if needed >= input.len() {
346                    // Need more bytes.
347                    p = input.len();
348                    break;
349                }
350                if input[needed] != CR {
351                    return finish_req_error(
352                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
353                        routing,
354                    );
355                }
356                let cmd_bytes = &input[start..needed];
357                p = needed + 1;
358                rlen = 0;
359                token = None;
360                let prev_ty = ty;
361                if prev_ty != MsgType::ReqRedisScript {
362                    ty = MsgType::Unknown;
363                }
364                if let Some((found, traits)) = lookup(cmd_bytes) {
365                    ty = found;
366                    is_read = traits.is_read;
367                    quit = traits.quit;
368                    routing = match traits.routing {
369                        RoutingOverride::None => routing,
370                        RoutingOverride::LocalNodeOnly => MsgRouting::LocalNodeOnly,
371                        RoutingOverride::TokenOwnerLocalRackOnly => {
372                            MsgRouting::TokenOwnerLocalRackOnly
373                        }
374                        RoutingOverride::AllNodesAllRacksAllDcs => {
375                            MsgRouting::AllNodesAllRacksAllDcs
376                        }
377                    };
378                    if ty == MsgType::ReqRedisExists && prev_ty == MsgType::ReqRedisScript {
379                        ty = MsgType::ReqRedisScriptExists;
380                        routing = MsgRouting::AllNodesAllRacksAllDcs;
381                        is_read = true;
382                    }
383                    if ty == MsgType::ReqRedisPing {
384                        // Short-circuit the inline form here.
385                        // p was advanced to needed+1 (the LF position).
386                        return finish_req_ok(
387                            r,
388                            p + 1,
389                            ty,
390                            true,
391                            quit,
392                            MsgRouting::LocalNodeOnly,
393                            ntokens,
394                            nkeys,
395                        );
396                    }
397                } else {
398                    return finish_req_error(
399                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
400                        routing,
401                    );
402                }
403                state = ReqState::ReqTypeLf;
404                continue;
405            }
406            ReqState::ReqTypeLf => {
407                if ty == MsgType::ReqRedisScript {
408                    state = ReqState::ReqTypeLen;
409                    p += 1;
410                    continue;
411                }
412                if ty == MsgType::HackSettingConnConsistency {
413                    state = ReqState::Arg1Len;
414                    p += 1;
415                    continue;
416                }
417                if ch != LF {
418                    return finish_req_error(
419                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
420                        routing,
421                    );
422                }
423                p += 1;
424                let class = classify(ty);
425                if matches!(class, CommandClass::Argz) && rntokens == 0 {
426                    return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
427                }
428                if class == CommandClass::ArgUpto1 && rntokens == 0 {
429                    return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
430                }
431                if class == CommandClass::ArgUpto1 && rntokens == 1 {
432                    state = ReqState::Arg1Len;
433                    continue;
434                }
435                if matches!(
436                    ty,
437                    MsgType::ReqRedisScriptLoad | MsgType::ReqRedisScriptExists
438                ) {
439                    state = ReqState::Arg1Len;
440                    continue;
441                }
442                if class == CommandClass::ArgEval {
443                    state = ReqState::Arg1Len;
444                    continue;
445                }
446                state = ReqState::KeyLen;
447            }
448            ReqState::KeyLen => {
449                if token.is_none() {
450                    if ch != b'$' {
451                        return finish_req_error(
452                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
453                            routing,
454                        );
455                    }
456                    token = Some(p);
457                    rlen = 0;
458                    p += 1;
459                } else if ch.is_ascii_digit() {
460                    rlen = rlen.saturating_mul(10).saturating_add(u32::from(ch - b'0'));
461                    p += 1;
462                } else if ch == CR {
463                    if rlen == 0 || rlen as usize >= MBUF_MAX_SIZE || rntokens == 0 {
464                        return finish_req_error(
465                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
466                            routing,
467                        );
468                    }
469                    rntokens -= 1;
470                    token = None;
471                    state = ReqState::KeyLenLf;
472                    p += 1;
473                } else {
474                    return finish_req_error(
475                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
476                        routing,
477                    );
478                }
479            }
480            ReqState::KeyLenLf => {
481                if ch == LF {
482                    state = ReqState::Key;
483                    p += 1;
484                } else {
485                    return finish_req_error(
486                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
487                        routing,
488                    );
489                }
490            }
491            ReqState::Key => {
492                if token.is_none() {
493                    token = Some(p);
494                }
495                let start = token.expect("token recorded");
496                let needed = start + rlen as usize;
497                if needed >= input.len() {
498                    p = input.len();
499                    break;
500                }
501                if input[needed] != CR {
502                    return finish_req_error(
503                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
504                        routing,
505                    );
506                }
507                let kbytes = input[start..needed].to_vec();
508                p = needed + 1;
509                rlen = 0;
510                let kp = if let Some(tag) = hash_tag {
511                    if let Some(open_idx) = kbytes.iter().position(|&b| b == tag.open) {
512                        if let Some(close_off) =
513                            kbytes[open_idx + 1..].iter().position(|&b| b == tag.close)
514                        {
515                            let s = open_idx + 1;
516                            let e = open_idx + 1 + close_off;
517                            KeyPos::new(kbytes, s..e)
518                        } else {
519                            KeyPos::without_tag(kbytes)
520                        }
521                    } else {
522                        KeyPos::without_tag(kbytes)
523                    }
524                } else {
525                    KeyPos::without_tag(kbytes)
526                };
527                r.push_key(kp);
528                token = None;
529                state = ReqState::KeyLf;
530            }
531            ReqState::KeyLf => {
532                if ch != LF {
533                    return finish_req_error(
534                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
535                        routing,
536                    );
537                }
538                p += 1;
539                let class = classify(ty);
540                match class {
541                    CommandClass::Arg0 => {
542                        if rntokens != 0 {
543                            return finish_req_error(
544                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
545                                quit, routing,
546                            );
547                        }
548                        return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
549                    }
550                    CommandClass::Arg1 => {
551                        if rntokens != 1 {
552                            return finish_req_error(
553                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
554                                quit, routing,
555                            );
556                        }
557                        state = ReqState::Arg1Len;
558                    }
559                    CommandClass::Arg2 => {
560                        if rntokens != 2 {
561                            return finish_req_error(
562                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
563                                quit, routing,
564                            );
565                        }
566                        state = ReqState::Arg1Len;
567                    }
568                    CommandClass::Arg3 => {
569                        if rntokens != 3 {
570                            return finish_req_error(
571                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
572                                quit, routing,
573                            );
574                        }
575                        state = ReqState::Arg1Len;
576                    }
577                    CommandClass::ArgN => {
578                        if rntokens == 0 {
579                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
580                        }
581                        state = ReqState::Arg1Len;
582                    }
583                    CommandClass::ArgX => {
584                        if rntokens == 0 {
585                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
586                        }
587                        state = ReqState::KeyLen;
588                    }
589                    CommandClass::ArgKvX => {
590                        if ntokens.is_multiple_of(2) {
591                            return finish_req_error(
592                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
593                                quit, routing,
594                            );
595                        }
596                        state = ReqState::Arg1Len;
597                    }
598                    CommandClass::ArgEval => {
599                        nkeys = nkeys.saturating_sub(1);
600                        if nkeys > 0 {
601                            state = ReqState::KeyLen;
602                        } else if rntokens > 0 {
603                            state = ReqState::ArgnLen;
604                        } else {
605                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
606                        }
607                    }
608                    _ => {
609                        return finish_req_error(
610                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
611                            routing,
612                        );
613                    }
614                }
615            }
616            ReqState::Arg1Len => {
617                match read_bulk_len(input, &mut p, &mut token, &mut rlen, &mut rntokens) {
618                    BulkLenStep::More => {}
619                    BulkLenStep::Done => state = ReqState::Arg1LenLf,
620                    BulkLenStep::Error => {
621                        return finish_req_error(
622                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
623                            routing,
624                        );
625                    }
626                    BulkLenStep::Eof => break,
627                }
628            }
629            ReqState::Arg1LenLf => {
630                if ch == LF {
631                    state = ReqState::Arg1;
632                    p += 1;
633                } else {
634                    return finish_req_error(
635                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
636                        routing,
637                    );
638                }
639            }
640            ReqState::Arg1 => match read_bulk_arg(input, p, rlen, record_args, r) {
641                ArgStep::More => {
642                    p = input.len();
643                    let consumed = (p - r.parser_pos()) as u32;
644                    let _ = consumed;
645                    break;
646                }
647                ArgStep::Done(new_p) => {
648                    p = new_p;
649                    rlen = 0;
650                    state = ReqState::Arg1Lf;
651                }
652                ArgStep::Error => {
653                    return finish_req_error(
654                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
655                        routing,
656                    );
657                }
658            },
659            ReqState::Arg1Lf => {
660                if ch != LF {
661                    return finish_req_error(
662                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
663                        routing,
664                    );
665                }
666                p += 1;
667                let class = classify(ty);
668                match class {
669                    CommandClass::ArgUpto1 | CommandClass::Arg1 => {
670                        if rntokens != 0 {
671                            return finish_req_error(
672                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
673                                quit, routing,
674                            );
675                        }
676                        return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
677                    }
678                    CommandClass::Arg2 => {
679                        if rntokens != 1 {
680                            return finish_req_error(
681                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
682                                quit, routing,
683                            );
684                        }
685                        state = ReqState::Arg2Len;
686                    }
687                    CommandClass::Arg3 => {
688                        if rntokens != 2 {
689                            return finish_req_error(
690                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
691                                quit, routing,
692                            );
693                        }
694                        state = ReqState::Arg2Len;
695                    }
696                    CommandClass::ArgN => {
697                        if rntokens == 0 {
698                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
699                        }
700                        state = ReqState::ArgnLen;
701                    }
702                    CommandClass::ArgEval => {
703                        // After parsing the script body we need at
704                        // least the `numkeys` argument to follow.
705                        // Valid Redis EVAL forms include `EVAL
706                        // "script" 0` (no keys, no args) which has
707                        // rntokens == 1 after the script. The
708                        // earlier `< 2` bound rejected that and
709                        // closed the client connection on every
710                        // EVAL-with-no-extra-args invocation.
711                        if rntokens < 1 {
712                            return finish_req_error(
713                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
714                                quit, routing,
715                            );
716                        }
717                        state = ReqState::Arg2Len;
718                    }
719                    CommandClass::ArgKvX => {
720                        if rntokens == 0 {
721                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
722                        }
723                        state = ReqState::KeyLen;
724                    }
725                    _ => {
726                        return finish_req_error(
727                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
728                            routing,
729                        );
730                    }
731                }
732            }
733            ReqState::Arg2Len => {
734                match read_bulk_len(input, &mut p, &mut token, &mut rlen, &mut rntokens) {
735                    BulkLenStep::More => {}
736                    BulkLenStep::Done => state = ReqState::Arg2LenLf,
737                    BulkLenStep::Error => {
738                        return finish_req_error(
739                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
740                            routing,
741                        );
742                    }
743                    BulkLenStep::Eof => break,
744                }
745            }
746            ReqState::Arg2LenLf => {
747                if ch == LF {
748                    state = ReqState::Arg2;
749                    p += 1;
750                } else {
751                    return finish_req_error(
752                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
753                        routing,
754                    );
755                }
756            }
757            ReqState::Arg2 => {
758                let class = classify(ty);
759                let token_start = if class == CommandClass::ArgEval && token.is_none() {
760                    Some(p)
761                } else {
762                    token
763                };
764                token = token_start;
765                match read_bulk_arg(input, p, rlen, record_args, r) {
766                    ArgStep::More => {
767                        p = input.len();
768                        break;
769                    }
770                    ArgStep::Done(new_p) => {
771                        p = new_p;
772                        if class == CommandClass::ArgEval {
773                            // Token holds the start of the integer.
774                            // `read_bulk_arg` returns `needed + 1`,
775                            // so position `p - 1` is the trailing
776                            // `\r`. The digit string is
777                            // `input[start..p - 1]`.
778                            let start = token.unwrap_or(0);
779                            if start >= p.saturating_sub(1) {
780                                return finish_req_error(
781                                    r, state, p, token, rlen, rntokens, ntokens, nkeys, ty,
782                                    is_read, quit, routing,
783                                );
784                            }
785                            let mut nkey: u32 = 0;
786                            for &b in &input[start..p - 1] {
787                                if b.is_ascii_digit() {
788                                    nkey =
789                                        nkey.saturating_mul(10).saturating_add(u32::from(b - b'0'));
790                                } else {
791                                    return finish_req_error(
792                                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty,
793                                        is_read, quit, routing,
794                                    );
795                                }
796                            }
797                            // numkeys may legitimately be 0 (script
798                            // takes no keys). Validate only that
799                            // there are at least `nkey` more bulk
800                            // tokens to follow.
801                            if rntokens < nkey {
802                                return finish_req_error(
803                                    r, state, p, token, rlen, rntokens, ntokens, nkeys, ty,
804                                    is_read, quit, routing,
805                                );
806                            }
807                            nkeys = nkey;
808                            token = None;
809                        }
810                        rlen = 0;
811                        state = ReqState::Arg2Lf;
812                    }
813                    ArgStep::Error => {
814                        return finish_req_error(
815                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
816                            routing,
817                        );
818                    }
819                }
820            }
821            ReqState::Arg2Lf => {
822                if ch != LF {
823                    return finish_req_error(
824                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
825                        routing,
826                    );
827                }
828                p += 1;
829                let class = classify(ty);
830                match class {
831                    CommandClass::Arg2 => {
832                        if rntokens != 0 {
833                            return finish_req_error(
834                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
835                                quit, routing,
836                            );
837                        }
838                        return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
839                    }
840                    CommandClass::Arg3 => {
841                        if rntokens != 1 {
842                            return finish_req_error(
843                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
844                                quit, routing,
845                            );
846                        }
847                        state = ReqState::Arg3Len;
848                    }
849                    CommandClass::ArgN => {
850                        if rntokens == 0 {
851                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
852                        }
853                        state = ReqState::ArgnLen;
854                    }
855                    CommandClass::ArgEval => {
856                        // After parsing the numkeys arg, the
857                        // remaining tokens are `numkeys` keys
858                        // followed by zero or more args. If
859                        // numkeys=0 AND no more args remain we
860                        // are done with a valid `EVAL "script"
861                        // 0` request.
862                        if rntokens == 0 {
863                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
864                        }
865                        // Move to the keys section if numkeys > 0
866                        // OR to the args section if numkeys == 0
867                        // and we still have arg tokens to read.
868                        // The state-transition itself does not
869                        // need to know which case we are in:
870                        // KeyLen reads the next bulk len whether
871                        // it represents a key (recorded with
872                        // routing tag) or an arg (recorded with
873                        // record_args). For EVAL with numkeys=0
874                        // we want the arg path, so route to
875                        // ArgnLen instead.
876                        if nkeys == 0 {
877                            state = ReqState::ArgnLen;
878                        } else {
879                            state = ReqState::KeyLen;
880                        }
881                    }
882                    _ => {
883                        return finish_req_error(
884                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
885                            routing,
886                        );
887                    }
888                }
889            }
890            ReqState::Arg3Len => {
891                match read_bulk_len(input, &mut p, &mut token, &mut rlen, &mut rntokens) {
892                    BulkLenStep::More => {}
893                    BulkLenStep::Done => state = ReqState::Arg3LenLf,
894                    BulkLenStep::Error => {
895                        return finish_req_error(
896                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
897                            routing,
898                        );
899                    }
900                    BulkLenStep::Eof => break,
901                }
902            }
903            ReqState::Arg3LenLf => {
904                if ch == LF {
905                    state = ReqState::Arg3;
906                    p += 1;
907                } else {
908                    return finish_req_error(
909                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
910                        routing,
911                    );
912                }
913            }
914            ReqState::Arg3 => match read_bulk_arg(input, p, rlen, record_args, r) {
915                ArgStep::More => {
916                    p = input.len();
917                    break;
918                }
919                ArgStep::Done(new_p) => {
920                    p = new_p;
921                    rlen = 0;
922                    state = ReqState::Arg3Lf;
923                }
924                ArgStep::Error => {
925                    return finish_req_error(
926                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
927                        routing,
928                    );
929                }
930            },
931            ReqState::Arg3Lf => {
932                if ch != LF {
933                    return finish_req_error(
934                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
935                        routing,
936                    );
937                }
938                p += 1;
939                let class = classify(ty);
940                match class {
941                    CommandClass::Arg3 => {
942                        if rntokens != 0 {
943                            return finish_req_error(
944                                r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read,
945                                quit, routing,
946                            );
947                        }
948                        return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
949                    }
950                    CommandClass::ArgN => {
951                        if rntokens == 0 {
952                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
953                        }
954                        state = ReqState::ArgnLen;
955                    }
956                    _ => {
957                        return finish_req_error(
958                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
959                            routing,
960                        );
961                    }
962                }
963            }
964            ReqState::ArgnLen => {
965                match read_bulk_len(input, &mut p, &mut token, &mut rlen, &mut rntokens) {
966                    BulkLenStep::More => {}
967                    BulkLenStep::Done => state = ReqState::ArgnLenLf,
968                    BulkLenStep::Error => {
969                        return finish_req_error(
970                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
971                            routing,
972                        );
973                    }
974                    BulkLenStep::Eof => break,
975                }
976            }
977            ReqState::ArgnLenLf => {
978                if ch == LF {
979                    state = ReqState::Argn;
980                    p += 1;
981                } else {
982                    return finish_req_error(
983                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
984                        routing,
985                    );
986                }
987            }
988            ReqState::Argn => match read_bulk_arg(input, p, rlen, record_args, r) {
989                ArgStep::More => {
990                    p = input.len();
991                    break;
992                }
993                ArgStep::Done(new_p) => {
994                    p = new_p;
995                    rlen = 0;
996                    state = ReqState::ArgnLf;
997                }
998                ArgStep::Error => {
999                    return finish_req_error(
1000                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
1001                        routing,
1002                    );
1003                }
1004            },
1005            ReqState::ArgnLf => {
1006                if ch != LF {
1007                    return finish_req_error(
1008                        r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
1009                        routing,
1010                    );
1011                }
1012                p += 1;
1013                let class = classify(ty);
1014                match class {
1015                    CommandClass::ArgN | CommandClass::ArgEval => {
1016                        if rntokens == 0 {
1017                            return finish_req_ok(r, p, ty, is_read, quit, routing, ntokens, nkeys);
1018                        }
1019                        state = ReqState::ArgnLen;
1020                    }
1021                    _ => {
1022                        return finish_req_error(
1023                            r, state, p, token, rlen, rntokens, ntokens, nkeys, ty, is_read, quit,
1024                            routing,
1025                        );
1026                    }
1027                }
1028            }
1029        }
1030    }
1031
1032    // Reached end of input without completing.
1033    r.set_parser_state(state as u32);
1034    r.set_parser_pos(p);
1035    r.set_parser_token(token);
1036    r.set_rlen(rlen);
1037    r.set_rntokens(rntokens);
1038    r.set_ntokens(ntokens);
1039    r.set_nkeys(nkeys);
1040    if ty != MsgType::Unknown {
1041        r.set_type(ty);
1042    }
1043    r.flags_mut().is_read = is_read;
1044    r.flags_mut().quit = quit;
1045    r.flags_mut().rewrite_with_ts_possible = false;
1046    r.set_routing(routing);
1047    r.set_parse_result(MsgParseResult::Again);
1048    MsgParseResult::Again
1049}
1050
1051#[derive(Debug)]
1052enum BulkLenStep {
1053    More,
1054    Done,
1055    Error,
1056    Eof,
1057}
1058
1059fn read_bulk_len(
1060    input: &[u8],
1061    p: &mut usize,
1062    token: &mut Option<usize>,
1063    rlen: &mut u32,
1064    rntokens: &mut u32,
1065) -> BulkLenStep {
1066    if *p >= input.len() {
1067        return BulkLenStep::Eof;
1068    }
1069    let ch = input[*p];
1070    if token.is_none() {
1071        if ch != b'$' {
1072            return BulkLenStep::Error;
1073        }
1074        *token = Some(*p);
1075        *rlen = 0;
1076        *p += 1;
1077        return BulkLenStep::More;
1078    }
1079    if ch.is_ascii_digit() {
1080        *rlen = rlen.saturating_mul(10).saturating_add(u32::from(ch - b'0'));
1081        *p += 1;
1082        BulkLenStep::More
1083    } else if ch == CR {
1084        let start = token.expect("token recorded");
1085        if *p - start <= 1 || *rntokens == 0 {
1086            return BulkLenStep::Error;
1087        }
1088        *rntokens -= 1;
1089        *token = None;
1090        *p += 1;
1091        BulkLenStep::Done
1092    } else {
1093        BulkLenStep::Error
1094    }
1095}
1096
1097#[derive(Debug)]
1098enum ArgStep {
1099    /// Need more input bytes.
1100    More,
1101    /// Argument fully consumed; new cursor `p` (pointing at CR).
1102    Done(usize),
1103    /// Bad framing.
1104    Error,
1105}
1106
1107fn read_bulk_arg(input: &[u8], p: usize, rlen: u32, record: bool, r: &mut Msg) -> ArgStep {
1108    let needed = match p.checked_add(rlen as usize) {
1109        Some(n) => n,
1110        None => return ArgStep::Error,
1111    };
1112    if needed >= input.len() {
1113        return ArgStep::More;
1114    }
1115    if input[needed] != CR {
1116        return ArgStep::Error;
1117    }
1118    if record && rlen > 0 {
1119        r.push_arg(ArgPos::new(input[p..needed].to_vec()));
1120    }
1121    ArgStep::Done(needed + 1)
1122}
1123
1124#[allow(clippy::too_many_arguments)]
1125fn finish_req_ok(
1126    r: &mut Msg,
1127    pos: usize,
1128    ty: MsgType,
1129    is_read: bool,
1130    quit: bool,
1131    routing: MsgRouting,
1132    ntokens: u32,
1133    nkeys: u32,
1134) -> MsgParseResult {
1135    if ty == MsgType::Unknown {
1136        r.set_parse_result(MsgParseResult::Error);
1137        return MsgParseResult::Error;
1138    }
1139    r.set_type(ty);
1140    r.flags_mut().is_read = is_read;
1141    r.flags_mut().quit = quit;
1142    r.set_routing(routing);
1143    r.set_ntokens(ntokens);
1144    r.set_nkeys(nkeys);
1145    r.set_parser_state(0);
1146    r.set_parser_pos(pos);
1147    r.set_parser_token(None);
1148    r.set_parse_result(MsgParseResult::Ok);
1149    MsgParseResult::Ok
1150}
1151
1152#[allow(clippy::too_many_arguments)]
1153fn finish_req_error(
1154    r: &mut Msg,
1155    state: ReqState,
1156    pos: usize,
1157    token: Option<usize>,
1158    rlen: u32,
1159    rntokens: u32,
1160    ntokens: u32,
1161    nkeys: u32,
1162    ty: MsgType,
1163    is_read: bool,
1164    quit: bool,
1165    routing: MsgRouting,
1166) -> MsgParseResult {
1167    r.set_parser_state(state as u32);
1168    r.set_parser_pos(pos);
1169    r.set_parser_token(token);
1170    r.set_rlen(rlen);
1171    r.set_rntokens(rntokens);
1172    r.set_ntokens(ntokens);
1173    r.set_nkeys(nkeys);
1174    if ty != MsgType::Unknown {
1175        r.set_type(ty);
1176    }
1177    r.flags_mut().is_read = is_read;
1178    r.flags_mut().quit = quit;
1179    r.set_routing(routing);
1180    r.set_parse_result(MsgParseResult::Error);
1181    MsgParseResult::Error
1182}
1183
1184/// Parse a Redis response from `input` and update `r` in place.
1185///
1186/// On success the response type tag is set, the integer payload
1187/// (for `:n\r\n` responses) is recorded on the message, and the
1188/// parser cursor advances just past the trailing LF. On truncated
1189/// input the function returns [`MsgParseResult::Again`]. Invalid
1190/// bytes return [`MsgParseResult::Error`].
1191///
1192/// # Examples
1193///
1194/// ```
1195/// use dynomite::msg::{Msg, MsgParseResult, MsgType};
1196/// use dynomite::proto::redis::redis_parse_rsp;
1197///
1198/// let mut r = Msg::new(0, MsgType::Unknown, false);
1199/// let res = redis_parse_rsp(&mut r, b"+OK\r\n");
1200/// assert_eq!(res, MsgParseResult::Ok);
1201/// assert_eq!(r.ty(), MsgType::RspRedisStatus);
1202/// ```
1203#[allow(clippy::too_many_lines)]
1204pub fn redis_parse_rsp(r: &mut Msg, input: &[u8]) -> MsgParseResult {
1205    if r.is_request() {
1206        r.set_parse_result(MsgParseResult::Error);
1207        return MsgParseResult::Error;
1208    }
1209    let mut state = RspState::from_u32(r.parser_state());
1210    let mut p = r.parser_pos();
1211    let mut token: Option<usize> = r.parser_token();
1212    let mut rlen = r.rlen();
1213    let mut rntokens = r.rntokens();
1214    let mut ty = r.ty();
1215    let mut integer = r.integer();
1216    let mut int_negative = false;
1217    let mut ntoken_start: Option<usize> = r.ntoken_start();
1218    let mut ntoken_end: Option<usize> = r.ntoken_end();
1219
1220    while p < input.len() {
1221        let ch = input[p];
1222        match state {
1223            RspState::Start => {
1224                ty = MsgType::Unknown;
1225                match ch {
1226                    b'+' => {
1227                        ty = MsgType::RspRedisStatus;
1228                        state = RspState::RuntoCrlf;
1229                        p += 1;
1230                    }
1231                    b'-' => {
1232                        ty = MsgType::RspRedisError;
1233                        state = RspState::Error;
1234                    }
1235                    b':' => {
1236                        ty = MsgType::RspRedisInteger;
1237                        state = RspState::IntegerStart;
1238                        integer = 0;
1239                        int_negative = false;
1240                        p += 1;
1241                    }
1242                    b'$' => {
1243                        ty = MsgType::RspRedisBulk;
1244                        state = RspState::Bulk;
1245                    }
1246                    b'*' => {
1247                        ty = MsgType::RspRedisMultibulk;
1248                        state = RspState::Multibulk;
1249                    }
1250                    _ => {
1251                        return finish_rsp_error(
1252                            r,
1253                            state,
1254                            p,
1255                            token,
1256                            rlen,
1257                            rntokens,
1258                            ty,
1259                            integer,
1260                            ntoken_start,
1261                            ntoken_end,
1262                        );
1263                    }
1264                }
1265            }
1266            // Resume-from-state entry: the parser API allows callers
1267            // to resume from a saved RspState across input chunks.
1268            // Status and Integer entries are no-op transitions to
1269            // their respective body states (RuntoCrlf and
1270            // IntegerStart) so a resumed parse picks up where the
1271            // caller left off without re-consuming a byte.
1272            RspState::Status => {
1273                state = RspState::RuntoCrlf;
1274            }
1275            RspState::Error => {
1276                if token.is_none() {
1277                    if ch != b'-' {
1278                        return finish_rsp_error(
1279                            r,
1280                            state,
1281                            p,
1282                            token,
1283                            rlen,
1284                            rntokens,
1285                            ty,
1286                            integer,
1287                            ntoken_start,
1288                            ntoken_end,
1289                        );
1290                    }
1291                    token = Some(p);
1292                    p += 1;
1293                } else if ch == b' ' || ch == CR {
1294                    let start = token.expect("token recorded");
1295                    if let Some(t) = error_lookup(&input[start..p]) {
1296                        ty = t;
1297                    }
1298                    state = RspState::RuntoCrlf;
1299                    token = None;
1300                    // Do not advance.
1301                } else {
1302                    p += 1;
1303                }
1304            }
1305            RspState::Integer => {
1306                // Resume-from-state entry: see RspState::Status.
1307                state = RspState::IntegerStart;
1308                integer = 0;
1309                int_negative = false;
1310            }
1311            RspState::IntegerStart => {
1312                if ch == CR {
1313                    if int_negative {
1314                        integer = -integer;
1315                    }
1316                    state = RspState::AlmostDone;
1317                    p += 1;
1318                } else if ch == b'-' {
1319                    int_negative = true;
1320                    p += 1;
1321                } else if ch.is_ascii_digit() {
1322                    integer = integer
1323                        .saturating_mul(10)
1324                        .saturating_add(i64::from(ch - b'0'));
1325                    p += 1;
1326                } else {
1327                    return finish_rsp_error(
1328                        r,
1329                        state,
1330                        p,
1331                        token,
1332                        rlen,
1333                        rntokens,
1334                        ty,
1335                        integer,
1336                        ntoken_start,
1337                        ntoken_end,
1338                    );
1339                }
1340            }
1341            RspState::Simple => {
1342                if ch == CR {
1343                    state = RspState::MultibulkArgnLf;
1344                    rntokens = rntokens.saturating_sub(1);
1345                }
1346                p += 1;
1347            }
1348            RspState::RuntoCrlf => {
1349                if ch == CR {
1350                    state = RspState::AlmostDone;
1351                    p += 1;
1352                } else {
1353                    p += 1;
1354                }
1355            }
1356            RspState::AlmostDone => {
1357                if ch != LF {
1358                    return finish_rsp_error(
1359                        r,
1360                        state,
1361                        p,
1362                        token,
1363                        rlen,
1364                        rntokens,
1365                        ty,
1366                        integer,
1367                        ntoken_start,
1368                        ntoken_end,
1369                    );
1370                }
1371                p += 1;
1372                return finish_rsp_ok(r, p, ty, integer, ntoken_start, ntoken_end);
1373            }
1374            RspState::Bulk => {
1375                if token.is_none() {
1376                    if ch != b'$' {
1377                        return finish_rsp_error(
1378                            r,
1379                            state,
1380                            p,
1381                            token,
1382                            rlen,
1383                            rntokens,
1384                            ty,
1385                            integer,
1386                            ntoken_start,
1387                            ntoken_end,
1388                        );
1389                    }
1390                    token = Some(p);
1391                    rlen = 0;
1392                    p += 1;
1393                } else if ch == b'-' {
1394                    state = RspState::RuntoCrlf;
1395                    p += 1;
1396                } else if ch.is_ascii_digit() {
1397                    rlen = rlen.saturating_mul(10).saturating_add(u32::from(ch - b'0'));
1398                    p += 1;
1399                } else if ch == CR {
1400                    let start = token.expect("token recorded");
1401                    if p - start <= 1 {
1402                        return finish_rsp_error(
1403                            r,
1404                            state,
1405                            p,
1406                            token,
1407                            rlen,
1408                            rntokens,
1409                            ty,
1410                            integer,
1411                            ntoken_start,
1412                            ntoken_end,
1413                        );
1414                    }
1415                    token = None;
1416                    state = RspState::BulkLf;
1417                    p += 1;
1418                } else {
1419                    return finish_rsp_error(
1420                        r,
1421                        state,
1422                        p,
1423                        token,
1424                        rlen,
1425                        rntokens,
1426                        ty,
1427                        integer,
1428                        ntoken_start,
1429                        ntoken_end,
1430                    );
1431                }
1432            }
1433            RspState::BulkLf => {
1434                if ch == LF {
1435                    state = RspState::BulkArg;
1436                    p += 1;
1437                } else {
1438                    return finish_rsp_error(
1439                        r,
1440                        state,
1441                        p,
1442                        token,
1443                        rlen,
1444                        rntokens,
1445                        ty,
1446                        integer,
1447                        ntoken_start,
1448                        ntoken_end,
1449                    );
1450                }
1451            }
1452            RspState::BulkArg => {
1453                let needed = match p.checked_add(rlen as usize) {
1454                    Some(n) => n,
1455                    None => {
1456                        return finish_rsp_error(
1457                            r,
1458                            state,
1459                            p,
1460                            token,
1461                            rlen,
1462                            rntokens,
1463                            ty,
1464                            integer,
1465                            ntoken_start,
1466                            ntoken_end,
1467                        );
1468                    }
1469                };
1470                if needed >= input.len() {
1471                    p = input.len();
1472                    break;
1473                }
1474                if input[needed] != CR {
1475                    return finish_rsp_error(
1476                        r,
1477                        state,
1478                        p,
1479                        token,
1480                        rlen,
1481                        rntokens,
1482                        ty,
1483                        integer,
1484                        ntoken_start,
1485                        ntoken_end,
1486                    );
1487                }
1488                p = needed + 1;
1489                rlen = 0;
1490                state = RspState::BulkArgLf;
1491            }
1492            RspState::BulkArgLf => {
1493                if ch != CR && ch != LF {
1494                    return finish_rsp_error(
1495                        r,
1496                        state,
1497                        p,
1498                        token,
1499                        rlen,
1500                        rntokens,
1501                        ty,
1502                        integer,
1503                        ntoken_start,
1504                        ntoken_end,
1505                    );
1506                }
1507                if ch == CR {
1508                    p += 1;
1509                    continue;
1510                }
1511                p += 1;
1512                return finish_rsp_ok(r, p, ty, integer, ntoken_start, ntoken_end);
1513            }
1514            RspState::Multibulk => {
1515                if token.is_none() {
1516                    if ch != b'*' {
1517                        return finish_rsp_error(
1518                            r,
1519                            state,
1520                            p,
1521                            token,
1522                            rlen,
1523                            rntokens,
1524                            ty,
1525                            integer,
1526                            ntoken_start,
1527                            ntoken_end,
1528                        );
1529                    }
1530                    token = Some(p);
1531                    ntoken_start = Some(p);
1532                    rntokens = 0;
1533                    p += 1;
1534                } else if ch == b'-' {
1535                    state = RspState::RuntoCrlf;
1536                    p += 1;
1537                } else if ch.is_ascii_digit() {
1538                    rntokens = rntokens
1539                        .saturating_mul(10)
1540                        .saturating_add(u32::from(ch - b'0'));
1541                    p += 1;
1542                } else if ch == CR {
1543                    let start = token.expect("token recorded");
1544                    if p - start <= 1 {
1545                        return finish_rsp_error(
1546                            r,
1547                            state,
1548                            p,
1549                            token,
1550                            rlen,
1551                            rntokens,
1552                            ty,
1553                            integer,
1554                            ntoken_start,
1555                            ntoken_end,
1556                        );
1557                    }
1558                    ntoken_end = Some(p);
1559                    token = None;
1560                    state = RspState::MultibulkNargLf;
1561                    p += 1;
1562                } else {
1563                    return finish_rsp_error(
1564                        r,
1565                        state,
1566                        p,
1567                        token,
1568                        rlen,
1569                        rntokens,
1570                        ty,
1571                        integer,
1572                        ntoken_start,
1573                        ntoken_end,
1574                    );
1575                }
1576            }
1577            RspState::MultibulkNargLf => {
1578                if ch != LF {
1579                    return finish_rsp_error(
1580                        r,
1581                        state,
1582                        p,
1583                        token,
1584                        rlen,
1585                        rntokens,
1586                        ty,
1587                        integer,
1588                        ntoken_start,
1589                        ntoken_end,
1590                    );
1591                }
1592                p += 1;
1593                if rntokens == 0 {
1594                    return finish_rsp_ok(r, p, ty, integer, ntoken_start, ntoken_end);
1595                }
1596                state = RspState::MultibulkArgnLen;
1597            }
1598            RspState::MultibulkArgnLen => {
1599                if token.is_none() {
1600                    if ch == b'*' {
1601                        state = RspState::Multibulk;
1602                        // Do not advance.
1603                        continue;
1604                    }
1605                    if ch == b':' || ch == b'+' || ch == b'-' {
1606                        state = RspState::Simple;
1607                        // Do not advance.
1608                        continue;
1609                    }
1610                    if ch != b'$' {
1611                        return finish_rsp_error(
1612                            r,
1613                            state,
1614                            p,
1615                            token,
1616                            rlen,
1617                            rntokens,
1618                            ty,
1619                            integer,
1620                            ntoken_start,
1621                            ntoken_end,
1622                        );
1623                    }
1624                    token = Some(p);
1625                    rlen = 0;
1626                    p += 1;
1627                } else if ch.is_ascii_digit() {
1628                    rlen = rlen.saturating_mul(10).saturating_add(u32::from(ch - b'0'));
1629                    p += 1;
1630                } else if ch == b'-' {
1631                    p += 1;
1632                } else if ch == CR {
1633                    let start = token.expect("token recorded");
1634                    if p - start <= 1 || rntokens == 0 {
1635                        return finish_rsp_error(
1636                            r,
1637                            state,
1638                            p,
1639                            token,
1640                            rlen,
1641                            rntokens,
1642                            ty,
1643                            integer,
1644                            ntoken_start,
1645                            ntoken_end,
1646                        );
1647                    }
1648                    if rlen == 1 && p - start == 3 {
1649                        rlen = 0;
1650                        state = RspState::MultibulkArgnLf;
1651                    } else {
1652                        state = RspState::MultibulkArgnLenLf;
1653                    }
1654                    rntokens = rntokens.saturating_sub(1);
1655                    token = None;
1656                    p += 1;
1657                } else {
1658                    return finish_rsp_error(
1659                        r,
1660                        state,
1661                        p,
1662                        token,
1663                        rlen,
1664                        rntokens,
1665                        ty,
1666                        integer,
1667                        ntoken_start,
1668                        ntoken_end,
1669                    );
1670                }
1671            }
1672            RspState::MultibulkArgnLenLf => {
1673                if ch == LF {
1674                    state = RspState::MultibulkArgn;
1675                    p += 1;
1676                } else {
1677                    return finish_rsp_error(
1678                        r,
1679                        state,
1680                        p,
1681                        token,
1682                        rlen,
1683                        rntokens,
1684                        ty,
1685                        integer,
1686                        ntoken_start,
1687                        ntoken_end,
1688                    );
1689                }
1690            }
1691            RspState::MultibulkArgn => {
1692                let needed = match p.checked_add(rlen as usize) {
1693                    Some(n) => n,
1694                    None => {
1695                        return finish_rsp_error(
1696                            r,
1697                            state,
1698                            p,
1699                            token,
1700                            rlen,
1701                            rntokens,
1702                            ty,
1703                            integer,
1704                            ntoken_start,
1705                            ntoken_end,
1706                        );
1707                    }
1708                };
1709                if needed >= input.len() {
1710                    p = input.len();
1711                    break;
1712                }
1713                if input[needed] != CR {
1714                    return finish_rsp_error(
1715                        r,
1716                        state,
1717                        p,
1718                        token,
1719                        rlen,
1720                        rntokens,
1721                        ty,
1722                        integer,
1723                        ntoken_start,
1724                        ntoken_end,
1725                    );
1726                }
1727                if rlen > 0 {
1728                    r.push_arg(ArgPos::new(input[p..needed].to_vec()));
1729                }
1730                p = needed + 1;
1731                rlen = 0;
1732                state = RspState::MultibulkArgnLf;
1733            }
1734            RspState::MultibulkArgnLf => {
1735                if ch != LF {
1736                    return finish_rsp_error(
1737                        r,
1738                        state,
1739                        p,
1740                        token,
1741                        rlen,
1742                        rntokens,
1743                        ty,
1744                        integer,
1745                        ntoken_start,
1746                        ntoken_end,
1747                    );
1748                }
1749                p += 1;
1750                if rntokens == 0 {
1751                    return finish_rsp_ok(r, p, ty, integer, ntoken_start, ntoken_end);
1752                }
1753                state = RspState::MultibulkArgnLen;
1754            }
1755        }
1756    }
1757
1758    r.set_parser_state(state as u32);
1759    r.set_parser_pos(p);
1760    r.set_parser_token(token);
1761    r.set_rlen(rlen);
1762    r.set_rntokens(rntokens);
1763    r.set_integer(integer);
1764    r.set_ntoken_span(ntoken_start, ntoken_end);
1765    if ty != MsgType::Unknown {
1766        r.set_type(ty);
1767    }
1768    r.flags_mut().is_error = super::commands::is_redis_error(ty);
1769    r.set_parse_result(MsgParseResult::Again);
1770    MsgParseResult::Again
1771}
1772
1773fn finish_rsp_ok(
1774    r: &mut Msg,
1775    pos: usize,
1776    ty: MsgType,
1777    integer: i64,
1778    ntoken_start: Option<usize>,
1779    ntoken_end: Option<usize>,
1780) -> MsgParseResult {
1781    if ty == MsgType::Unknown {
1782        r.set_parse_result(MsgParseResult::Error);
1783        return MsgParseResult::Error;
1784    }
1785    r.set_type(ty);
1786    r.set_integer(integer);
1787    r.set_ntoken_span(ntoken_start, ntoken_end);
1788    r.set_parser_state(0);
1789    r.set_parser_pos(pos);
1790    r.set_parser_token(None);
1791    r.flags_mut().is_error = super::commands::is_redis_error(ty);
1792    r.set_parse_result(MsgParseResult::Ok);
1793    MsgParseResult::Ok
1794}
1795
1796#[allow(clippy::too_many_arguments)]
1797fn finish_rsp_error(
1798    r: &mut Msg,
1799    state: RspState,
1800    pos: usize,
1801    token: Option<usize>,
1802    rlen: u32,
1803    rntokens: u32,
1804    ty: MsgType,
1805    integer: i64,
1806    ntoken_start: Option<usize>,
1807    ntoken_end: Option<usize>,
1808) -> MsgParseResult {
1809    r.set_parser_state(state as u32);
1810    r.set_parser_pos(pos);
1811    r.set_parser_token(token);
1812    r.set_rlen(rlen);
1813    r.set_rntokens(rntokens);
1814    r.set_integer(integer);
1815    r.set_ntoken_span(ntoken_start, ntoken_end);
1816    if ty != MsgType::Unknown {
1817        r.set_type(ty);
1818    }
1819    r.set_parse_result(MsgParseResult::Error);
1820    MsgParseResult::Error
1821}
1822
1823#[cfg(test)]
1824mod tests {
1825    use super::*;
1826
1827    fn parse_req(input: &[u8]) -> Msg {
1828        let mut m = Msg::new(0, MsgType::Unknown, true);
1829        let _ = redis_parse_req(&mut m, input);
1830        m
1831    }
1832
1833    fn parse_rsp(input: &[u8]) -> Msg {
1834        let mut m = Msg::new(0, MsgType::Unknown, false);
1835        let _ = redis_parse_rsp(&mut m, input);
1836        m
1837    }
1838
1839    #[test]
1840    fn parse_get() {
1841        let m = parse_req(b"*2\r\n$3\r\nGET\r\n$3\r\nfoo\r\n");
1842        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1843        assert_eq!(m.ty(), MsgType::ReqRedisGet);
1844        assert_eq!(m.keys()[0].key(), b"foo");
1845    }
1846
1847    #[test]
1848    fn parse_set() {
1849        let m = parse_req(b"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n");
1850        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1851        assert_eq!(m.ty(), MsgType::ReqRedisSet);
1852        assert_eq!(m.keys()[0].key(), b"foo");
1853        assert_eq!(m.args()[0].bytes(), b"bar");
1854    }
1855
1856    #[test]
1857    fn parse_del_multikey() {
1858        let m = parse_req(b"*4\r\n$3\r\nDEL\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n");
1859        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1860        assert_eq!(m.ty(), MsgType::ReqRedisDel);
1861        let keys: Vec<&[u8]> = m.keys().iter().map(crate::msg::KeyPos::key).collect();
1862        assert_eq!(keys, vec![&b"a"[..], b"b", b"c"]);
1863    }
1864
1865    #[test]
1866    fn parse_mset() {
1867        let m = parse_req(b"*5\r\n$4\r\nMSET\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n");
1868        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1869        assert_eq!(m.ty(), MsgType::ReqRedisMset);
1870        let keys: Vec<&[u8]> = m.keys().iter().map(crate::msg::KeyPos::key).collect();
1871        assert_eq!(keys, vec![&b"a"[..], b"b"]);
1872    }
1873
1874    #[test]
1875    fn parse_ping_inline() {
1876        let m = parse_req(b"PING\r\n");
1877        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1878        assert_eq!(m.ty(), MsgType::ReqRedisPing);
1879    }
1880
1881    #[test]
1882    fn parse_ping_resp() {
1883        let m = parse_req(b"*1\r\n$4\r\nPING\r\n");
1884        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1885        assert_eq!(m.ty(), MsgType::ReqRedisPing);
1886    }
1887
1888    #[test]
1889    fn parse_unknown_command_errors() {
1890        let m = parse_req(b"*1\r\n$3\r\nFOO\r\n");
1891        assert_eq!(m.parse_result(), MsgParseResult::Error);
1892    }
1893
1894    #[test]
1895    fn parse_truncated_returns_again() {
1896        let m = parse_req(b"*2\r\n$3\r\nGET\r\n$3\r\nfo");
1897        assert_eq!(m.parse_result(), MsgParseResult::Again);
1898    }
1899
1900    #[test]
1901    fn parse_status_response() {
1902        let m = parse_rsp(b"+OK\r\n");
1903        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1904        assert_eq!(m.ty(), MsgType::RspRedisStatus);
1905    }
1906
1907    #[test]
1908    fn parse_error_response_classifies() {
1909        let m = parse_rsp(b"-WRONGTYPE Operation not allowed\r\n");
1910        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1911        assert_eq!(m.ty(), MsgType::RspRedisErrorWrongtype);
1912        assert!(m.flags().is_error);
1913    }
1914
1915    #[test]
1916    fn parse_integer_response() {
1917        let m = parse_rsp(b":42\r\n");
1918        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1919        assert_eq!(m.ty(), MsgType::RspRedisInteger);
1920        assert_eq!(m.integer(), 42);
1921    }
1922
1923    #[test]
1924    fn parse_bulk_response() {
1925        let m = parse_rsp(b"$5\r\nhello\r\n");
1926        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1927        assert_eq!(m.ty(), MsgType::RspRedisBulk);
1928    }
1929
1930    #[test]
1931    fn parse_null_bulk_response() {
1932        let m = parse_rsp(b"$-1\r\n");
1933        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1934        assert_eq!(m.ty(), MsgType::RspRedisBulk);
1935    }
1936
1937    #[test]
1938    fn parse_empty_multibulk_response() {
1939        let m = parse_rsp(b"*0\r\n");
1940        assert_eq!(m.parse_result(), MsgParseResult::Ok);
1941        assert_eq!(m.ty(), MsgType::RspRedisMultibulk);
1942    }
1943}
dynomite/proto/redis/parser.rs

dynomite/proto/redis/
parser.rs
