Dynamic-dim shape inference, upload conflict reporting, and libcurl lifetime

.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 `<skipped/>`, 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

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

native/csrc/catalog/clickhouse_client.cpp

@@ -2,6 +2,7 @@
 
 #include <cstdio>
 
+#include "../common/curl_init.h"
 #include "sql_escape.h"
 
 namespace dmi_catalog {
@@ -109,10 +110,14 @@ std::map<std::string, std::string> 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<Row> ClickHouseClient::execute(
     const std::string& query, const Params& params,

native/csrc/common/curl_init.cpp

@@ -0,0 +1,26 @@
+#include "curl_init.h"
+
+#include <mutex>
+#include <stdexcept>
+#include <string>
+
+#include <curl/curl.h>
+
+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

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: <curl/curl.h> 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_

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<int32_t>();
   slice.logical_shape = row["shape"].cast<std::vector<int64_t>>();
+  // 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<int32_t>();
+  }
   return slice;
 }
 

native/csrc/sink/native_pack_sink.cpp

@@ -3,6 +3,7 @@
 #include <ATen/ATen.h>
 
 #include <chrono>
+#include <limits>
 #include <stdexcept>
 
 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<int64_t>* shape_out, std::string* error) {
+                  uint64_t element_bytes, std::vector<int64_t>* 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<int>(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<int>(i) == dim) continue;
-    fixed *= static_cast<uint64_t>((*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<uint64_t>(dim_value);
+    if (dimension != 0 &&
+        fixed > std::numeric_limits<uint64_t>::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<int64_t>(length_bytes / fixed);
+  const uint64_t inferred = elements / fixed;
+  if (inferred >
+      static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
+    if (error) *error = "inferred tensor dimension exceeds int64";
+    return false;
+  }
+  (*shape_out)[dim] = static_cast<int64_t>(inferred);
   return true;
 }
 
@@ -125,7 +164,8 @@ void NativePackSink::submit(ring::RecordEnvelope envelope) {
     }
     std::vector<int64_t> shape;
     std::string shape_error;
-    if (!ResolveShape(*slice, length, &shape, &shape_error)) {
+    if (!ResolveShape(*slice, length, static_cast<uint64_t>(width), &shape,
+                      &shape_error)) {
       invalid(shape_error);
     }
     RowInput input;

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<std::mutex> 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<std::mutex> lock(mutex_);
+    ++counters_.oversized_records;
+    return Admission::kTooLarge;
+  }
   const bool block = config_.overload == Overload::kBlock;
   const double deadline =
       config_.admission_timeout_s < 0

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<SinkRecord, std::shared_ptr<FlushBarrier>>;