From 39fc2859711760bf7f44a96974b15f6a906b0c47 Mon Sep 17 00:00:00 2001 From: Mark Rattle Date: Tue, 7 Jul 2026 16:36:56 -0400 Subject: [PATCH] Handle standard errors in meta protocol parser Assisted-By: devx/2de6bd4a-aeaa-4463-b734-383caca64107 --- src/parser/meta_parser.rs | 90 ++++++- src/proto/meta_protocol.rs | 56 ++++- tests/meta_proto_integration_tests.rs | 349 ++++++++++++++++++++++++++ 3 files changed, 487 insertions(+), 8 deletions(-) diff --git a/src/parser/meta_parser.rs b/src/parser/meta_parser.rs index 9480d95..522dc41 100644 --- a/src/parser/meta_parser.rs +++ b/src/parser/meta_parser.rs @@ -1,10 +1,11 @@ use nom::{ branch::alt, - bytes::streaming::{tag, take, take_while1}, + bytes::streaming::{tag, take, take_until, take_while1}, character::streaming::{crlf, space1}, - combinator::{map, opt, value}, + combinator::{map, map_res, opt, value}, error::ErrorKind::Fail, multi::many0, + sequence::{preceded, terminated}, IResult, Parser, }; @@ -13,12 +14,32 @@ use std::num::NonZero; use super::{parse_u32, ErrorKind, MetaResponse, MetaValue, Status}; use crate::Error; +fn parse_meta_error(buf: &[u8]) -> IResult<&[u8], MetaResponse> { + let parser = terminated( + alt(( + value(ErrorKind::NonexistentCommand, tag(&b"ERROR"[..])), + map_res( + preceded(tag(&b"CLIENT_ERROR "[..]), take_until("\r\n")), + |s| std::str::from_utf8(s).map(|s| ErrorKind::Client(s.to_string())), + ), + map_res( + preceded(tag(&b"SERVER_ERROR "[..]), take_until("\r\n")), + |s| std::str::from_utf8(s).map(|s| ErrorKind::Server(s.to_string())), + ), + )), + crlf, + ); + + map(parser, |e| MetaResponse::Status(Status::Error(e))).parse(buf) +} + pub fn parse_meta_get_status(buf: &[u8]) -> IResult<&[u8], MetaResponse> { alt(( value(MetaResponse::Status(Status::Value), tag(&b"VA "[..])), value(MetaResponse::Status(Status::Exists), tag(&b"HD"[..])), value(MetaResponse::Status(Status::NotFound), tag(&b"EN"[..])), value(MetaResponse::Status(Status::NoOp), tag(&b"MN\r\n"[..])), + parse_meta_error, )) .parse(buf) } @@ -30,6 +51,7 @@ pub fn parse_meta_set_status(buf: &[u8]) -> IResult<&[u8], MetaResponse> { value(MetaResponse::Status(Status::Exists), tag(&b"EX"[..])), value(MetaResponse::Status(Status::NotFound), tag(&b"NF"[..])), value(MetaResponse::Status(Status::NoOp), tag(&b"MN\r\n"[..])), + parse_meta_error, )) .parse(buf) } @@ -40,6 +62,7 @@ pub fn parse_meta_delete_status(buf: &[u8]) -> IResult<&[u8], MetaResponse> { value(MetaResponse::Status(Status::NotFound), tag(&b"NF"[..])), value(MetaResponse::Status(Status::Exists), tag(&b"EX"[..])), value(MetaResponse::Status(Status::NoOp), tag(&b"MN\r\n"[..])), + parse_meta_error, )) .parse(buf) } @@ -52,6 +75,7 @@ pub fn parse_meta_arithmetic_status(buf: &[u8]) -> IResult<&[u8], MetaResponse> value(MetaResponse::Status(Status::NotStored), tag(&b"NS"[..])), value(MetaResponse::Status(Status::Exists), tag(&b"EX"[..])), value(MetaResponse::Status(Status::NoOp), tag(&b"MN\r\n"[..])), + parse_meta_error, )) .parse(buf) } @@ -196,6 +220,8 @@ fn parse_meta_get_data_value(buf: &[u8]) -> IResult<&[u8], MetaResponse> { } // match arm for "MN\r\n" response MetaResponse::Status(Status::NoOp) => Ok((input, MetaResponse::Status(Status::NoOp))), + // match arm for standard ASCII error responses: ERROR, CLIENT_ERROR, SERVER_ERROR + MetaResponse::Status(Status::Error(_)) => Ok((input, status)), _ => { // unexpected response code, should never happen, bail Err(nom::Err::Error(nom::error::Error::new( @@ -234,6 +260,8 @@ fn parse_meta_set_data_value(buf: &[u8]) -> IResult<&[u8], MetaResponse> { match status { // match arm for "MN\r\n" response MetaResponse::Status(Status::NoOp) => Ok((input, MetaResponse::Status(Status::NoOp))), + // match arm for standard ASCII error responses: ERROR, CLIENT_ERROR, SERVER_ERROR + MetaResponse::Status(Status::Error(_)) => Ok((input, status)), // match arm for "HD", "NS", "EX" & "NF" responses MetaResponse::Status(s) => process_meta_response_without_data_payload(input, s), _ => Err(nom::Err::Error(nom::error::Error::new( @@ -271,6 +299,8 @@ fn parse_meta_delete_data_value(buf: &[u8]) -> IResult<&[u8], MetaResponse> { match status { // match arm for "MN\r\n" response MetaResponse::Status(Status::NoOp) => Ok((input, MetaResponse::Status(Status::NoOp))), + // match arm for standard ASCII error responses: ERROR, CLIENT_ERROR, SERVER_ERROR + MetaResponse::Status(Status::Error(_)) => Ok((input, status)), // match arm for "HD", "NF" & "EX" responses MetaResponse::Status(s) => process_meta_response_without_data_payload(input, s), _ => Err(nom::Err::Error(nom::error::Error::new( @@ -299,6 +329,8 @@ fn parse_meta_arithmetic_data_value(buf: &[u8]) -> IResult<&[u8], MetaResponse> } // match arm for "MN\r\n" response MetaResponse::Status(Status::NoOp) => Ok((input, MetaResponse::Status(Status::NoOp))), + // match arm for standard ASCII error responses: ERROR, CLIENT_ERROR, SERVER_ERROR + MetaResponse::Status(Status::Error(_)) => Ok((input, status)), // match arm for "HD", "NF" & "EX" responses MetaResponse::Status(s) => process_meta_response_without_data_payload(input, s), _ => Err(nom::Err::Error(nom::error::Error::new( @@ -988,4 +1020,58 @@ mod tests { _ => panic!("Expected Response::Data, got something else"), } } + + #[test] + fn test_parse_meta_get_response_handles_error() { + let response = parse_meta_get_response(b"ERROR\r\n").unwrap().unwrap(); + + assert_eq!(response.0, 7); + assert_eq!( + response.1, + MetaResponse::Status(Status::Error(ErrorKind::NonexistentCommand)) + ); + } + + #[test] + fn test_parse_meta_set_response_handles_client_error() { + let response = parse_meta_set_response(b"CLIENT_ERROR bad command line format\r\n") + .unwrap() + .unwrap(); + + assert_eq!(response.0, 38); + assert_eq!( + response.1, + MetaResponse::Status(Status::Error(ErrorKind::Client( + "bad command line format".to_string() + ))) + ); + } + + #[test] + fn test_parse_meta_delete_response_handles_server_error() { + let response = parse_meta_delete_response(b"SERVER_ERROR out of memory\r\n") + .unwrap() + .unwrap(); + + assert_eq!(response.0, 28); + assert_eq!( + response.1, + MetaResponse::Status(Status::Error(ErrorKind::Server( + "out of memory".to_string() + ))) + ); + } + + #[test] + fn test_parse_meta_arithmetic_response_handles_client_error() { + let response = parse_meta_arithmetic_response(b"CLIENT_ERROR bad flag\r\n") + .unwrap() + .unwrap(); + + assert_eq!(response.0, 23); + assert_eq!( + response.1, + MetaResponse::Status(Status::Error(ErrorKind::Client("bad flag".to_string()))) + ); + } } diff --git a/src/proto/meta_protocol.rs b/src/proto/meta_protocol.rs index 299d37e..0ea53dc 100644 --- a/src/proto/meta_protocol.rs +++ b/src/proto/meta_protocol.rs @@ -1,4 +1,4 @@ -use crate::{AsMemcachedValue, Client, Error, Status}; +use crate::{AsMemcachedValue, Client, Error, ErrorKind, Status}; use crate::parser::{ parse_meta_arithmetic_response, parse_meta_delete_response, parse_meta_get_response, @@ -10,6 +10,25 @@ use std::future::Future; use tokio::io::AsyncWriteExt; +type MetaResponseParser = fn(&[u8]) -> Result, ErrorKind>; + +async fn drain_quiet_noop_response( + client: &mut Client, + parser: MetaResponseParser, +) -> Result<(), Error> { + match client.drive_receive(parser).await? { + MetaResponse::Status(Status::NoOp) => Ok(()), + MetaResponse::Status(status) => Err(Status::Error(ErrorKind::Protocol(Some(format!( + "Expected quiet-mode no-op response, got status {status}" + )))) + .into()), + MetaResponse::Data(_) => Err(Status::Error(ErrorKind::Protocol(Some( + "Expected quiet-mode no-op response, got data response".to_string(), + ))) + .into()), + } +} + /// Trait defining Meta protocol-specific methods for the Client. pub trait MetaProtocol { /// Gets the given key with additional metadata. @@ -174,7 +193,12 @@ impl MetaProtocol for Client { self.conn.flush().await?; - match self.drive_receive(parse_meta_get_response).await? { + let response = self.drive_receive(parse_meta_get_response).await?; + if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) { + drain_quiet_noop_response(self, parse_meta_get_response).await?; + } + + match response { MetaResponse::Status(Status::NotFound) => Ok(None), MetaResponse::Status(Status::NoOp) => Ok(None), MetaResponse::Status(s) => Err(s.into()), @@ -232,7 +256,12 @@ impl MetaProtocol for Client { self.conn.flush().await?; - match self.drive_receive(parse_meta_set_response).await? { + let response = self.drive_receive(parse_meta_set_response).await?; + if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) { + drain_quiet_noop_response(self, parse_meta_set_response).await?; + } + + match response { MetaResponse::Status(Status::Stored) => Ok(None), MetaResponse::Status(Status::NoOp) => Ok(None), MetaResponse::Status(s) => Err(s.into()), @@ -269,7 +298,12 @@ impl MetaProtocol for Client { self.conn.flush().await?; - match self.drive_receive(parse_meta_delete_response).await? { + let response = self.drive_receive(parse_meta_delete_response).await?; + if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) { + drain_quiet_noop_response(self, parse_meta_delete_response).await?; + } + + match response { MetaResponse::Status(Status::Deleted) => Ok(None), MetaResponse::Status(Status::Exists) => Err(Error::Protocol(Status::Exists)), MetaResponse::Status(Status::NoOp) => Ok(None), @@ -331,7 +365,12 @@ impl MetaProtocol for Client { self.conn.flush().await?; - match self.drive_receive(parse_meta_arithmetic_response).await? { + let response = self.drive_receive(parse_meta_arithmetic_response).await?; + if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) { + drain_quiet_noop_response(self, parse_meta_arithmetic_response).await?; + } + + match response { MetaResponse::Status(Status::Stored) => Ok(None), MetaResponse::Status(Status::NoOp) => Ok(None), MetaResponse::Status(s) => Err(s.into()), @@ -392,7 +431,12 @@ impl MetaProtocol for Client { self.conn.flush().await?; - match self.drive_receive(parse_meta_arithmetic_response).await? { + let response = self.drive_receive(parse_meta_arithmetic_response).await?; + if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) { + drain_quiet_noop_response(self, parse_meta_arithmetic_response).await?; + } + + match response { MetaResponse::Status(Status::Stored) => Ok(None), MetaResponse::Status(Status::NoOp) => Ok(None), MetaResponse::Status(s) => Err(s.into()), diff --git a/tests/meta_proto_integration_tests.rs b/tests/meta_proto_integration_tests.rs index 46dc752..423ecbf 100644 --- a/tests/meta_proto_integration_tests.rs +++ b/tests/meta_proto_integration_tests.rs @@ -1,5 +1,7 @@ use async_memcached::{AsciiProtocol, Client, Error, ErrorKind, MetaProtocol, Status}; use serial_test::parallel; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; // NOTE: Each test should run with keys unique to that test to avoid async conflicts. Because these tests run concurrently, // it's possible to delete/overwrite keys created by another test before they're read. @@ -26,6 +28,353 @@ async fn setup_client(keys: &[&str]) -> Client { client } +async fn read_until_contains(socket: &mut tokio::net::TcpStream, needle: &[u8]) -> Vec { + let mut buf = Vec::new(); + let mut tmp = [0; 1024]; + + loop { + // Bounded test with timeout to avoid any potential hangs + let n = tokio::time::timeout(std::time::Duration::from_secs(5), socket.read(&mut tmp)) + .await + .expect("timed out waiting for expected command") + .unwrap(); + assert_ne!(n, 0, "socket closed before expected command was received"); + buf.extend_from_slice(&tmp[..n]); + + if buf.windows(needle.len()).any(|window| window == needle) { + return buf; + } + } +} + +#[derive(Clone, Copy, Debug)] +enum QuietMetaOperation { + GetError, + GetData, + GetNoop, + SetStored, + SetError, + DeleteDeleted, + DeleteError, + IncrementValue, + IncrementError, + DecrementValue, + DecrementError, +} + +impl QuietMetaOperation { + async fn run(self, client: &mut Client) { + match self { + Self::GetError => { + let result = client + .meta_get("bad-client-error", true, None, Some(&["badflag"])) + .await; + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Client( + "bad flag".to_string() + )))) + ); + } + Self::GetData => { + let result = client + .meta_get("quiet-hit", true, None, Some(&["v"])) + .await + .unwrap() + .unwrap(); + assert_eq!(result.data, Some(b"value".to_vec())); + } + Self::GetNoop => { + let result = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.meta_get("quiet-miss", true, None, Some(&["v"])), + ) + .await + .expect("quiet no-op response should not wait for an extra drain") + .unwrap(); + assert_eq!(result, None); + } + Self::SetStored => { + let result = client + .meta_set("quiet-set", "value", true, None, None) + .await + .unwrap(); + assert_eq!(result, None); + } + Self::SetError => { + let result = client + .meta_set("quiet-set-error", "value", true, None, None) + .await; + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Client( + "bad set".to_string() + )))) + ); + } + Self::DeleteDeleted => { + let result = client + .meta_delete("quiet-delete", true, None, None) + .await + .unwrap(); + assert_eq!(result, None); + } + Self::DeleteError => { + let result = client + .meta_delete("quiet-delete-error", true, None, None) + .await; + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Client( + "bad delete".to_string() + )))) + ); + } + Self::IncrementValue => { + let result = client + .meta_increment("quiet-increment", true, None, None, Some(&["v"])) + .await + .unwrap() + .unwrap(); + assert_eq!(result.data, Some(b"2".to_vec())); + } + Self::IncrementError => { + let result = client + .meta_increment( + "quiet-increment-error", + true, + None, + None, + Some(&["badflag"]), + ) + .await; + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Client( + "bad increment".to_string() + )))) + ); + } + Self::DecrementValue => { + let result = client + .meta_decrement("quiet-decrement", true, None, None, Some(&["v"])) + .await + .unwrap() + .unwrap(); + assert_eq!(result.data, Some(b"1".to_vec())); + } + Self::DecrementError => { + let result = client + .meta_decrement( + "quiet-decrement-error", + true, + None, + None, + Some(&["badflag"]), + ) + .await; + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Client( + "bad decrement".to_string() + )))) + ); + } + } + } +} + +#[derive(Clone, Copy)] +struct QuietMetaCase { + name: &'static str, + operation: QuietMetaOperation, + expected_command: &'static [u8], + response: &'static [u8], +} + +#[tokio::test] +async fn test_quiet_meta_operations_drain_internal_noop_before_next_response() { + let cases = [ + QuietMetaCase { + name: "meta_get error", + operation: QuietMetaOperation::GetError, + expected_command: b"mg bad-client-error badflag q\r\nmn\r\n", + response: b"CLIENT_ERROR bad flag\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_get data", + operation: QuietMetaOperation::GetData, + expected_command: b"mg quiet-hit v q\r\nmn\r\n", + response: b"VA 5\r\nvalue\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_get no-op", + operation: QuietMetaOperation::GetNoop, + expected_command: b"mg quiet-miss v q\r\nmn\r\n", + response: b"MN\r\n", + }, + QuietMetaCase { + name: "meta_set stored", + operation: QuietMetaOperation::SetStored, + expected_command: b"ms quiet-set 5 q\r\nvalue\r\nmn\r\n", + response: b"HD\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_set error", + operation: QuietMetaOperation::SetError, + expected_command: b"ms quiet-set-error 5 q\r\nvalue\r\nmn\r\n", + response: b"CLIENT_ERROR bad set\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_delete deleted", + operation: QuietMetaOperation::DeleteDeleted, + expected_command: b"md quiet-delete q\r\nmn\r\n", + response: b"HD\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_delete error", + operation: QuietMetaOperation::DeleteError, + expected_command: b"md quiet-delete-error q\r\nmn\r\n", + response: b"CLIENT_ERROR bad delete\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_increment value", + operation: QuietMetaOperation::IncrementValue, + expected_command: b"ma quiet-increment v q\r\nmn\r\n", + response: b"VA 1\r\n2\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_increment error", + operation: QuietMetaOperation::IncrementError, + expected_command: b"ma quiet-increment-error badflag q\r\nmn\r\n", + response: b"CLIENT_ERROR bad increment\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_decrement value", + operation: QuietMetaOperation::DecrementValue, + expected_command: b"ma quiet-decrement MD v q\r\nmn\r\n", + response: b"VA 1\r\n1\r\nMN\r\n", + }, + QuietMetaCase { + name: "meta_decrement error", + operation: QuietMetaOperation::DecrementError, + expected_command: b"ma quiet-decrement-error MD badflag q\r\nmn\r\n", + response: b"CLIENT_ERROR bad decrement\r\nMN\r\n", + }, + ]; + + for case in cases { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + + let first_command = read_until_contains(&mut socket, b"mn\r\n").await; + assert_eq!( + first_command.as_slice(), + case.expected_command, + "{} sent unexpected first command", + case.name + ); + socket.write_all(case.response).await.unwrap(); + + let second_command = read_until_contains(&mut socket, b"\r\n").await; + assert_eq!( + second_command.as_slice(), + b"mg next-key v\r\n", + "{} left the stream misaligned before the second command", + case.name + ); + socket.write_all(b"VA 4\r\nnext\r\n").await.unwrap(); + }); + + let mut client = Client::new(format!("tcp://{addr}")).await.unwrap(); + + case.operation.run(&mut client).await; + + let result = client + .meta_get("next-key", false, None, Some(&["v"])) + .await + .unwrap() + .unwrap(); + assert_eq!( + result.data, + Some(b"next".to_vec()), + "{} did not leave the connection reusable", + case.name + ); + + server.await.unwrap(); + } +} + +#[derive(Clone, Copy)] +struct UnexpectedQuietDrainCase { + name: &'static str, + expected_command: &'static [u8], + response: &'static [u8], + expected_error: &'static str, +} + +#[tokio::test] +async fn test_quiet_meta_unexpected_drain_response_reports_response_kind() { + let cases = [ + UnexpectedQuietDrainCase { + name: "status response", + expected_command: b"ms quiet-set 5 q\r\nvalue\r\nmn\r\n", + response: b"HD\r\nNS\r\n", + expected_error: "Expected quiet-mode no-op response, got status not stored", + }, + UnexpectedQuietDrainCase { + name: "data response", + expected_command: b"mg quiet-hit v q\r\nmn\r\n", + response: b"VA 5\r\nvalue\r\nVA 4\r\noops\r\n", + expected_error: "Expected quiet-mode no-op response, got data response", + }, + ]; + + for case in cases { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + + let command = read_until_contains(&mut socket, b"mn\r\n").await; + assert_eq!( + command.as_slice(), + case.expected_command, + "{} sent unexpected command", + case.name + ); + socket.write_all(case.response).await.unwrap(); + }); + + let mut client = Client::new(format!("tcp://{addr}")).await.unwrap(); + let result = match case.name { + "status response" => { + client + .meta_set("quiet-set", "value", true, None, None) + .await + } + "data response" => client.meta_get("quiet-hit", true, None, Some(&["v"])).await, + _ => unreachable!(), + }; + + assert_eq!( + result, + Err(Error::Protocol(Status::Error(ErrorKind::Protocol(Some( + case.expected_error.to_string() + ))))), + "{} reported an unexpected error", + case.name + ); + + server.await.unwrap(); + } +} + #[ignore = "Relies on a running memcached server"] #[tokio::test] #[parallel]