From c7295cad1dfe1c6bd07850f5cf2c93a1fe7c6ab4 Mon Sep 17 00:00:00 2001 From: jichaowang02-lang Date: Mon, 22 Jun 2026 21:53:41 +0100 Subject: [PATCH] fix(ccr): store opaque blobs from lossless:table compaction (#1083) (#1182) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description SmartCrusher's `lossless:table` compaction path emits opaque-blob CCR markers (`<>`) but never wrote the original payload to the CCR store. As a result `GET /v1/retrieve/{hash}` and the `headroom_retrieve` tool return **404** for those hashes. The opaque-*string* path (`walker::emit_opaque_ccr_marker`) already stores its payload; the table compactor diverged simply because no store was threaded into it. Closes #1083 ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `compaction/compactor.rs`: add `compact_with_store(items, cfg, store)` and a private `compact_inner`; thread `Option<&Arc>` through `build_homogeneous_table` → `build_row` → `cell_from_value` and the recursive bucket/nested calls. In the `Opaque` branch, `store.put(&hash, payload)` under the **same** `hash_opaque` value that becomes the marker hash (mirrors `walker::emit_opaque_ccr_marker`). Public `compact` is unchanged — it delegates with `None`. - `compaction/mod.rs`: add `CompactionStage::run_with_store`; `run` is unchanged. - `crusher.rs`: the lossless branch now calls `stage.run_with_store(items, self.ccr_store.as_ref())` instead of `stage.run(items)`. - Two new unit tests in `compactor.rs` (see below). The IR (and therefore the rendered marker text) is identical whether or not a store is supplied — the store only gains the write that should already have happened, so existing output stays byte-for-byte the same. ## Testing - [ ] Unit tests pass (`pytest`) - [ ] Linting passes (`ruff check .`) - [ ] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed > Note: this change is in the Rust core (`crates/headroom-core`), so the > Python-specific checks above are N/A. The Rust equivalents were run: ### Test Output ```text $ cargo test -p headroom-core --lib compaction test result: ok. 70 passed; 0 failed; 0 ignored; 0 measured; 766 filtered out; finished in 0.01s $ cargo fmt -p headroom-core -- --check # clean (exit 0) ``` New tests: - `opaque_payload_is_stored_under_marker_hash` — after `compact_with_store`, the original blob is retrievable via `store.get(marker_hash)`, and the stored key equals `hash_opaque(payload)` (locks the key↔marker contract). - `store_presence_does_not_change_the_ir` — `compact` and `compact_with_store` produce identical IR; only the store write is added. (The full `cargo test -p headroom-core --lib` run has 18 pre-existing failures, all in `transforms::magika_detector` — they require the ONNX runtime/model and are unrelated to this change. All 70 compaction + crusher tests pass.) ## Real Behavior Proof - Environment: Windows, Rust 1.95.0, `cargo test -p headroom-core` (no live proxy). - Exact command / steps: build a 2-item array with a long opaque-blob field → `compact_with_store(&items, &cfg, Some(&InMemoryCcrStore))` → read the `OpaqueRef.ccr_hash` from the IR → `store.get(ccr_hash)`. - Observed result: before the fix the store is empty (retrieval would 404); after the fix `store.get(ccr_hash) == Some(original_payload)` and the marker hash is unchanged. - Not tested: end-to-end through a running proxy / a real `GET /v1/retrieve/{hash}` HTTP round-trip. Verified at the unit level that the store now receives the payload under the exact marker hash, which is the write that was missing. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - Docs/CHANGELOG checklist items are N/A — this is an internal correctness fix with no user-facing API change. - Scope is intentionally minimal: public `compact`/`run` signatures are preserved (delegating with `None`), so all existing callers and the 68 in-crate compaction tests are unaffected. Only the lossless `crush_array` branch opts into the store-threading via `run_with_store`. --- .../smart_crusher/compaction/compactor.rs | 145 ++++++++++++++++-- .../smart_crusher/compaction/mod.rs | 20 ++- .../src/transforms/smart_crusher/crusher.rs | 6 +- 3 files changed, 155 insertions(+), 16 deletions(-) diff --git a/crates/headroom-core/src/transforms/smart_crusher/compaction/compactor.rs b/crates/headroom-core/src/transforms/smart_crusher/compaction/compactor.rs index 9c21e0a93..496188175 100644 --- a/crates/headroom-core/src/transforms/smart_crusher/compaction/compactor.rs +++ b/crates/headroom-core/src/transforms/smart_crusher/compaction/compactor.rs @@ -40,12 +40,14 @@ //! [`CellClass::Opaque`]: super::classifier::CellClass::Opaque use std::collections::BTreeMap; +use std::sync::Arc; use serde_json::Value; use sha2::{Digest, Sha256}; use super::classifier::{classify_cell, CellClass, ClassifyConfig}; use super::ir::{Bucket, CellValue, Compaction, FieldSpec, Row, Schema}; +use crate::ccr::CcrStore; /// Config for the compactor. #[derive(Debug, Clone)] @@ -95,7 +97,37 @@ impl Default for CompactConfig { } /// Top-level compaction entry point. +/// +/// Opaque blobs become CCR pointers, but the original payload is **not** +/// stored — callers that need `<>` markers to resolve on +/// retrieval must use [`compact_with_store`] instead. pub fn compact(items: &[Value], cfg: &CompactConfig) -> Compaction { + compact_inner(items, cfg, None) +} + +/// Like [`compact`], but stash every opaque-blob payload in `store` under +/// the same 12-char hash that ends up in its `<>` marker, so +/// the runtime can serve the original back on a `headroom_retrieve` call / +/// `GET /v1/retrieve/{hash}`. Mirrors the contract already honored by the +/// opaque-string path in [`super::walker::emit_opaque_ccr_marker`]. +/// +/// The IR (and therefore the rendered marker text) is identical whether or +/// not a store is supplied — the store only adds a side-effecting write, so +/// `compact(items, cfg)` and `compact_with_store(items, cfg, Some(store))` +/// return the same [`Compaction`]. +pub fn compact_with_store( + items: &[Value], + cfg: &CompactConfig, + store: Option<&Arc>, +) -> Compaction { + compact_inner(items, cfg, store) +} + +fn compact_inner( + items: &[Value], + cfg: &CompactConfig, + store: Option<&Arc>, +) -> Compaction { if items.len() < cfg.min_items { return Compaction::Untouched(Value::Array(items.to_vec())); } @@ -117,14 +149,14 @@ pub fn compact(items: &[Value], cfg: &CompactConfig) -> Compaction { if core_ratio < cfg.heterogeneous_core_ratio { if let Some(disc) = detect_discriminator(items, &key_freqs, cfg) { - return bucket_by(items, &disc, cfg); + return bucket_by(items, &disc, cfg, store); } // No clean discriminator — fall through to a sparse Table // rather than refusing. A sparse table is still better than // letting the lossy path drop fields wholesale. } - build_homogeneous_table(items, &key_freqs, cfg) + build_homogeneous_table(items, &key_freqs, cfg, store) } fn compute_key_freqs(items: &[Value]) -> BTreeMap { @@ -143,6 +175,7 @@ fn build_homogeneous_table( items: &[Value], key_freqs: &BTreeMap, cfg: &CompactConfig, + store: Option<&Arc>, ) -> Compaction { // Order: descending frequency, then alphabetical for stability. let mut keys: Vec<(&String, &usize)> = key_freqs.iter().collect(); @@ -165,7 +198,7 @@ fn build_homogeneous_table( let mut rows: Vec = items .iter() - .map(|item| build_row(item, &ordered_keys, cfg)) + .map(|item| build_row(item, &ordered_keys, cfg, store)) .collect(); flatten_uniform_nested(&mut field_specs, &mut rows, cfg); @@ -179,7 +212,12 @@ fn build_homogeneous_table( } } -fn build_row(item: &Value, ordered_keys: &[String], cfg: &CompactConfig) -> Row { +fn build_row( + item: &Value, + ordered_keys: &[String], + cfg: &CompactConfig, + store: Option<&Arc>, +) -> Row { let obj = match item.as_object() { Some(o) => o, None => return Row::new(vec![]), @@ -188,13 +226,13 @@ fn build_row(item: &Value, ordered_keys: &[String], cfg: &CompactConfig) -> Row .iter() .map(|k| match obj.get(k) { None => CellValue::Missing, - Some(v) => cell_from_value(v, cfg), + Some(v) => cell_from_value(v, cfg, store), }) .collect(); Row::new(cells) } -fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue { +fn cell_from_value(v: &Value, cfg: &CompactConfig, store: Option<&Arc>) -> CellValue { match classify_cell(v, &cfg.classify) { CellClass::Scalar => CellValue::Scalar(v.clone()), CellClass::JsonObject => CellValue::Scalar(v.clone()), // flatten pass may promote @@ -202,7 +240,7 @@ fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue { // Recurse if the inner array is array-of-objects; else scalar. if let Value::Array(items) = v { if items.iter().all(|i| matches!(i, Value::Object(_))) && items.len() >= 2 { - return CellValue::Nested(Box::new(compact(items, cfg))); + return CellValue::Nested(Box::new(compact_inner(items, cfg, store))); } } CellValue::Scalar(v.clone()) @@ -212,19 +250,28 @@ fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue { // store the parsed value as a Scalar (un-escapes for free). if let Value::Array(items) = &parsed { if items.iter().all(|i| matches!(i, Value::Object(_))) && items.len() >= 2 { - return CellValue::Nested(Box::new(compact(items, cfg))); + return CellValue::Nested(Box::new(compact_inner(items, cfg, store))); } } CellValue::Scalar(parsed) } CellClass::Opaque(kind) => { - let bytes = match v { - Value::String(s) => s.as_bytes(), + let s = match v { + Value::String(s) => s, _ => return CellValue::Scalar(v.clone()), }; + let ccr_hash = hash_opaque(s.as_bytes()); + // Stash the original so `GET /v1/retrieve/{hash}` and the + // `headroom_retrieve` tool can serve it back — mirrors + // `walker::emit_opaque_ccr_marker`. Without this write the + // marker points at a key that was never stored and retrieval + // 404s (issue #1083). + if let Some(store) = store { + store.put(&ccr_hash, s); + } CellValue::OpaqueRef { - ccr_hash: hash_opaque(bytes), - byte_size: bytes.len(), + ccr_hash, + byte_size: s.len(), kind, } } @@ -451,7 +498,12 @@ fn detect_discriminator( best.map(|(k, _)| k) } -fn bucket_by(items: &[Value], discriminator: &str, cfg: &CompactConfig) -> Compaction { +fn bucket_by( + items: &[Value], + discriminator: &str, + cfg: &CompactConfig, + store: Option<&Arc>, +) -> Compaction { let mut groups: BTreeMap> = BTreeMap::new(); for item in items { let key = item @@ -465,7 +517,7 @@ fn bucket_by(items: &[Value], discriminator: &str, cfg: &CompactConfig) -> Compa let buckets: Vec = groups .into_iter() .map(|(key, group_items)| { - let inner = compact(&group_items, cfg); + let inner = compact_inner(&group_items, cfg, store); match inner { Compaction::Table { schema, rows, .. } => Bucket { key: Value::String(key), @@ -730,4 +782,69 @@ mod tests { assert_ne!(h1, h3); assert_eq!(h1.len(), 12); } + + #[test] + fn opaque_payload_is_stored_under_marker_hash() { + use crate::ccr::{CcrStore, InMemoryCcrStore}; + use std::sync::Arc; + + // Same blob the `opaque_cell_becomes_ccr_ref` test uses — known to + // classify as Opaque, so the OpaqueRef / `<>` path runs. + let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8); + let items = vec![ + json!({"id": 1, "blob": big.clone()}), + json!({"id": 2, "blob": big.clone()}), + ]; + + let store: Arc = Arc::new(InMemoryCcrStore::new()); + let c = compact_with_store(&items, &cfg(), Some(&store)); + + // Pull the hash the rendered marker will carry out of the IR. + let hash = match &c { + Compaction::Table { rows, schema, .. } => { + let blob_idx = schema + .fields + .iter() + .position(|f| f.name == "blob") + .expect("blob col"); + match &rows[0].0[blob_idx] { + CellValue::OpaqueRef { ccr_hash, .. } => ccr_hash.clone(), + other => panic!("expected OpaqueRef, got {other:?}"), + } + } + other => panic!("expected Table, got {other:?}"), + }; + + // Issue #1083: the original payload must be retrievable under the + // exact hash the marker carries (before the fix the store was empty). + assert_eq!(store.get(&hash).as_deref(), Some(big.as_str())); + // Lock the key<->marker contract: stored key == hash_opaque(payload). + assert_eq!(hash, hash_opaque(big.as_bytes())); + } + + #[test] + fn store_presence_does_not_change_the_ir() { + use crate::ccr::{CcrStore, InMemoryCcrStore}; + use std::sync::Arc; + + let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8); + let items = vec![ + json!({"id": 1, "blob": big.clone()}), + json!({"id": 2, "blob": big.clone()}), + ]; + + let without = compact(&items, &cfg()); + let store: Arc = Arc::new(InMemoryCcrStore::new()); + let with = compact_with_store(&items, &cfg(), Some(&store)); + + // Marker text is store-independent — only a side-effecting write is + // added, so the two IRs render identically. Compare a deterministic + // formatter's output rather than Debug formatting, which is not a + // stable contract. + use super::super::{Formatter, JsonFormatter}; + let fmt = JsonFormatter::new(); + assert_eq!(fmt.format(&without), fmt.format(&with)); + // ...and that write actually happened. + assert!(!store.is_empty()); + } } diff --git a/crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs b/crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs index b3ce01e8c..3fe57693e 100644 --- a/crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs +++ b/crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs @@ -30,7 +30,7 @@ pub mod ir; pub mod walker; pub use classifier::{classify_cell, CellClass, ClassifyConfig}; -pub use compactor::{compact, CompactConfig}; +pub use compactor::{compact, compact_with_store, CompactConfig}; pub use formatter::{CsvSchemaFormatter, Formatter, JsonFormatter, MarkdownKvFormatter}; pub use ir::{Bucket, CellValue, Compaction, FieldSpec, OpaqueKind, Row, Schema}; pub use walker::{ @@ -115,6 +115,24 @@ impl CompactionStage { let rendered = self.formatter.format(&c); (c, rendered) } + + /// Like [`Self::run`], but stash every opaque-blob payload into `store` + /// under the same hash the rendered `<>` marker carries, + /// so `GET /v1/retrieve/{hash}` and the `headroom_retrieve` tool can + /// serve the original back. `SmartCrusher::crush_array`'s lossless + /// branch passes the proxy's CCR store here; previously it called + /// [`Self::run`], which rendered markers whose payload was never stored + /// (issue #1083). When `store` is `None`, behaves exactly like + /// [`Self::run`]. + pub fn run_with_store( + &self, + items: &[serde_json::Value], + store: Option<&std::sync::Arc>, + ) -> (Compaction, String) { + let c = compact_with_store(items, &self.config, store); + let rendered = self.formatter.format(&c); + (c, rendered) + } } impl std::fmt::Debug for CompactionStage { diff --git a/crates/headroom-core/src/transforms/smart_crusher/crusher.rs b/crates/headroom-core/src/transforms/smart_crusher/crusher.rs index 98d986780..6af0fdc15 100644 --- a/crates/headroom-core/src/transforms/smart_crusher/crusher.rs +++ b/crates/headroom-core/src/transforms/smart_crusher/crusher.rs @@ -672,7 +672,11 @@ impl SmartCrusher { // ship it — nothing dropped, no CCR retrieval needed. // Otherwise fall through to the lossy path. if let Some(stage) = &self.compaction { - let (c, rendered) = stage.run(items); + // Thread the CCR store so opaque-blob `<>` markers + // emitted by lossless:table compaction are actually retrievable + // (issue #1083); the row-drop lossy path below stores its own + // payload separately. + let (c, rendered) = stage.run_with_store(items, self.ccr_store.as_ref()); if c.was_compacted() { let input_bytes = estimate_array_bytes(&item_strings); let savings_ratio = if input_bytes > 0 {