diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e4dfce5..a9e5f3ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,10 @@ # Changelog ## [Unreleased] +### Fixed + +- zstd compression uses a fresh `zstandard` compressor/decompressor per call instead of shared module-level singletons, fixing process segfaults when ingester workers compress documents concurrently (Python < 3.14). + ## [0.61.0] - 2026-06-23 ### Added diff --git a/haiku_rag_slim/haiku/rag/store/compression.py b/haiku_rag_slim/haiku/rag/store/compression.py index 5aecf464..2202cba8 100644 --- a/haiku_rag_slim/haiku/rag/store/compression.py +++ b/haiku_rag_slim/haiku/rag/store/compression.py @@ -10,15 +10,18 @@ try: # pragma: no cover except ImportError: from zstandard import ZstdCompressor, ZstdDecompressor, get_frame_parameters - _zstd_compressor = ZstdCompressor() - _zstd_decompressor = ZstdDecompressor() - + # ZstdCompressor/ZstdDecompressor are not thread-safe: each wraps a single + # reused ZSTD_CCtx/ZSTD_DCtx, and concurrent .compress()/.decompress() calls + # corrupt that context and segfault in the C backend. Ingestion drives this + # path from multiple worker threads (asyncio.to_thread in + # _prepare_document_from_docling), so construct a fresh instance per call + # rather than sharing a module-level singleton. def _zstd_compress(data: bytes) -> bytes: - return _zstd_compressor.compress(data) + return ZstdCompressor().compress(data) def _zstd_decompress(data: bytes) -> bytes: content_size = get_frame_parameters(data).content_size - return _zstd_decompressor.decompress(data, max_output_size=content_size) + return ZstdDecompressor().decompress(data, max_output_size=content_size) def compress_json(json_str: str) -> bytes: diff --git a/tests/store/test_compression.py b/tests/store/test_compression.py index 9532b312..45d3f4b2 100644 --- a/tests/store/test_compression.py +++ b/tests/store/test_compression.py @@ -1,4 +1,5 @@ import json +from concurrent.futures import ThreadPoolExecutor from haiku.rag.store.compression import ( compress_docling_split, @@ -24,6 +25,27 @@ class TestJsonCompression: compressed = compress_json('{"test": true}') assert compressed[:4] == b"\x28\xb5\x2f\xfd" + def test_concurrent_compress_decompress_is_safe(self): + """Compression must be safe under concurrent threads. + + Ingestion offloads compression to worker threads via + asyncio.to_thread; sharing a single zstandard compressor/decompressor + across threads corrupts its internal C context and segfaults the + process. Hammer both paths from many threads to guard against a + regression to module-level singletons. + """ + payloads = [ + json.dumps({"i": i, "text": f"document body {i} " * 200}) for i in range(64) + ] + + def roundtrip(json_str: str) -> str: + return decompress_json(compress_json(json_str)) + + with ThreadPoolExecutor(max_workers=16) as pool: + results = list(pool.map(roundtrip, payloads * 8)) + + assert results == (payloads * 8) + class TestDoclingCompressionSplit: def test_split_with_pages(self):