From 05868f96fdbb29d8190268df1e09968e138f822a Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:24:28 -0400 Subject: [PATCH 01/16] Infer a dynamic tensor dim in elements, not bytes ResolveShape divided the raw payload-slice byte count by the product of the fixed dims, so the inferred dimension came out too large by exactly the dtype width: a float32 [-1, 4] slice of 32 bytes resolved to [8, 4] where the reference (clickhouse_record_sink.cpp resolve_shape) says [2, 4]. SubmitRow then compared that against the row's own metadata, answered kShapeMismatch, and submit() threw -- so every production capture with a dynamic dim and a dtype wider than one byte failed hard. The element size comes from DtypeWidth(dtype_name), which submit() already computes and proves non-zero one line above the call, and which SubmitRow uses for the same purpose. The reference's two other guards come with it: a slice length that is not a whole number of elements, and an inferred dimension past int64. The non-dynamic branch needs no equivalent of the reference's 'fixed tensor shape does not match payload-slice bytes': SubmitRow already refuses that row as kSizeMismatch against elements * DtypeWidth. None of this was reachable from a test, because bindings_sink.cpp pinned inferred_dynamic_dim to -1 while production (bindings.cpp) sets it from the -1 in a PayloadSlice's shape. The driver now takes the field, and the new tests drive the float32 case the reference specifies with a uint8 control that passed before. --- native/csrc/sink/bindings_sink.cpp | 9 ++++ native/csrc/sink/native_pack_sink.cpp | 34 ++++++++++++--- tests/test_native_adapter_torch.py | 59 ++++++++++++++++++++++++++- 3 files changed, 95 insertions(+), 7 deletions(-) diff --git a/native/csrc/sink/bindings_sink.cpp b/native/csrc/sink/bindings_sink.cpp index 9ef2fa72f..427edd63d 100644 --- a/native/csrc/sink/bindings_sink.cpp +++ b/native/csrc/sink/bindings_sink.cpp @@ -32,7 +32,16 @@ ring::PayloadSlice ParseSlice(const py::dict& row) { slice.materialization = ring::PayloadMaterialization::TENSOR; slice.dtype = row["dtype"].cast(); slice.logical_shape = row["shape"].cast>(); + // Optional, default -1 ("no dynamic dimension"). The production encoder + // (bindings.cpp) sets this from the -1 it finds in the Python slice's + // shape, so pinning it here to -1 left the whole dynamic-dim branch of + // the adapter unreachable from this driver. A caller drives it the way + // production presents it: `shape=[-1, 4], inferred_dynamic_dim=0`. slice.inferred_dynamic_dim = -1; + if (row.contains("inferred_dynamic_dim") && + !row["inferred_dynamic_dim"].is_none()) { + slice.inferred_dynamic_dim = row["inferred_dynamic_dim"].cast(); + } return slice; } diff --git a/native/csrc/sink/native_pack_sink.cpp b/native/csrc/sink/native_pack_sink.cpp index 964ce4109..ffbb79ff9 100644 --- a/native/csrc/sink/native_pack_sink.cpp +++ b/native/csrc/sink/native_pack_sink.cpp @@ -3,6 +3,7 @@ #include #include +#include #include namespace dmi_sink { @@ -14,9 +15,19 @@ namespace { } // Logical shape of the slice, with the dynamic dim resolved from the byte -// count exactly like the reference checked_payload_view does. +// count exactly like the reference resolve_shape does. +// +// `element_bytes` is the slice dtype's width (DtypeWidth of the mapped name, +// which the caller has already proved non-zero). A dynamic dim counts +// ELEMENTS, not bytes: the reference divides the slice length by the element +// size first and only then factors it over the fixed dims. Dividing the raw +// byte count resolved every dtype wider than one byte too large by exactly +// that width -- a float32 [-1, 4] slice of 32 bytes came out [8, 4] where the +// reference says [2, 4] -- and SubmitRow refused the row against its own +// metadata, so every capture with a dynamic dim failed hard. bool ResolveShape(const ring::PayloadSlice& slice, uint64_t length_bytes, - std::vector* shape_out, std::string* error) { + uint64_t element_bytes, std::vector* shape_out, + std::string* error) { *shape_out = slice.logical_shape; if (slice.inferred_dynamic_dim < 0) return true; const int dim = slice.inferred_dynamic_dim; @@ -24,17 +35,27 @@ bool ResolveShape(const ring::PayloadSlice& slice, uint64_t length_bytes, if (error) *error = "inferred dynamic dim exceeds shape rank"; return false; } + if (element_bytes == 0 || length_bytes % element_bytes != 0) { + if (error) *error = "payload-slice bytes are not divisible by dtype size"; + return false; + } + const uint64_t elements = length_bytes / element_bytes; uint64_t fixed = 1; for (size_t i = 0; i < shape_out->size(); ++i) { if (static_cast(i) == dim) continue; fixed *= static_cast((*shape_out)[i]); } - if (fixed == 0 || length_bytes % fixed != 0) { + if (fixed == 0 || elements % fixed != 0) { if (error) *error = "payload bytes do not factor over the fixed dims"; return false; } - (*shape_out)[dim] = - static_cast(length_bytes / fixed); + const uint64_t inferred = elements / fixed; + if (inferred > + static_cast(std::numeric_limits::max())) { + if (error) *error = "inferred tensor dimension exceeds int64"; + return false; + } + (*shape_out)[dim] = static_cast(inferred); return true; } @@ -125,7 +146,8 @@ void NativePackSink::submit(ring::RecordEnvelope envelope) { } std::vector shape; std::string shape_error; - if (!ResolveShape(*slice, length, &shape, &shape_error)) { + if (!ResolveShape(*slice, length, static_cast(width), &shape, + &shape_error)) { invalid(shape_error); } RowInput input; diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py index 2236fe0f8..abaedb1c2 100644 --- a/tests/test_native_adapter_torch.py +++ b/tests/test_native_adapter_torch.py @@ -93,13 +93,14 @@ def _meta(index: int, dtype: str = "float32", session: str = "s") -> dict: def _row(meta_json: dict, offset: int, length: int | None, dtype: str, - shape=(16,)) -> dict: + shape=(16,), inferred_dynamic_dim: int | None = None) -> dict: return { "metadata_json": json.dumps(meta_json), "offset": offset, "length": length, "dtype": ATEN[dtype], "shape": list(shape), + "inferred_dynamic_dim": inferred_dynamic_dim, } @@ -274,6 +275,62 @@ def test_a_scalar_envelope_is_admitted(tmp_path): assert payload == torch.tensor(3.5, dtype=torch.float32).numpy().tobytes() +@pytest.mark.parametrize("dtype, resolved", [("float32", 2), ("uint8", 8)]) +def test_a_dynamic_dimension_is_inferred_in_elements_not_bytes( + tmp_path, dtype, resolved +): + """shape=[-1, 4] over 32 payload bytes resolves in ELEMENTS, not bytes. + + The production encoder (bindings.cpp) sets `inferred_dynamic_dim` from + the -1 it finds in a PayloadSlice's shape, and the reference sink + (clickhouse_record_sink.cpp, resolve_shape) divides the slice length by + the dtype's element size before factoring it over the fixed dims. The + adapter factored the raw BYTE count instead, so a float32 [-1, 4] slice + of 32 bytes resolved to [8, 4] where the reference says [2, 4]; SubmitRow + then refused it against its own metadata and `submit()` threw. Every + capture with a dynamic dim and a dtype wider than one byte failed. + + uint8 is the control: element size 1, so bytes and elements agree and it + resolved correctly before this fix as well. + """ + meta = _meta(0, dtype=dtype) + meta["shape"] = [resolved, 4] + CaptureMetadata.from_mapping(meta) # the shape the reference infers + sink, lease = _make_sink(tmp_path) + payload = torch.zeros(32 // WIDTH[dtype], dtype=TORCH_DTYPE[dtype]) + sink.submit_envelope( + LAYOUT, + [_row(meta, 0, 32, dtype, shape=(-1, 4), inferred_dynamic_dim=0)], + payload, + ) + assert sink.flush_and_wait(30.0) + sink.rethrow_if_failed() + assert sink.snapshot()["persisted_records"] == 1 + ((staged, staged_payload),) = _read_staged(tmp_path) + assert staged.shape == (resolved, 4) + assert len(staged_payload) == 32 + + +def test_a_slice_that_is_not_a_whole_number_of_elements_is_refused(tmp_path): + """30 bytes is not a whole number of float32s: the reference's guard. + + resolve_shape divides by the element size and refuses a remainder before + it ever factors over the fixed dims, so the refusal names the dtype size + rather than the fixed dims it never reached. + """ + meta = _meta(0) + meta["shape"] = [2, 4] + sink, lease = _make_sink(tmp_path) + with pytest.raises(RuntimeError, match="divisible"): + sink.submit_envelope( + LAYOUT, + [_row(meta, 0, 30, "float32", shape=(-1, 4), + inferred_dynamic_dim=0)], + torch.zeros(8, dtype=torch.float32), + ) + assert sink.snapshot()["submitted_records"] == 0 + + @pytest.mark.parametrize("field, value", [ ("layer_number", 0.5), ("captured_at_ns", 123.5), From 47adecb2f14e495f5a513b997b3ddaee9d026dcd Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:27:40 -0400 Subject: [PATCH 02/16] Report the pack conflict that refuses to overwrite The pack-conflict branch composed its diagnostic into last_error and then returned straight out of the retry loop, skipping the `*error = last_error` at the end of UploadOne. UploadPending recorded {pack_id, key, attempts=0, error=""} -- an empty string for the ONE failure that means 'the object store already holds someone else's immutable object at this key, do not overwrite it'. An operator saw a pack that had stopped uploading with no stated reason. The exit now assigns both outputs, like the other two: attempts carries the HEAD that found the conflict (1), matching the retry-exhausted exit rather than the byte-gate refusal's 0, which is recorded before any attempt is made. --- native/csrc/store/uploader.cpp | 8 +++++ tests/test_native_uploader.py | 59 ++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/native/csrc/store/uploader.cpp b/native/csrc/store/uploader.cpp index 73d2d85cd..8f8de47b4 100644 --- a/native/csrc/store/uploader.cpp +++ b/native/csrc/store/uploader.cpp @@ -156,6 +156,14 @@ bool SpoolUploader::UploadOne(const StagedPack& staged, PackRef* ref, "the staged pack " + staged.pack_id + "). The staged pack is " "retained in the spool for inspection; do not overwrite the " "existing object."; + // Reported the way every other exit reports: the diagnostic is the + // whole point of this branch, and returning straight out of the + // loop skipped the `*error = last_error` below, so the caller was + // handed an empty string for the ONE failure that means "do not + // overwrite". The attempt count goes with it, for the same reason + // the retry-exhausted exit carries one. + if (attempts_out) *attempts_out = attempts; + if (error) *error = last_error; return false; // NOT retryable } } diff --git a/tests/test_native_uploader.py b/tests/test_native_uploader.py index dbf71960c..c17d9f809 100644 --- a/tests/test_native_uploader.py +++ b/tests/test_native_uploader.py @@ -370,6 +370,65 @@ def test_mixed_batch_reports_oversized_pack_at_its_position(fake_s3, tmp_path): store.close() +def test_a_pack_conflict_reports_why_it_will_not_overwrite(fake_s3, tmp_path): + """The one non-retryable failure must reach the caller as WORDS. + + A different object already at the key is the only outcome that means + "someone else's immutable object is here, do not overwrite". UploadOne + composed that diagnostic into `last_error` and then returned straight + out of the loop, skipping the `*error = last_error` at the end of the + function -- so UploadPending recorded the failure with an empty error + string and an attempt count of zero, and an operator saw a pack that + had simply stopped uploading for no stated reason. + """ + sink = DriverSession(SINK_DRIVER) + store = DriverSession(STORE_DRIVER) + try: + spool_root = tmp_path / "spool" + staged = _stage(sink, spool_root, 7) + # A DIFFERENT object at this key: neither its size nor its + # dmi-sha256 agrees with the staged pack. + squatter = b"someone else's pack" * 7 + assert squatter != Path(staged["path"]).read_bytes() + put = _store_call( + "put", **_client_base(fake_s3), key=staged["object_key"], + data_b64=base64.b64encode(squatter).decode(), + metadata={"dmi-format": "dmi-pack-v1", "dmi-sha256": "ab" * 32}, + content_type="application/vnd.dmi.pack", + ) + assert put["ok"], put + puts_before = sum( + 1 for c in STATE.calls + if c["method"] == "PUT" and staged["object_key"] in c["path"] + ) + + result = _upload_pending(store, fake_s3, spool_root) + assert result["ok"], result + assert result["snapshot"]["failed_packs"] == 1 + assert result["snapshot"]["uploaded_packs"] == 0 + failure = result["failures"][0] + assert failure["pack_id"] == staged["pack_id"], failure + assert failure["object_key"] == staged["object_key"], failure + assert "pack conflict" in failure["error"], failure + assert staged["object_key"] in failure["error"], failure + assert "do not overwrite" in failure["error"], failure + # Counted like every other attempted failure: the HEAD that found + # the conflict was an attempt, and it is not retried. + assert failure["attempts"] == 1, failure + + # The refusal is real: nothing was written over the existing object, + # and the staged pack is still there to inspect. + puts_after = sum( + 1 for c in STATE.calls + if c["method"] == "PUT" and staged["object_key"] in c["path"] + ) + assert puts_after == puts_before + assert Path(staged["path"]).exists() + finally: + sink.close() + store.close() + + def test_uploader_head_to_head_with_python_reference(fake_s3, tmp_path): """Same staged bytes through both uploaders; identical objects land. From e2617f5d6037b2d4cca3a76c6e2d785afd2c724e Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:32:15 -0400 Subject: [PATCH 03/16] Refuse a record larger than the queue instead of waiting for room Submit screened max_pack_bytes and nothing else, so a record between max_queue_bytes and max_pack_bytes passed admission and reached a wait loop whose condition (queue_bytes_ + n <= max_queue_bytes) cannot hold even on an empty queue. Under kDropNewest it came back dropped, with an admission timeout timed out, and under kBlock with a negative timeout it blocked forever -- all three reachable from the conformance driver, and with the shipped defaults (16 MiB queue, 128 MiB pack) that is every record over 16 MiB. The oracle, _BoundedQueue.put (pipeline.py), answers TOO_LARGE for that record before entering its wait loop, and HostCapturePipeline.submit counts it oversized -- the same answer and the same counter as the max_pack_bytes screen it applies first. Both screens now sit before the loop, in that order. test_block_with_timeout_counts_timeouts asserted the defect: it drove a 64-byte payload under a 63-byte cap and expected timed_out, which the oracle calls too_large. Its refusal moved to the new parity test, and what remains under that name is the boundary the new screen must not overshoot (a record of exactly the cap is still admitted). The timeout path needs a queue held full by an undrained consumer, which the oracle pins by stalling its sink (BlockingPackSink) and this driver has no equivalent for. --- native/csrc/sink/pack_sink.cpp | 16 ++++++++ tests/test_native_pack_sink.py | 69 ++++++++++++++++++++++++++++++---- 2 files changed, 78 insertions(+), 7 deletions(-) diff --git a/native/csrc/sink/pack_sink.cpp b/native/csrc/sink/pack_sink.cpp index f42d7eba0..b015a2f6f 100644 --- a/native/csrc/sink/pack_sink.cpp +++ b/native/csrc/sink/pack_sink.cpp @@ -171,11 +171,27 @@ Admission PackSink::Submit(dmi_pack::RecordMetadata metadata, worker = RouteWorker(metadata.tenant_id, metadata.session_id, metadata.producer_rank); } + // Both bounds screen before the wait loop, in the oracle's order: + // HostCapturePipeline.submit tests max_pack_bytes first and only then + // hands the record to _BoundedQueue.put, which tests its own byte cap + // before it ever waits. Both answer TOO_LARGE and both count the record + // as oversized. if (n > config_.max_pack_bytes) { std::lock_guard lock(mutex_); ++counters_.oversized_records; return Admission::kTooLarge; } + // Without this, a record between the two bounds reached a loop whose + // condition (queue_bytes_ + n <= max_queue_bytes) cannot hold even on an + // empty queue: kDropNewest called it dropped, a timeout called it timed + // out, and kBlock with a negative timeout waited forever. With the + // shipped defaults (16 MiB queue, 128 MiB pack) that is every record + // over 16 MiB. + if (n > config_.max_queue_bytes) { + std::lock_guard lock(mutex_); + ++counters_.oversized_records; + return Admission::kTooLarge; + } const bool block = config_.overload == Overload::kBlock; const double deadline = config_.admission_timeout_s < 0 diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py index 9bac1ad1b..fd7df1a39 100644 --- a/tests/test_native_pack_sink.py +++ b/tests/test_native_pack_sink.py @@ -229,15 +229,33 @@ def test_drop_newest_counts_drops(sink, tmp_path): assert outcomes.count("accepted") == snapshot["admitted_records"] -def test_block_with_timeout_counts_timeouts(sink, tmp_path): - # A queue that can never fit a 64-byte payload (63-byte cap): BLOCK - # waits, then times out. Deterministic — no thread timing involved. +def test_block_with_timeout_admits_a_record_that_exactly_fits(sink, tmp_path): + """A 64-byte payload under a 64-byte cap is admitted, not refused. + + This test used to submit the same 64-byte payload under a 63-byte cap + and assert "timed_out": with no screening of max_queue_bytes at + admission, a record the queue could never hold entered the wait loop + and aged out of it. The oracle (_BoundedQueue.put) answers TOO_LARGE + for that record before it ever waits, so the refusal moved to + test_a_record_the_queue_can_never_hold_is_too_large and what is left + here is the boundary the new screen must not overshoot: the bound is + `n <= max_queue_bytes`, so a record of exactly the cap still fits. + + The timeout path itself needs a queue that is full of records a + consumer has not drained yet. The oracle pins it by stalling its sink + (tests/_faults.BlockingPackSink, test_capture_pipeline.py); this + driver has no equivalent stall, and every alternative here would be a + race against the packer thread. + """ _open(sink, tmp_path / "spool", max_queue_records=256, - max_queue_bytes=63, overload="block", admission_timeout=0.05) - assert _submit(sink, _record(0)) == "timed_out" + max_queue_bytes=64, overload="block", admission_timeout=0.05) + assert _submit(sink, _record(0)) == "accepted" + assert sink.call(op="flush", timeout=30)["ok"] snapshot = sink.call(op="close", timeout=30)["snapshot"] - assert snapshot["timed_out_records"] == 1 - assert snapshot["persisted_records"] == 0 + assert snapshot["admitted_records"] == 1 + assert snapshot["oversized_records"] == 0 + assert snapshot["timed_out_records"] == 0 + assert snapshot["persisted_records"] == 1 def test_duplicate_capture_is_dropped_not_failed(sink, tmp_path): @@ -262,6 +280,43 @@ def test_oversized_record_is_rejected_up_front(sink, tmp_path): assert snapshot["packs_persisted"] == 0 +@pytest.mark.parametrize("overload, timeout", [ + ("drop_newest", -1), + # A timeout, so a regression cannot hang the suite: under BLOCK with + # admission_timeout=-1 this record waits for room that can never exist. + ("block", 0.05), +]) +def test_a_record_the_queue_can_never_hold_is_too_large(sink, tmp_path, + overload, timeout): + """max_queue_bytes is an admission bound, not just a wait condition. + + The oracle (_BoundedQueue.put, pipeline.py) answers TOO_LARGE for a + record bigger than the queue's byte cap BEFORE it enters the wait loop, + and HostCapturePipeline.submit counts it as oversized -- exactly as it + does for a record past max_pack_bytes, which it checks first. The native + sink screened only max_pack_bytes, so a record between the two bounds + passed admission and reached a loop whose condition + (queue_bytes_ + n <= max_queue_bytes) is unsatisfiable even on an empty + queue: DROP_NEWEST called it dropped, BLOCK with a timeout called it + timed out, and BLOCK without one waited forever. With the shipped + defaults (16 MiB queue, 128 MiB pack) that is every record over 16 MiB. + """ + _open(sink, tmp_path / "spool", max_queue_records=256, + max_queue_bytes=1024, max_pack_bytes=1 << 20, + overload=overload, admission_timeout=timeout) + big = CaptureRecord( + metadata=_meta(0, dtype="uint8", shape=(2048,)), payload=bytes(2048) + ) + assert _submit(sink, big) == "too_large" + snapshot = sink.call(op="close", timeout=30)["snapshot"] + assert snapshot["oversized_records"] == 1 + assert snapshot["dropped_records"] == 0 + assert snapshot["timed_out_records"] == 0 + assert snapshot["admitted_records"] == 0 + assert snapshot["persisted_records"] == 0 + assert snapshot["packs_persisted"] == 0 + + def test_small_packs_split_by_size(sink, tmp_path): _open(sink, tmp_path / "spool", max_pack_bytes=4096) for i in range(8): From 139abca2e38e7411dc1f9922fe708756ef9d3b71 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:54:04 -0400 Subject: [PATCH 04/16] Initialise libcurl once per process, not per client object ClickHouseClient called curl_global_init in its constructor and curl_global_cleanup in its DESTRUCTOR. Both are process-global, and libcurl documents them as unsafe to call while another thread is inside the library -- SpoolUploader::UploadPending runs up to max_workers concurrent curl_easy_perform calls, so destroying a client on the main thread during an upload batch tore the library down underneath them, and the workers' next curl_easy_init raced on re-initialisation. S3Client made this worse by calling no init at all: it leaned on the implicit one inside curl_easy_init, which carries the same caveat. One process-lifetime init, behind a std::once_flag, in the shared common/ helper both clients already link (json.cpp is there, and the store's link line stays free of the catalog). No teardown: there is no correct moment for a library-wide cleanup in a process that still has threads, and the OS reclaims the allocations at exit. Neither client's public API changes -- ~ClickHouseClient stays declared, and is now defaulted. No deterministic red test exists for the race itself; a test that reproduces it sometimes is a flake. Pinned instead: the source invariant (no per-object teardown anywhere, one init in one place), which fails on the pre-fix tree, and the lifetime behaviour through the catalog driver (three clients constructed and destroyed in one process, the third still reaching libcurl). --- native/Makefile | 8 +- native/csrc/catalog/clickhouse_client.cpp | 9 +- native/csrc/common/curl_init.cpp | 14 ++ native/csrc/common/curl_init.h | 34 +++++ native/csrc/store/s3_client.cpp | 6 + tests/test_native_curl_global_lifetime.py | 156 ++++++++++++++++++++++ 6 files changed, 221 insertions(+), 6 deletions(-) create mode 100644 native/csrc/common/curl_init.cpp create mode 100644 native/csrc/common/curl_init.h create mode 100644 tests/test_native_curl_global_lifetime.py diff --git a/native/Makefile b/native/Makefile index 14256df1c..571210b8f 100644 --- a/native/Makefile +++ b/native/Makefile @@ -285,22 +285,22 @@ build/conformance_sign: csrc/store/s3_sign.cpp csrc/store/conformance_sign.cpp c # The link needs only -L; the runtime libcurl.so.4 ships with the OS. Override: # make CURL_INCDIR=/usr/include/x86_64-linux-gnu CURL_LIBDIR=/usr/lib/x86_64-linux-gnu ... CURL_LIBDIR ?= /tmp/opencode/sysroot/usr/lib/x86_64-linux-gnu -build/conformance_store: csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/store/s3_sign.h csrc/store/s3_client.h csrc/store/spool.h csrc/store/uploader.h csrc/common/json.cpp csrc/common/json.h +build/conformance_store: csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/store/s3_sign.h csrc/store/s3_client.h csrc/store/spool.h csrc/store/uploader.h csrc/common/json.cpp csrc/common/json.h csrc/common/curl_init.cpp csrc/common/curl_init.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ - csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/common/json.cpp \ + csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/common/json.cpp csrc/common/curl_init.cpp \ -Icsrc/store -Icsrc/common -I$(CURL_INCDIR) -L$(CURL_LIBDIR) -lcrypto -lcurl -lpthread # B1 catalog conformance driver: lease coordinator + version allocator over # ClickHouse's HTTP interface. No torch/pybind; needs libcurl like the store. -build/conformance_catalog: csrc/catalog/clickhouse_client.cpp csrc/catalog/lease_coordinator.cpp csrc/catalog/version_allocator.cpp csrc/catalog/catalog_writer.cpp csrc/catalog/pack_index.cpp csrc/catalog/indexer.cpp csrc/catalog/schema.cpp csrc/catalog/reader.cpp csrc/catalog/hydration.cpp csrc/catalog/conformance_catalog.cpp csrc/common/json.cpp csrc/pack/pack_builder.cpp csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/catalog/clickhouse_client.h csrc/catalog/lease_coordinator.h csrc/catalog/version_allocator.h csrc/catalog/catalog_writer.h csrc/catalog/sql_escape.h csrc/catalog/pack_index.h csrc/catalog/indexer.h csrc/catalog/schema.h csrc/catalog/reader.h csrc/catalog/hydration.h csrc/common/json.h csrc/store/s3_client.h +build/conformance_catalog: csrc/catalog/clickhouse_client.cpp csrc/catalog/lease_coordinator.cpp csrc/catalog/version_allocator.cpp csrc/catalog/catalog_writer.cpp csrc/catalog/pack_index.cpp csrc/catalog/indexer.cpp csrc/catalog/schema.cpp csrc/catalog/reader.cpp csrc/catalog/hydration.cpp csrc/catalog/conformance_catalog.cpp csrc/common/json.cpp csrc/pack/pack_builder.cpp csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/catalog/clickhouse_client.h csrc/catalog/lease_coordinator.h csrc/catalog/version_allocator.h csrc/catalog/catalog_writer.h csrc/catalog/sql_escape.h csrc/catalog/pack_index.h csrc/catalog/indexer.h csrc/catalog/schema.h csrc/catalog/reader.h csrc/catalog/hydration.h csrc/common/json.h csrc/store/s3_client.h csrc/common/curl_init.cpp csrc/common/curl_init.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ csrc/catalog/clickhouse_client.cpp csrc/catalog/lease_coordinator.cpp \ csrc/catalog/version_allocator.cpp csrc/catalog/catalog_writer.cpp \ csrc/catalog/pack_index.cpp csrc/catalog/indexer.cpp csrc/catalog/schema.cpp csrc/catalog/reader.cpp csrc/catalog/hydration.cpp \ csrc/catalog/conformance_catalog.cpp \ - csrc/common/json.cpp csrc/pack/pack_builder.cpp \ + csrc/common/json.cpp csrc/common/curl_init.cpp csrc/pack/pack_builder.cpp \ csrc/store/s3_sign.cpp csrc/store/s3_client.cpp \ -Icsrc/catalog -Icsrc/common -Icsrc/pack -Icsrc/store \ -I$(CURL_INCDIR) -L$(CURL_LIBDIR) -lcrypto -lcurl -lpthread diff --git a/native/csrc/catalog/clickhouse_client.cpp b/native/csrc/catalog/clickhouse_client.cpp index db6e5530e..c40d48e0f 100644 --- a/native/csrc/catalog/clickhouse_client.cpp +++ b/native/csrc/catalog/clickhouse_client.cpp @@ -2,6 +2,7 @@ #include +#include "../common/curl_init.h" #include "sql_escape.h" namespace dmi_catalog { @@ -109,10 +110,14 @@ std::map deciding_read() { ClickHouseClient::ClickHouseClient(std::string host, uint16_t port) : host_(std::move(host)), port_(port) { - curl_global_init(CURL_GLOBAL_DEFAULT); + // Process-lifetime, not per-object: the matching curl_global_cleanup used + // to run in the destructor below, which tore libcurl down for the WHOLE + // process while the uploader's worker threads were inside + // curl_easy_perform. See common/curl_init.h. + dmi_common::EnsureCurlGlobalInit(); } -ClickHouseClient::~ClickHouseClient() { curl_global_cleanup(); } +ClickHouseClient::~ClickHouseClient() = default; std::vector ClickHouseClient::execute( const std::string& query, const Params& params, diff --git a/native/csrc/common/curl_init.cpp b/native/csrc/common/curl_init.cpp new file mode 100644 index 000000000..a2f50fe9c --- /dev/null +++ b/native/csrc/common/curl_init.cpp @@ -0,0 +1,14 @@ +#include "curl_init.h" + +#include + +#include + +namespace dmi_common { + +void EnsureCurlGlobalInit() { + static std::once_flag once; + std::call_once(once, [] { curl_global_init(CURL_GLOBAL_DEFAULT); }); +} + +} // namespace dmi_common diff --git a/native/csrc/common/curl_init.h b/native/csrc/common/curl_init.h new file mode 100644 index 000000000..025f3e8bb --- /dev/null +++ b/native/csrc/common/curl_init.h @@ -0,0 +1,34 @@ +// One process-lifetime libcurl initialization, shared by every client. +// +// libcurl's global init/cleanup pair is PROCESS-global and documented as +// unsafe to call while any other thread is inside the library. A client +// that initialized in its constructor and cleaned up in its destructor +// therefore tore the library down under whatever else was using it: the +// uploader runs up to `max_workers` concurrent `curl_easy_perform` calls, +// and destroying an unrelated client on the main thread dropped the +// refcount to zero beneath them — after which those workers' implicit +// re-initialization inside `curl_easy_init` raced. +// +// So: initialize once, never clean up. The OS reclaims libcurl's +// allocations at exit, which is what a process-lifetime dependency is for; +// there is no correct moment for a library-wide teardown in a process that +// still has threads. Every client calls this from its constructor rather +// than relying on the implicit init inside `curl_easy_init`, which carries +// the same thread-safety caveat. +// +// Header-level curl dependency deliberately avoided: needs +// CURL_INCDIR on the compile line, and this declaration should be usable +// from anywhere in the tree. + +#ifndef DMI_COMMON_CURL_INIT_H_ +#define DMI_COMMON_CURL_INIT_H_ + +namespace dmi_common { + +// Runs curl_global_init(CURL_GLOBAL_DEFAULT) exactly once per process. +// Safe to call from any thread, any number of times. +void EnsureCurlGlobalInit(); + +} // namespace dmi_common + +#endif // DMI_COMMON_CURL_INIT_H_ diff --git a/native/csrc/store/s3_client.cpp b/native/csrc/store/s3_client.cpp index 55d119f5a..7024e5bf6 100644 --- a/native/csrc/store/s3_client.cpp +++ b/native/csrc/store/s3_client.cpp @@ -7,6 +7,7 @@ #include #include +#include "../common/curl_init.h" #include "s3_sign.h" namespace dmi_store { @@ -111,6 +112,11 @@ std::string XmlTag(const std::string& xml, const std::string& tag) { } // namespace S3Client::S3Client(S3Config config) : config_(std::move(config)) { + // Explicit, rather than leaning on the implicit init inside + // curl_easy_init: that implicit path carries libcurl's thread-safety + // caveat, and Exchange() runs on the uploader's worker threads. Once per + // process, and never torn down. See common/curl_init.h. + dmi_common::EnsureCurlGlobalInit(); // endpoint := scheme://host[:port]; bucket and key are appended per call // (path style, matching the Python store's addressing_style="path"). std::string rest = config_.endpoint; diff --git a/tests/test_native_curl_global_lifetime.py b/tests/test_native_curl_global_lifetime.py new file mode 100644 index 000000000..95dcceb5d --- /dev/null +++ b/tests/test_native_curl_global_lifetime.py @@ -0,0 +1,156 @@ +"""libcurl is initialised once per process and never torn down by an object. + +``curl_global_init``/``curl_global_cleanup`` are PROCESS-global, and libcurl +documents both as unsafe to call while another thread is inside the library. +``ClickHouseClient`` used to call the init in its constructor and the cleanup +in its **destructor**, so destroying one client dropped the library's refcount +for everything else in the process -- including ``SpoolUploader``'s worker +threads, which run up to ``max_workers`` concurrent ``curl_easy_perform`` +calls, and ``S3Client``, which called no init of its own and relied on the +implicit one inside ``curl_easy_init``. + +There is deliberately NO red test for the failure itself: reproducing it +needs a data race between a destructor on one thread and ``curl_easy_perform`` +on another, and a race is not something to pin in CI -- a test that reproduces +it sometimes is a flake, and one that reproduces it never is decorative. What +is pinned instead is the invariant the fix establishes, in the two places it +is observable without a race: + + * the SOURCE invariant -- no per-object teardown exists anywhere in the + native tree, and the one init lives in the one shared helper. This half + fails on the pre-fix tree, at clickhouse_client.cpp's destructor. + * the LIFETIME behaviour -- clients constructed and destroyed in sequence + leave libcurl usable for the next one. Single-threaded, libcurl's own + refcounting made this hold before the fix too, so it is a guard rather + than a reproduction; it is what would break first if the teardown came + back in a form the source scan did not recognise. + +Build: make -C native build/conformance_catalog +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +CSRC = REPO_ROOT / "native" / "csrc" +DRIVER = REPO_ROOT / "native" / "build" / "conformance_catalog" + +# The single home for the process-lifetime init. Anywhere else is a client +# initialising (and, historically, tearing down) libcurl on its own schedule. +INIT_HOME = CSRC / "common" / "curl_init.cpp" + +pytestmark = pytest.mark.cpu + + +# Comments are stripped before scanning: these files DOCUMENT why the global +# calls are absent, and a prose mention is not a call site. Crude but exact +# enough for this -- no string literal in the tree contains "//" or "/*". +_COMMENT = re.compile(r"//[^\n]*|/\*.*?\*/", re.DOTALL) + + +def _native_sources() -> list[Path]: + return sorted( + path + for suffix in ("*.cpp", "*.h", "*.cu") + for path in CSRC.rglob(suffix) + ) + + +def _code_of(path: Path) -> str: + return _COMMENT.sub("", path.read_text(encoding="utf-8")) + + +def test_no_native_source_tears_down_libcurl_globally(): + """``curl_global_cleanup`` has no correct caller in a threaded process.""" + offenders = [ + str(path.relative_to(REPO_ROOT)) + for path in _native_sources() + if "curl_global_cleanup" in _code_of(path) + ] + assert not offenders, ( + "curl_global_cleanup() is process-global and unsafe while another " + "thread is inside libcurl; the uploader's workers are. Found in: " + + ", ".join(offenders) + ) + + +def test_the_global_init_lives_in_exactly_one_place(): + """One init, behind one ``std::once_flag``, shared by every client.""" + callers = [ + str(path.relative_to(REPO_ROOT)) + for path in _native_sources() + if "curl_global_init" in _code_of(path) + ] + assert callers == [str(INIT_HOME.relative_to(REPO_ROOT))], ( + "curl_global_init() belongs to the shared once-only helper in " + f"{INIT_HOME.relative_to(REPO_ROOT)}; clients call " + "dmi_common::EnsureCurlGlobalInit(). Found in: " + ", ".join(callers) + ) + + assert "std::call_once" in _code_of(INIT_HOME), ( + "the shared init must be once-only: a second curl_global_init is " + "the same re-initialisation race the per-object version had" + ) + + +@pytest.mark.skipif( + not DRIVER.exists(), + reason="native/build/conformance_catalog is not built; run " + "`make -C native build/conformance_catalog`", +) +def test_libcurl_survives_clients_constructed_and_destroyed_in_sequence(): + """Three clients in one process: the third still reaches libcurl. + + Each ``open`` builds a fresh ``ClickHouseClient`` and destroys the one + before it, which is where the teardown used to run. The final ``execute`` + dials a port nothing listens on, so no server is needed: what it proves is + that libcurl answered at all -- a clean, named ``ClickHouseError`` from + ``curl_easy_perform`` rather than a crash or a hang inside a library some + earlier destructor had dismantled. + """ + request = { + "op": "open", + "database": "curl_lifetime_db", + "table_prefix": "curl_lifetime", + "lease_ttl_ns": 30_000_000_000, + "publish_timeout_ns": 1_000_000_000, + "clock_skew_ns": 1_000_000, + "allocation_attempts": 3, + } + lines = [json.dumps(request)] * 3 + lines.append(json.dumps({"op": "execute", "query": "SELECT 1"})) + + env = dict(os.environ) + env["DMI_CLICKHOUSE_HOST"] = "127.0.0.1" + # Port 1: privileged and unbound, so the connect is refused immediately + # rather than waiting out a timeout, and an operator's real ClickHouse on + # 8123 cannot turn this into a live test by accident. + env["DMI_CLICKHOUSE_HTTP_PORT"] = "1" + + proc = subprocess.run( + [str(DRIVER)], + input="\n".join(lines) + "\n", + capture_output=True, + text=True, + timeout=120, + env=env, + ) + answers = [ + json.loads(line) for line in proc.stdout.splitlines() if line.strip() + ] + assert proc.returncode == 0, proc.stderr + assert len(answers) == 4, proc.stdout + proc.stderr + for answer in answers[:3]: + assert answer == {"op": "open", "ok": True} + # libcurl was reached and reported for itself; the transport error is the + # closed port, not a dismantled library. + assert answers[3]["ok"] is False + assert answers[3]["error"] == "ClickHouseError" + assert answers[3]["message"].startswith("curl: ") From 5832e842817fa22e5ee97109f34037ac1ebb39b4 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:57:01 -0400 Subject: [PATCH 05/16] Refuse a capture sink config the backend can never read capture_sink_config is consumed in exactly one branch of create_record_runtime, guarded by `storage_backend == "capture"`, and storage_backend defaults to "auto". So a caller who set the sink config and left the backend alone passed the engine's type check at the boundary and then fell through with record_sink=None: no packs written at all, and nothing said so anywhere. Checked before changing it: nothing RESOLVES "auto" into "capture". "auto" is the pre-field behaviour -- _reject_a_sink_the_config_did_not _ask_for returns early on it and the default-writer branch compares the literal -- so the sink config is not merely unused under "auto", it is unreachable. And no test, example or doc in the tree passes capture_sink_config with a non-capture backend: the two tests that use it either set storage_backend="capture" or assign engine. _capture_sink_config directly, past MonitoringConfig entirely. So the naive check is the correct one, in __post_init__ where the field's own design note ("a mismatch becomes an error instead of a silent choice") already lives. This is a deliberate behaviour change: the combination now fails at config construction instead of silently capturing nothing. Narrow -- only a sink config that was actually passed is refused, and only under a backend that cannot read it. --- src/dmi/config.py | 26 ++++++++++++++++++++ tests/test_engine_runtime_api.py | 42 ++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/src/dmi/config.py b/src/dmi/config.py index c136293dd..0dfe407a7 100644 --- a/src/dmi/config.py +++ b/src/dmi/config.py @@ -111,3 +111,29 @@ def __post_init__(self) -> None: + ", ".join(repr(name) for name in backends) + f"; got {self.storage_backend!r}" ) + # ``capture_sink_config`` is read in exactly one place -- the + # engine's default-writer branch, which tests ``storage_backend == + # "capture"`` literally. Nothing RESOLVES "auto" into "capture": + # "auto" is the pre-field behaviour, and it reaches that branch as + # "auto" and falls straight through. So under any other backend a + # configured sink is not merely unused, it is unreachable, and the + # engine's type check at the boundary passes a config that then + # writes no packs at all -- the silent no-op this field's own design + # note ("a mismatch becomes an error instead of a silent choice") + # exists to prevent. + # + # Deliberate behaviour change: a caller who passes the sink config + # with a non-capture backend gets a startup error where they used to + # get silence. That is the trade the note asks for -- the alternative + # is a run that captures nothing and says so nowhere. + if self.capture_sink_config is not None and ( + self.storage_backend != "capture" + ): + raise ValueError( + "capture_sink_config configures the capture backend's " + "default pack writer and is read only when " + "storage_backend='capture'; got storage_backend=" + f"{self.storage_backend!r}, under which it would be silently " + "ignored and nothing would be written. Set " + "storage_backend='capture', or drop capture_sink_config" + ) diff --git a/tests/test_engine_runtime_api.py b/tests/test_engine_runtime_api.py index e42069d1b..84c5fb09e 100644 --- a/tests/test_engine_runtime_api.py +++ b/tests/test_engine_runtime_api.py @@ -553,6 +553,48 @@ def test_storage_backend_rejects_an_unknown_name(): MonitoringConfig(storage_backend="object-store") +@pytest.mark.parametrize("backend", ["auto", "native", "none"]) +def test_a_sink_config_under_a_backend_that_cannot_use_it_is_refused( + backend: str, +): + """The combination that used to write nothing and say nothing. + + ``capture_sink_config`` is consumed in exactly one branch, guarded by + ``storage_backend == "capture"``. Under any other backend -- the DEFAULT + "auto" included, which is never resolved into "capture" anywhere -- the + config passed the engine's type check and then fell through + ``create_record_runtime`` with ``record_sink=None``: no packs written, no + diagnostic. A startup error instead, naming both fields. + """ + from dmi.config import MonitoringConfig + from dmi.storage.capture.native_sink import NativeSinkConfig + + with pytest.raises(ValueError) as caught: + MonitoringConfig( + storage_backend=backend, + capture_sink_config=NativeSinkConfig(spool_root="/tmp/unused"), + ) + message = str(caught.value) + assert "capture_sink_config" in message + assert "storage_backend" in message + assert repr(backend) in message + + +def test_the_capture_backend_still_takes_its_sink_config(): + """The control: the one combination that works keeps working.""" + from dmi.config import MonitoringConfig + from dmi.storage.capture.native_sink import NativeSinkConfig + + sink_config = NativeSinkConfig(spool_root="/tmp/unused") + config = MonitoringConfig( + storage_backend="capture", capture_sink_config=sink_config + ) + assert config.capture_sink_config is sink_config + # And a bare config is untouched: the refusal is about a sink config that + # was actually passed, not about the backend on its own. + assert MonitoringConfig(storage_backend="none").capture_sink_config is None + + def test_native_backend_requires_a_host_engine(): """Declaring the C++ path without one is a configuration error, not a silently transport-only engine.""" From 41313f36781864d02b5d5e2a5f2682b3f5dbcdd1 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 12:58:00 -0400 Subject: [PATCH 06/16] Select the live CI suite by marker, not by filename The live job ran `pytest tests/*_live.py -m "clickhouse and manual and not garage"`, so the naming convention was load-bearing: a new `manual`/`clickhouse` test written under any other filename is silently NOT COLLECTED, and the "Fail if any live test was skipped" gate below cannot catch it -- uncollected is not skipped, so the job stays green over coverage that never ran. That is the same silent-green the gate exists to end. Measured before widening, as the risk is pulling in files that are expensive or need another service: the marker expression alone over `tests/` selects 199 tests in 9 files, identical to what the glob selected, and every one of those files is already `*_live.py`. Collection of the whole tree takes ~2s with no errors, and the cpu job already collects it the same way (`pytest -m cpu`, no path) on the same image and dependency set. Verified with a temporary `manual`/`clickhouse` test named outside the convention (`tests/test_..._e2e.py`): the glob collected 199 and missed it; marker selection collected 200 and named it. Temporary file removed. --- .github/workflows/python-checks.yml | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml index dceb67c8a..b34c8b512 100644 --- a/.github/workflows/python-checks.yml +++ b/.github/workflows/python-checks.yml @@ -248,8 +248,20 @@ jobs: # `python -m pytest`, matching the Makefile. `not garage` drops the one # suite that also needs a Garage S3 endpoint; everything else under # these markers needs nothing but the server above. + # + # The whole `tests` tree, selected BY MARKER -- not `tests/*_live.py`. + # A filename glob makes the naming convention load-bearing: a new + # `manual`/`clickhouse` test written as `test_..._e2e.py` is not + # collected, and the skip gate below cannot catch it, because + # uncollected is not skipped -- it is the exact silent-green this job + # exists to end. Measured before widening: the marker expression alone + # over `tests/` selects 199 tests in 9 files, which is byte for byte + # what the glob selected, so nothing expensive or needing another + # service is pulled in. Collecting the tree costs ~2s and is what the + # cpu job above already does (`pytest -m cpu`, no path), on the same + # image and the same dependency set, so the imports are proven. run: | - python -m pytest tests/*_live.py -m "clickhouse and manual and not garage" \ + python -m pytest tests -m "clickhouse and manual and not garage" \ -q --junit-xml=live-results.xml - name: Fail if any live test was skipped From 6eada78252a16929c329a1179b662081bcf9ba02 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 13:23:34 -0400 Subject: [PATCH 07/16] Run the live suite by glob again; the tree walk imports what it cannot build 41313f3 widened the live job's selection from `tests/*_live.py` to the whole `tests` tree so a `manual`/`clickhouse` test named outside the convention could not hide. The goal was right, the mechanism was not: walking the tree makes pytest IMPORT every module, and this job cannot import all of them. It builds only the four conformance drivers. It does NOT build build/_dmi_native_sink, which the cpu job gets from `make -C native cpu-goals` (CPU_ONLY_GOALS includes it). tests/test_native_adapter_torch.py and tests/test_native_rollback.py import that .so at module scope and so take a module-level `pytest.skip` when it is absent -- they are the two permanent exceptions tests/test_ci_guard_skips.py already records as unable to convert to a per-test skipif. A collection skip is written to the JUnit as ``, and the "Fail if any live test was skipped" gate then failed the job over two files this suite never wanted: 201 collected, 2 skipped, across 1 suite(s) skipped: tests.test_native_adapter_torch: collection skipped skipped: tests.test_native_rollback: collection skipped The gate behaved exactly as designed and is untouched here -- byte for byte what it was before 41313f3. What changes is that the RUN goes back to importing only the files it needs. Note the local measurement that cleared 41313f3 was not wrong about torch: BOTH jobs install torch. The difference is the native build step, which is why collecting on a dev box with the .so present showed 199/199 and no skips. Reproduced here by hiding _dmi_native_sink*.so with torch fully installed: the tree walk emits both skips at tests/..._torch.py:34 and tests/..._rollback.py:33, the glob emits none, and the glob's JUnit contains no "collection skipped" at all. The hole 41313f3 set out to close is real and stays closed -- by a separate explicit check, in the next commit, rather than by widening this run. --- .github/workflows/python-checks.yml | 31 ++++++++++++++++++----------- 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml index b34c8b512..8710825fa 100644 --- a/.github/workflows/python-checks.yml +++ b/.github/workflows/python-checks.yml @@ -249,19 +249,26 @@ jobs: # suite that also needs a Garage S3 endpoint; everything else under # these markers needs nothing but the server above. # - # The whole `tests` tree, selected BY MARKER -- not `tests/*_live.py`. - # A filename glob makes the naming convention load-bearing: a new - # `manual`/`clickhouse` test written as `test_..._e2e.py` is not - # collected, and the skip gate below cannot catch it, because - # uncollected is not skipped -- it is the exact silent-green this job - # exists to end. Measured before widening: the marker expression alone - # over `tests/` selects 199 tests in 9 files, which is byte for byte - # what the glob selected, so nothing expensive or needing another - # service is pulled in. Collecting the tree costs ~2s and is what the - # cpu job above already does (`pytest -m cpu`, no path), on the same - # image and the same dependency set, so the imports are proven. + # The GLOB, not the whole `tests` tree. Walking the tree here makes + # pytest IMPORT every test module, including ones this job never + # builds for: it builds the four conformance drivers above, NOT + # build/_dmi_native_sink, and tests/test_native_adapter_torch.py and + # tests/test_native_rollback.py take a module-level `pytest.skip` + # when that .so is absent (they import it at module scope, so they + # cannot convert to a per-test skipif -- tests/test_ci_guard_skips.py + # records exactly those two as the permanent exceptions). A + # collection skip lands in the JUnit as ``, and the gate + # below then fails the job -- correctly, by its own doctrine -- over + # two files this suite does not even want. Measured in CI: `201 + # collected, 2 skipped`. + # + # The hole a glob leaves -- a `manual`/`clickhouse` test named + # outside the convention is silently UNCOLLECTED, and this gate + # cannot catch it because uncollected is not skipped -- is closed by + # a separate check rather than by widening the run. See the next + # commit. run: | - python -m pytest tests -m "clickhouse and manual and not garage" \ + python -m pytest tests/*_live.py -m "clickhouse and manual and not garage" \ -q --junit-xml=live-results.xml - name: Fail if any live test was skipped From d0859c983acb2ccc8645c26850176c20b7481c2e Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 11 Sep 2026 13:24:03 -0400 Subject: [PATCH 08/16] Fail the cpu job when a live-marked test hides outside tests/*_live.py The previous commit put the live run back on `tests/*_live.py`, which leaves the naming convention load-bearing: a `manual`/`clickhouse` test written as `test_..._e2e.py` is never collected by that job, and its skip gate cannot see the gap, because uncollected is not skipped. That is the real hole 41313f3 found. This closes it without making the live job import anything new. The check collects the marker expression over the whole `tests` tree and compares the collected node ids' FILES against what `tests/*_live.py` matches. Anything in the first set and not the second fails the job and names the file. It runs in the CPU job, not the live job, and the placement is the whole point. The tree walk needs an environment that can import the tree, and only this job has one: `make -C native cpu-goals` builds CPU_ONLY_GOALS, including build/_dmi_native_sink, so the two suites that import that .so at module scope collect normally here. The live job builds four conformance drivers and cannot. This job also already collects the same tree for `make check` (`pytest -m cpu`, no path), so the walk adds no import that is not already happening. Comparing FILES, not node ids, is what makes an unimportable module a no-op: it contributes no node ids, so it simply is not in the set. Nothing special-cases "collection skipped", and nothing here can be widened into ignoring a skip. An unimportable module in this job is still a failure -- the cpu skip gate above catches it -- so the two checks compose rather than overlap. Collection errors are handled deliberately rather than with `|| true`, which would silently disable the gate. `--collect-only` exits 2 on a collection error and 5 on an empty selection; both are failures that name themselves. The gate also refuses to pass on an empty collected set, so a marker expression that stops matching cannot make every comparison vacuously true. Verified both directions on the current tree: - with a temporary `manual`/`clickhouse` test added as tests/test_capture_catalog_probe_e2e.py, the check exits 1 with "outside the glob: tests/test_capture_catalog_probe_e2e.py", while the live job's glob collects 199 and never sees the file. Temporary file removed. - with the tree as it stands, the check exits 0: "9 file(s) hold 'clickhouse and manual and not garage' tests; 9 matched by tests/*_live.py". - with _dmi_native_sink*.so hidden to mimic the live job, the check still exits 0 and still reports 9/9 -- the two collection-skipped modules are invisible to it, as designed. - with a deliberately unimportable module in tests/, the check exits 1 with "collecting ... exited 2; this gate cannot run, which is indistinguishable from it passing". The live job's skip gate is unchanged and byte-identical to pre-41313f3. --- .github/workflows/python-checks.yml | 93 +++++++++++++++++++++++++++-- 1 file changed, 88 insertions(+), 5 deletions(-) diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml index 8710825fa..45c4d38b2 100644 --- a/.github/workflows/python-checks.yml +++ b/.github/workflows/python-checks.yml @@ -127,6 +127,87 @@ jobs: "that is not absent hardware; a skip is not a pass") EOF + - name: Fail if a live-marked test lives outside tests/*_live.py + # The live job runs `pytest tests/*_live.py -m "clickhouse and manual + # and not garage"`, which makes the FILENAME load-bearing: a new test + # under those markers written as `test_..._e2e.py` is never collected + # there, and that job's skip gate cannot catch it, because uncollected + # is not skipped. This step is that missing half -- it fails when a + # test the live job is meant to run is sitting where the glob cannot + # see it. + # + # It runs HERE, in the cpu job, and not in the live job, because it + # has to collect the WHOLE tree and only this job's environment can + # import the whole tree: `make -C native cpu-goals` above builds + # CPU_ONLY_GOALS, which includes build/_dmi_native_sink, while the + # live job builds only the four conformance drivers and so hits a + # module-level skip in the two suites that import that .so. Doing the + # tree walk in the live job is precisely the regression this replaces. + # This job already collects the same tree (`make check` -> `pytest -m + # cpu`, no path), so the walk costs seconds and adds no new imports. + # + # A module that cannot be imported contributes NO node ids and is + # therefore invisible here -- ignored by construction rather than by a + # special case on "collection skipped". That is deliberate: an + # unimportable module in THIS job is already a failure, caught by the + # skip gate above, so the two checks compose instead of overlapping. + run: | + python - <<'EOF' + import glob + import os + import subprocess + import sys + + MARKERS = "clickhouse and manual and not garage" + LIVE_GLOB = "tests/*_live.py" + + proc = subprocess.run( + [sys.executable, "-m", "pytest", "tests", "-m", MARKERS, + "--collect-only", "-q", "-p", "no:cacheprovider"], + capture_output=True, text=True, + ) + # NOT `|| true`. `--collect-only` still exits non-zero on a + # collection ERROR (exit 2) and on an empty selection (exit 5), and + # swallowing either turns this gate into a no-op that reports green + # -- the same silent-green it exists to end. Both are failures, and + # both name themselves. + if proc.returncode != 0: + print(proc.stdout) + print(proc.stderr, file=sys.stderr) + sys.exit(f"collecting '{MARKERS}' over tests/ exited " + f"{proc.returncode}; this gate cannot run, which is " + "indistinguishable from it passing") + + collected = set() + for line in proc.stdout.splitlines(): + node = line.strip() + if "::" not in node: + continue + path = node.split("::", 1)[0] + if path.endswith(".py") and os.path.exists(path): + collected.add(os.path.normpath(path)) + + # Self-check: an expression that matches nothing would make every + # comparison below vacuously true. + if not collected: + sys.exit(f"'{MARKERS}' selected no tests at all; the live job " + "would run nothing and this gate would pass vacuously") + + covered = {os.path.normpath(p) for p in glob.glob(LIVE_GLOB)} + outside = sorted(collected - covered) + print(f"{len(collected)} file(s) hold '{MARKERS}' tests; " + f"{len(collected & covered)} matched by {LIVE_GLOB}") + if outside: + for path in outside: + print(f" outside the glob: {path}") + sys.exit( + f"{len(outside)} file(s) hold '{MARKERS}' tests but do not " + f"match {LIVE_GLOB}, so the clickhouse-live job never " + "collects them and its skip gate cannot see the gap. Rename " + "them to *_live.py." + ) + EOF + clickhouse-live: # The `manual`-marked suites, which `make check` cannot run: `addopts` in # pyproject.toml carries `-m 'not manual'`, so every test that needs a real @@ -262,11 +343,13 @@ jobs: # two files this suite does not even want. Measured in CI: `201 # collected, 2 skipped`. # - # The hole a glob leaves -- a `manual`/`clickhouse` test named - # outside the convention is silently UNCOLLECTED, and this gate - # cannot catch it because uncollected is not skipped -- is closed by - # a separate check rather than by widening the run. See the next - # commit. + # So the RUN stays scoped to the files it needs to import. The hole + # a glob leaves -- a `manual`/`clickhouse` test named outside the + # convention is silently UNCOLLECTED, and this gate cannot catch it + # because uncollected is not skipped -- is closed instead by the + # "Fail if a live-marked test lives outside tests/*_live.py" step in + # the cpu job, which collects the whole tree in an environment that + # can actually import it. run: | python -m pytest tests/*_live.py -m "clickhouse and manual and not garage" \ -q --junit-xml=live-results.xml From e7cd00d2ae701c43327f7260bfc41699b22b05c1 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 11:58:38 -0400 Subject: [PATCH 09/16] Pin the kBlock admission timeout natively by wedging the pipeline The A3b rework's admission screen made the old 64-byte-record/63-byte-cap timeout test unreachable (that record is now TOO_LARGE before the wait loop), and the replacement suite pinned timed_out_records == 0 everywhere: the deadline wait in PackSink::Submit shipped with no native coverage at all, so a hang, a never-times-out, or a miscount there would go green. The claimed reason -- "every alternative is a race against the packer thread" -- was wrong. The Spool::SetStageHookForTesting seam that test_spool_reservations.cpp already uses can park a stager inside Stage(); with stage_queue_packs = 1 the stage queue then fills, the packer blocks in SealForStager, and the admission queue (one record) fills behind it. A further kBlock submit waits on space that provably cannot free until the hook is released, so it must return kTimedOut and count timed_out_records -- no sleep decides the outcome, only whether the wedge is released. Releasing it drains everything admitted and closes cleanly, which the test also pins. tests/native/test_pack_sink_timeout.cpp is compiled and run by tests/test_native_pack_sink_timeout.py, the same mechanism as the spool reservation test; pack_sink.h gains a SpoolForTesting() accessor so the test can reach the sink's spool. Verified red first: stubbing the ++timed_out_records increment fails both snapshot checks. --- native/csrc/sink/pack_sink.h | 6 + tests/native/test_pack_sink_timeout.cpp | 196 ++++++++++++++++++++++++ tests/test_native_pack_sink.py | 7 +- tests/test_native_pack_sink_timeout.py | 84 ++++++++++ 4 files changed, 291 insertions(+), 2 deletions(-) create mode 100644 tests/native/test_pack_sink_timeout.cpp create mode 100644 tests/test_native_pack_sink_timeout.py diff --git a/native/csrc/sink/pack_sink.h b/native/csrc/sink/pack_sink.h index 1a6be5b9c..f9c48bb88 100644 --- a/native/csrc/sink/pack_sink.h +++ b/native/csrc/sink/pack_sink.h @@ -153,6 +153,12 @@ class PackSink { // Latched async failure, if any (for RecordSink::rethrow_if_failed). std::string LastError() const; + // Test seam: the sink's spool, so a test can park a stager inside + // Spool::Stage through Spool::SetStageHookForTesting and wedge the + // pipeline deterministically (tests/native/test_pack_sink_timeout.cpp). + // Valid only after Start(); stagers call Stage() on it concurrently. + dmi_store::Spool& SpoolForTesting() { return spool_; } + private: using Item = std::variant>; diff --git a/tests/native/test_pack_sink_timeout.cpp b/tests/native/test_pack_sink_timeout.cpp new file mode 100644 index 000000000..795e7c2b8 --- /dev/null +++ b/tests/native/test_pack_sink_timeout.cpp @@ -0,0 +1,196 @@ +// The kBlock admission timeout, exercised without racing the packer. +// +// The refusal a timeout guards is "the queue is full and nobody is +// draining it". Filling the queue with a live packer thread is a race: +// the packer pops as fast as we push. So the pipeline is wedged bottom-up +// instead, through the same seam test_spool_reservations.cpp uses: +// +// Spool::SetStageHookForTesting blocks the STAGER inside Stage() (1) +// the stage queue (stage_queue_packs = 1) then fills with one pack (2) +// the packer seals its next pack and blocks in SealForStager (3) +// the admission queue (max_queue_records = 1) fills behind it (4) +// +// After (4) no admission can succeed until the hook is released: the only +// thread that could free queue space is the packer, the packer can only +// move once the stager pops, and the stager is parked in the hook. A +// further kBlock submit with a finite admission_timeout_s must therefore +// return kTimedOut and count timed_out_records — deterministically, with +// no sleep-based guessing about where the packer is. +// +// Every earlier submit is admitted on its FIRST bounds check (the test +// waits for the pipeline to visibly quiesce between submits), so the +// finite timeout never touches them. +// +// Built and run by tests/test_native_pack_sink_timeout.py. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "sink/pack_sink.h" + +namespace fs = std::filesystem; + +namespace { + +int g_failures = 0; + +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + std::cerr << __FILE__ << ":" << __LINE__ << ": CHECK failed: " #cond \ + << "\n"; \ + ++g_failures; \ + } \ + } while (0) + +dmi_pack::RecordMetadata Metadata(int n) { + dmi_pack::RecordMetadata m; + m.capture_id = "capture-" + std::to_string(n); + m.tenant_id = "tenant-a"; + m.experiment_id = "exp-1"; + m.run_id = "run-1"; + m.session_id = "session-1"; + m.request_id = "req-1"; + m.sequence_id = "seq-1"; + m.model_id = "model-1"; + m.model_revision = "rev-1"; + m.capture_policy_version = "policy-1"; + m.hook_name = "hook.0"; + m.dtype = "float32"; + m.shape = {4}; + m.captured_at_ns = 1700000000000000000ull + n; + return m; +} + +dmi_sink::Admission SubmitRecord(dmi_sink::PackSink& sink, int n) { + const uint8_t payload[16] = {static_cast(n)}; + return sink.Submit(Metadata(n), payload, sizeof(payload)); +} + +// Poll the snapshot until `done` holds. Bounded patience, not a race: the +// state waited for is reached by threads that are runnable, and failing to +// reach it within 30s is itself a pipeline bug worth failing on. +bool WaitForSnapshot( + dmi_sink::PackSink& sink, + const std::function& done) { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(30); + while (std::chrono::steady_clock::now() < deadline) { + if (done(sink.Snapshot())) return true; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return false; +} + +void TestBlockedPipelineTimesOutAndCountsIt() { + const char* base = std::getenv("SPOOL_TEST_ROOT"); + const std::string root = + std::string(base != nullptr ? base : "/tmp") + "/sink-timeout"; + fs::remove_all(root); + + dmi_sink::SinkConfig config; + config.spool_root = root; + config.num_workers = 1; + config.max_queue_records = 1; // (4) one record fills admission + config.max_pack_records = 1; // every record seals into its own pack + config.stage_queue_packs = 1; // (2) one sealed pack fills the stage + config.max_linger_ns = 3600ull * 1000 * 1000 * 1000; // linger never fires + config.overload = dmi_sink::Overload::kBlock; + config.admission_timeout_s = 0.2; + + dmi_sink::PackSink sink(config); + std::string start_error = sink.Start(); + CHECK(start_error.empty()); + if (!start_error.empty()) { + std::cerr << "start failed: " << start_error << "\n"; + return; + } + + std::mutex mutex; + std::condition_variable cv; + bool in_hook = false, release = false; + sink.SpoolForTesting().SetStageHookForTesting([&] { + std::unique_lock lock(mutex); + in_hook = true; + cv.notify_all(); + cv.wait(lock, [&] { return release; }); + }); + + // R1 seals into pack 1; the stager pops it and parks in the hook (1). + CHECK(SubmitRecord(sink, 1) == dmi_sink::Admission::kAccepted); + { + std::unique_lock lock(mutex); + cv.wait(lock, [&] { return in_hook; }); + } + CHECK(WaitForSnapshot(sink, [](const dmi_sink::SinkSnapshot& s) { + return s.queue_records == 0 && s.stage_packs == 0; + })); + + // R2 seals into pack 2, which fills the stage queue (2). + CHECK(SubmitRecord(sink, 2) == dmi_sink::Admission::kAccepted); + CHECK(WaitForSnapshot(sink, [](const dmi_sink::SinkSnapshot& s) { + return s.queue_records == 0 && s.stage_packs == 1; + })); + + // R3 seals into pack 3; the packer blocks handing it over (3). Once the + // packer has popped R3 it cannot pop again before pushing pack 3, and it + // cannot push until the wedged stager pops — so waiting for the pop is + // enough; no guess about scheduler timing decides the outcome. + CHECK(SubmitRecord(sink, 3) == dmi_sink::Admission::kAccepted); + CHECK(WaitForSnapshot(sink, [](const dmi_sink::SinkSnapshot& s) { + return s.queue_records == 0; + })); + + // R4 fills the admission queue behind the wedged packer (4). + CHECK(SubmitRecord(sink, 4) == dmi_sink::Admission::kAccepted); + + // R5 must wait, and the wait can never be satisfied: kTimedOut. + CHECK(SubmitRecord(sink, 5) == dmi_sink::Admission::kTimedOut); + { + const dmi_sink::SinkSnapshot snap = sink.Snapshot(); + CHECK(snap.timed_out_records == 1); + CHECK(snap.submitted_records == 5); + CHECK(snap.admitted_records == 4); + CHECK(snap.dropped_records == 0); + CHECK(snap.oversized_records == 0); + CHECK(snap.queue_records == 1); // R4 still waiting on the packer + } + + // Release the stager and drain: every ADMITTED record still lands, the + // timed-out one stays counted, and close neither hangs nor latches. + { + std::lock_guard lock(mutex); + release = true; + cv.notify_all(); + } + std::string close_error; + const dmi_sink::SinkSnapshot final_snap = sink.Close(-1.0, &close_error); + CHECK(close_error.empty()); + if (!close_error.empty()) std::cerr << close_error << "\n"; + CHECK(final_snap.timed_out_records == 1); + CHECK(final_snap.admitted_records == 4); + CHECK(final_snap.persisted_records == 4); + CHECK(final_snap.packs_persisted == 4); + CHECK(final_snap.failures == 0); + CHECK(final_snap.queue_records == 0); + CHECK(final_snap.stage_packs == 0); +} + +} // namespace + +int main() { + TestBlockedPipelineTimesOutAndCountsIt(); + if (g_failures != 0) { + std::cerr << g_failures << " check(s) failed\n"; + return 1; + } + std::cout << "ok\n"; + return 0; +} diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py index fd7df1a39..3e23e24ba 100644 --- a/tests/test_native_pack_sink.py +++ b/tests/test_native_pack_sink.py @@ -244,8 +244,11 @@ def test_block_with_timeout_admits_a_record_that_exactly_fits(sink, tmp_path): The timeout path itself needs a queue that is full of records a consumer has not drained yet. The oracle pins it by stalling its sink (tests/_faults.BlockingPackSink, test_capture_pipeline.py); this - driver has no equivalent stall, and every alternative here would be a - race against the packer thread. + driver has no equivalent stall, so the native tier pins it in C++ + instead: tests/native/test_pack_sink_timeout.cpp (run by + test_native_pack_sink_timeout.py) parks a stager inside Spool::Stage + through Spool::SetStageHookForTesting and wedges the whole pipeline, + so the kBlock wait deterministically times out and counts. """ _open(sink, tmp_path / "spool", max_queue_records=256, max_queue_bytes=64, overload="block", admission_timeout=0.05) diff --git a/tests/test_native_pack_sink_timeout.py b/tests/test_native_pack_sink_timeout.py new file mode 100644 index 000000000..1cae7f12f --- /dev/null +++ b/tests/test_native_pack_sink_timeout.py @@ -0,0 +1,84 @@ +"""The kBlock admission timeout in the native sink, pinned deterministically. + +Compiles and runs tests/native/test_pack_sink_timeout.cpp against the real +pack_sink.cpp. The C++ test wedges the whole pipeline bottom-up — the +stager parked inside Spool::Stage through Spool::SetStageHookForTesting, +the stage queue full behind it, the packer blocked in SealForStager, the +admission queue full behind the packer — so a further kBlock submit waits +on space that provably cannot free, times out, and must return kTimedOut +with timed_out_records == 1. No admission along the way ever depends on +where the packer thread happens to be. + +This is the native-tier coverage of the deadline wait loop in +PackSink::Submit; the conformance-driver suite (test_native_pack_sink.py) +covers the admission bounds around it. + +Needs a C++17 compiler and libcrypto (the same as conformance_sink). +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +from pathlib import Path + +import pytest + + +@pytest.mark.cpu +def test_blocked_pipeline_times_out_and_counts_it(tmp_path): + compiler = shutil.which("g++") or shutil.which("c++") + if compiler is None: + pytest.skip("a C++17 compiler is required") + + root = Path(__file__).resolve().parents[1] + csrc = root / "native" / "csrc" + source = root / "tests" / "native" / "test_pack_sink_timeout.cpp" + executable = tmp_path / "test_pack_sink_timeout" + extra: list[str] = [] + # Homebrew OpenSSL is not on the default search path on macOS; Linux + # hosts (CI) find libcrypto without help. + for candidate in ("/opt/homebrew/opt/openssl@3", "/usr/local/opt/openssl@3"): + if Path(candidate, "include", "openssl", "sha.h").exists(): + extra += [f"-I{candidate}/include", f"-L{candidate}/lib"] + break + compile_result = subprocess.run( + [ + compiler, + "-std=c++17", + "-O0", + "-pthread", + f"-I{csrc}", + *extra, + str(source), + str(csrc / "sink" / "pack_sink.cpp"), + str(csrc / "sink" / "object_key.cpp"), + str(csrc / "pack" / "pack_builder.cpp"), + str(csrc / "store" / "spool.cpp"), + "-o", + str(executable), + "-lcrypto", + ], + capture_output=True, + text=True, + check=False, + ) + if compile_result.returncode != 0 and "openssl" in ( + compile_result.stdout + compile_result.stderr + ): + pytest.skip("libcrypto headers are unavailable: " + + compile_result.stderr[-400:]) + assert compile_result.returncode == 0, ( + compile_result.stdout + compile_result.stderr + ) + + run_result = subprocess.run( + [str(executable)], + capture_output=True, + text=True, + check=False, + env={**os.environ, "SPOOL_TEST_ROOT": str(tmp_path)}, + timeout=120, + ) + assert run_result.returncode == 0, run_result.stdout + run_result.stderr From eb97cb16eb84b9c36c27bd098d36130a733a8c25 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:09:11 -0400 Subject: [PATCH 10/16] Link curl_init.cpp into both extensions, with the curl flags it needs EnsureCurlGlobalInit() added a symbol the normal extension builds did not link: only the conformance recipes compiled common/curl_init.cpp, so _native_backend/_host_backend carried an undefined symbol. The helper and its header are now in SRCS, HOST_SRCS and HEADERS, and the extension flags gain curl's include/link dirs (guarded on the dev tree existing, so a system curl-dev in the default search path needs nothing) plus -lcurl. Verified: make -n all shows csrc/common/curl_init.cpp in the compile and link commands of the extension, build/host/common/curl_init.o compiles, and the conformance drivers still build. (A full extension link cannot run on this host: the branch's -std=c++17 predates the installed torch 2.14 headers, which require C++20 -- unrelated to this change.) --- native/Makefile | 42 ++++++++++++++++++++++++++++++++---------- 1 file changed, 32 insertions(+), 10 deletions(-) diff --git a/native/Makefile b/native/Makefile index 571210b8f..73a76aa2d 100644 --- a/native/Makefile +++ b/native/Makefile @@ -11,14 +11,17 @@ SRCS := csrc/bindings.cpp \ csrc/clickhouse_record_sink.cpp \ csrc/reference_python_capture_sink.cpp \ csrc/clickhouse_client.cpp \ + csrc/common/curl_init.cpp \ csrc/ring/drain_thread.cpp \ csrc/ring/p2p_thread.cpp \ csrc/ring/record_consumer.cpp \ csrc/ring/ring_torch_op.cpp HOST_SRCS := csrc/bindings.cpp csrc/clickhouse_client.cpp \ + csrc/common/curl_init.cpp \ csrc/pack/pack_builder.cpp HEADERS := csrc/dmx_host_utils.h \ csrc/clickhouse_client.h csrc/clickhouse_record_sink.h \ + csrc/common/curl_init.h \ csrc/reference_python_capture_sink.h \ csrc/dmx_host_engine.h csrc/record_schema.h \ csrc/ring/ring_config.h csrc/ring/ring_state.h csrc/ring/ring_alloc.h \ @@ -122,6 +125,16 @@ PYTHON_INCLUDE_FLAGS := $(shell $(PYTHON) -c "import sysconfig; paths = sysconfi # Pybind11 ABI flags PYBIND_FLAGS := $(shell $(PYTHON) -c "import pybind11; print(' '.join([f'-DPYBIND11_COMPILER_TYPE=\\\"{pybind11.get_compiler_type()}\\\"', f'-DPYBIND11_STDLIB=\\\"{pybind11.get_stdlib()}\\\"', f'-DPYBIND11_BUILD_ABI=\\\"{pybind11.get_build_abi()}\\\"']))" 2>/dev/null || echo "") +# libcurl headers/dev symlink for the extension builds: the shared +# process-lifetime init (common/curl_init.cpp) is compiled into both. The +# runtime libcurl.so.4 ships with the OS; the dev symlink comes from +# CURL_INCDIR/CURL_LIBDIR (a dev .deb extract by default, no root needed): +# apt-get download libcurl4-openssl-dev && dpkg-deb -x +# Override on machines with system curl-dev: +# make CURL_INCDIR=/usr/include/x86_64-linux-gnu CURL_LIBDIR=/usr/lib/x86_64-linux-gnu ... +CURL_INCDIR ?= /tmp/opencode/sysroot/usr/include/x86_64-linux-gnu +CURL_LIBDIR ?= /tmp/opencode/sysroot/usr/lib/x86_64-linux-gnu + CXXFLAGS := -std=c++17 -O3 -fPIC -Wall -Wextra -Wno-unused-parameter -Wno-unused-variable \ -DTORCH_API_INCLUDE_EXTENSION_H -DTORCH_EXTENSION_NAME=$(TARGET) \ $(ABI_FLAG) $(PYBIND_FLAGS) \ @@ -168,6 +181,20 @@ HOST_LDFLAGS += -L$(CLICKHOUSE_CONTRIB)/lz4/lz4 -llz4 \ -L$(CLICKHOUSE_CONTRIB)/absl/absl -labsl_int128 \ -L$(CLICKHOUSE_CONTRIB)/zstd/zstd -lzstdstatic +# ---- libcurl (common/curl_init.cpp is in both extension source lists) ------ +# The include/link dirs are only added when the configured dev tree exists: +# a system with curl-dev in the default search path needs neither. +ifneq ($(wildcard $(CURL_INCDIR)/curl/curl.h),) +CXXFLAGS += -I$(CURL_INCDIR) +HOST_CXXFLAGS += -I$(CURL_INCDIR) +endif +ifneq ($(wildcard $(CURL_LIBDIR)/libcurl.so),) +LDFLAGS += -L$(CURL_LIBDIR) +HOST_LDFLAGS += -L$(CURL_LIBDIR) +endif +LDFLAGS += -lcurl +HOST_LDFLAGS += -lcurl + # ---- CUDA kernel compilation (nvcc) ---------------------------------------- SM_ARCH ?= native @@ -269,22 +296,17 @@ build/bench_sink: csrc/sink/pack_sink.cpp csrc/sink/object_key.cpp csrc/sink/rec -Icsrc/sink -Icsrc/pack -Icsrc/store -Icsrc/common -lcrypto -lpthread # A2a SigV4 conformance driver: no torch/pybind/network; runs in any CI job. -# libcurl headers come from CURL_INCDIR (dev .deb extract); the runtime -# libcurl.so.4 ships with the OS. Override on machines with system curl-dev: -# make CURL_INCDIR=/usr/include/x86_64-linux-gnu ... -CURL_INCDIR ?= /tmp/opencode/sysroot/usr/include/x86_64-linux-gnu +# libcurl headers come from CURL_INCDIR (defined with the extension flags +# above); the runtime libcurl.so.4 ships with the OS. build/conformance_sign: csrc/store/s3_sign.cpp csrc/store/conformance_sign.cpp csrc/store/s3_sign.h csrc/common/json.cpp csrc/common/json.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ csrc/store/s3_sign.cpp csrc/store/conformance_sign.cpp csrc/common/json.cpp \ -Icsrc/store -Icsrc/common -lcrypto -# A2 store conformance driver + fault-matrix target. libcurl headers come from -# the dev .deb extracted to a sysroot (no root needed): -# apt-get download libcurl4-openssl-dev && dpkg-deb -x -# The link needs only -L; the runtime libcurl.so.4 ships with the OS. Override: -# make CURL_INCDIR=/usr/include/x86_64-linux-gnu CURL_LIBDIR=/usr/lib/x86_64-linux-gnu ... -CURL_LIBDIR ?= /tmp/opencode/sysroot/usr/lib/x86_64-linux-gnu +# A2 store conformance driver + fault-matrix target. libcurl headers/libs come +# from CURL_INCDIR/CURL_LIBDIR (defined with the extension flags above); the +# link needs only -L; the runtime libcurl.so.4 ships with the OS. build/conformance_store: csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/store/s3_sign.h csrc/store/s3_client.h csrc/store/spool.h csrc/store/uploader.h csrc/common/json.cpp csrc/common/json.h csrc/common/curl_init.cpp csrc/common/curl_init.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ From eca6647b58ec89277c17a7d2a4602f170d179335 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:09:18 -0400 Subject: [PATCH 11/16] Propagate a failed curl_global_init instead of latching it as success curl_global_init returns a CURLcode, but std::call_once still marked the helper complete when it failed, so every later client proceeded as though libcurl were initialized and the real failure surfaced as misleading handle/transport errors. The code is stored and thrown as std::runtime_error, so the constructing client fails with the actual reason. --- native/csrc/common/curl_init.cpp | 14 +++++++++++++- native/csrc/common/curl_init.h | 4 +++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/native/csrc/common/curl_init.cpp b/native/csrc/common/curl_init.cpp index a2f50fe9c..5e670fea4 100644 --- a/native/csrc/common/curl_init.cpp +++ b/native/csrc/common/curl_init.cpp @@ -1,6 +1,8 @@ #include "curl_init.h" #include +#include +#include #include @@ -8,7 +10,17 @@ namespace dmi_common { void EnsureCurlGlobalInit() { static std::once_flag once; - std::call_once(once, [] { curl_global_init(CURL_GLOBAL_DEFAULT); }); + static CURLcode status = CURLE_OK; + std::call_once(once, [] { status = curl_global_init(CURL_GLOBAL_DEFAULT); }); + // call_once only means "ran"; it says nothing about whether the init + // worked. Discarding the code turned a failed process-global init into + // misleading downstream handle/transport errors, so the failure is + // propagated to the constructing client instead. + if (status != CURLE_OK) { + throw std::runtime_error( + std::string("curl_global_init failed: ") + + curl_easy_strerror(status)); + } } } // namespace dmi_common diff --git a/native/csrc/common/curl_init.h b/native/csrc/common/curl_init.h index 025f3e8bb..0a956994e 100644 --- a/native/csrc/common/curl_init.h +++ b/native/csrc/common/curl_init.h @@ -26,7 +26,9 @@ namespace dmi_common { // Runs curl_global_init(CURL_GLOBAL_DEFAULT) exactly once per process. -// Safe to call from any thread, any number of times. +// Safe to call from any thread, any number of times; throws std::runtime_error +// if libcurl's process-global initialization failed, so the constructing +// client does not proceed on a library that was never initialized. void EnsureCurlGlobalInit(); } // namespace dmi_common From e036b96c3b12377d058e75c0f843a434bd968183 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:09:21 -0400 Subject: [PATCH 12/16] Check the fixed-dimension product before inferring a dynamic dim fixed *= dim was unchecked: three 2^31-1 dims (the largest CaptureMetadata admits) wrap to a nonzero product, and an empty payload divided evenly by it, so a row was admitted with an inferred dimension whose fixed product never fit. The multiplication now refuses a negative dimension and checks overflow exactly like the reference's checked_product. The new case fails against the pre-fix adapter (the row is admitted) and passes with the check. --- native/csrc/sink/native_pack_sink.cpp | 18 +++++++++++++++++- tests/test_native_adapter_torch.py | 24 ++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/native/csrc/sink/native_pack_sink.cpp b/native/csrc/sink/native_pack_sink.cpp index ffbb79ff9..5cb17cb27 100644 --- a/native/csrc/sink/native_pack_sink.cpp +++ b/native/csrc/sink/native_pack_sink.cpp @@ -40,10 +40,26 @@ bool ResolveShape(const ring::PayloadSlice& slice, uint64_t length_bytes, return false; } const uint64_t elements = length_bytes / element_bytes; + // Checked like the reference's checked_product: a negative fixed dim has no + // uint64 meaning, and an unchecked multiply can wrap before the modulo + // below -- a wrapped fixed can divide elements evenly, admitting a shape + // whose fixed product never fit (a zero inferred dim made `elements % fixed` + // pass on an overflowed fixed). uint64_t fixed = 1; for (size_t i = 0; i < shape_out->size(); ++i) { if (static_cast(i) == dim) continue; - fixed *= static_cast((*shape_out)[i]); + const int64_t dim_value = (*shape_out)[i]; + if (dim_value < 0) { + if (error) *error = "negative logical tensor dimension"; + return false; + } + const uint64_t dimension = static_cast(dim_value); + if (dimension != 0 && + fixed > std::numeric_limits::max() / dimension) { + if (error) *error = "logical tensor shape overflows uint64"; + return false; + } + fixed *= dimension; } if (fixed == 0 || elements % fixed != 0) { if (error) *error = "payload bytes do not factor over the fixed dims"; diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py index abaedb1c2..8774bdd8a 100644 --- a/tests/test_native_adapter_torch.py +++ b/tests/test_native_adapter_torch.py @@ -311,6 +311,30 @@ def test_a_dynamic_dimension_is_inferred_in_elements_not_bytes( assert len(staged_payload) == 32 +def test_a_fixed_dimension_product_that_overflows_is_refused(tmp_path): + """The fixed dims are multiplied in uint64 before the divide. + + Unchecked, three 2^31-1 dims (the largest CaptureMetadata admits) wrap to + a nonzero product, and an empty payload (elements == 0) divided evenly by + it -- so the row was admitted with an inferred dimension whose fixed + product never fit. The reference's checked_product refuses exactly this; + the adapter must too. + """ + dim = 2**31 - 1 + meta = _meta(0) + meta["shape"] = [0, dim, dim, dim] + CaptureMetadata.from_mapping(meta) # the shape the unfixed sink resolved to + sink, lease = _make_sink(tmp_path) + with pytest.raises(RuntimeError, match="overflows uint64"): + sink.submit_envelope( + LAYOUT, + [_row(meta, 0, 0, "float32", shape=(-1, dim, dim, dim), + inferred_dynamic_dim=0)], + torch.zeros(0, dtype=torch.float32), + ) + assert sink.snapshot()["submitted_records"] == 0 + + def test_a_slice_that_is_not_a_whole_number_of_elements_is_refused(tmp_path): """30 bytes is not a whole number of float32s: the reference's guard. From bdd6e7380ff87e8e4800b6f74a50a95176a7de6d Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:09:27 -0400 Subject: [PATCH 13/16] Make the curl-teardown source scan string-aware The comment-stripping regex treated '//' inside a string literal as a comment opener, and the tree already has one (s3_client.cpp's 'http://'), so a forbidden call placed later on that line was stripped before the scan and the guard would pass. Strings, char literals, and both comment forms are now matched in one alternation: a string is kept whole, so a '//' in it cannot hide code, and only real comments are removed. --- tests/test_native_curl_global_lifetime.py | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/tests/test_native_curl_global_lifetime.py b/tests/test_native_curl_global_lifetime.py index 95dcceb5d..accf6e9b6 100644 --- a/tests/test_native_curl_global_lifetime.py +++ b/tests/test_native_curl_global_lifetime.py @@ -50,9 +50,20 @@ # Comments are stripped before scanning: these files DOCUMENT why the global -# calls are absent, and a prose mention is not a call site. Crude but exact -# enough for this -- no string literal in the tree contains "//" or "/*". -_COMMENT = re.compile(r"//[^\n]*|/\*.*?\*/", re.DOTALL) +# calls are absent, and a prose mention is not a call site. Strings and +# comments are matched in one alternation so a `//` INSIDE a string literal +# (s3_client.cpp's "http://") does not open a comment that hides the rest of +# the line -- a forbidden call placed after such a literal must still be +# seen. Strings are kept: their contents are not call sites, but dropping +# them would silently swallow whatever follows, and a false positive fails +# closed. +_TOKEN = re.compile( + r'"(?:\\.|[^"\\])*"' # "..." string literal + r"|'(?:\\.|[^'\\])*'" # '...' char literal + r"|//[^\n]*" # // line comment + r"|/\*.*?\*/", # /* block comment */ + re.DOTALL, +) def _native_sources() -> list[Path]: @@ -64,7 +75,10 @@ def _native_sources() -> list[Path]: def _code_of(path: Path) -> str: - return _COMMENT.sub("", path.read_text(encoding="utf-8")) + return _TOKEN.sub( + lambda match: "" if match.group(0)[0] == "/" else match.group(0), + path.read_text(encoding="utf-8"), + ) def test_no_native_source_tears_down_libcurl_globally(): From 50969e5615ae7e5dc1958edfe3075bc41906b547 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:27:24 -0400 Subject: [PATCH 14/16] Revert 'Link curl_init.cpp into both extensions' The review comment behind that commit (native/csrc/catalog/ clickhouse_client.cpp:117) assumed the catalog client is part of the normal extension build. It is not: SRCS/HOST_SRCS compile the top-level csrc/clickhouse_client.cpp, which speaks clickhouse-cpp's native protocol and never calls EnsureCurlGlobalInit. Preprocessing bindings.cpp and clickhouse_client.cpp with the extension flags finds zero references to the symbol; the only call sites are catalog/clickhouse_client.cpp and store/s3_client.cpp, both compiled solely by the conformance drivers, whose recipes already include common/curl_init.cpp. The shipped _host_backend/ _native_backend .so contain no curl symbol. Adding the helper to the extensions therefore fixed no undefined symbol and instead made both extension builds hard-depend on libcurl dev files (the sysroot guard covers only -I/-L; -lcurl was unconditional). Reverted; the conformance drivers are unchanged and still link the helper. The curl_global_init result propagation (eca6647) stays: that file is real and compiled by the drivers. --- native/Makefile | 42 ++++++++++-------------------------------- 1 file changed, 10 insertions(+), 32 deletions(-) diff --git a/native/Makefile b/native/Makefile index 73a76aa2d..571210b8f 100644 --- a/native/Makefile +++ b/native/Makefile @@ -11,17 +11,14 @@ SRCS := csrc/bindings.cpp \ csrc/clickhouse_record_sink.cpp \ csrc/reference_python_capture_sink.cpp \ csrc/clickhouse_client.cpp \ - csrc/common/curl_init.cpp \ csrc/ring/drain_thread.cpp \ csrc/ring/p2p_thread.cpp \ csrc/ring/record_consumer.cpp \ csrc/ring/ring_torch_op.cpp HOST_SRCS := csrc/bindings.cpp csrc/clickhouse_client.cpp \ - csrc/common/curl_init.cpp \ csrc/pack/pack_builder.cpp HEADERS := csrc/dmx_host_utils.h \ csrc/clickhouse_client.h csrc/clickhouse_record_sink.h \ - csrc/common/curl_init.h \ csrc/reference_python_capture_sink.h \ csrc/dmx_host_engine.h csrc/record_schema.h \ csrc/ring/ring_config.h csrc/ring/ring_state.h csrc/ring/ring_alloc.h \ @@ -125,16 +122,6 @@ PYTHON_INCLUDE_FLAGS := $(shell $(PYTHON) -c "import sysconfig; paths = sysconfi # Pybind11 ABI flags PYBIND_FLAGS := $(shell $(PYTHON) -c "import pybind11; print(' '.join([f'-DPYBIND11_COMPILER_TYPE=\\\"{pybind11.get_compiler_type()}\\\"', f'-DPYBIND11_STDLIB=\\\"{pybind11.get_stdlib()}\\\"', f'-DPYBIND11_BUILD_ABI=\\\"{pybind11.get_build_abi()}\\\"']))" 2>/dev/null || echo "") -# libcurl headers/dev symlink for the extension builds: the shared -# process-lifetime init (common/curl_init.cpp) is compiled into both. The -# runtime libcurl.so.4 ships with the OS; the dev symlink comes from -# CURL_INCDIR/CURL_LIBDIR (a dev .deb extract by default, no root needed): -# apt-get download libcurl4-openssl-dev && dpkg-deb -x -# Override on machines with system curl-dev: -# make CURL_INCDIR=/usr/include/x86_64-linux-gnu CURL_LIBDIR=/usr/lib/x86_64-linux-gnu ... -CURL_INCDIR ?= /tmp/opencode/sysroot/usr/include/x86_64-linux-gnu -CURL_LIBDIR ?= /tmp/opencode/sysroot/usr/lib/x86_64-linux-gnu - CXXFLAGS := -std=c++17 -O3 -fPIC -Wall -Wextra -Wno-unused-parameter -Wno-unused-variable \ -DTORCH_API_INCLUDE_EXTENSION_H -DTORCH_EXTENSION_NAME=$(TARGET) \ $(ABI_FLAG) $(PYBIND_FLAGS) \ @@ -181,20 +168,6 @@ HOST_LDFLAGS += -L$(CLICKHOUSE_CONTRIB)/lz4/lz4 -llz4 \ -L$(CLICKHOUSE_CONTRIB)/absl/absl -labsl_int128 \ -L$(CLICKHOUSE_CONTRIB)/zstd/zstd -lzstdstatic -# ---- libcurl (common/curl_init.cpp is in both extension source lists) ------ -# The include/link dirs are only added when the configured dev tree exists: -# a system with curl-dev in the default search path needs neither. -ifneq ($(wildcard $(CURL_INCDIR)/curl/curl.h),) -CXXFLAGS += -I$(CURL_INCDIR) -HOST_CXXFLAGS += -I$(CURL_INCDIR) -endif -ifneq ($(wildcard $(CURL_LIBDIR)/libcurl.so),) -LDFLAGS += -L$(CURL_LIBDIR) -HOST_LDFLAGS += -L$(CURL_LIBDIR) -endif -LDFLAGS += -lcurl -HOST_LDFLAGS += -lcurl - # ---- CUDA kernel compilation (nvcc) ---------------------------------------- SM_ARCH ?= native @@ -296,17 +269,22 @@ build/bench_sink: csrc/sink/pack_sink.cpp csrc/sink/object_key.cpp csrc/sink/rec -Icsrc/sink -Icsrc/pack -Icsrc/store -Icsrc/common -lcrypto -lpthread # A2a SigV4 conformance driver: no torch/pybind/network; runs in any CI job. -# libcurl headers come from CURL_INCDIR (defined with the extension flags -# above); the runtime libcurl.so.4 ships with the OS. +# libcurl headers come from CURL_INCDIR (dev .deb extract); the runtime +# libcurl.so.4 ships with the OS. Override on machines with system curl-dev: +# make CURL_INCDIR=/usr/include/x86_64-linux-gnu ... +CURL_INCDIR ?= /tmp/opencode/sysroot/usr/include/x86_64-linux-gnu build/conformance_sign: csrc/store/s3_sign.cpp csrc/store/conformance_sign.cpp csrc/store/s3_sign.h csrc/common/json.cpp csrc/common/json.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ csrc/store/s3_sign.cpp csrc/store/conformance_sign.cpp csrc/common/json.cpp \ -Icsrc/store -Icsrc/common -lcrypto -# A2 store conformance driver + fault-matrix target. libcurl headers/libs come -# from CURL_INCDIR/CURL_LIBDIR (defined with the extension flags above); the -# link needs only -L; the runtime libcurl.so.4 ships with the OS. +# A2 store conformance driver + fault-matrix target. libcurl headers come from +# the dev .deb extracted to a sysroot (no root needed): +# apt-get download libcurl4-openssl-dev && dpkg-deb -x +# The link needs only -L; the runtime libcurl.so.4 ships with the OS. Override: +# make CURL_INCDIR=/usr/include/x86_64-linux-gnu CURL_LIBDIR=/usr/lib/x86_64-linux-gnu ... +CURL_LIBDIR ?= /tmp/opencode/sysroot/usr/lib/x86_64-linux-gnu build/conformance_store: csrc/store/s3_sign.cpp csrc/store/s3_client.cpp csrc/store/spool.cpp csrc/store/uploader.cpp csrc/store/conformance_store.cpp csrc/store/s3_sign.h csrc/store/s3_client.h csrc/store/spool.h csrc/store/uploader.h csrc/common/json.cpp csrc/common/json.h csrc/common/curl_init.cpp csrc/common/curl_init.h mkdir -p $(BUILD_DIR) $(CXX) -std=c++17 -O2 -Wall -Wextra -o $@ \ From 4734db7bbe534fa5a280bcb7e7b62cb13b454b0d Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:30:13 -0400 Subject: [PATCH 15/16] Check the fixed product when no dim is dynamic, too ResolveShape returned before the checked walk when inferred_dynamic_dim was absent, and SubmitRow's own element-count multiply is unchecked: three 2^30 dims wrap to 0, match an empty payload and the row is admitted with a shape whose product never fit. The walk now always runs (over the whole shape when nothing is inferred), so SubmitRow never sees a wrapped product. New case fails against the previous adapter (row admitted) and passes with the walk; all 31 adapter tests pass. --- native/csrc/sink/native_pack_sink.cpp | 16 +++++++++------- tests/test_native_adapter_torch.py | 22 ++++++++++++++++++++++ 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/native/csrc/sink/native_pack_sink.cpp b/native/csrc/sink/native_pack_sink.cpp index 5cb17cb27..ad6c5e54a 100644 --- a/native/csrc/sink/native_pack_sink.cpp +++ b/native/csrc/sink/native_pack_sink.cpp @@ -29,22 +29,18 @@ bool ResolveShape(const ring::PayloadSlice& slice, uint64_t length_bytes, uint64_t element_bytes, std::vector* shape_out, std::string* error) { *shape_out = slice.logical_shape; - if (slice.inferred_dynamic_dim < 0) return true; const int dim = slice.inferred_dynamic_dim; if (dim >= static_cast(shape_out->size())) { if (error) *error = "inferred dynamic dim exceeds shape rank"; return false; } - if (element_bytes == 0 || length_bytes % element_bytes != 0) { - if (error) *error = "payload-slice bytes are not divisible by dtype size"; - return false; - } - const uint64_t elements = length_bytes / element_bytes; // Checked like the reference's checked_product: a negative fixed dim has no // uint64 meaning, and an unchecked multiply can wrap before the modulo // below -- a wrapped fixed can divide elements evenly, admitting a shape // whose fixed product never fit (a zero inferred dim made `elements % fixed` - // pass on an overflowed fixed). + // pass on an overflowed fixed). With no dynamic dim the walk covers the + // whole shape, so SubmitRow's own unchecked multiply never sees a product + // that wrapped to match an empty payload. uint64_t fixed = 1; for (size_t i = 0; i < shape_out->size(); ++i) { if (static_cast(i) == dim) continue; @@ -61,6 +57,12 @@ bool ResolveShape(const ring::PayloadSlice& slice, uint64_t length_bytes, } fixed *= dimension; } + if (dim < 0) return true; + if (element_bytes == 0 || length_bytes % element_bytes != 0) { + if (error) *error = "payload-slice bytes are not divisible by dtype size"; + return false; + } + const uint64_t elements = length_bytes / element_bytes; if (fixed == 0 || elements % fixed != 0) { if (error) *error = "payload bytes do not factor over the fixed dims"; return false; diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py index 8774bdd8a..9268d03f7 100644 --- a/tests/test_native_adapter_torch.py +++ b/tests/test_native_adapter_torch.py @@ -335,6 +335,28 @@ def test_a_fixed_dimension_product_that_overflows_is_refused(tmp_path): assert sink.snapshot()["submitted_records"] == 0 +def test_a_fixed_shape_whose_product_overflows_is_refused(tmp_path): + """No dynamic dim: the fixed product must still be checked. + + ResolveShape returned early when nothing had to be inferred, and + SubmitRow's own element-count multiply is unchecked, so three 2^30 dims + wrapped to 0 and matched an empty payload -- admitted with a shape whose + product never fit. The reference's checked_product refuses it. + """ + dim = 2**30 + meta = _meta(0) + meta["shape"] = [dim, dim, dim] + CaptureMetadata.from_mapping(meta) + sink, lease = _make_sink(tmp_path) + with pytest.raises(RuntimeError, match="overflows uint64"): + sink.submit_envelope( + LAYOUT, + [_row(meta, 0, 0, "float32", shape=(dim, dim, dim))], + torch.zeros(0, dtype=torch.float32), + ) + assert sink.snapshot()["submitted_records"] == 0 + + def test_a_slice_that_is_not_a_whole_number_of_elements_is_refused(tmp_path): """30 bytes is not a whole number of float32s: the reference's guard. From 81475ed8e460cdaba8ae7ca2f8a03fe397211c37 Mon Sep 17 00:00:00 2001 From: Alan Liu Date: Fri, 18 Sep 2026 14:30:16 -0400 Subject: [PATCH 16/16] Say what the curl source scan is not It is a regression guard against a teardown being reintroduced, not an obfuscation boundary: preprocessor line splicing or token pasting can still spell an identifier across lines, which a textual scan cannot see. --- tests/test_native_curl_global_lifetime.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_native_curl_global_lifetime.py b/tests/test_native_curl_global_lifetime.py index accf6e9b6..18bef6e13 100644 --- a/tests/test_native_curl_global_lifetime.py +++ b/tests/test_native_curl_global_lifetime.py @@ -57,6 +57,10 @@ # seen. Strings are kept: their contents are not call sites, but dropping # them would silently swallow whatever follows, and a false positive fails # closed. +# +# This is a regression guard against a teardown being reintroduced, not an +# obfuscation boundary: preprocessor line splicing or token pasting can +# still spell an identifier across lines, which a textual scan cannot see. _TOKEN = re.compile( r'"(?:\\.|[^"\\])*"' # "..." string literal r"|'(?:\\.|[^'\\])*'" # '...' char literal