fix(ccr): key Rust search/diff/log markers with explicit_hash (#852)

Fixes #816

## What

The three `_persist_to_python_ccr` shims (`search_compressor.py`,
`diff_compressor.py`, `log_compressor.py`) called `store.store(original,
compressed)` with the default key — `SHA-256(original)[:24]` since PR
#395 — while the Rust side embeds `MD5(original)[:24]` in the emitted
`Retrieve more: hash=...` marker. Marker key and storage key never
matched, so **every retrieval of a Rust search/diff/log marker returned
"Entry not found or expired"** (inside any TTL — the symptom class
reported in #714).

Fix is exactly what #816 proposed: pass the marker's key via
`explicit_hash=cache_key` at all three call sites, the same contract
SmartCrusher has used since PR #395. No store changes needed — `store()`
already validates and honors `explicit_hash`. Also corrected the stale
comment in `search_compressor.py` that still claimed "both use
MD5(original)[:24]".

## Tests

`tests/test_ccr_rust_marker_hash_bridge.py` (companion to
`test_ccr_row_drop_store_bridge.py`, which pinned the same bug class for
SmartCrusher in #389): for each shim, the store entry must be
retrievable under the Rust marker key AND absent under the SHA-256
default key.

Verified red→green: all 3 tests fail on main with the exact issue
symptom ("store has no entry under the Rust marker key ...; the marker
dangles") and pass with the fix. `test_ccr_row_drop_store_bridge.py`
still green. `ruff check` + `ruff format --check` clean.

---------

Co-authored-by: Ash Rhodes <ashley.rhodes@king.com>

headroom/transforms/diff_compressor.py

@@ -139,7 +139,11 @@ class DiffCompressor:
             return
         try:
             store: Any = get_compression_store()
-            store.store(original, compressed)
+            # The Rust-emitted marker embeds MD5(original)[:24], but
+            # store() has defaulted to SHA-256(original)[:24] since
+            # PR #395. Pass the marker's key explicitly so retrieving
+            # the marker hash actually finds the entry (issue #816).
+            store.store(original, compressed, explicit_hash=cache_key)
         except Exception as e:
             logger.warning(
                 "CCR store write failed; cache_key %s remains in-marker only: %s",

headroom/transforms/log_compressor.py

@@ -497,7 +497,11 @@ class LogCompressor:
             return
         try:
             store: Any = get_compression_store()
-            store.store(original, compressed)
+            # The Rust-emitted marker embeds MD5(original)[:24], but
+            # store() has defaulted to SHA-256(original)[:24] since
+            # PR #395. Pass the marker's key explicitly so retrieving
+            # the marker hash actually finds the entry (issue #816).
+            store.store(original, compressed, explicit_hash=cache_key)
         except Exception as e:
             logger.warning(
                 "CCR store write failed; cache_key %s remains in-marker only: %s",

headroom/transforms/search_compressor.py

@@ -340,10 +340,10 @@ class SearchCompressor:
         `CompressionStore`. Failures are surfaced via logging instead of
         being silently swallowed (see no-silent-fallbacks rule).
 
-        Note: the Rust path computes the hash and the Python store
-        accepts the original directly. We do not rely on the Python
-        store's hash matching Rust's — the Rust hash IS the canonical
-        one (MD5(original)[:24]).
+        Note: the Rust path computes the hash and embeds it in the
+        emitted marker text — the Rust hash IS the canonical one
+        (MD5(original)[:24]). The store must be keyed by that exact
+        hash or the marker dangles.
         """
         try:
             from ..cache.compression_store import get_compression_store
@@ -353,11 +353,11 @@ class SearchCompressor:
 
         try:
             store: Any = get_compression_store()
-            # The Python `CompressionStore.store` API takes original,
-            # compressed, and an optional original_item_count. The
-            # cache_key it returns will be the same as Rust's because
-            # both use MD5(original)[:24].
-            store.store(original, compressed)
+            # The Rust-emitted marker embeds MD5(original)[:24], but
+            # store() has defaulted to SHA-256(original)[:24] since
+            # PR #395. Pass the marker's key explicitly so retrieving
+            # the marker hash actually finds the entry (issue #816).
+            store.store(original, compressed, explicit_hash=cache_key)
         except Exception as e:
             logger.warning(
                 "CCR store write failed; cache_key %s remains in-marker only: %s", cache_key, e

tests/test_ccr_rust_marker_hash_bridge.py

@@ -0,0 +1,85 @@
+"""Issue #816: Rust search/diff/log CCR markers must be retrievable.
+
+The Rust side embeds ``MD5(original)[:24]`` in the emitted
+``Retrieve more: hash=...`` marker, but since PR #395
+``CompressionStore.store()`` defaults to ``SHA-256(original)[:24]``.
+PR #395 fixed the SmartCrusher path by passing ``explicit_hash``
+(see ``test_ccr_row_drop_store_bridge.py``); the three
+``_persist_to_python_ccr`` shims on the Rust-accelerated transforms
+were never migrated, so every marker they emitted dangled —
+retrieval returned "Entry not found or expired" inside any TTL.
+
+These tests pin the cross-language contract at the shim layer: the
+store entry must be keyed by the exact hash the marker embeds.
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from headroom.cache.compression_store import (
+    get_compression_store,
+    reset_compression_store,
+)
+from headroom.transforms.diff_compressor import DiffCompressor
+from headroom.transforms.log_compressor import LogCompressor
+from headroom.transforms.search_compressor import SearchCompressor
+
+pytest.importorskip("headroom._core", reason="Rust extension required")
+
+
+@pytest.fixture(autouse=True)
+def _fresh_store():
+    reset_compression_store()
+    yield
+    reset_compression_store()
+
+
+def _rust_marker_key(original: str) -> str:
+    """The hash the Rust side embeds in emitted markers."""
+    return hashlib.md5(original.encode()).hexdigest()[:24]
+
+
+def _assert_round_trip(original: str) -> None:
+    """Entry must be retrievable under the marker's key, not SHA-256."""
+    marker_key = _rust_marker_key(original)
+    store = get_compression_store()
+
+    entry = store.retrieve(marker_key)
+    assert entry is not None, (
+        f"store has no entry under the Rust marker key {marker_key!r}; "
+        f"the marker dangles (issue #816)"
+    )
+    assert entry.original_content == original
+
+    sha_key = hashlib.sha256(original.encode()).hexdigest()[:24]
+    assert store.retrieve(sha_key) is None, (
+        "entry stored under the SHA-256 default key instead of the "
+        "marker's MD5 key — explicit_hash was not passed through"
+    )
+
+
+def test_search_compressor_shim_stores_under_marker_key() -> None:
+    original = "src/app.py:12: def handle_request(payload):\n" * 40
+    SearchCompressor()._persist_to_python_ccr(
+        original, "compressed search output", _rust_marker_key(original)
+    )
+    _assert_round_trip(original)
+
+
+def test_diff_compressor_shim_stores_under_marker_key() -> None:
+    original = "+added line of code\n-removed line of code\n" * 40
+    DiffCompressor()._persist_to_python_ccr(
+        original, "compressed diff output", _rust_marker_key(original)
+    )
+    _assert_round_trip(original)
+
+
+def test_log_compressor_shim_stores_under_marker_key() -> None:
+    original = "2026-06-11T09:00:00Z INFO worker heartbeat ok seq=1\n" * 40
+    LogCompressor()._persist_to_python_ccr(
+        original, "compressed log output", _rust_marker_key(original)
+    )
+    _assert_round_trip(original)

tests/test_transforms_search_compressor.py

@@ -104,20 +104,22 @@ def test_search_compressor_persist_to_python_ccr(monkeypatch: pytest.MonkeyPatch
     logged (not silently swallowed) — this pins both paths."""
     compressor = SearchCompressor()
 
-    seen: dict[str, tuple[str, str]] = {}
+    seen: dict[str, tuple[str, str, str | None]] = {}
     monkeypatch.setitem(
         __import__("sys").modules,
         "headroom.cache.compression_store",
         SimpleNamespace(
             get_compression_store=lambda: SimpleNamespace(
-                store=lambda original, compressed, original_item_count=0: (
-                    seen.setdefault("call", (original, compressed)) or "stored-key"
+                store=lambda original, compressed, original_item_count=0, explicit_hash=None: (
+                    seen.setdefault("call", (original, compressed, explicit_hash)) or "stored-key"
                 )
             )
         ),
     )
     compressor._persist_to_python_ccr("orig", "comp", "abc123")
-    assert seen["call"] == ("orig", "comp")
+    # explicit_hash carries the Rust marker key so retrieval of the
+    # marker hash finds the entry (issue #816).
+    assert seen["call"] == ("orig", "comp", "abc123")
 
     # Loud failure: the store raises, but persist swallows + logs (no
     # exception propagates to the compress callsite).