From b5cc274a4c3e96e4123847b05d70f06c62e09f68 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Tue, 5 May 2026 15:24:59 -0400 Subject: [PATCH] fix(cold-store): tolerate torn trailing line in parse_jsonl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under concurrent producer write + reader scan (the e2e harness shape), the cold-fallback reader saw a part-file whose final line was a partial JSON record and errored the whole scan, manifesting as 10 s cold-fallback timeouts on `sum(http_requests_total)` in E0 (PR #261, 34/300 replay queries). Tolerate the torn last line specifically: if a parse error hits the *final* line *and* the blob has no terminating newline (the producer-mid-flush shape per design-sketch-db.md §5.2), warn + drop instead of failing the part. Mid-file parse errors still hard-error — that signals real corruption, not a concurrent write. Threads `Option<&Path>` through a new `parse_jsonl_at` helper so the warn log includes the offending file + a 200-char line preview; existing `parse_jsonl(bytes, start, end)` callers (`local_fs.rs`) unchanged. Pins both shapes in tests: - parse_jsonl_ignores_torn_trailing_line - parse_jsonl_errors_on_mid_file_corruption Unblocks PROGRESS.md "Outstanding follow-ups #5" and the E0 exit-criterion (1) cold-routed query family. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../query/fallback/cold_store/format.rs | 76 ++++++++++++++++--- 1 file changed, 67 insertions(+), 9 deletions(-) diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/format.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/format.rs index 0359ab5c..8258db43 100644 --- a/asap-query-engine/src/drivers/query/fallback/cold_store/format.rs +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/format.rs @@ -13,6 +13,8 @@ use chrono::{DateTime, Datelike, Timelike, Utc}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::path::Path; +use tracing::warn; use super::ColdStoreError; @@ -67,26 +69,47 @@ pub fn hour_prefixes(metric: &str, start_ms: i64, end_ms: i64) -> Vec { } /// Parse a `.jsonl` blob into `RawSample`s, filtering to the -/// half-open range `[start_ms, end_ms)`. Malformed lines fail the -/// whole parse — partial results from a corrupted part are worse -/// than an error a caller can route around. +/// half-open range `[start_ms, end_ms)`. Mid-file malformed lines +/// are hard errors. A malformed *trailing* line is tolerated +/// (warn + drop) only when the blob has no terminating newline — +/// that's the producer-mid-flush shape, see `docs/design-sketch-db.md` +/// §5.2. `path` is threaded through purely for the warn log. pub fn parse_jsonl( bytes: &[u8], start_ms: i64, end_ms: i64, +) -> Result, ColdStoreError> { + parse_jsonl_at(bytes, start_ms, end_ms, None) +} + +pub fn parse_jsonl_at( + bytes: &[u8], + start_ms: i64, + end_ms: i64, + path: Option<&Path>, ) -> Result, ColdStoreError> { let text = std::str::from_utf8(bytes) .map_err(|e| ColdStoreError::Malformed(format!("non-utf8: {e}")))?; + let trailing_torn = !text.is_empty() && !text.ends_with('\n'); + let lines: Vec<&str> = text.lines().collect(); + let last = lines.len().saturating_sub(1); let mut out = Vec::new(); - for (lineno, line) in text.lines().enumerate() { - let line = line.trim(); + for (i, raw) in lines.iter().enumerate() { + let line = raw.trim(); if line.is_empty() { continue; } - let sample: RawSample = serde_json::from_str(line) - .map_err(|e| ColdStoreError::Malformed(format!("line {}: {}", lineno + 1, e)))?; - if sample.ts_ms >= start_ms && sample.ts_ms < end_ms { - out.push(sample); + match serde_json::from_str::(line) { + Ok(s) if s.ts_ms >= start_ms && s.ts_ms < end_ms => out.push(s), + Ok(_) => {} + Err(e) if i == last && trailing_torn => warn!( + path = %path.map(|p| p.display().to_string()).unwrap_or_default(), + line = %line.chars().take(200).collect::(), + error = %e, + "cold-store: dropping torn trailing line in JSONL part \ + (no terminating newline; likely producer mid-flush)" + ), + Err(e) => return Err(ColdStoreError::Malformed(format!("line {}: {}", i + 1, e))), } } Ok(out) @@ -147,4 +170,39 @@ mod tests { let blob = "not-json\n"; assert!(parse_jsonl(blob.as_bytes(), 0, i64::MAX).is_err()); } + + /// Pins follow-up #5: producer is mid-flush, reader sees a + /// part file whose last line is partial (no terminating + /// newline). We must surface the N preceding records, not + /// fail the whole scan. + #[test] + fn parse_jsonl_ignores_torn_trailing_line() { + let blob = "{\"ts_ms\":100,\"labels\":{\"a\":\"1\"},\"value\":1.0}\n\ + {\"ts_ms\":200,\"labels\":{\"a\":\"2\"},\"value\":2.0}\n\ + {\"ts_ms\":300,\"labels\":{\"a\":\"3\"},\"value\":3.0}\n\ + {\"ts_ms\":400,\"labels\":{\"a\":\"4\"},\"valu"; + let out = parse_jsonl(blob.as_bytes(), 0, i64::MAX).expect("torn tail must not error"); + assert_eq!(out.len(), 3); + assert_eq!(out[0].ts_ms, 100); + assert_eq!(out[2].ts_ms, 300); + } + + /// A malformed line in the *middle* of the file is real + /// corruption, not a concurrent-write torn tail — must still + /// hard-error so callers can route around the bad part. + #[test] + fn parse_jsonl_errors_on_mid_file_corruption() { + let blob = "{\"ts_ms\":100,\"labels\":{\"a\":\"1\"},\"value\":1.0}\n\ + not-json\n\ + {\"ts_ms\":300,\"labels\":{\"a\":\"3\"},\"value\":3.0}\n"; + let err = parse_jsonl(blob.as_bytes(), 0, i64::MAX) + .expect_err("mid-file corruption must surface as an error"); + match err { + ColdStoreError::Malformed(msg) => assert!( + msg.contains("line 2"), + "expected lineno in error, got: {msg}" + ), + other => panic!("expected Malformed, got {other:?}"), + } + } }