Reject malformed record queries instead of silently ignoring them

server/src/axum_util/query.rs

@@ -3,13 +3,15 @@ use axum::http;
 use axum::response;
 use serde::de;
 
+use crate::http::ErrorReply;
+
 #[derive(Debug)]
 pub struct Query<T>(pub T);
 
 #[derive(Debug)]
 #[non_exhaustive]
 pub enum QueryRejection {
-    FailedToDeserializeQueryString,
+    FailedToDeserializeQueryString(String),
 }
 
 #[axum::async_trait]
@@ -24,20 +26,20 @@ where
         parts: &mut http::request::Parts,
         _state: &S,
     ) -> Result<Self, Self::Rejection> {
-        let query = parts
-            .uri
-            .query()
-            .ok_or(QueryRejection::FailedToDeserializeQueryString)?;
+        // A request without a query string carries the same information as one
+        // with an empty query string: every parameter takes its default.
+        let query = parts.uri.query().unwrap_or_default();
         let config = serde_qs::Config::new(2, false);
         config
             .deserialize_str(query)
             .map(Query)
-            .map_err(|_| QueryRejection::FailedToDeserializeQueryString)
+            .map_err(|e| QueryRejection::FailedToDeserializeQueryString(e.to_string()))
     }
 }
 
 impl response::IntoResponse for QueryRejection {
     fn into_response(self) -> response::Response {
-        http::StatusCode::NOT_ACCEPTABLE.into_response()
+        let Self::FailedToDeserializeQueryString(detail) = self;
+        ErrorReply::InvalidQueryString(detail).into_response()
     }
 }

server/src/http.rs

@@ -311,12 +311,20 @@ async fn topic_get_info(
 async fn topic_iterate_route(
     State(AppState(catalog, config)): State<AppState>,
     Path(topic_name): Path<String>,
-    query: Option<Query<TopicIterationQuery>>,
+    query: Query<TopicIterationQuery>,
     headers: HeaderMap,
     position: Option<Json<TopicIterator>>,
 ) -> Result<axum::response::Response, ErrorReply> {
     let max_page = config.http.max_page;
-    topic_iterate(topic_name, query, headers, position, catalog, max_page).await
+    topic_iterate(
+        topic_name,
+        Some(query),
+        headers,
+        position,
+        catalog,
+        max_page,
+    )
+    .await
 }
 
 pub async fn topic_iterate(

server/src/http/error.rs

@@ -11,6 +11,7 @@ pub enum ErrorReply {
     EmptyBody,
     WriterBusy,
     InvalidQuery,
+    InvalidQueryString(String),
     InvalidSchema,
     NullTypes,
     BadEncoding,
@@ -30,6 +31,9 @@ impl axum::response::IntoResponse for ErrorReply {
             Self::Chunk(e) => (StatusCode::BAD_REQUEST, format!("chunk error: {e}")),
             Self::Path(e) => (StatusCode::BAD_REQUEST, format!("invalid path: {e}")),
             Self::InvalidQuery => (StatusCode::BAD_REQUEST, "invalid query".to_string()),
+            Self::InvalidQueryString(detail) => {
+                (StatusCode::BAD_REQUEST, format!("invalid query: {detail}"))
+            }
             Self::InvalidSchema => (StatusCode::BAD_REQUEST, "invalid schema".to_string()),
             Self::NullTypes => (
                 StatusCode::BAD_REQUEST,

server/tests/server.rs

@@ -771,6 +771,130 @@ async fn topic_time_query() -> Result<()> {
     Ok(())
 }
 
+/// Fill two partitions of `topic_name` with `count` records each.
+async fn append_two_partitions(
+    client: &Client,
+    server: &TestServer,
+    topic_name: &str,
+    count: usize,
+) {
+    for partition in ["alpha", "beta"] {
+        repeat_append(
+            client,
+            append_url(server, topic_name, partition).as_str(),
+            TEST_MESSAGE,
+            count,
+        )
+        .await;
+    }
+    server.catalog.checkpoint().await;
+}
+
+async fn assert_error_reply(response: Response, expected: reqwest::StatusCode) {
+    assert_eq!(response.status(), expected);
+    let body: json::Value = response.json().await.unwrap();
+    let body = body.as_object().expect("expected an error object");
+    assert_eq!(
+        body.get("code").and_then(json::Value::as_u64),
+        Some(expected.as_u16() as u64)
+    );
+    assert!(body.contains_key("message"), "expected a message: {body:?}");
+}
+
+#[test_log::test(tokio::test)]
+async fn topic_iterate_partition_filter() -> Result<()> {
+    let (client, topic_name, server) = setup().await;
+    append_two_partitions(&client, &server, &topic_name, 5).await;
+    let url = topic_records_url(&server, &topic_name);
+
+    // no query string at all: every partition is iterated
+    let response = client.post(&url).json(&json::json!({})).send().await?;
+    assert_response_length(response.error_for_status()?, 10).await;
+
+    // an exact partition name, and the equivalent regex, select one partition
+    for filter in ["alpha", "regex:^alpha$"] {
+        let response = client
+            .post(&url)
+            .query(&[("partition_filter[]", filter)])
+            .json(&json::json!({}))
+            .send()
+            .await?;
+        assert_response_length(response.error_for_status()?, 5).await;
+    }
+
+    Ok(())
+}
+
+#[test_log::test(tokio::test)]
+async fn topic_iterate_rejects_malformed_query() -> Result<()> {
+    let (client, topic_name, server) = setup().await;
+    append_two_partitions(&client, &server, &topic_name, 5).await;
+    let url = topic_records_url(&server, &topic_name);
+
+    // `descending` is not a valid order. Rejecting the request keeps the
+    // partition filter that came with it from being silently discarded, which
+    // would widen the query to every partition.
+    let response = client
+        .post(&url)
+        .query(&[("partition_filter[]", "alpha"), ("order", "descending")])
+        .json(&json::json!({}))
+        .send()
+        .await?;
+    assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await;
+
+    // the same request with a valid order still reads just that partition
+    let response = client
+        .post(&url)
+        .query(&[("partition_filter[]", "alpha"), ("order", "desc")])
+        .json(&json::json!({}))
+        .send()
+        .await?;
+    assert_response_length(response.error_for_status()?, 5).await;
+
+    Ok(())
+}
+
+#[test_log::test(tokio::test)]
+async fn topic_iterate_rejects_uncompilable_partition_regex() -> Result<()> {
+    let (client, topic_name, server) = setup().await;
+    append_two_partitions(&client, &server, &topic_name, 5).await;
+    let url = topic_records_url(&server, &topic_name);
+
+    // look-around is not supported by the regex crate, so this selector cannot
+    // compile: report it rather than matching it as a literal partition name,
+    // which reads as an empty topic.
+    for filter in ["regex:(?=alpha)", "regex:alpha("] {
+        let response = client
+            .post(&url)
+            .query(&[("partition_filter[]", filter)])
+            .json(&json::json!({}))
+            .send()
+            .await?;
+        assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await;
+    }
+
+    Ok(())
+}
+
+#[test_log::test(tokio::test)]
+async fn partition_records_rejects_malformed_query() -> Result<()> {
+    let (client, topic_name, server) = setup().await;
+    append_two_partitions(&client, &server, &topic_name, 5).await;
+    let url = partition_records_url(&server, &topic_name, "alpha");
+
+    let response = client
+        .get(&url)
+        .query(&[("start", "0"), ("page_size", "many")])
+        .send()
+        .await?;
+    assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await;
+
+    let response = client.get(&url).query(&[("start", "0")]).send().await?;
+    assert_response_length(response.error_for_status()?, 5).await;
+
+    Ok(())
+}
+
 #[test_log::test(tokio::test)]
 async fn topic_iterate_pandas_records() -> Result<()> {
     let (client, topic_name, server) = setup().await;

transport/src/lib.rs

@@ -336,15 +336,20 @@ pub struct TopicIterationQuery {
 /// begin with "regex:" to signify that any partition matching the following string can be converted.
 pub type PartitionFilter = Option<Vec<PartitionSelector>>;
 
-#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, utoipa::ToSchema)]
-#[serde(from = "String", into = "String")]
+#[derive(Clone, Debug, serde::Serialize, utoipa::ToSchema)]
+#[serde(into = "String")]
 pub enum PartitionSelector {
     /// The exact name of a partition.
     String(String),
     /// A string beginning with `regex:`. The suffix will be matched against partition names.
     Regex(Regex),
 }
 
+/// Returned when a `regex:` prefixed [`PartitionSelector`] does not compile.
+#[derive(Clone, Debug, Error)]
+#[error("invalid partition selector: {0}")]
+pub struct PartitionSelectorError(String);
+
 #[cfg(feature = "rweb")]
 impl Entity for PartitionSelector {
     fn type_name() -> std::borrow::Cow<'static, str> {
@@ -377,25 +382,46 @@ impl From<PartitionSelector> for String {
     }
 }
 
+/// Note that this conversion cannot fail: a `regex:` selector that does not
+/// compile falls back to the exact partition name. Use
+/// [`PartitionSelector::parse`] to reject such a selector instead.
 impl<T> From<T> for PartitionSelector
 where
     T: AsRef<str>,
 {
     fn from(text: T) -> Self {
-        let build_regex = |pattern| {
-            regex::RegexBuilder::new(pattern)
-                .size_limit(2 << 12)
-                .build()
-                .ok()
-        };
-        text.as_ref()
-            .strip_prefix("regex:")
-            .and_then(build_regex)
-            .map_or_else(|| Self::String(text.as_ref().to_string()), Self::Regex)
+        let text = text.as_ref();
+        Self::parse(text).unwrap_or_else(|_| Self::String(text.to_string()))
+    }
+}
+
+impl<'de> Deserialize<'de> for PartitionSelector {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let text = String::deserialize(deserializer)?;
+        Self::parse(&text).map_err(serde::de::Error::custom)
     }
 }
 
 impl PartitionSelector {
+    /// Compiled size limit for a `regex:` selector.
+    const REGEX_SIZE_LIMIT: usize = 2 << 12;
+
+    /// Build a selector from its string form, reporting a `regex:` pattern that
+    /// does not compile rather than silently treating it as a partition name.
+    pub fn parse(text: &str) -> Result<Self, PartitionSelectorError> {
+        match text.strip_prefix("regex:") {
+            Some(pattern) => regex::RegexBuilder::new(pattern)
+                .size_limit(Self::REGEX_SIZE_LIMIT)
+                .build()
+                .map(Self::Regex)
+                .map_err(|e| PartitionSelectorError(e.to_string())),
+            None => Ok(Self::String(text.to_string())),
+        }
+    }
+
     pub fn matches(&self, name: &str) -> bool {
         match self {
             Self::String(s) => s == name,
@@ -1078,6 +1104,8 @@ impl fmt::Display for PartitionId {
 mod tests {
     use super::*;
     use arrow_array::{Int32Array, Int64Array};
+    use serde::de::value::{Error as ValueError, StrDeserializer};
+    use serde::de::IntoDeserializer;
 
     fn nested_chunk() -> (
         SchemaChunk<SchemaRef>,
@@ -1252,6 +1280,42 @@ mod tests {
         assert_eq!(ps.to_string(), r"regex:\p{Greek}");
     }
 
+    #[test]
+    fn partition_selector_reports_uncompilable_regex() {
+        // look-around is unsupported, and this pattern is far past the size limit
+        for pattern in [
+            "regex:(?=alpha)",
+            "regex:alpha(",
+            &format!("regex:{}", "a{100}".repeat(100)),
+        ] {
+            assert!(
+                PartitionSelector::parse(pattern).is_err(),
+                "expected {pattern} to be rejected"
+            );
+
+            let deserializer: StrDeserializer<'_, ValueError> = pattern.into_deserializer();
+            assert!(
+                PartitionSelector::deserialize(deserializer).is_err(),
+                "expected {pattern} to fail deserialization"
+            );
+
+            // the infallible conversion keeps its documented fallback
+            assert!(matches!(
+                PartitionSelector::from(pattern),
+                PartitionSelector::String(_)
+            ));
+        }
+    }
+
+    #[test]
+    fn partition_selector_deserializes_valid_selectors() {
+        for (pattern, name) in [("alpha", "alpha"), ("regex:^alpha$", "alpha")] {
+            let deserializer: StrDeserializer<'_, ValueError> = pattern.into_deserializer();
+            let selector = PartitionSelector::deserialize(deserializer).unwrap();
+            assert!(selector.matches(name));
+        }
+    }
+
     #[test]
     fn test_focus() {
         let (test, (time, _child_struct, _grandchild_struct, _index)) = nested_chunk();