Capture storage: immutable packs through bounded analysis (Phases 1-5) by zaoxing · Pull Request #113 · ProjectDMX/DMI

zaoxing · 2026-08-26T02:56:53Z

What this is, plainly

When we watch a model think, we capture a lot of tensors very fast. The obvious way to store them is to write each one into the database as it arrives — but then capture runs only as fast as the database accepts writes, and a slow database slows down the thing we were trying to observe.

So this does it the other way around. Captures are packed into files, the file is sealed and written to object storage once, and nothing waits on a database. Afterwards, a separate step walks those files and builds a searchable catalog in ClickHouse describing what is in them. Nobody is blocked while that catalog is being built.

The payoff comes at read time. The catalog knows exactly where each tensor sits — which file, which byte offset, how long it is — so answering "give me these forty tensors" fetches only those byte ranges, not the files containing them. The files are large; the tensors you want usually are not.

This PR adds all of that: Phases 1 through 5 of the plan in docs/capture-storage-design.md. None of it is on main today — the whole src/dmi/storage/capture/ package is new, about 4,600 lines across 14 modules.

It is opt-in and host-only. There are no CUDA changes, the existing ClickHouse payload sink is still the default, and nothing outside storage/capture/ imports any of this yet. Merging it changes no current behaviour.

What each phase adds

Phase 1 — the file format. Defines what a capture pack looks like on disk: identifiers, checksums, and a footer listing everything inside it. The footer sits at a known place, so a reader can learn a file's whole contents with two small reads instead of downloading it. Tested against truncation, corruption, unknown versions, and repeated writes.

Phase 2 — the host pipeline. Takes captures as they arrive and turns them into packs, sealing a pack when it gets big enough or old enough. Memory and disk are bounded: there is a fixed-size queue with an explicit policy for what to drop when it fills, so sustained overload cannot grow without limit. Packs can go straight out, or through a durable local spool that survives a restart.

Phase 3 — object storage. Uploads packs to any S3-compatible store, with multipart uploads, safe retries, listing, and the two small range reads used to load a footer. Verified against a pinned Garage 2.3.0 release.

Phase 4 — the catalog. Reads the footer of each uploaded pack and writes a row per capture into ClickHouse. It only reads footers, never payloads, so cataloguing stays cheap no matter how large the tensors are. Duplicate notifications are collapsed before any work happens, and packs whose notification went missing are found by scanning the bucket.

Phase 5 — search and summaries. Phase 4 could write the catalog but not read it, so this adds the read side: searching the catalog, paging through results stably, fetching only the byte ranges you selected, and computing summary statistics over the tensors you fetched.

The one correctness problem worth reading

ClickHouse's FINAL keyword is the normal way to collapse duplicate rows, and Phase 4's views use it. It cannot be used for a point-in-time read, and the reason is subtle enough to be worth stating.

Each catalog row carries a version. To read the catalog as it stood at version W, the obvious query is "collapse duplicates, then keep rows at or below W". But FINAL collapses to the newest version that exists, and only then applies the filter. So a capture that was re-catalogued after W collapses to its new version, gets filtered out, and vanishes from the result entirely — rather than falling back to the version that was current at W.

Measured on ClickHouse 26.9.1, asking for the catalog as of version 1, where one capture was re-catalogued at version 2:

FINAL, then filter  -> [('capture-b', 128)]                      # capture-a has disappeared
correct query       -> [('capture-a', 64), ('capture-b', 128)]

Reads therefore go to the underlying tables and pick the newest version at or below the watermark explicitly. The existing FINAL views are untouched and still fine for ad-hoc queries.

Numbers

All measured on an Apple Silicon laptop against local services. These are regression baselines for the Python reference implementation — not production capacity claims, and every one should be repeated on real hardware before anyone plans against it.

Phase	What was measured	Result
1	Building packs, 2,000 × 64 KiB captures	0.521 GiB/s, 0.9% size overhead
2	Pipeline, direct vs. durable spool	0.328 / 0.345 GiB/s, nothing dropped
3	Upload to Garage, 64 MiB packs, 4 workers	0.502 GiB/s
4	Catalog inserts, 50,000-row batches	157,567 rows/s
5	Point-in-time read vs. plain `FINAL` read	22.3 ms vs. 12.1 ms
5	One page of results, first page vs. 25th	21.5 ms → 24.4 ms

Two of these are easy to misread:

The point-in-time read is about 1.85× slower than the FINAL read, but they are not two ways of doing the same thing. As shown above, FINAL returns the wrong answer. The gap is what correctness costs, not a menu option.
Page 25 costs about the same as page 1. That is the point of how paging is implemented here — cost depends on page size, not on how deep you have gone. A more common approach would get steadily slower the further you page.

Testing

425 tests in the standard CPU suite, 20 tests against a real ClickHouse, and one against a real Garage.

Two commits fix gaps in what was being tested, rather than adding more tests:

CI cannot run on Phase 4 as it stands. Two old test files came back that import a package removed in #95. Pytest imports every test file before it applies marker filtering, so the failure happens during collection and aborts the whole run — a file nobody meant to run takes everything down with it:

!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
58 deselected, 1 error in 0.94s
make: *** [test-cpu] Error 2

That is make check, which is the whole of the CI job, and its test-cpu step is the standard CPU command. To be clear about the scope: the CPU tests themselves are perfectly healthy — skip the two files and all 277 of them pass. The problem is that the default command exits non-zero having run nothing. Those files were already replaced under new names in #95; the stale copies are removed here.

Nothing tested the two halves together. The Garage tests never touched ClickHouse, and the ClickHouse tests used made-up catalog entries pointing at a file that does not exist — so nothing they returned could actually be fetched, and the component that reads footers and writes catalog rows was covered by no live test at all. The new end-to-end test runs the whole path on real data: real tensors, packed, uploaded, catalogued, searched, fetched, decoded, and compared against the originals. To check it was worth adding, I introduced a deliberate bug that records byte offsets four bytes off — the old tests all still passed; the new one failed five ways.

Problems found and fixed

Four, all in code from earlier phases:

One oversized capture could kill the pipeline permanently. The queue checks a capture against the queue's size limit, which has nothing to do with the maximum pack size. So a capture too big for any pack was accepted, then rejected later on the background thread, where the error was treated as fatal — the pack being assembled was thrown away and the pipeline stopped for good. It is now rejected up front, and the remaining edge case is counted and skipped instead of being fatal.
The upload integrity check could not fail. Uploads hand the file to the S3 transfer manager, so the upload code never sees the bytes. It recorded a checksum taken from the file's own metadata, and the later verification compared the stored object against that same recorded value — so it was comparing a number to itself. This ran immediately before deleting the local durable copy. Uploads now compute the checksum from the bytes actually sent.
A batch-size limit was reported as a broken pack. Exceeding a limit on total batch size was recorded as a failure of whichever pack happened to trip it, and the loop then carried on and skipped every remaining pack — while still reporting success.
The Garage test harness could hang. The server's error output went to a pipe nobody read; once the pipe filled, the server blocked, with no message explaining why.
The CUDA ring tests had the same problem — they pointed at monitoring/csrc, moved to native/csrc in Refactor project into a clean src-based layout #95, so make there failed immediately. That suite has since been dropped from this branch entirely, so the branch is now CPU-only: no .cu, .cuh, or native/ file is touched or added.

Phase 6 groundwork

Two of the instruments Phase 6 requires are included here, since neither existed and both define what a future native writer must satisfy.

Fault injection (tests/_faults.py) wraps the object store, the ClickHouse client, and the pack sink. Faults are scripted rather than random — a schedule names which calls fail and how — so a failure reproduces exactly. Nine tests characterise the required behaviour: a short read is refused rather than silently truncating, an immutable key written twice converges rather than conflicting, a sink failure fails loudly and then refuses admission, an insert failure leaves the pack uncommitted and the batch replayable, a duplicated insert is absorbed by replay semantics, and one corrupt pack fails only itself.

A conformance manifest (tests/tools/golden_workload.py) produces the golden-workload comparison the phase asks for: one JSON document over a deterministic corpus covering every dtype, recording pack identity and checksum, per-capture payload and decoded-tensor digests, placement, and the full summary contract. Every value is language-neutral, so a native writer is conformant exactly when the same corpus yields the same document.

Generating that manifest found a real defect in the Phase 5 summarizer on its first run: l2_norm used sqrt(sum(x**2)), which overflows float64 for large-magnitude tensors and returns infinity where the true norm is finite. It now factors out the largest magnitude before squaring.

What this does not do yet

Stated here rather than buried: Phase 3's production capacity target is still unverified, because no representative hardware or target rate has been supplied. The catalog assumes a single writer, since row versions come from that process's own clock. Paging tokens are checked for tampering but not cryptographically signed. Search filters are applied before duplicate rows are collapsed, which is safe only because catalog entries are derived from immutable files. Each search costs one extra round trip. Looking up captures by ID does not use the table's sort order. Summary statistics ignore non-finite values, and report how many there were separately. A selection is a single page — callers page explicitly.

The live tests need a running ClickHouse and Garage, so they do not run in CI as configured.

3f95d87 (#95) renamed monitoring/internal_mapper.py to src/dmi/storage/internals.py and moved both of its test modules to tests/test_storage_internals*.py with updated imports. 46e270b re-added the pre-refactor copies, which still import the removed `monitoring` package. Because a collection error aborts the whole run, this broke the PR gate outright: `python -m pytest -m "cpu" -q` failed before executing a single test. The current modules already cover this code and pass.

CaptureQuery.query_hash hashed asdict(self), which includes `cursor` and `limit`. Page one and page two of the same logical query therefore hashed differently, so a cursor could not be bound to the query that issued it and two selections over identical filters were not comparable. Add CaptureQuery.filter_hash, computed over the filter fields only, and bind CaptureSelection to it (envelope bumped to version 2). query_hash is unchanged for callers that want whole-request identity. Also give `cursor` its own 2048-byte limit. It was validated against the 512-byte identifier limit, which a keyset cursor carrying the full five-part sort key can exceed -- and it would have failed deep into a pagination walk.

A cursor addresses a position in the catalog's own sort order -- (tenant_id, experiment_id, run_id, captured_at_ns, capture_id) -- so page latency does not grow with depth and a concurrent insert cannot shift a page boundary. Each cursor carries the pinned watermark and the issuing query's filter_hash, so it cannot be replayed against different filters or against a catalog that cannot serve its snapshot. Decoding is strict: unknown version, missing or unexpected envelope fields, wrong key arity, wrong element types, empty strings and out-of-range timestamps all raise InvalidCursorError. Decode with validate=True. Python's default base64 decoder silently discards characters outside the alphabet, so a cursor with four injected junk characters decoded to a byte-identical payload and was accepted. That matters beyond tidiness: cursors are deliberately unauthenticated, on the reasoning that a forged cursor can only reach the caller's own keyspace -- which only holds if tampering is detected as malformed.

ClickHouseCatalogWriter was write-only; the sole CaptureCatalog implementation in the tree was a test fake. ClickHouseCaptureCatalog implements search and get_by_ids so CaptureReader can run against the real catalog. Reads go to *_capture_raw with explicit argMax, not the FINAL views. FINAL collapses duplicates to the highest version present and only then applies predicates, so `FINAL ... WHERE index_version <= W` drops any capture re-indexed above W instead of falling back to its value at W. Measured on 26.9.1, with capture-a re-indexed at v2 and a snapshot at watermark 1: FINAL + predicate -> [('capture-b', 128)] # a vanishes argMax at watermark -> [('capture-a', 64), ('capture-b', 128)] # correct Aggregates are deliberately unaliased. Naming one after its own source column shadows that column across the statement, and ClickHouse then rejects any filter on it with "Aggregate function ... is found in WHERE in query". Rows map onto descriptors positionally, so server-side names are never read. The keyset comparison sits in WHERE rather than HAVING: all five sort-key columns are the table's ORDER BY prefix, so it prunes granules on the primary index before grouping. Filters likewise apply pre-group, which is safe only because a descriptor derives from an immutable pack footer -- re-indexing a capture rewrites identical values. ClickHouseReaderConfig carries per-query scan-row, scan-byte and execution-time ceilings with read_overflow_mode=throw, so a breach surfaces as a server exception rather than a long scan.

Five MATERIALIZED columns on *_capture_raw -- facet_version, element_count, tensor_rank, token_span, compression_ratio -- making the catalog filterable and sortable server-side. Every one is a pure function of columns the writer already stores, so they are computed at insert with no indexer change and no extra object reads. That constraint is not incidental: CatalogIndexer.index range-reads pack footers only, so anything requiring payload bytes would force it to download every payload. The casts are load-bearing, and were verified against 26.9.1 rather than assumed. arrayProduct returns Float64, UInt64 - UInt64 returns Int64, and nullIf makes an expression Nullable -- none of which fit these column types. The uncast DDL fails at CREATE TABLE. CREATE TABLE IF NOT EXISTS will not add columns to an existing table, so ensure_schema also issues ADD COLUMN IF NOT EXISTS per facet. Verified live: a table built with the exact pre-facet schema and then populated upgrades in place, and rows written before the ALTER still resolve correct facet values. The FINAL views are left untouched; facets are queryable on the raw table.

Summaries run on payloads the caller has already fetched under the hydration byte and request limits, so summarising adds no object-store traffic and the phase gate holds by construction. That placement is forced: the indexer reads footers only, so nothing tensor-derived can be computed at index time without downloading every payload. Descriptor-derived facts live in catalog facets. CoreTensorSummaryV1 computes its statistics over the finite elements only, reporting nan_count, inf_count and finite_count alongside. Plain mean() over a tensor holding one NaN returns NaN for the whole tensor, destroying the signal exactly when it is most wanted; finite_count == 0 still distinguishes all-NaN from all-zero. Everything accumulates in float64, so abs(int64.min) and squared sums do not overflow. bfloat16 has no native NumPy dtype: it is read as uint16 and widened to float32 by a 16-bit shift, which is exact for every bit pattern. Tested against ten hand-chosen patterns including both NaN encodings and both infinities. Extension points are registries, not a fixed metric list. ScalarMetric and ArtifactProducer are bounded at 32 registrations, reject duplicate names and malformed identities, and are contained on failure -- a raising metric, one returning a non-number, or a producer returning the wrong shape becomes a typed ExtensionFailure while the core summary still returns. Producers hand back bytes and a content type; the framework writes through an ArtifactSink, so producers never touch a store. The per-extension time budget is checked after each call, so a runaway extension is reported rather than interrupted. Real preemption needs a worker boundary. Gate, both halves. Identical decoded tensors: a full round trip per dtype asserting array and byte equality. No unrelated payload bytes: with max_coalesce_gap_bytes=0, reads fall exactly inside selected extents (4 reads, 256 bytes for 256 stored); with the 4 KiB default, coalescing pulls in 192 unrelated bytes -- inside the gap-per-join bound and exactly matching HydrationEstimate. The strict gate was mutation-tested by patching _plan to fetch whole objects, and fails as it should.

benchmarks.bench_capture_search covers the read side of the catalog: snapshot cost, page latency against size and depth, filter selectivity, and core summary throughput per dtype. On local ClickHouse 26.9.1 with 50,000 rows written twice: argMax snapshot read 22.3 ms FINAL read (not a snapshot) 12.1 ms max(index_version) watermark 1.7 ms page limit=100 / 1000 / 5000 21.4 / 78.8 / 145.2 ms page 1 vs page 25 @ limit=100 21.5 vs 24.4 ms The two snapshot rows are not alternatives -- FINAL drops captures re-indexed above the watermark -- so the 1.85x gap is what correctness costs, and it is the figure Phase 6 needs before deciding what the public views project. Page cost is flat with depth, which is the property keyset pagination exists to provide. The watermark aggregate is a second round trip per search but under a tenth of a page's cost, so caching it is not yet worth the staleness. The design doc recorded the Phase 5 gate as "analysis reads only explicitly selected payload ranges", losing the identical-decoded-tensors half of the original plan. Restore it, record status and the measurements, and add a Phase 5 limitations section: single-indexer clock assumption, unauthenticated cursors, pre-aggregation filtering, the watermark round trip, get_by_ids not reaching the primary index, facets absent from the FINAL views, post-hoc extension budgets, finite-only statistics, and single-page selections.

The live suites were disjoint. The `garage` tests never touched ClickHouse, and the `clickhouse` tests indexed fabricated descriptors pointing at "packs/synthetic.dmi-pack" -- an object that does not exist -- so nothing they returned could ever hydrate. Each half was well covered; the seam between them was not covered at all. Add an end-to-end suite driving the whole chain against real bytes: tensors -> pack -> object store -> CatalogIndexer footer read -> ClickHouse -> search -> hydrate -> decode -> compare, with one capture per supported dtype. It asserts descriptor equality field by field, byte-identical payloads, identical decoded tensors, read locality against the estimate, and summary agreement with the source tensors. This closes a real blind spot rather than adding redundancy. Injecting an indexer bug that records payload offsets 4 bytes off from reality leaves the existing live suite fully green (6 passed) because it bypasses the indexer entirely, while the new suite fails on 5 tests. It is also Phase 6's golden-workload comparison in miniature. Also make synthetic_descriptors discriminating. producer_rank == batch_position and step_number == token_start held for every row, so a projection swap between either pair was undetectable; each independent field now occupies a disjoint value range. (stored_length == decoded_length is left alone: dmi-pack-v1 enforces it for uncompressed records, so it is an invariant rather than a fixture accident -- and a hazard only once a compressing codec exists.)

Co-authored-by: zaoxing <[email protected]>

Copilot

Pull request overview

Adds the missing read/analysis side of the capture catalog (Phase 5) on top of the Phase 4 write path: snapshot-correct ClickHouse reads pinned to a watermark (no FINAL snapshotting), keyset-pagination cursors, bounded hydration planning, and tensor summaries with extension points. The PR also expands CPU/live test coverage, benchmarks, and documentation to validate the end-to-end gate (selected hydration returns identical decoded tensors while not fetching unrelated bytes).

Changes:

Implement ClickHouse catalog reader semantics (watermark snapshot via argMax), cursor encoding/validation, and bounded selection/hydration planning.
Add core tensor summary + extension registries for scalar metrics and derived artifacts.
Add comprehensive CPU + live integration tests, plus benchmarks and docs for performance/semantics.

Reviewed changes

Copilot reviewed 54 out of 54 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/tools/run_garage_live.py	Manual harness to spin up Garage and run live Garage tests/benchmarks.
tests/tools/check_package.py	Ensures new `dmi.storage.capture` modules are packaged/importable.
tests/test_garage_upload_benchmark.py	CPU tests for Garage benchmark config validation + dry-run output.
tests/test_garage_live.py	Live Garage pack-store conformance test (multipart, listing, footer reads).
tests/test_clickhouse_reader_live.py	Live ClickHouse reader tests for snapshot/pagination/watermark semantics.
tests/test_clickhouse_host_benchmark.py	Regression test for ClickHouse host sampler query compatibility.
tests/test_clickhouse_facets_live.py	Live tests for facet columns + idempotent schema upgrade behavior.
tests/test_clickhouse_catalog_live.py	Live ClickHouse writer tests for replay dedupe and committed-pack queries.
tests/test_clickhouse_capture_reader.py	CPU tests for ClickHouse reader SQL shape, projection, cursors, limits.
tests/test_clickhouse_capture_catalog.py	CPU tests for writer DDL/DML and facet schema upgrades.
tests/test_capture_spool.py	CPU tests for durable spool atomicity, recovery, bounds, and pipeline integration.
tests/test_capture_s3.py	CPU tests for S3 pack store metadata, retries, listing, bounded range reads.
tests/test_capture_query_contract.py	CPU contract tests for `filter_hash`, cursor bounds, selection identity.
tests/test_capture_pipeline.py	CPU tests for bounded queue + pack assembly + pipeline behaviors.
tests/test_capture_pipeline_benchmark.py	CPU tests for pipeline benchmark config and dry-run output.
tests/test_capture_parallel_upload.py	CPU tests for bounded parallel upload workers/in-flight bytes and retries.
tests/test_capture_pack_benchmark.py	CPU tests for pack benchmark config + determinism + verification.
tests/test_capture_end_to_end_live.py	Live end-to-end test from pack bytes → indexer → ClickHouse → hydrate/decode/summary.
tests/test_capture_cursor.py	CPU tests for strict cursor encode/decode and filter/watermark binding.
tests/test_capture_catalog_indexer.py	CPU tests for indexer batching, footer-only reads, rebuild/reconcile behaviors.
tests/test_capture_catalog_benchmark.py	CPU tests for bounded insert batching in catalog benchmark.
tests/ring/test_null_mode.cu	CUDA ring test for producer “null mode” behavior and toggle correctness.
tests/ring/Makefile	Build rules for CUDA ring tests.
src/dmi/storage/capture/summary.py	Tensor decode + core summary computation (incl. `bfloat16`) with lazy NumPy import.
src/dmi/storage/capture/reader.py	CaptureReader orchestration: bounded selection, planning, hydration, summaries, extensions.
src/dmi/storage/capture/model.py	Data model + validation, query hashes, selection identity, store/catalog protocols.
src/dmi/storage/capture/filesystem.py	Filesystem pack store with key validation and integrity checks.
src/dmi/storage/capture/extensions.py	Extension registries for scalar metrics and artifact producers with failure isolation.
src/dmi/storage/capture/cursor.py	Strict keyset cursor encoding/decoding with filter/watermark binding.
src/dmi/storage/capture/clickhouse_catalog.py	ClickHouse schema writer: raw tables, FINAL views, facet columns, idempotent upgrades.
src/dmi/storage/capture/catalog.py	Catalog indexer + reconciler: footer-only reads, bounded batching, event emission.
src/dmi/storage/capture/init.py	Public API exports for capture storage, reader, catalog, cursor, summary, extensions.
pyproject.toml	Adds optional `s3` extras and registers `garage` pytest marker.
docs/capture-storage-pipeline.html	HTML explainer of capture storage pipeline and measured tradeoffs.
docs/benchmarks.md	Documents new catalog/search benchmarks and interpretation guidance.
docs/architecture.md	Updates architecture overview to include opt-in capture storage pipeline.
benchmarks/bench_garage_upload.py	Garage upload scaling benchmark with bounded concurrency/bytes.
benchmarks/bench_clickhouse_host.py	Updates sampler query to avoid removed `tables` column dependency.
benchmarks/bench_capture_search.py	Benchmark for snapshot shapes, pagination latency, selectivity, summary throughput.
benchmarks/bench_capture_pipeline.py	CPU benchmark for packing/persistence pipeline with verification.
benchmarks/bench_capture_pack.py	CPU benchmark for pack writer with deterministic payload pools.
benchmarks/bench_capture_catalog.py	ClickHouse metadata insert throughput benchmark with bounded batch sizing.

💡 Configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

An oversized capture no longer kills the pipeline. Queue admission bounds max_queue_bytes, which is unrelated to max_pack_bytes, so a payload no pack could ever hold was admitted and then rejected on the persistence thread, where `except BaseException` treated it as fatal: the buffered pack was discarded and the pipeline permanently failed. submit() now rejects such a payload up front with TOO_LARGE, and the residual case -- a payload that clears admission but is pushed over by header, footer and trailer -- is caught in the persistence loop, counted, and dropped. PackAssembler was already written to survive this with its buffered pack intact; its only caller could not reach that. S3PackStore.put now hashes the bytes it uploads. upload_fileobj hands the stream to the transfer manager, so unlike FilesystemPackStore.put nothing ever saw the content, and the object metadata it wrote came from pack.checksum -- making SpoolUploader's later stat() comparison against that same metadata tautological. It could not detect a source that lied about its own contents, and it runs immediately before the durable local spool copy is deleted. The stat() docstring now says plainly that its checksum proves identity and size, not that stored bytes are intact. The catalog's batch-level max_estimated_bytes check moves out of the per-pack try block. Raised inside it, a caller error was reported as a per-pack IndexFailure -- blaming an innocent pack -- and because the loop continued it silently skipped every remaining pack while index() still returned normally, so CatalogReconciler.rebuild reported success over an incomplete catalog. The Garage harness writes server stderr to a file instead of an unread PIPE whose buffer a long run fills, blocking the server with no diagnostic. The log tail is now included when the server exits during startup.

3f95d87 (#95) moved the native sources from monitoring/csrc to native/csrc, but the ring test harness that 46e270b added still refers to the old tree: the Makefile's include path and its producer/drain sources, plus four #includes in test_rings.cu. None of those paths exist, so `make` in tests/ring fails outright on missing files and headers. Every referenced file is present under native/csrc/ring/ with an unchanged name, so this is a path substitution only. Verified by resolving each Makefile path and each -I-relative include; not compile-verified, since this machine has no nvcc. Same regression family as the resurrected Python test modules removed earlier in this branch -- pre-refactor paths that survived the src-layout move.

This reverts commit d35a0b6. Out of scope: this branch is the CPU-side capture path, and the CUDA ring tests belong to the GPU workstream. The defect is real and still there -- tests/ring/Makefile and tests/ring/test_rings.cu carry seven references to monitoring/csrc, which #95 moved to native/csrc, so `make` in tests/ring fails on missing files -- but it should be fixed alongside the rest of the ring work by someone who can actually compile it. Nothing here can: the machine has no nvcc, so the previous commit was path-verified only. Recorded in the PR description as a finding rather than a fix.

3f95d87 (#95) moved the native sources from monitoring/csrc to native/csrc, but the ring test harness that 46e270b added still refers to the old tree: the Makefile's include path and its producer/drain sources, plus four #includes in test_rings.cu. None of those paths exist, so `make` in tests/ring fails outright on missing files and headers. Nothing but the directory name changes. All fourteen changed lines collapse to seven once monitoring/csrc and native/csrc are normalised -- no CUDA code, no build flags, no logic. Every referenced file is present under native/csrc/ring/ with an unchanged name. Verified by resolving each Makefile path and each -I-relative include. Not compile-verified: this machine has no nvcc. Same regression family as the resurrected Python test modules removed earlier in this branch -- pre-refactor paths that survived the src-layout move.

Phase 6 requires fault injection before the default sink can be switched, and comparison of golden workloads "by identity, logical bytes, checksums, decoded tensors, and query results". Neither instrument existed. Both are built here as the specification a native writer has to satisfy, since the Python implementation is the reference rather than the production writer. tests/_faults.py wraps the three boundaries that can misbehave -- object store, ClickHouse client, pack sink -- with scripted, never random, faults: a schedule names which 1-based call numbers fail and how, so a failing test reproduces exactly. Short reads, hard failures, duplicated writes and transient outages are each expressible in one call. The nine tests that use it assert observable consequences rather than internals: a short trailer read is refused rather than silently truncating; a read failure aborts indexing instead of producing a partial pack; an immutable key written twice converges rather than conflicting; a sink failure fails the pipeline loudly and then refuses further admission; an insert failure leaves the pack uncommitted and the batch replayable; a duplicated insert is absorbed by replay semantics; and one corrupt pack fails only itself rather than poisoning its batch. tests/tools/golden_workload.py produces the conformance manifest: one JSON document covering every dtype the format accepts, with pack identity and checksum, per-capture payload sha256 and crc32, decoded-tensor sha256, placement, and the full summary contract. Everything in it is language-neutral -- byte counts, hex digests, integers -- so a C++ writer can be checked against it directly. `verify` diffs field by field, naming the capture and field that moved rather than only reporting inequality. Generating that manifest immediately found a real defect in the Phase 5 summarizer: l2_norm computed sqrt(sum(x**2)), which overflows float64 for large-magnitude tensors and returns inf where the true norm is finite -- 1e200 needs 1e400 to square. Now scaled by the largest magnitude before squaring, so every squared term stays at or below 1. Verified to agree with the plain formula at normal scale, and the manifest comparison is mutation-tested: regressing the norm produces 10 named differences.

…ision Phase 6 was the only phase with no status line, and the two instruments built for it were undocumented. Record what exists, what it guarantees, and what is still missing before the default sink can move. Write down the architectural decision, which was not in the doc at all: the Python implementation is the reference and conformance suite permanently, and the production pack-and-upload plane will be native C++. The reason is structural rather than performance-related -- the ring reconstructs tensors on a native callback thread specifically to avoid touching Python or the GIL, and DMXHostEngine consumes pre-assembled rows from that thread, so a Python pack sink has no hot-path caller and creating one would reintroduce the per-tensor GIL contention the ring exists to avoid. This supersedes the Phase 2 limitation that described the Python pipeline as an interim stand-in; that limitation now points at the decision. Document the fault-injection contract as a table of required behaviours rather than a description of the harness, since a native writer has to reproduce the behaviour and not the implementation. Same for the conformance manifest: what makes a writer conformant is that the same corpus yields the same document. Note the l2_norm scaling in the Phase 5 limitations. The summary contract's numerical behaviour changed when the direct sqrt(sum(x**2)) form was found to overflow float64 and return infinity where the true norm is finite. Every command the doc now advertises was run before committing.

Samfisheryu · 2026-08-26T04:37:47Z

PR #113 exposes an interface-boundary issue before the new capture storage can be wired into the GPU path.

The CUDA ring/drain is mostly storage-agnostic, but the p2p → host boundary is hard-coded to DMXHostEngine and ClickHouseRow. We should refactor it to:

p2p → backend-neutral CapturedTensor → CaptureSink

Then keep the current behavior through ClickHouseSink, and connect #113 through PackSink. @XbzOnGit

zaoxing · 2026-08-26T04:52:22Z

Agree with the direction.

One thing worth flagging: the coupling goes a bit deeper than the sink class. dmx_host_queue_item holds a ClickHouseRow, and bindings.cpp already flattens model_id / req_id / act_name / layer_no / shard_rank into row cells before the handoff. So most of the work is moving serialization to after CaptureSink, with CapturedTensor being raw bytes plus metadata.

Two things I would nail down early:

CapturedTensor needs the full descriptor. CaptureMetadata wants 21 fields, the row today carries about five. Whatever is missing, PackSink cannot rebuild.
Ownership at the seam. PackSink appends into reusable slabs, so a borrowed view with a clear lifetime keeps us from adding a memcpy on the hot path.

Also, if we plan to run both sinks side by side to compare: index_version comes from the indexing process own clock, so two writers can go non-monotone and a watermark can drop one side rows. It is in the Phase 5 limitations. Probably worth deciding before we start the comparison rather than during.

FWIW PackSink already has something to validate against. tests/tools/golden_workload.py emits a manifest over a corpus covering every dtype, and tests/_faults.py covers the failure behaviour.

Samfisheryu · 2026-08-26T04:52:55Z

+payload_offset UInt64, stored_length UInt64, decoded_length UInt64,
+codec LowCardinality(String), payload_checksum FixedString(8), index_version UInt64,
+{_facet_ddl()}
+) ENGINE = ReplacingMergeTree(index_version)


Blocker: pinned snapshots disappear after a normal ClickHouse merge.

This table is used as version history by the watermark reader, but ReplacingMergeTree(index_version) eventually keeps only the highest version for each sorting key and removes the older rows.

I reproduced this on ClickHouse 25.12 with two identical copies of one descriptor:

Insert the descriptor at versions 1 and 2.

get_by_ids(..., watermark="1") returns one row.

Run OPTIMIZE TABLE ... FINAL to force the normal merge behavior.

The same watermark-1 lookup returns zero rows.

Observed result:

raw rows: 2 → 1

watermark-1 result: 1 → 0

OPTIMIZE only makes the reproduction deterministic; background merges can do the same automatically. This means a saved cursor or selection can stop resolving at an unpredictable time.

If old watermarks must remain readable, the history needs to be append-only, or the snapshot design must not depend on versions that this engine deletes.

Confirmed, reproduced on 26.9.1:

BEFORE merge: raw rows = 2 watermark 1 -> [(capture-a, 64)] AFTER merge: raw rows = 1 watermark 1 -> []

You are right that this is a design problem rather than a bug. argMax(...) WHERE index_version <= W needs rows that ReplacingMergeTree is defined to delete, so the snapshot has an expiry nobody controls.

My live test missed it because it wrote two versions and queried immediately, never forcing a merge. It confirmed the mechanism I built instead of testing what the engine does to the rows underneath it. My fault.

One angle for the fix: descriptors come from immutable pack footers, so a re-indexed row is byte-identical to the one it replaces. The version history has no content difference in it. The watermark only really needs to exclude captures indexed after the snapshot, which may not require keeping superseded versions at all. Will come back with options rather than patch it blind.

Samfisheryu · 2026-08-26T04:53:06Z

+        not free, and it runs once per search.
+        """
+        rows = self._client.execute(
+            f"SELECT max(index_version) FROM {self._qualified()}",


Blocker: this watermark becomes visible before the logical index batch is committed.

CatalogIndexer assigns one index_version, then writes descriptors through multiple INSERTs of up to 10,000 rows before writing the pack markers. Reading max(index_version) from the raw descriptor table exposes that version after the first INSERT, while the rest of the same batch is still being written.

Reproduction:

Write 2 of 4 descriptors at version 123.

Read page 1 with limit=1; it returns watermark 123 while the table contains 2 rows.

Write the remaining 2 descriptors at version 123.

Continue using the pinned cursor.

Observed result:

rows visible when the snapshot started: 2

rows returned by the pinned walk: 4

Each ClickHouse INSERT may be atomic, but this Python loop is not one transaction. The reader can therefore observe and pin a half-written batch.

We need a separate published watermark: write every descriptor batch and pack marker first, then publish version 123 as the final step. Readers should use only that published version.

Confirmed, reproduced against the real indexer path with four descriptors written as two INSERTs at one version:

rows visible when the snapshot was taken : 2 rows returned by the pinned watermark : 4

Agreed on the fix: write every descriptor batch and pack marker first, publish the version as a separate final step, and have readers use only the published value.

This was unreachable in my tests because max_rows_per_insert defaults to 10,000 and every corpus I used was small enough to fit one INSERT, so a batch never split. Adding coverage that forces multiple inserts per version.

Samfisheryu found that a reader could pin a half-written batch. CatalogIndexer assigns one index_version and then writes descriptors across several INSERTs before the pack markers, so a reader sampling max(index_version) on the descriptor table observes that version mid-batch. Reproduced against the real indexer path with four descriptors written as two INSERTs: rows visible when the snapshot was taken : 2 rows returned by the pinned watermark : 4 Each INSERT is atomic on its own; the loop around them is not, so "pinned snapshot" was not true. The version is now published as a separate final step, after every descriptor batch and pack marker is durable, into an append-only *_index_watermark table. Readers take the watermark only from that log. The table is plain MergeTree rather than ReplacingMergeTree: it is a history, and Replacing would eventually collapse exactly what a pinned snapshot reads. Add tests/test_clickhouse_snapshot_live.py for the corner cases the first round of live tests could not reach. Those tests confirmed the mechanism rather than challenging the engine underneath it: ReplacingMergeTree deduplicates "at an unknown time", so reading immediately after two writes only ever exercises the pre-merge state, and max_rows_per_insert defaults to 10,000, so no corpus I used ever split a batch across INSERTs. Merges are now forced with OPTIMIZE FINAL and batches are written with a deliberately small insert size. The second blocker Samfisheryu raised is not fixed here and is recorded as a strict xfail: argMax(...) WHERE index_version <= W reads rows that ReplacingMergeTree is defined to delete on merge, so a pinned snapshot expires at a time nobody controls. That needs a design change rather than a patch, and the test will flag as soon as one lands.

…ions Second blocker from Samfisheryu: a pinned watermark stopped resolving after a merge. Reproduced on 26.9.1 -- BEFORE merge: raw rows = 2, watermark 1 -> [(capture-a, 64)] AFTER merge: raw rows = 1, watermark 1 -> [] argMax(...) WHERE index_version <= W reads rows that ReplacingMergeTree is defined to delete: it keeps only the highest version per sorting key. So the snapshot had an expiry nobody controlled, and a saved cursor could stop resolving at an unpredictable time. The snapshot boundary moves to the packs committed at or before the watermark, recorded in an append-only *_pack_commit_log that no merge rewrites. A capture belongs to exactly one immutable pack, so this is the same question asked of durable data instead of collapsible data. What makes it sound is that a descriptor is derived from an immutable pack footer, so re-indexing rewrites byte-identical rows. There is no content to choose between, which is why argMax is now deduplication rather than version selection and why it no longer matters which duplicate survives a merge. Two existing live tests encoded the old semantics and were updated rather than the code bent to fit them: - test_watermark_isolates_rows_indexed_after_the_first_page gave its "later" captures the same pack_id as the corpus, so it simulated adding captures to an already-committed pack. Packs are sealed before commit, so that cannot happen; the later captures now get their own pack. - test_snapshot_returns_the_version_at_the_watermark_not_the_latest mutated a descriptor's offset between versions to prove version selection worked. That scenario is outside the contract now, so it is replaced by test_replay_is_invisible_because_it_rewrites_identical_descriptors, which asserts the invariant the design depends on and fails loudly if a future change ever makes a re-indexed descriptor differ. The limitation this trades into: the snapshot can no longer distinguish content across versions, and relies on that immutability invariant instead. The new test is the guard.

…hat missed them Each fix comes with a test that names the gap in the existing suites, because the pattern matters more than the individual bug: all five sit just outside a boundary the tests already exercised. Object keys. quote() treats "~" as always-safe per RFC 3986 and ignores its `safe` argument for it, but the object-key pattern rejects "~". A tenant or session id containing one produced a key every store refuses, the sink raised, and the persistence thread treated that as fatal -- so one tenant name took down capture for every tenant. The gap: key generation and key validation were tested separately and never against each other. Now parametrised over every character the encoder passes through unchanged. Catalog commit ordering. commit_packs wrote the inventory before the commit log, but committed_pack_ids reads the inventory to skip replays while readers bound the snapshot by the log. A crash between the two writes left a pack skipped forever and never visible -- silent, permanent loss. The log is now written first, so the same crash only costs redundant work. The gap: the fault tests failed whole operations, never the window between two writes that must agree. Reconciliation. index_object_keys inspected every key outside CatalogIndexer's per-pack failure handling, so one foreign object in the bucket aborted a whole rebuild instead of being recorded as one failure. A bucket holds whatever anyone put in it. The gap: reconciliation was only ever tested over buckets containing valid packs. Version monotonicity. index_version comes from a wall clock, which steps backwards on NTP correction and can repeat on a coarse clock, letting a batch land under a watermark a reader already pinned -- the same defect the published watermark fixed for the mid-batch case. The indexer now advances past its last published version rather than trusting the clock. It advances rather than raising: a version is an ordering token, not a timestamp, and refusing would stop capture over a clock correction. published_at_ns still records real time, and cross-process skew remains documented under Phase 5 limitations. Search benchmark. It wrote descriptors but never committed packs or published a watermark, so after the snapshot change every search returned zero rows and the page-latency table in docs/benchmarks.md was no longer reproducible. It also leaked the two new tables on cleanup. Both fixed; verified it returns rows again and drops everything it creates. Two existing tests changed with the behaviour rather than around it: the commit-order assertion now expects the log first and says why, and the backwards-clock test asserts published versions stay strictly increasing instead of expecting a refusal.

Samfisheryu · 2026-08-26T14:27:26Z

+        }
+        sql = (
+            f"SELECT {self._projection()} FROM {self._qualified()} "
+            "WHERE capture_id IN %(capture_ids)s AND pack_id IN "


Production scaling blocker: this point lookup cannot use the table ordering.

The capture table is ordered by (tenant_id, experiment_id, run_id, captured_at_ns, capture_id), but this query only knows capture_id. Because the ID is the last key, ClickHouse cannot jump directly to it.

I reproduced the lookup shape on ClickHouse 25.12 with 100,000 rows and the same ORDER BY. Looking up one ID produced:

PrimaryKey Keys: capture_id Parts: 1/1 Granules: 12/12

So finding one capture scanned every granule. At one billion captures, hydrating ten selected IDs can still scan a large part of the billion-row catalog. This matters on every estimate() and hydrate(), because both resolve the selection through get_by_ids.

The lookup needs an access path ordered by identity, for example a projection or companion table on (tenant_id, capture_id). The selection should carry the tenant as well.

Samfisheryu · 2026-08-26T14:27:38Z

+            descriptors.extend(pack_descriptors)
+            valid_refs.append(ref)
+
+        version = self._clock_ns()


Production blocker for multiple indexers: version ordering is only local to one process.

The new _published_version guard fixes repeated or backward clock values inside one CatalogIndexer instance, but a second process starts with no knowledge of the first one. Its wall clock can publish a lower version later.

I reproduced this against the updated implementation with two independent indexers:

Indexer A publishes two captures at version 200.

A reader pins watermark 200.

Indexer B runs later but publishes another capture at version 100.

The reader continues the original watermark-200 cursor.

Observed result:

rows when snapshot started : 2 rows returned by the pinned walk: 3

The new row is version 100, so it falls underneath the already-pinned watermark 200 and appears inside the old snapshot.

This limitation is already mentioned in the design notes and in the PR discussion, but it is still a correctness blocker for multiple workers, failover, or a restarted worker after clock rollback. Versions need to come from one shared monotonic allocator or a database-owned catalog generation, not each process wall clock.

zaoxing added 9 commits August 25, 2026 17:25

Phase 4

46e270b

Copilot AI lite review requested due to automatic review settings August 26, 2026 02:56

Copilot started reviewing on behalf of zaoxing August 26, 2026 02:57 View session

Copilot started work on behalf of zaoxing August 26, 2026 02:59 View session

ci: install numpy in cpu-and-package-layout job

ad35b83

Co-authored-by: zaoxing <[email protected]>

Copilot AI reviewed Aug 26, 2026

View reviewed changes

Comment thread tests/tools/run_garage_live.py

Comment thread src/dmi/storage/capture/catalog.py

Comment thread tests/ring/Makefile Outdated

Copilot finished work on behalf of zaoxing August 26, 2026 03:01

zaoxing added 2 commits August 25, 2026 23:07

zaoxing changed the title ~~Phase 5: bounded analysis over the capture catalog~~ Capture storage: immutable packs through bounded analysis (Phases 1-5) Aug 26, 2026

zaoxing added 2 commits August 25, 2026 23:24

zaoxing requested a review from Samfisheryu August 26, 2026 03:34

zaoxing and others added 3 commits August 26, 2026 00:15

test: drop superseded CUDA ring suite

3ceb813

Samfisheryu reviewed Aug 26, 2026

View reviewed changes

zaoxing added 2 commits August 26, 2026 01:01

Samfisheryu reviewed Aug 26, 2026

View reviewed changes

Samfisheryu requested a review from XbzOnGit August 26, 2026 15:35

Navigation Menu

Uh oh!

Capture storage: immutable packs through bounded analysis (Phases 1-5) - #113

Capture storage: immutable packs through bounded analysis (Phases 1-5)#113
zaoxing wants to merge 20 commits into
mainfrom
alan/clickhouse-optimization

zaoxing commented Aug 26, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Samfisheryu commented Aug 26, 2026 •

edited

Loading

Uh oh!

zaoxing commented Aug 26, 2026 •

edited

Loading

Uh oh!

Samfisheryu Aug 26, 2026

Uh oh!

zaoxing Aug 26, 2026

Uh oh!

Samfisheryu Aug 26, 2026

Uh oh!

zaoxing Aug 26, 2026

Uh oh!

Samfisheryu Aug 26, 2026

Uh oh!

Samfisheryu Aug 26, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

zaoxing commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What this is, plainly

What each phase adds

The one correctness problem worth reading

Numbers

Testing

Problems found and fixed

Phase 6 groundwork

What this does not do yet

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Samfisheryu commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zaoxing commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Samfisheryu Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

zaoxing Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

Samfisheryu Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

zaoxing Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

Samfisheryu Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

Samfisheryu Aug 26, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zaoxing commented Aug 26, 2026 •

edited

Loading

Samfisheryu commented Aug 26, 2026 •

edited

Loading

zaoxing commented Aug 26, 2026 •

edited

Loading