diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml
index dceb67c8a..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
@@ -248,6 +329,27 @@ 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 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`.
+ #
+ # 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
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..5e670fea4
--- /dev/null
+++ b/native/csrc/common/curl_init.cpp
@@ -0,0 +1,26 @@
+#include "curl_init.h"
+
+#include
+#include
+#include
+
+#include
+
+namespace dmi_common {
+
+void EnsureCurlGlobalInit() {
+ static std::once_flag once;
+ 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
new file mode 100644
index 000000000..0a956994e
--- /dev/null
+++ b/native/csrc/common/curl_init.h
@@ -0,0 +1,36 @@
+// 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; 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
+
+#endif // DMI_COMMON_CURL_INIT_H_
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..ad6c5e54a 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,27 +15,65 @@ 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;
if (dim >= static_cast(shape_out->size())) {
if (error) *error = "inferred dynamic dim exceeds shape rank";
return false;
}
+ // 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). 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;
- 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 (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;
}
- if (fixed == 0 || length_bytes % fixed != 0) {
+ 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;
}
- (*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 +164,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/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/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/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/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/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/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_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."""
diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py
index 2236fe0f8..9268d03f7 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,108 @@ 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_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_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.
+
+ 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),
diff --git a/tests/test_native_curl_global_lifetime.py b/tests/test_native_curl_global_lifetime.py
new file mode 100644
index 000000000..18bef6e13
--- /dev/null
+++ b/tests/test_native_curl_global_lifetime.py
@@ -0,0 +1,174 @@
+"""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. 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.
+#
+# 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
+ r"|//[^\n]*" # // line comment
+ r"|/\*.*?\*/", # /* block comment */
+ 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 _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():
+ """``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: ")
diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py
index 9bac1ad1b..3e23e24ba 100644
--- a/tests/test_native_pack_sink.py
+++ b/tests/test_native_pack_sink.py
@@ -229,15 +229,36 @@ 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, 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=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 +283,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):
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
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.