Bound request snapshots in the slow requests log (#9726)

* Bound request snapshots in the slow requests log Entries in the slow requests log retain a full serde_json::Value snapshot of the (internal) shard request indefinitely. The distance matrix API internally generates a batch of `sample` queries, each carrying a has_id filter with all `sample` sampled ids, so a single log entry ballooned to sample^2 ids expanded into a JSON tree (~90 bytes/id): ~90MB per entry for sample=1000, ~2GB for sample=5000. Random sample ids give every request a fresh content hash, so each call added a new entry until the 32-slot queue filled — OOM long before that for larger samples. Truncate all arrays in logged request bodies to 64 elements plus an omission marker. Query batches are serialized per element up to the cap, so the full untruncated JSON tree is never materialized even transiently. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Fix rustfmt Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-09-21 13:37:46 -05:00 · 2026-08-04 11:16:59 +02:00
parent 508b9ed38a
commit 7cc755c225
1 changed files with 123 additions and 6 deletions
@@ -10,6 +10,13 @@ use shard::scroll::ScrollRequestInternal;
 use crate::operations::types::PointRequestInternal;
 use crate::operations::universal_query::shard_query::ShardQueryRequest;

+/// Log entries are retained in memory indefinitely, so a snapshot must not keep
+/// unbounded request data alive. E.g. internally generated distance matrix
+/// queries carry `has_id` filters with sample² point ids in total, which blows
+/// up multi-fold when expanded into a JSON tree. Arrays longer than this are
+/// cut down to this many elements plus an omission marker.
+const MAX_LOGGED_ARRAY_LEN: usize = 64;
+
 pub trait Loggable {
    fn to_log_value(&self) -> serde_json::Value;

@@ -19,9 +26,41 @@ pub trait Loggable {
    fn request_hash(&self) -> u64;
 }

+/// Serialize a request for the slow-requests log, bounding all arrays.
+fn bounded_log_value<T: serde::Serialize>(request: &T) -> Value {
+    let mut value = serde_json::to_value(request).unwrap_or_default();
+    truncate_long_arrays(&mut value);
+    value
+}
+
+fn omission_marker(omitted: usize) -> Value {
+    Value::String(format!("... ({omitted} more items)"))
+}
+
+fn truncate_long_arrays(value: &mut Value) {
+    match value {
+        Value::Array(items) => {
+            if items.len() > MAX_LOGGED_ARRAY_LEN {
+                let omitted = items.len() - MAX_LOGGED_ARRAY_LEN;
+                items.truncate(MAX_LOGGED_ARRAY_LEN);
+                items.push(omission_marker(omitted));
+            }
+            for item in items.iter_mut() {
+                truncate_long_arrays(item);
+            }
+        }
+        Value::Object(map) => {
+            for (_key, item) in map.iter_mut() {
+                truncate_long_arrays(item);
+            }
+        }
+        Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {}
+    }
+}
+
 impl Loggable for CollectionUpdateOperations {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        bounded_log_value(self)
    }

    fn request_name(&self) -> &'static str {
@@ -38,7 +77,19 @@ impl Loggable for CollectionUpdateOperations {

 impl Loggable for Vec<ShardQueryRequest> {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        // Serialize requests one by one, only up to the cap: serializing the
+        // whole batch at once would materialize the full JSON tree (gigabytes
+        // for internally generated matrix batches) before truncation could
+        // trim it.
+        let mut items: Vec<Value> = self
+            .iter()
+            .take(MAX_LOGGED_ARRAY_LEN)
+            .map(bounded_log_value)
+            .collect();
+        if self.len() > MAX_LOGGED_ARRAY_LEN {
+            items.push(omission_marker(self.len() - MAX_LOGGED_ARRAY_LEN));
+        }
+        Value::Array(items)
    }

    fn request_name(&self) -> &'static str {
@@ -55,7 +106,7 @@ impl Loggable for Vec<ShardQueryRequest> {

 impl Loggable for ScrollRequestInternal {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        bounded_log_value(self)
    }

    fn request_name(&self) -> &'static str {
@@ -86,7 +137,7 @@ impl<T: Loggable> Loggable for Arc<T> {

 impl Loggable for FacetParams {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        bounded_log_value(self)
    }

    fn request_name(&self) -> &'static str {
@@ -103,7 +154,7 @@ impl Loggable for FacetParams {

 impl Loggable for CountRequestInternal {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        bounded_log_value(self)
    }

    fn request_name(&self) -> &'static str {
@@ -120,7 +171,7 @@ impl Loggable for CountRequestInternal {

 impl Loggable for PointRequestInternal {
    fn to_log_value(&self) -> Value {
-        serde_json::to_value(self).unwrap_or_default()
+        bounded_log_value(self)
    }

    fn request_name(&self) -> &'static str {
@@ -134,3 +185,69 @@ impl Loggable for PointRequestInternal {
        hasher.finish()
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use segment::types::{Condition, Filter, HasIdCondition, WithPayloadInterface, WithVector};
+    use serde_json::json;
+
+    use super::*;
+
+    #[test]
+    fn test_truncate_long_arrays() {
+        let mut value = json!({
+            "short": [1, 2, 3],
+            "long": (0..MAX_LOGGED_ARRAY_LEN + 100).collect::<Vec<_>>(),
+            "nested": {
+                "long": [(0..MAX_LOGGED_ARRAY_LEN + 1).collect::<Vec<_>>()],
+            },
+        });
+
+        truncate_long_arrays(&mut value);
+
+        assert_eq!(value["short"], json!([1, 2, 3]));
+
+        let long = value["long"].as_array().unwrap();
+        assert_eq!(long.len(), MAX_LOGGED_ARRAY_LEN + 1);
+        assert_eq!(long[MAX_LOGGED_ARRAY_LEN], json!("... (100 more items)"));
+
+        let nested = value["nested"]["long"][0].as_array().unwrap();
+        assert_eq!(nested.len(), MAX_LOGGED_ARRAY_LEN + 1);
+        assert_eq!(nested[MAX_LOGGED_ARRAY_LEN], json!("... (1 more items)"));
+    }
+
+    #[test]
+    fn test_query_batch_log_value_is_bounded() {
+        // Mimics an internally generated distance matrix batch: many queries,
+        // each carrying a has_id filter with all sampled point ids.
+        let ids: ahash::AHashSet<_> = (0..MAX_LOGGED_ARRAY_LEN as u64 + 36)
+            .map(segment::types::PointIdType::from)
+            .collect();
+        let filter = Filter::new_must(Condition::HasId(HasIdCondition::from(ids)));
+
+        let request = ShardQueryRequest {
+            prefetches: vec![],
+            query: None,
+            filter: Some(filter),
+            score_threshold: None,
+            limit: 10,
+            offset: 0,
+            params: None,
+            with_vector: WithVector::Bool(false),
+            with_payload: WithPayloadInterface::Bool(false),
+        };
+        let batch = vec![request; MAX_LOGGED_ARRAY_LEN + 100];
+
+        let value = batch.to_log_value();
+
+        let queries = value.as_array().unwrap();
+        assert_eq!(queries.len(), MAX_LOGGED_ARRAY_LEN + 1);
+        assert_eq!(queries[MAX_LOGGED_ARRAY_LEN], json!("... (100 more items)"));
+
+        let has_id = queries[0]["filter"]["must"][0]["has_id"]
+            .as_array()
+            .unwrap();
+        assert_eq!(has_id.len(), MAX_LOGGED_ARRAY_LEN + 1);
+        assert_eq!(has_id[MAX_LOGGED_ARRAY_LEN], json!("... (36 more items)"));
+    }
+}