diff --git a/native/csrc/catalog/reader.cpp b/native/csrc/catalog/reader.cpp index e47a9c5b8..b7f93e41a 100644 --- a/native/csrc/catalog/reader.cpp +++ b/native/csrc/catalog/reader.cpp @@ -372,10 +372,28 @@ std::vector parse_tsv_tuple(const std::string& text) { // turned `packs/a\b.dmi-pack` into a backspace. Grouped columns // are the opposite case -- plain TSV fields, unescaped where the // row is flattened, never here. + // + // The cases below are the escapes ClickHouse actually EMITS when it + // renders a string inside a tuple, measured with + // `SELECT tuple(concat('block', char(N), 'resid')) FORMAT TSV` over + // the control range: \0 \b \t \n \f \r \' \\ and nothing else. In + // particular 0x07 and 0x0B travel RAW here, so they need no case -- + // this set is deliberately NOT sql_quote's, which also writes \a and + // \v, and the footer decoder (unquote_sql) covers that one. + // + // Both sets have to DECODE to the same bytes even though they differ, + // because hydrate binds the two field by field. \b and \f were + // missing and fell to the default, so they decoded to the letters "b" + // and "f" while the footer side decoded them to 0x08 and 0x0C: a + // hook_name carrying either (legal -- _validate_text asks only for + // non-empty UTF-8 under 512 bytes) staged, uploaded and indexed and + // then failed with "does not match the pack footer: field 12". switch (next) { case 'n': current.push_back('\n'); break; case 't': current.push_back('\t'); break; case 'r': current.push_back('\r'); break; + case 'b': current.push_back('\b'); break; + case 'f': current.push_back('\f'); break; case '0': current.push_back('\0'); break; default: current.push_back(next); break; // \' and \\ included } diff --git a/native/csrc/common/json.cpp b/native/csrc/common/json.cpp index 973fd4f2b..9b0229df9 100644 --- a/native/csrc/common/json.cpp +++ b/native/csrc/common/json.cpp @@ -2,8 +2,19 @@ namespace dmi_common { -std::string Unescape(const std::string& text, size_t& q) { +std::string Unescape(const std::string& text, size_t& q, bool* ok) { std::string raw; + // An escape with no code point behind it. The decoder stays TOTAL -- it has + // to, since most callers read fields whose contents nothing validates -- so + // it emits U+FFFD and lets `ok` carry the refusal to the callers that have + // an error channel. U+FFFD matters on its own: whatever a caller does next, + // no sequence that is not valid UTF-8 leaves this function, which is what + // put CESU-8 in a pack. `ok` is only ever cleared, never set, so one flag + // can latch a whole object's worth of fields. + const auto reject = [&ok, &raw] { + if (ok != nullptr) *ok = false; + raw.append("\xEF\xBF\xBD"); + }; while (q < text.size() && text[q] != '"') { if (text[q] == '\\' && q + 1 < text.size()) { ++q; @@ -15,19 +26,46 @@ std::string Unescape(const std::string& text, size_t& q) { case 'f': raw.push_back('\f'); ++q; break; case 'u': { // Four hex digits at text[at+1..at+4], or -1 when they are not - // there (a truncated escape at the end of the text). + // there: a truncated escape at the end of the text, or a character + // that is not a hex digit at all. The digits were NOT checked, so + // `h <= '9' ? h - '0' : (h | 0x20) - 'a' + 10` turned \uZZZZ into + // U+25553 and the field decoded to F0 A5 95 93, where json.loads + // raises "Invalid \uXXXX escape" and the record never lands. const auto hex4 = [&text](size_t at) -> long { if (at + 4 >= text.size()) return -1; unsigned v = 0; for (int k = 1; k <= 4; ++k) { const char h = text[at + k]; - v = v * 16 + (h <= '9' ? h - '0' : (h | 0x20) - 'a' + 10); + unsigned digit = 0; + if (h >= '0' && h <= '9') { + digit = static_cast(h - '0'); + } else if ((h | 0x20) >= 'a' && (h | 0x20) <= 'f') { + digit = static_cast((h | 0x20) - 'a' + 10); + } else { + return -1; + } + v = v * 16 + digit; } return static_cast(v); }; const long first = hex4(q); - unsigned v = first < 0 ? 0u : static_cast(first); + // Advance past the four digits only when they ARE four hex + // digits. On failure the "digits" can include the field's real + // closing quote -- `"x\u12"` -- and stepping over them stepped + // over that quote too, so the loop kept copying the FOLLOWING + // JSON text (comma, separator, the next field's key) into this + // field's value until the next quote; a truncated escape at the + // end of the buffer left q strictly past text.size(). Consuming + // only the `u` keeps the decoder on the string: whatever failed + // hex4 is copied as literal text, the real closing quote + // terminates the value, and the latch still carries the refusal. + if (first < 0) { + ++q; + reject(); + break; + } q += 5; + unsigned v = static_cast(first); // json.dumps (ensure_ascii, the default) writes a code point // outside the BMP as a UTF-16 surrogate PAIR: U+1F600 is // \ud83d\ude00. The two halves are one character, so they have to @@ -35,16 +73,33 @@ std::string Unescape(const std::string& text, size_t& q) { // half on its own produced six bytes (ED A0 BD ED B8 80) that are // not UTF-8 at all, and the identifier reached the catalog // corrupted (reproduced with hook_name "block" + U+1F600: Python's reader - // then failed to decode byte 0xED). A lone surrogate keeps the - // old three-byte form -- there is nothing to combine it with. - if (v >= 0xD800 && v <= 0xDBFF && q + 1 < text.size() && - text[q] == '\\' && text[q + 1] == 'u') { - const long second = hex4(q + 1); - if (second >= 0xDC00 && second <= 0xDFFF) { - v = 0x10000 + ((v - 0xD800) << 10) + - (static_cast(second) - 0xDC00); - q += 6; + // then failed to decode byte 0xED). + // + // A surrogate with no partner has NOTHING to combine with, and + // keeping its three-byte form is the same corruption in the small: + // \ud83d decoded to ED A0 BD, which ValidText admits (it checks + // lead/continuation byte SHAPE, not the surrogate range) and the + // record was packed and inserted. The oracle refuses it -- + // CaptureMetadata's _validate_text calls .encode("utf-8"), which + // raises UnicodeEncodeError on a lone surrogate -- so it is + // refused here rather than encoded. + if (v >= 0xD800 && v <= 0xDBFF) { + const long second = + (q + 1 < text.size() && text[q] == '\\' && text[q + 1] == 'u') + ? hex4(q + 1) + : -1; + if (second < 0xDC00 || second > 0xDFFF) { + reject(); + break; } + v = 0x10000 + ((v - 0xD800) << 10) + + (static_cast(second) - 0xDC00); + q += 6; + } else if (v >= 0xDC00 && v <= 0xDFFF) { + // A trailing surrogate reached on its own: the leading half it + // belongs to is not there, so this one has no partner either. + reject(); + break; } if (v < 0x80) { raw.push_back(static_cast(v)); @@ -73,7 +128,7 @@ std::string Unescape(const std::string& text, size_t& q) { } std::string FindString(const std::string& text, const std::string& key, - size_t from) { + size_t from, bool* ok) { for (const char* sep : {": \"", ":\""}) { // NOTE: sep includes the opening quote; search for key+sep. const std::string needle = "\"" + key + "\"" + sep; @@ -81,8 +136,11 @@ std::string FindString(const std::string& text, const std::string& key, if (at == std::string::npos) continue; size_t q = at + needle.size() - 1; // back up onto the opening quote ++q; - return Unescape(text, q); + return Unescape(text, q, ok); } + // An ABSENT key is not a decode failure, so `ok` is left alone: a caller + // latching one flag over a whole object must not be told the field it did + // not send was malformed. return ""; } diff --git a/native/csrc/common/json.h b/native/csrc/common/json.h index 1717b53be..742741b74 100644 --- a/native/csrc/common/json.h +++ b/native/csrc/common/json.h @@ -16,12 +16,24 @@ namespace dmi_common { // JSON string escape sequences back to raw bytes; \uXXXX to UTF-8, with a // UTF-16 surrogate pair combined into its one non-BMP code point. // `q` must be positioned just after the opening quote; returns with `q` on -// the closing quote. -std::string Unescape(const std::string& text, size_t& q); +// the closing quote (at text.size() for an unterminated text, never past +// it -- a malformed escape does not advance over the quote or the end). +// +// `ok`, when given, is CLEARED (never set) for an escape that has no code +// point behind it: a \uXXXX whose digits are not hex, and a surrogate with +// no partner. Both are things the oracle refuses -- json.loads raises +// "Invalid \uXXXX escape" for the first, and _validate_text's +// .encode("utf-8") raises UnicodeEncodeError for the second -- and this +// decoder has no way to refuse on its own. The returned text stays valid +// UTF-8 either way (U+FFFD stands in), so a caller that cannot check still +// never receives the CESU-8 that reached a pack. Initialise it to true and +// pass the same flag to every field of one object to latch the whole parse. +std::string Unescape(const std::string& text, size_t& q, bool* ok = nullptr); // First string value for `key` anywhere in `text` (either separator style). +// `ok` is as for Unescape; an absent key leaves it alone. std::string FindString(const std::string& text, const std::string& key, - size_t from = 0); + size_t from = 0, bool* ok = nullptr); // Outcome of a bounded integer scan. enum class IntFind { diff --git a/native/csrc/sink/conformance_sink.cpp b/native/csrc/sink/conformance_sink.cpp index bb866ce82..47196a0ff 100644 --- a/native/csrc/sink/conformance_sink.cpp +++ b/native/csrc/sink/conformance_sink.cpp @@ -13,11 +13,14 @@ // {"op":"snapshot"} -> {"ok":true,"snapshot":{...}} // {"op":"object_key","tenant_id":"...","session_id":"...","producer_rank":N, // "captured_at_ns":N,"pack_id":"..."} -> {"ok":true,"object_key":"..."} +// {"op":"parse_metadata","metadata_json":"..."} +// -> {"ok":true,"hook_name_hex":"..."} // Errors: {"ok":false,"what":"..."}. #include "pack_sink.h" #include +#include #include #include #include @@ -44,6 +47,14 @@ namespace { // are the driver's remaining decodes. std::string g_out_of_range; +// Latched the same way, for a metadata string the JSON decoder could not turn +// into code points: non-hex \uXXXX digits, or a surrogate with no partner. +// The oracle refuses both (json.loads raises "Invalid \uXXXX escape"; +// _validate_text's .encode("utf-8") raises UnicodeEncodeError), and this op is +// the native side of CaptureMetadata.from_mapping, so it has to as well -- +// SubmitRow's own parse already does (record_row.cpp). +bool g_bad_text = false; + // A field the oracle types as SIGNED must pass IntDomain::kSigned: the // union's two's-complement bit pattern is a legal-looking negative to such a // field, and 18446744073709551615 arrived at layer_number as -1 -- its own @@ -62,20 +73,25 @@ int64_t Integer(const std::string& text, const char* key, dmi_pack::RecordMetadata ParseMetadata(const std::string& obj) { dmi_pack::RecordMetadata m; - m.capture_id = jc::FindString(obj, "capture_id"); - m.tenant_id = jc::FindString(obj, "tenant_id"); - m.experiment_id = jc::FindString(obj, "experiment_id"); - m.run_id = jc::FindString(obj, "run_id"); - m.session_id = jc::FindString(obj, "session_id"); - m.request_id = jc::FindString(obj, "request_id"); - m.sequence_id = jc::FindString(obj, "sequence_id"); - m.model_id = jc::FindString(obj, "model_id"); - m.model_revision = jc::FindString(obj, "model_revision"); + bool text_ok = true; + const auto text_field = [&obj, &text_ok](const char* name) { + return jc::FindString(obj, name, 0, &text_ok); + }; + m.capture_id = text_field("capture_id"); + m.tenant_id = text_field("tenant_id"); + m.experiment_id = text_field("experiment_id"); + m.run_id = text_field("run_id"); + m.session_id = text_field("session_id"); + m.request_id = text_field("request_id"); + m.sequence_id = text_field("sequence_id"); + m.model_id = text_field("model_id"); + m.model_revision = text_field("model_revision"); if (!jc::FindNull(obj, "adapter_revision")) { - m.adapter_revision = jc::FindString(obj, "adapter_revision"); + m.adapter_revision = text_field("adapter_revision"); } - m.capture_policy_version = jc::FindString(obj, "capture_policy_version"); - m.hook_name = jc::FindString(obj, "hook_name"); + m.capture_policy_version = text_field("capture_policy_version"); + m.hook_name = text_field("hook_name"); + if (!text_ok) g_bad_text = true; m.layer_number = Integer(obj, "layer_number", jc::IntDomain::kSigned); m.producer_rank = static_cast(Integer(obj, "producer_rank")); m.step_number = static_cast(Integer(obj, "step_number")); @@ -161,8 +177,18 @@ int main() { &out); std::cout << out << "}\n"; }; + const auto refuse_bad_text = [] { + std::string out = "{\"ok\":false,\"what\":"; + // Same wording as ParseMetadataJson's refusal: the decoder replaces a + // malformed \uXXXX or an unpaired surrogate with U+FFFD (valid UTF-8), + // so the failure is the escape, not the encoding. + jc::EscapeJson("capture metadata text has an invalid or unpaired Unicode escape", + &out); + std::cout << out << "}\n"; + }; while (std::getline(std::cin, line)) { g_out_of_range.clear(); + g_bad_text = false; const std::string op = jc::FindString(line, "op"); if (op == "open") { dmi_sink::SinkConfig config; @@ -233,6 +259,38 @@ int main() { std::cout << out << "}\n"; continue; } + if (op == "parse_metadata") { + // Sessionless like `object_key`: the row path's metadata decoder alone, + // handing hook_name back as RAW BYTES in hex. + // + // Why the bytes, and why not read them off a staged pack: the footer + // writer decodes each UTF-8 sequence and re-emits \uXXXX + // (pack_builder.cpp, EncodeJsonString), which is an exact inverse of + // this decoder. A broken CESU-8 surrogate pair written by the decoder + // comes back out of the footer as a correct 😀 that + // json.loads recombines, so every assertion made on a hook_name read + // back through the pack holds whether or not the decoder combines -- + // the surrogate tests passed with the combine branch disabled. These + // bytes never meet that serializer, so they can tell the two apart. + dmi_pack::RecordMetadata metadata; + std::string error; + if (!dmi_sink::ParseMetadataJson(jc::FindString(line, "metadata_json"), + &metadata, &error)) { + std::string out = "{\"ok\":false,\"what\":"; + jc::EscapeJson(error, &out); + std::cout << out << "}\n"; + continue; + } + std::string hex; + for (const char byte : metadata.hook_name) { + char buf[3]; + std::snprintf(buf, sizeof(buf), "%02x", + static_cast(static_cast(byte))); + hex.append(buf, 2); + } + std::cout << "{\"ok\":true,\"hook_name_hex\":\"" << hex << "\"}\n"; + continue; + } if (!sink) { std::cout << "{\"ok\":false,\"what\":\"sink is not open\"}\n"; continue; @@ -247,6 +305,10 @@ int main() { refuse_out_of_range(); continue; } + if (g_bad_text) { + refuse_bad_text(); + continue; + } std::vector payload; jc::DecodeBase64(jc::FindString(line, "payload_b64"), &payload); const dmi_sink::Admission admission = diff --git a/native/csrc/sink/record_row.cpp b/native/csrc/sink/record_row.cpp index 7c8f788fe..932e0332c 100644 --- a/native/csrc/sink/record_row.cpp +++ b/native/csrc/sink/record_row.cpp @@ -37,22 +37,42 @@ bool ParseMetadataJson(const std::string& text, return false; }; // The object may carry surrounding whitespace; Find* scans for keys. - out->capture_id = jc::FindString(text, "capture_id"); - out->tenant_id = jc::FindString(text, "tenant_id"); - out->experiment_id = jc::FindString(text, "experiment_id"); - out->run_id = jc::FindString(text, "run_id"); - out->session_id = jc::FindString(text, "session_id"); - out->request_id = jc::FindString(text, "request_id"); - out->sequence_id = jc::FindString(text, "sequence_id"); - out->model_id = jc::FindString(text, "model_id"); - out->model_revision = jc::FindString(text, "model_revision"); + // + // One latch across every identifier: an escape the decoder cannot turn into + // a code point (non-hex \uXXXX digits, or a surrogate with no partner) is a + // refusal, not a value. Both are things the oracle refuses -- json.loads + // raises "Invalid \uXXXX escape" for the first and _validate_text's + // .encode("utf-8") raises UnicodeEncodeError for the second -- while native + // encoded them as U+25553 and as three-byte CESU-8, which ValidText admits + // (it checks byte SHAPE, not the surrogate range) and the record was packed + // and inserted. + bool text_ok = true; + const auto text_field = [&text, &text_ok](const char* name) { + return jc::FindString(text, name, 0, &text_ok); + }; + out->capture_id = text_field("capture_id"); + out->tenant_id = text_field("tenant_id"); + out->experiment_id = text_field("experiment_id"); + out->run_id = text_field("run_id"); + out->session_id = text_field("session_id"); + out->request_id = text_field("request_id"); + out->sequence_id = text_field("sequence_id"); + out->model_id = text_field("model_id"); + out->model_revision = text_field("model_revision"); if (!jc::FindNull(text, "adapter_revision")) { - out->adapter_revision = jc::FindString(text, "adapter_revision"); + out->adapter_revision = text_field("adapter_revision"); } else { out->adapter_revision.reset(); } - out->capture_policy_version = jc::FindString(text, "capture_policy_version"); - out->hook_name = jc::FindString(text, "hook_name"); + out->capture_policy_version = text_field("capture_policy_version"); + out->hook_name = text_field("hook_name"); + if (!text_ok) { + // The decoder replaces a malformed \uXXXX (non-hex or truncated digits) + // and an unpaired surrogate with U+FFFD, which IS valid UTF-8, so "not + // encodable UTF-8" named the wrong failure. These are the only two ways + // `ok` is cleared; name them. + return fail("capture metadata text has an invalid or unpaired Unicode escape"); + } // Presence first: layer_number == -1 is legal (logits-style captures), // so FindInt's missing sentinel cannot stand in for absence. for (const char* name : diff --git a/native/csrc/store/spool.cpp b/native/csrc/store/spool.cpp index 52a39c398..3dab9f242 100644 --- a/native/csrc/store/spool.cpp +++ b/native/csrc/store/spool.cpp @@ -262,11 +262,14 @@ SpoolStatus Spool::Open(SpoolConfig config, Spool* out, std::string* error) { out->committed_entries_ = 0; out->reserved_bytes_ = 0; out->reserved_entries_ = 0; + out->accounted_ready_.clear(); out->inflight_temps_.clear(); out->peak_bytes_ = 0; out->generation_ = 0; // Account pre-existing files exactly like the Python constructor: ready - // files plus stale .open files both count until Recover() runs. + // files plus stale .open files both count until Recover() runs, and only + // the ready PATHS are remembered (_accounted_ready = ready_bytes), so a + // later retry of one of them is recognised as already counted. for (const auto& entry : fs::recursive_directory_iterator(out->root_, ec)) { if (ec) break; @@ -276,12 +279,82 @@ SpoolStatus Spool::Open(SpoolConfig config, Spool* out, std::string* error) { const bool is_open = HasSuffix(name, kOpenSuffix); if (!is_ready && !is_open) continue; out->committed_bytes_ += entry.file_size(); - if (is_ready) ++out->committed_entries_; + if (is_ready) { + ++out->committed_entries_; + out->accounted_ready_.emplace(entry.path().string(), entry.file_size()); + } } out->peak_bytes_ = out->committed_bytes_; return SpoolStatus::kOk; } +bool Spool::AccountReadyLocked(const std::string& path, + uint64_t object_bytes) { + if (accounted_ready_.count(path) != 0) return false; + accounted_ready_.emplace(path, object_bytes); + committed_bytes_ += object_bytes; + ++committed_entries_; + peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_); + return true; +} + +bool Spool::UnaccountReadyLocked(const std::string& path) { + const auto found = accounted_ready_.find(path); + if (found == accounted_ready_.end()) return false; + // Subtract what was RECORDED for this path, not what the caller believes it + // to be: the recorded value is the one that went into the aggregate. + if (committed_bytes_ >= found->second) committed_bytes_ -= found->second; + if (committed_entries_ > 0) --committed_entries_; + accounted_ready_.erase(found); + return true; +} + +void Spool::ReconcileCommittedLocked() { + // The committed counter is only as fresh as the Remove calls THIS Spool + // object has seen. An uploader running through a second Spool object on + // the same root (or another process) removes ready files and updates its + // OWN counter -- this object's never learns about the released capacity, + // so staging eventually refuses on a directory that is actually empty + // (reproduced: sink with a 1500-byte limit, serial upload-and-remove + // between two records). Reconcile the COMMITTED account from the + // directory -- the durable truth for what is committed. + // + // The scan is authoritative for committed files only. The reservations + // other stagers hold right now are not on disk (or are, as a temp file + // this object already accounts for), so they are kept as they are and + // added back on top: replacing a single combined counter with the scan + // admitted a concurrent stage the reservation should have refused. + uint64_t actual = 0; + uint64_t ready_count = 0; + std::unordered_map seen_ready; + std::error_code walk_ec; + for (const auto& entry : + fs::recursive_directory_iterator(root_, walk_ec)) { + if (walk_ec) break; + if (!entry.is_regular_file()) continue; + const std::string name = entry.path().filename().string(); + if (HasSuffix(name, kReadySuffix)) { + actual += entry.file_size(); + ++ready_count; + seen_ready.emplace(entry.path().string(), entry.file_size()); + } else if (HasSuffix(name, kOpenSuffix) && + inflight_temps_.count(entry.path().string()) == 0) { + // Someone else's in-progress write (another process, or a stale + // leftover Recover() has not swept yet): counts, as it does at + // Open(). This object's own temps are reservations. + actual += entry.file_size(); + } + } + committed_bytes_ = actual; + committed_entries_ = ready_count; + // The path ledger is rebuilt with the aggregate it describes, so the two + // never disagree about which files the committed account holds. + accounted_ready_ = std::move(seen_ready); + // The scan can raise the committed total (files another object wrote), and + // peak_bytes_ must never read below what the account holds right now. + peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_); +} + void Spool::SetStageHookForTesting(std::function hook) { std::lock_guard lock(mutex_); stage_hook_for_testing_ = std::move(hook); @@ -361,43 +434,7 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns, } } if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) { - // The committed counter is only as fresh as the Remove calls THIS - // Spool object has seen. An uploader running through a second - // Spool object on the same root (or another process) removes - // ready files and updates its OWN counter — this object's never - // learns about the released capacity, so staging eventually - // refuses on a directory that is actually empty (reproduced: sink - // with a 1500-byte limit, serial upload-and-remove between two - // records). Reconcile the COMMITTED account from the directory — - // the durable truth for what is committed — before refusing. - // - // The scan is authoritative for committed files only. The - // reservations other stagers hold right now are not on disk (or - // are, as a temp file this object already accounts for), so they - // are kept as they are and added back on top: replacing a single - // combined counter with the scan admitted a concurrent stage the - // reservation should have refused. - uint64_t actual = 0; - uint64_t ready_count = 0; - std::error_code walk_ec; - for (const auto& entry : - fs::recursive_directory_iterator(root_, walk_ec)) { - if (walk_ec) break; - if (!entry.is_regular_file()) continue; - const std::string name = entry.path().filename().string(); - if (HasSuffix(name, kReadySuffix)) { - actual += entry.file_size(); - ++ready_count; - } else if (HasSuffix(name, kOpenSuffix) && - inflight_temps_.count(entry.path().string()) == 0) { - // Someone else's in-progress write (another process, or a - // stale leftover Recover() has not swept yet): counts, as it - // does at Open(). This object's own temps are reservations. - actual += entry.file_size(); - } - } - committed_bytes_ = actual; - committed_entries_ = ready_count; + ReconcileCommittedLocked(); if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) { if (error) { *error = "spool byte limit exceeded: " + @@ -432,10 +469,40 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns, if (reserved_bytes_ >= n) reserved_bytes_ -= n; if (reserved_entries_ > 0) --reserved_entries_; inflight_temps_.erase(temp); - committed_bytes_ += n; - ++committed_entries_; + AccountReadyLocked(ready, n); ++generation_; }; + // The retry and EEXIST-loser paths both end on a ready file that already + // exists. Whether it costs anything depends on whether THIS object has + // counted that path before, which only the path ledger can answer. A path + // it has NOT counted is judged against the durable truth plus in-flight + // reservations -- the same capacity decision a fresh stage makes, except + // that the reconciliation scan counts the file's bytes rather than + // charging them on top of the committed total. Without the decision, a + // retry could be added on top of an in-flight reservation: with a + // 1000-byte reservation paused and a 1000-byte ready file created by a + // second Spool object, admitting the retry put committed + reserved at + // 2000 under a 1500 cap and both stages completed. + // + // The verdict must not depend on which call ran the scan. An + // already-counted path skips the scan, so it re-checks the same sum + // instead of returning kOk unconditionally: otherwise the first retry + // (which reconciles, sees the overage and refuses) and the second (which + // finds the path ledgered) would disagree about identical state. + auto account_existing = [&]() -> SpoolStatus { + std::lock_guard lock(mutex_); + if (accounted_ready_.count(ready) == 0) ReconcileCommittedLocked(); + if (committed_bytes_ + reserved_bytes_ > max_bytes_) { + if (error) { + *error = "spool byte limit exceeded: " + + std::to_string(committed_bytes_ + reserved_bytes_) + " > " + + std::to_string(max_bytes_); + } + return SpoolStatus::kFull; + } + if (AccountReadyLocked(ready, n)) ++generation_; + return SpoolStatus::kOk; + }; auto fill_out = [&] { out->pack_id = pack_id; out->created_at_ns = created_at_ns; @@ -468,9 +535,19 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns, if (error) *error = "spool contains different content: " + object_key; return SpoolStatus::kConflict; } - // A retry adds no accounting: the file was counted by the stage (or - // process start) that created it. Every successful retry still closes - // the durability window itself with a fresh fsync chain. + // A retry of a file THIS object created adds no accounting; a retry of + // one created by a second Spool object on the same root after this + // one's Open() adds its full size, because nothing here has counted it. + // Python makes exactly that distinction -- spool.py:124 runs the retry + // through _account_ready_locked, which is a no-op only for a path + // already in _accounted_ready. Adding nothing unconditionally judged the + // cap against 0: with max 1500, a 1000-byte file staged through another + // object and retried here still admitted another 1000. + // + // Every successful retry still closes the durability window itself with + // a fresh fsync chain. + const SpoolStatus accounted = account_existing(); + if (accounted != SpoolStatus::kOk) return accounted; if (!FsyncChain(root_, dir, error)) return SpoolStatus::kIo; fill_out(); return SpoolStatus::kOk; @@ -493,15 +570,19 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns, } if (::link(temp.c_str(), ready.c_str()) != 0) { if (errno == EEXIST) { - // Lost the race: the winner's file is already counted (by its stage - // or by process start), so release this reservation, then validate - // the winner exactly like the retry path. + // Lost the race: release this reservation, then validate the winner + // exactly like the retry path. The winner's file is counted here only + // if this object has not counted that path already -- the winner may + // be another Spool object entirely, whose stage touched nothing in + // this account. Python does the same (spool.py:159). ::unlink(temp.c_str()); unreserve(); if (!validate_ready()) { if (error) *error = "spool contains different content: " + object_key; return SpoolStatus::kConflict; } + const SpoolStatus accounted = account_existing(); + if (accounted != SpoolStatus::kOk) return accounted; // The winner may still be between link() and its own fsync: do not // acknowledge its dirent before independently making the chain // durable (mirrors the Python loser's fsync). @@ -542,6 +623,7 @@ SpoolStatus Spool::Scan(std::vector* out, bool discard_open_files, std::error_code ec; std::vector readies; uint64_t bytes = 0; + std::unordered_map seen_ready; for (const auto& entry : fs::recursive_directory_iterator(root_, ec)) { if (ec) break; @@ -594,11 +676,16 @@ SpoolStatus Spool::Scan(std::vector* out, bool discard_open_files, staged.object_bytes = size; out->push_back(std::move(staged)); bytes += size; + seen_ready.emplace(path, size); } // Recovery rebuilds the committed account only; a stage in flight on - // another thread keeps its reservation. + // another thread keeps its reservation. The path ledger is rebuilt with + // it (_commit_recovery_locked does the same), so the surviving entries are + // exactly the ones a later retry will recognise as already counted, and + // the quarantined ones are simply absent. committed_bytes_ = bytes; committed_entries_ = out->size(); + accounted_ready_ = std::move(seen_ready); peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_); ++generation_; (void)error; @@ -615,8 +702,12 @@ SpoolStatus Spool::Remove(const StagedPack& staged, std::string* error) { if (error) *error = "cannot inspect staged pack: " + ec.message(); return SpoolStatus::kIo; } - // Removal retries must not release another pack's capacity. A stale - // count after an external removal is reconciled before refusing Stage. + // The file is already gone -- removed through another Spool object, or + // by an earlier call here. Drop it from this object's account only if + // this object was counting that PATH: a removal RETRY finds nothing + // recorded and costs nothing, so it cannot release another pack's + // capacity (which a blind subtraction did). + if (UnaccountReadyLocked(staged.path)) ++generation_; return SpoolStatus::kOk; } const std::string name = fs::path(staged.path).filename().string(); @@ -626,18 +717,36 @@ SpoolStatus Spool::Remove(const StagedPack& staged, std::string* error) { id != staged.pack_id || created != staged.created_at_ns || records != staged.record_count || sum != staged.checksum || fs::file_size(staged.path, ec) != staged.object_bytes) { + if (ec == std::errc::no_such_file_or_directory) { + // The file vanished between the exists() check and this one -- + // another Spool object removed it. Uncount the path if THIS object + // was charging it, exactly like the already-missing branch, or the + // ledger keeps bytes for a file that is gone and a later retry + // treats a recreated path as already accounted. + if (UnaccountReadyLocked(staged.path)) ++generation_; + return SpoolStatus::kOk; + } if (error) *error = "staged pack identity changed before removal"; return SpoolStatus::kIntegrity; } if (::unlink(staged.path.c_str()) != 0) { - if (errno == ENOENT) return SpoolStatus::kOk; + if (errno == ENOENT) { + // Same removal race, now between the checks above and the unlink: + // release this object's charge for the path that another Spool + // object just removed. + if (UnaccountReadyLocked(staged.path)) ++generation_; + return SpoolStatus::kOk; + } if (error) *error = "cannot remove staged pack: " + std::string(strerror(errno)); return SpoolStatus::kIo; } - if (committed_bytes_ >= staged.object_bytes) { - committed_bytes_ -= staged.object_bytes; - } - if (committed_entries_ > 0) --committed_entries_; + // Uncount by PATH: the bytes that leave the aggregate are the ones this + // object recorded for it, and a path this object never counted (removed + // on behalf of another Spool object) costs nothing, exactly as + // _unaccount_ready_locked does. Uncounting by anything else would let a + // path stay in the ledger after its file is gone, and the next stage of + // the same pack would then be treated as already accounted. + UnaccountReadyLocked(staged.path); ++generation_; parent = fs::path(staged.path).parent_path().string(); } diff --git a/native/csrc/store/spool.h b/native/csrc/store/spool.h index 051527b29..ad3558653 100644 --- a/native/csrc/store/spool.h +++ b/native/csrc/store/spool.h @@ -19,6 +19,7 @@ #include #include #include +#include #include #include @@ -105,6 +106,26 @@ class Spool { private: SpoolStatus Scan(std::vector* out, bool discard_open_files, std::string* error); + // Count/uncount one ready path in the committed account, at most once each + // -- Python's _account_ready_locked / _unaccount_ready_locked. `mutex_` + // must be held. Both return whether they actually changed the account. + // + // The aggregate alone cannot decide this. A ready file reached by the retry + // or EEXIST-loser path may have been created by THIS object (already + // counted, so adding would double it) or by a second Spool object on the + // same root after this one's Open() (never counted, so adding nothing + // leaves the cap judged against 0 -- with max 1500, one 1000-byte file + // staged elsewhere then retried here left room for another 1000). + bool AccountReadyLocked(const std::string& path, uint64_t object_bytes); + bool UnaccountReadyLocked(const std::string& path); + // Rebuild the COMMITTED account (bytes, entries, path ledger) from the + // directory, leaving in-flight reservations alone. Run under `mutex_` + // before refusing a stage: this object's counter is only as fresh as the + // Remove calls it has seen, so a second Spool object's removals make it + // stale-high. The retry/EEXIST-loser paths run it too, so a charge that + // would exceed the cap is judged against the same durable truth. + void ReconcileCommittedLocked(); + std::string root_; uint64_t max_bytes_ = 0; mutable std::mutex mutex_; @@ -124,6 +145,15 @@ class Spool { uint64_t committed_entries_ = 0; uint64_t reserved_bytes_ = 0; uint64_t reserved_entries_ = 0; + // Which ready paths the committed account currently includes, and for how + // many bytes -- Python's _accounted_ready. It is what makes the accounting + // idempotent per PATH rather than per call, so the same ready file can be + // met more than once (retry, EEXIST loser, a scan) and be counted exactly + // once. Rebuilt wholesale wherever committed_* is. Stale .open bytes are + // in committed_bytes_ without being here, exactly as in Python's + // constructor, so this map is a ledger of ready paths, not a second copy + // of the aggregate. + std::unordered_map accounted_ready_; std::unordered_set inflight_temps_; uint64_t peak_bytes_ = 0; uint64_t generation_ = 0; diff --git a/tests/native/test_json_unescape.cpp b/tests/native/test_json_unescape.cpp new file mode 100644 index 000000000..b1c196c51 --- /dev/null +++ b/tests/native/test_json_unescape.cpp @@ -0,0 +1,172 @@ +// Unescape's \uXXXX FAILURE path must leave the decoder on the string it is +// decoding. hex4 validates the length and the four digits, but the advance +// (`q += 5`) ran BEFORE the verdict was looked at, so a malformed escape +// whose "digits" include the field's real closing quote stepped past that +// quote, and the while loop kept copying the FOLLOWING JSON text -- comma, +// separator, the next field's key -- into this field's value until the next +// quote. Every FindString caller that passes no `ok` flag (pack_index.cpp's +// footer parsing, reader.cpp's cursor "fh") silently received a value +// containing structural JSON. A `\u12` at the very end of the buffer left +// `q` strictly past text.size(), breaking json.h's contract that Unescape +// "returns with `q` on the closing quote" (or at the end of an unterminated +// text). +// +// On failure the decoder consumes only the `\u` and resumes at the +// character after it: the characters that failed hex4 are copied as +// literals, the real closing quote terminates the string, and the `ok` +// latch still records the refusal. +// +// Built and run by tests/test_native_json_unescape.py. + +#include +#include + +#include "common/json.h" + +namespace jc = dmi_common; + +namespace { + +int g_failures = 0; + +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + std::cerr << __FILE__ << ":" << __LINE__ << ": CHECK failed: " #cond \ + << "\n"; \ + ++g_failures; \ + } \ + } while (0) + +// U+FFFD, the replacement character the decoder emits for a refusal. +const std::string kFFFD = "\xEF\xBF\xBD"; + +// A short escape whose "digits" run into the closing quote, spaced +// separator style (json.dumps's default). The refusal must latch AND the +// decoder must stop on the field's real closing quote: the two characters +// that failed hex4 come through as literals, and nothing past the quote -- +// no comma, no separator, no next key -- leaks into the value. +void ShortEscapeStopsOnTheClosingQuote() { + const std::string text = + "{\"hook_name\": \"x\\u12\", \"model_id\": \"m\"}"; + const size_t start = text.find("x\\u12"); + size_t q = start; + bool ok = true; + const std::string value = jc::Unescape(text, q, &ok); + CHECK(!ok); + CHECK(value == "x" + kFFFD + "12"); + CHECK(q == start + 5); // the field's real closing quote + CHECK(q < text.size() && text[q] == '"'); + + // The no-ok FindString path, as pack_index.cpp and reader.cpp call it: + // the value must not contain structural JSON from beyond the quote. + const std::string hook = jc::FindString(text, "hook_name"); + CHECK(hook.find(',') == std::string::npos); + CHECK(hook.find("model_id") == std::string::npos); + + // A subsequent field in the same object still decodes correctly. + bool ok_model = true; + CHECK(jc::FindString(text, "model_id", 0, &ok_model) == "m"); + CHECK(ok_model); +} + +// Compact separators: the unconditional advance landed exactly on the +// first character of the NEXT KEY, so the no-ok value came back as +// "x�model_id" -- the neighbouring field's name, spelled by nothing +// in the input's own value. +void CompactShortEscapeDoesNotSwallowTheNextKey() { + const std::string text = "{\"hook_name\":\"x\\u1\",\"model_id\":\"m\"}"; + const std::string hook = jc::FindString(text, "hook_name"); + CHECK(hook == "x" + kFFFD + "1"); + CHECK(hook.find("model_id") == std::string::npos); + CHECK(jc::FindString(text, "model_id") == "m"); +} + +// Sequential decoding: after a rejected escape in the FIRST field, `q` must +// sit on that field's closing quote so later fields scanned from `q` stay +// aligned with the object. +void SequentialFieldsStayAlignedAfterARejectedEscape() { + const std::string text = + "{\"a\": \"p\\uZZ\", \"b\": \"second\", \"c\": \"third\"}"; + const size_t start = text.find("p\\uZZ"); + size_t q = start; + bool ok = true; + const std::string a = jc::Unescape(text, q, &ok); + CHECK(!ok); + CHECK(a == "p" + kFFFD + "ZZ"); + CHECK(q == start + 5); // on the closing quote of "a"'s value + CHECK(jc::FindString(text, "b", q) == "second"); + CHECK(jc::FindString(text, "c", q) == "third"); +} + +// A truncated escape at the very end of the buffer: `q` must come to rest +// AT text.size() (the unterminated-string exit), never past it. +void TruncatedEscapeAtEndOfBufferKeepsQInBounds() { + { + std::string text = "\"x\\u12"; // opening quote, then x\u12, no close + size_t q = 1; + bool ok = true; + const std::string value = jc::Unescape(text, q, &ok); + CHECK(!ok); + CHECK(value == "x" + kFFFD + "12"); + CHECK(q == text.size()); + } + { + std::string text = "\"x\\u1"; + size_t q = 1; + bool ok = true; + const std::string value = jc::Unescape(text, q, &ok); + CHECK(!ok); + CHECK(value == "x" + kFFFD + "1"); + CHECK(q == text.size()); + } +} + +// A lead surrogate whose PARTNER escape is the malformed short one: the +// lone-surrogate refusal fires first, then the second escape is decoded on +// its own and refused too -- and the decoder still ends on the real +// closing quote. +void LeadSurrogateWithMalformedPartnerStaysAligned() { + const std::string text = "{\"k\": \"a\\ud83d\\u12\", \"m\": \"v\"}"; + const size_t start = text.find("a\\ud83d"); + size_t q = start; + bool ok = true; + jc::Unescape(text, q, &ok); + CHECK(!ok); + CHECK(q < text.size() && text[q] == '"'); + CHECK(q == text.find("\", \"m\"", start)); + CHECK(jc::FindString(text, "m", q) == "v"); +} + +// Valid escapes are untouched: BMP, a combined surrogate pair, and the +// two-character escapes decode as before, `ok` stays latched true, and `q` +// ends on the closing quote. +void ValidEscapesAreUnchanged() { + const std::string text = + "{\"k\": \"a\\u2603\\ud83d\\ude00\\n\", \"m\": \"v\"}"; + const size_t start = text.find("a\\u2603"); + size_t q = start; + bool ok = true; + const std::string value = jc::Unescape(text, q, &ok); + CHECK(ok); + CHECK(value == "a\xE2\x98\x83\xF0\x9F\x98\x80\n"); + CHECK(q < text.size() && text[q] == '"'); + CHECK(jc::FindString(text, "m", q) == "v"); +} + +} // namespace + +int main() { + ShortEscapeStopsOnTheClosingQuote(); + CompactShortEscapeDoesNotSwallowTheNextKey(); + SequentialFieldsStayAlignedAfterARejectedEscape(); + TruncatedEscapeAtEndOfBufferKeepsQInBounds(); + LeadSurrogateWithMalformedPartnerStaysAligned(); + ValidEscapesAreUnchanged(); + if (g_failures != 0) { + std::cerr << g_failures << " check(s) failed\n"; + return 1; + } + std::cout << "ok\n"; + return 0; +} diff --git a/tests/native/test_spool_reservations.cpp b/tests/native/test_spool_reservations.cpp index 5a399fa1a..8acd5cba4 100644 --- a/tests/native/test_spool_reservations.cpp +++ b/tests/native/test_spool_reservations.cpp @@ -1,5 +1,5 @@ -// Spool capacity accounting under the two access patterns the counters have -// to survive at once: +// Spool capacity accounting under the access patterns the counters have to +// survive at once: // // 1. A SECOND Spool object (an uploader in another thread or process) on // the same root removes a ready file. The first object's committed @@ -8,12 +8,21 @@ // 2. While one stager holds a reservation but has written NOTHING yet, // another stager trips that reconciliation. The scan sees an empty // directory; it must not erase the first stager's reservation. +// 3. A ready file created by a SECOND Spool object after this one's +// Open() is reached by this one's retry or EEXIST-loser path. This +// object's committed account has never counted it, so it has to count +// it now -- but exactly once, however many times that path is met. +// 4. That same retry charge arrives while this object holds an in-flight +// reservation. It is a charge like any other and must pass the +// capacity decision, not be added on top of the reservation. // -// A single counter cannot satisfy both: replacing it with the scan fixes -// (1) and breaks (2) -- with max 1500, A reserved 1000, B's 1000 scanned an -// empty root, and both were admitted (2000 bytes on disk against a 1500 -// limit). Keeping committed bytes and in-flight reservations as separate -// accounts, and letting the scan overwrite only the first, satisfies both. +// A single counter cannot satisfy (1) and (2) at once: replacing it with +// the scan fixes (1) and breaks (2) -- with max 1500, A reserved 1000, B's +// 1000 scanned an empty root, and both were admitted (2000 bytes on disk +// against a 1500 limit). Keeping committed bytes and in-flight reservations +// as separate accounts, and letting the scan overwrite only the first, +// satisfies both. Neither aggregate can decide (3) on its own, so the +// committed account also carries the set of ready PATHS it includes. // // Built and run by tests/test_native_spool_reservations.py. @@ -193,9 +202,11 @@ void TestReconciliationKeepsAnInflightReservation() { CHECK(spool.Snapshot().bytes == 1500); } -// A refused stage and a lost link() race both release their reservation -// (the reserved account must return to zero, or the spool leaks capacity). -void TestARefusedStageLeavesNoReservationBehind() { +// A stage that the cap refuses, and a retry of a pack this object already +// staged, both leave the accounts exactly where they were. The refusal is the +// cheap half -- the capacity check returns before reserved_bytes_ moves, so +// there is no reservation to give back. +void TestARefusedStageAndARetryLeaveTheAccountsAlone() { const std::string root = FreshRoot("refused"); dmi_store::SpoolConfig config{root, 1500}; dmi_store::Spool spool; @@ -213,6 +224,215 @@ void TestARefusedStageLeavesNoReservationBehind() { dmi_store::SpoolStatus::kOk); CHECK(spool.Snapshot().bytes == 1000); CHECK(spool.Snapshot().entries == 1); + // And the refusal really did leave room: the 500 bytes still under the cap + // are stageable. + CHECK(StageBytes(spool, 3, 500, &out, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(spool.Snapshot().bytes == 1500); +} + +// The one reachable path that TAKES a reservation and then has to give it +// back: the EEXIST loser. It reserves, loses link() to a second Spool object +// on the same root, releases its reservation and accounts the winner's file +// instead. A release that leaked leaves the bytes on disk unchanged -- the +// file is there either way -- and shows up only where the reservation is +// still charged: the snapshot, and the room left under the cap. So the +// observable that matters is that a stage which FITS is still admitted. +void TestALostLinkRaceReleasesItsReservation() { + const std::string root = FreshRoot("lost-link-release"); + dmi_store::SpoolConfig config{root, 1500}; + dmi_store::Spool loser, winner; + std::string error; + CHECK(dmi_store::Spool::Open(config, &loser, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(dmi_store::Spool::Open(config, &winner, &error) == + dmi_store::SpoolStatus::kOk); + + dmi_store::StagedPack winner_out; + loser.SetStageHookForTesting([&] { + std::string inner; + CHECK(StageBytes(winner, 1, 1000, &winner_out, &inner) == + dmi_store::SpoolStatus::kOk); + }); + dmi_store::StagedPack out; + CHECK(StageBytes(loser, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + loser.SetStageHookForTesting(nullptr); + CHECK(loser.Snapshot().bytes == 1000); + CHECK(loser.Snapshot().entries == 1); + + // 1000 committed under a 1500 cap: 500 fits, and only fits if the loser's + // reservation went back. + const dmi_store::SpoolStatus status = + StageBytes(loser, 2, 500, &out, &error); + CHECK(status == dmi_store::SpoolStatus::kOk); + if (status != dmi_store::SpoolStatus::kOk) { + std::cerr << "a stage that fits was refused: " + << dmi_store::SpoolStatusName(status) << " " << error << "\n"; + } + CHECK(BytesOnDisk(root) == 1500); + CHECK(loser.Snapshot().bytes == 1500); + CHECK(loser.Snapshot().entries == 2); +} + +// (3) A ready file this object NEVER accounted for. The retry above is the +// easy half: the object had already counted that file, so adding nothing is +// right. When a SECOND Spool object on the same root creates the file after +// this one's Open(), the retry path meets a file worth 1000 bytes that this +// object's committed account has never seen -- and adding nothing there +// leaves the cap judged against 0. +// +// Python takes the retry and the EEXIST-loser paths through +// _account_ready_locked (spool.py:124 and :159), which is a no-op only when +// the path is already in _accounted_ready. Its measured answer: the retry +// snapshot reads 1000, and the next 1000-byte stage raises SpoolFullError +// ("2000 > 1500") with 1000 bytes on disk. +void TestARetryOfAnotherObjectsReadyFileIsAccounted() { + const std::string root = FreshRoot("foreign-retry"); + dmi_store::SpoolConfig config{root, 1500}; + dmi_store::Spool writer, other; + std::string error; + CHECK(dmi_store::Spool::Open(config, &writer, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(dmi_store::Spool::Open(config, &other, &error) == + dmi_store::SpoolStatus::kOk); + + // `other` writes the ready file; `writer` has never seen it. + dmi_store::StagedPack out; + CHECK(StageBytes(other, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(writer.Snapshot().bytes == 0); + + // The same pack through `writer`: the ready file is already there, so this + // is the retry path. It succeeds -- the content matches -- and it has to + // leave the 1000 bytes IN the writer's account. + CHECK(StageBytes(writer, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(writer.Snapshot().bytes == 1000); + CHECK(writer.Snapshot().entries == 1); + + // So a second, different pack of 1000 does not fit under the 1500 cap. + const dmi_store::SpoolStatus status = + StageBytes(writer, 2, 1000, &out, &error); + CHECK(status == dmi_store::SpoolStatus::kFull); + if (status != dmi_store::SpoolStatus::kFull) { + std::cerr << "second pack was admitted: " + << dmi_store::SpoolStatusName(status) << "\n"; + } + CHECK(BytesOnDisk(root) == 1000); + + // And a retry stays idempotent: running it twice more must not count the + // same path again. + CHECK(StageBytes(writer, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(StageBytes(writer, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(writer.Snapshot().bytes == 1000); + CHECK(writer.Snapshot().entries == 1); +} + +// (4) A retry while THIS object holds an in-flight reservation. The retry +// path used to charge a foreign ready file without passing the capacity +// decision, so a paused reservation and a retry could both be admitted past +// max_bytes_ (with a 1000-byte reservation and a 1000-byte foreign file +// under a 1500 cap, both completed at 2000). The retry must now be refused +// like any other charge that does not fit. +void TestARetryCannotOversubscribeAnInflightReservation() { + const std::string root = FreshRoot("retry-vs-reservation"); + dmi_store::SpoolConfig config{root, 1500}; + dmi_store::Spool spool, other; + std::string error; + CHECK(dmi_store::Spool::Open(config, &spool, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(dmi_store::Spool::Open(config, &other, &error) == + dmi_store::SpoolStatus::kOk); + + dmi_store::SpoolStatus retry_status = dmi_store::SpoolStatus::kIo; + dmi_store::SpoolStatus second_retry_status = dmi_store::SpoolStatus::kIo; + std::string retry_error; + spool.SetStageHookForTesting([&] { + // A's 1000-byte reservation is held and nothing of A's is on disk. A + // second Spool object stages a 1000-byte pack; the retry of that pack + // must be refused: committed + reserved + 1000 = 2000 > 1500. + dmi_store::StagedPack foreign; + std::string inner; + CHECK(StageBytes(other, 2, 1000, &foreign, &inner) == + dmi_store::SpoolStatus::kOk); + dmi_store::StagedPack retried; + retry_status = StageBytes(spool, 2, 1000, &retried, &retry_error); + // And the SAME call must answer the same way after its own + // reconciliation scan ledgered the foreign path: an identical retry + // cannot flip from kFull to kOk without any state change but its own. + second_retry_status = StageBytes(spool, 2, 1000, &retried, &retry_error); + }); + dmi_store::StagedPack out; + CHECK(StageBytes(spool, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + spool.SetStageHookForTesting(nullptr); + + CHECK(retry_status == dmi_store::SpoolStatus::kFull); + if (retry_status != dmi_store::SpoolStatus::kFull) { + std::cerr << "a concurrent retry was admitted: " + << dmi_store::SpoolStatusName(retry_status) << " " + << retry_error << "\n"; + } + CHECK(second_retry_status == dmi_store::SpoolStatus::kFull); + if (second_retry_status != dmi_store::SpoolStatus::kFull) { + std::cerr << "an identical retry flipped to " + << dmi_store::SpoolStatusName(second_retry_status) << " " + << retry_error << "\n"; + } + // What is refused is acknowledging the foreign pack ON TOP of the + // outstanding reservation. The capacity decision's reconciliation scan + // then records the foreign file anyway -- it IS on disk, and no object can + // stop the second writer -- so once A commits the account holds both, at + // the physical truth. 2000 bytes on disk against a 1500 cap is the second + // writer's doing, not this object admitting a stage it should have + // refused. + CHECK(spool.Snapshot().bytes == 2000); + CHECK(spool.Snapshot().entries == 2); + CHECK(BytesOnDisk(root) == 2000); +} + +// The EEXIST-loser path reaches the same file by a different route: both +// objects write the same pack, and the one whose link() loses must still +// end up with those bytes in its account. +void TestAnEexistLoserAccountsForTheWinnersFile() { + const std::string root = FreshRoot("eexist-loser"); + dmi_store::SpoolConfig config{root, 1500}; + dmi_store::Spool loser, winner; + std::string error; + CHECK(dmi_store::Spool::Open(config, &loser, &error) == + dmi_store::SpoolStatus::kOk); + CHECK(dmi_store::Spool::Open(config, &winner, &error) == + dmi_store::SpoolStatus::kOk); + + // The loser reserves, and while it is paused the winner links the ready + // file. The loser's own link() then fails with EEXIST -- the retry branch + // at the top of Stage ran before the file existed, so this is the only way + // to that code path. + dmi_store::StagedPack winner_out; + loser.SetStageHookForTesting([&] { + std::string inner; + CHECK(StageBytes(winner, 1, 1000, &winner_out, &inner) == + dmi_store::SpoolStatus::kOk); + }); + dmi_store::StagedPack out; + CHECK(StageBytes(loser, 1, 1000, &out, &error) == + dmi_store::SpoolStatus::kOk); + loser.SetStageHookForTesting(nullptr); + + CHECK(BytesOnDisk(root) == 1000); + CHECK(loser.Snapshot().bytes == 1000); + CHECK(loser.Snapshot().entries == 1); + const dmi_store::SpoolStatus status = + StageBytes(loser, 2, 1000, &out, &error); + CHECK(status == dmi_store::SpoolStatus::kFull); + if (status != dmi_store::SpoolStatus::kFull) { + std::cerr << "loser admitted a second pack: " + << dmi_store::SpoolStatusName(status) << "\n"; + } + CHECK(BytesOnDisk(root) == 1000); } void TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity() { @@ -237,7 +457,11 @@ void TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity() { int main() { TestSerialRemoveThroughAnotherSpoolIsReconciled(); TestReconciliationKeepsAnInflightReservation(); - TestARefusedStageLeavesNoReservationBehind(); + TestARefusedStageAndARetryLeaveTheAccountsAlone(); + TestALostLinkRaceReleasesItsReservation(); + TestARetryOfAnotherObjectsReadyFileIsAccounted(); + TestARetryCannotOversubscribeAnInflightReservation(); + TestAnEexistLoserAccountsForTheWinnersFile(); TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity(); if (g_failures != 0) { std::cerr << g_failures << " check(s) failed\n"; diff --git a/tests/test_engine_runtime_api.py b/tests/test_engine_runtime_api.py index e42069d1b..eaf9e7862 100644 --- a/tests/test_engine_runtime_api.py +++ b/tests/test_engine_runtime_api.py @@ -780,4 +780,9 @@ def test_the_sink_loader_tolerates_a_transport_module_without_the_loader( ) monkeypatch.setitem(sys.modules, "dmi.transport.native", fake) monkeypatch.setattr(dmi.transport, "native", fake, raising=False) - assert native_sink._load_native_sink_extension() is not None + # The module the loader returns, not merely "not None": the loader can + # only raise or hand back _load_named_extension's result, so `is not + # None` was satisfied by construction and would have held for any other + # object too. Same assertion style as the two tests above. + assert native_sink._load_native_sink_extension().RING_TYPES_ARE_STANDINS \ + is True diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py index 2236fe0f8..9208bc2a9 100644 --- a/tests/test_native_adapter_torch.py +++ b/tests/test_native_adapter_torch.py @@ -274,16 +274,25 @@ def test_a_scalar_envelope_is_admitted(tmp_path): assert payload == torch.tensor(3.5, dtype=torch.float32).numpy().tobytes() -@pytest.mark.parametrize("field, value", [ - ("layer_number", 0.5), - ("captured_at_ns", 123.5), - ("step_number", -1), +@pytest.mark.parametrize("field, value, reason", [ + ("layer_number", 0.5, "capture metadata field is not an integer"), + ("captured_at_ns", 123.5, "capture metadata field is not an integer"), + ("step_number", -1, "capture metadata integer is out of range"), ]) def test_invalid_numeric_metadata_is_refused_not_converted(tmp_path, field, - value): + value, reason): """What CaptureMetadata refuses, the sink refuses -- nothing is persisted. These used to be persisted as 0, 123 and 18446744073709551615. + + The exact `reason` is pinned, not just the word "metadata": every + metadata refusal carries RowStatusName(kBadMetadata) == "invalid capture + metadata", so matching on "metadata" alone would be satisfied by any + other refusal -- including the unrelated "descriptor requires metadata + JSON followed by one payload slice". The -1 case in particular has to + land on OUT OF RANGE, which is what the kUnsigned domain buys: without + it step_number=-1 parses fine and is packed as 18446744073709551615. + (Same style as tests/test_native_pack_sink.py's parametrised `reason`.) """ from dmi.storage.capture.model import CaptureStorageError @@ -292,11 +301,14 @@ def test_invalid_numeric_metadata_is_refused_not_converted(tmp_path, field, with pytest.raises((CaptureStorageError, ValueError, TypeError)): CaptureMetadata.from_mapping(meta) sink, lease = _make_sink(tmp_path) - with pytest.raises(RuntimeError, match="metadata"): + with pytest.raises(RuntimeError) as refusal: sink.submit_envelope( LAYOUT, [_row(meta, 0, 64, "float32")], torch.zeros(16, dtype=torch.float32), ) + assert str(refusal.value) == ( + f"NativePackSink: invalid capture metadata: {reason}" + ), (field, value, str(refusal.value)) assert sink.snapshot()["submitted_records"] == 0 @@ -308,18 +320,41 @@ def test_a_fractional_shape_dimension_is_refused_not_truncated(tmp_path): with pytest.raises((CaptureStorageError, ValueError, TypeError)): CaptureMetadata.from_mapping(meta) sink, lease = _make_sink(tmp_path) - with pytest.raises(RuntimeError, match="metadata"): + # The specific refusal, not merely "metadata": a truncated 16.5 would + # have been admitted as (16,), matching the envelope shape, so the + # message that has to come back is the shape parser's own. + with pytest.raises(RuntimeError) as refusal: sink.submit_envelope( LAYOUT, [_row(meta, 0, 64, "float32", shape=(16,))], torch.zeros(16, dtype=torch.float32), ) + assert str(refusal.value) == ( + "NativePackSink: invalid capture metadata: " + "capture shape must be an integer list" + ), str(refusal.value) assert sink.snapshot()["submitted_records"] == 0 -def test_a_non_bmp_hook_name_round_trips(tmp_path): +@pytest.mark.parametrize("hook_name", [ + "block\U0001F600", + # Exactly 512 UTF-8 bytes -- the limit _validate_text and ValidText both + # apply. This is the case the round-trip above cannot see: reading the + # name back off the pack proves nothing, because the footer writer decodes + # each UTF-8 sequence and re-emits \uXXXX, so a decoder that left the + # surrogate pair as two three-byte CESU-8 sequences produced a footer + # holding exactly the same 😀 a correct decoder does, and + # json.loads recombined it either way (measured: this test passed with the + # combine branch compiled out). The BYTE COUNT does differ -- 514 against + # 512 -- so at the limit the broken decoder is refused outright as + # "invalid capture metadata" where Python admits the record. + "b" * 508 + "\U0001F600", +]) +def test_a_non_bmp_hook_name_round_trips(tmp_path, hook_name): """json.dumps writes U+1F600 as a surrogate pair; the sink must combine it.""" meta = _meta(0) - meta["hook_name"] = "block\U0001F600" + meta["hook_name"] = hook_name + # The oracle admits both, so native must too. + assert CaptureMetadata.from_mapping(meta).hook_name == hook_name row = _row(meta, 0, 64, "float32") assert "\\ud83d\\ude00" in row["metadata_json"] sink, lease = _make_sink(tmp_path) @@ -327,7 +362,7 @@ def test_a_non_bmp_hook_name_round_trips(tmp_path): assert sink.flush_and_wait(30.0) sink.rethrow_if_failed() ((staged, _),) = _read_staged(tmp_path) - assert staged.hook_name == "block\U0001F600" + assert staged.hook_name == hook_name def test_the_sink_derives_from_the_engines_record_sink(tmp_path): diff --git a/tests/test_native_hydration_footer_fields.py b/tests/test_native_hydration_footer_fields.py index 83b76ebd1..ac3969220 100644 --- a/tests/test_native_hydration_footer_fields.py +++ b/tests/test_native_hydration_footer_fields.py @@ -1,8 +1,9 @@ """The footer-binding decoder: rendered footer rows against decoded catalog text. hydrate binds every catalog descriptor to the pack footer by comparing the -32 fields of the footer's rendered VALUES row (pack_index's renderer, SQL -escaping and all) against the catalog row the reader returns (TSV escapes +fields of the footer's rendered VALUES row -- one per CAPTURE_COLUMNS entry +barring index_version (pack_index's renderer, SQL +escaping and all) -- against the catalog row the reader returns (TSV escapes already undone, a NULL arriving as the empty string). The two sides only compare correctly in ONE representation, so the decoder is pinned here on the CPU gate through the driver's session-less `footer_row_fields` op: @@ -26,9 +27,15 @@ import pytest +from dmi.storage.capture.clickhouse_schema import CAPTURE_COLUMNS + REPO_ROOT = Path(__file__).resolve().parents[1] DRIVER = REPO_ROOT / "native" / "build" / "conformance_catalog" +# The footer's VALUES row is every capture column but the trailing +# index_version, which the indexer supplies rather than the pack. +FOOTER_COLUMNS = CAPTURE_COLUMNS[:-1] + pytestmark = [ pytest.mark.cpu, pytest.mark.skipif( @@ -122,24 +129,67 @@ def test_uuid_literals_arrays_and_numbers_split_at_top_level(): assert all(f["matches"] for f in fields), fields -def test_a_full_32_field_row_decodes_to_32_fields(): - """A rank-2 shape's inner comma must not split the row.""" - strings = ["capture-0", "t", "e", "r", "s", "q", "n", "m", "mr"] - row = ",".join( - [_quote(s) for s in strings] - + ["NULL", _quote("v"), _quote("block\\resid"), "3", "0", "0", "0", - "1", "0", _quote("float32"), "[2,8]", "1700000000000000000", - "toUUID('0190e8d0-4b2a-7c3e-9f00-0123456789ab')", _quote("local"), - _quote("packs/a\\b.dmi-pack"), "4096", _quote("c" * 64), "1", "64", - "64", "64", _quote("none"), _quote("deadbeef")] +def test_a_full_footer_row_decodes_to_one_field_per_footer_column(): + """A rank-2 shape's inner comma must not split the row. + + Width and positions come from CAPTURE_COLUMNS, the oracle this decoder + is a port of, not from a hand-counted 32. The footer carries every + capture column except the trailing index_version, and + hydration.cpp:708 guards on that same width -- so with the count + hard-coded, adding a column left this row at 32 tokens, the test + passing, and the native guard silently out of date. The row is + assembled per column name for the same reason: a new column has no + rendering here and says so by name. + """ + pack_id = "0190e8d0-4b2a-7c3e-9f00-0123456789ab" + rendered = { + "capture_id": _quote("capture-0"), + "tenant_id": _quote("t"), + "experiment_id": _quote("e"), + "run_id": _quote("r"), + "session_id": _quote("s"), + "request_id": _quote("q"), + "sequence_id": _quote("n"), + "model_id": _quote("m"), + "model_revision": _quote("mr"), + "adapter_revision": "NULL", + "capture_policy_version": _quote("v"), + "hook_name": _quote("block\\resid"), + "layer_number": "3", + "producer_rank": "0", + "step_number": "0", + "token_start": "0", + "token_end": "1", + "batch_position": "0", + "dtype": _quote("float32"), + "shape": "[2,8]", + "captured_at_ns": "1700000000000000000", + "pack_id": f"toUUID('{pack_id}')", + "store_id": _quote("local"), + "object_key": _quote("packs/a\\b.dmi-pack"), + "object_bytes": "4096", + "pack_checksum": _quote("c" * 64), + "pack_record_count": "1", + "payload_offset": "64", + "stored_length": "64", + "decoded_length": "64", + "codec": _quote("none"), + "payload_checksum": _quote("deadbeef"), + } + assert sorted(rendered) == sorted(FOOTER_COLUMNS), ( + sorted(set(FOOTER_COLUMNS) - set(rendered)), + sorted(set(rendered) - set(FOOTER_COLUMNS)), ) + + row = ",".join(rendered[column] for column in FOOTER_COLUMNS) fields = _fields(row) - assert len(fields) == 32, fields - assert fields[11]["text"] == "block\\resid" - assert fields[19]["text"] == "[2,8]" - assert fields[21]["text"] == "0190e8d0-4b2a-7c3e-9f00-0123456789ab" - assert fields[23]["text"] == "packs/a\\b.dmi-pack" - assert fields[9]["null"] is True + assert len(fields) == len(FOOTER_COLUMNS), fields + by_column = dict(zip(FOOTER_COLUMNS, fields)) + assert by_column["hook_name"]["text"] == "block\\resid" + assert by_column["shape"]["text"] == "[2,8]" + assert by_column["pack_id"]["text"] == pack_id + assert by_column["object_key"]["text"] == "packs/a\\b.dmi-pack" + assert by_column["adapter_revision"]["null"] is True def test_a_mismatching_decoded_value_is_still_a_mismatch(): @@ -203,3 +253,39 @@ def test_the_resolved_tuple_keeps_values_the_footer_binding_compares(): assert _tuple_fields("('h',[2,8],0,'none')") == ["h", "[2,8]", "0", "none"] # A value that merely CONTAINS the token is untouched. assert _tuple_fields("('NULLABLE','a NULL b')") == ["NULLABLE", "a NULL b"] + + +# --- the two decoders have to agree, byte for byte --------------------------- +# +# hydrate compares the tuple decoder's output (above) against the footer +# decoder's (unquote_sql, sql_quote's map entry for entry). Where the two +# disagree on ONE byte the binding refuses a legal capture with "catalog +# descriptor does not match the pack footer: field N", and nothing but a +# live hydrate of a capture carrying that byte could see it. +# +# The rendering below is MEASURED, not assumed: `SELECT tuple(concat('block', +# char(N), 'resid')) FORMAT TSV` was driven against the live server for the +# whole 1..127 range (plus char(0)). ClickHouse escapes exactly eight bytes +# inside a tuple -- \0 \b \t \n \f \r \' \\ -- and leaves every other one +# RAW, including 0x07 (BEL) and 0x0B (VT), which sql_quote does rewrite (as +# \a and \v). So the two escape SETS are deliberately different, and it is +# the decoded bytes, not the spellings, that have to match. +TUPLE_RENDERING = { + 0x00: "\\0", 0x07: "\x07", 0x08: "\\b", 0x09: "\\t", 0x0A: "\\n", + 0x0B: "\x0b", 0x0C: "\\f", 0x0D: "\\r", 0x27: "\\'", 0x5C: "\\\\", +} + + +@pytest.mark.parametrize("byte", sorted(TUPLE_RENDERING)) +def test_the_tuple_and_footer_decoders_agree_on_every_byte_clickhouse_rewrites(byte): + """Both halves of the footer binding decode to the same bytes. + + \\b and \\f decoded to the letters "b" and "f" on the catalog side while + the footer side decoded them to 0x08 and 0x0C, so a hook_name carrying + either (legal: _validate_text only asks for non-empty UTF-8 under 512 + bytes) staged, uploaded and indexed, and then failed to hydrate. + """ + value = "block" + chr(byte) + "resid" + (catalog,) = _tuple_fields(f"('block{TUPLE_RENDERING[byte]}resid')") + assert catalog == value + assert _fields(_quote(value), [catalog])[0]["matches"] is True diff --git a/tests/test_native_json_unescape.py b/tests/test_native_json_unescape.py new file mode 100644 index 000000000..3109f2212 --- /dev/null +++ b/tests/test_native_json_unescape.py @@ -0,0 +1,61 @@ +"""Unescape's \\uXXXX failure path: the decoder must stay on the string. + +Compiles and runs tests/native/test_json_unescape.cpp against the real +common/json.cpp. hex4 validates the four digits, but the advance (`q += 5`) +ran BEFORE its verdict, so a malformed short escape whose "digits" include +the field's real closing quote (`"x\\u12", ...`) stepped past that quote and +the value silently swallowed following JSON text -- every FindString caller +with no `ok` flag (pack footer parsing, the reader's cursor) received a +value containing structural JSON -- and a `\\u12` at the end of the buffer +left q strictly past text.size(), breaking json.h's "returns with `q` on +the closing quote" contract. + +Needs a C++17 compiler (no libcrypto, unlike conformance_spool). +""" + +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path + +import pytest + + +@pytest.mark.cpu +def test_unescape_failure_keeps_the_decoder_on_the_string(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_json_unescape.cpp" + executable = tmp_path / "test_json_unescape" + compile_result = subprocess.run( + [ + compiler, + "-std=c++17", + "-O0", + f"-I{csrc}", + str(source), + str(csrc / "common" / "json.cpp"), + "-o", + str(executable), + ], + capture_output=True, + text=True, + check=False, + ) + assert compile_result.returncode == 0, ( + compile_result.stdout + compile_result.stderr + ) + + run_result = subprocess.run( + [str(executable)], + capture_output=True, + text=True, + check=False, + timeout=120, + ) + assert run_result.returncode == 0, run_result.stdout + run_result.stderr diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py index 9bac1ad1b..4c7c122ab 100644 --- a/tests/test_native_pack_sink.py +++ b/tests/test_native_pack_sink.py @@ -1124,17 +1124,86 @@ def _read_staged_metadata(root): @pytest.mark.parametrize("hook_name", ["block\U0001F600", "block中文"]) def test_a_non_bmp_identifier_survives_the_metadata_decoder(sink, tmp_path, hook_name): - """A surrogate pair is ONE code point; the BMP name is the control.""" + """A surrogate pair is ONE code point; the BMP name is the control. + + The assertion that matters is on the DECODER's bytes, taken straight out + of `parse_metadata` before anything re-serializes them. Reading hook_name + back off the staged pack cannot see this bug at all: the footer writer + decodes each UTF-8 sequence and re-emits \\uXXXX, so the decoder's broken + CESU-8 (ED A0 BD ED B8 80) is written back out as a correct \\ud83d\\ude00 + pair that json.loads recombines into 😀. Measured -- both this test and + test_native_adapter_torch's round-trip passed with the surrogate combine + branch compiled out. + """ _open(sink, tmp_path / "spool") metadata = _row_meta(0, dtype="uint8", shape=[64], hook_name=hook_name) assert "\\ud83d\\ude00" in metadata or "\\u4e2d" in metadata, metadata + decoded = sink.call(op="parse_metadata", metadata_json=metadata) + assert decoded["ok"], decoded + assert decoded["hook_name_hex"] == hook_name.encode("utf-8").hex() response = _submit_row(sink, metadata, bytes(64), "uint8", [64]) assert response["ok"], response assert sink.call(op="flush", timeout=30)["ok"] assert sink.call(op="close", timeout=30)["snapshot"]["persisted_records"] == 1 (staged,) = _read_staged_metadata(tmp_path / "spool") assert staged.hook_name == hook_name - assert staged.hook_name.encode("utf-8") == hook_name.encode("utf-8") + + +@pytest.mark.parametrize("hook_name", ["block\ud83d", "block\udcff"]) +def test_a_lone_surrogate_is_refused_not_persisted_as_cesu8(sink, tmp_path, + hook_name): + """A surrogate with no partner has no UTF-8 encoding at all. + + The oracle cannot even build the record: _validate_text calls + .encode("utf-8"), which raises UnicodeEncodeError. Native kept the + three-byte form -- \\ud83d became ED A0 BD -- which ValidText admits, + because it checks lead/continuation byte SHAPE and not the surrogate + range, so the record was packed and inserted as CESU-8. + """ + mapping = _meta(0, dtype="uint8", shape=(64,)).to_mapping() + mapping["hook_name"] = hook_name + with pytest.raises(PackFormatError, match="surrogates not allowed"): + CaptureMetadata.from_mapping(mapping) + + _open(sink, tmp_path / "spool") + metadata = _row_meta(0, dtype="uint8", shape=[64], hook_name=hook_name) + assert "\\ud83d" in metadata or "\\udcff" in metadata, metadata + assert sink.call(op="parse_metadata", metadata_json=metadata)["ok"] is False + assert _submit_row(sink, metadata, bytes(64), "uint8", [64])["ok"] is False + # The mapping path is the native side of CaptureMetadata.from_mapping and + # refuses it too, rather than admitting a name the oracle cannot hold. + refused = sink.call(op="submit", metadata=json.loads(metadata), + payload_b64=base64.b64encode(bytes(64)).decode()) + assert refused["ok"] is False, refused + assert sink.call(op="snapshot")["snapshot"]["submitted_records"] == 0 + + +@pytest.mark.parametrize("bad_name", ["block\\uZZZZ", "block\\u12"]) +def test_a_malformed_unicode_escape_is_refused_not_decoded(sink, tmp_path, + bad_name): + """\\uZZZZ has no code point; json.loads raises "Invalid \\uXXXX escape". + + The four digits were never checked -- `h <= '9' ? h - '0' : (h | 0x20) - + 'a' + 10` accepts anything -- so \\uZZZZ decoded to U+25553 and the field + was packed as F0 A5 95 93, a name nothing in the request ever spelled. + \\u12 is the SHORT variant: its "digits" run into the field's closing + quote, which the decoder's unconditional `q += 5` used to step past, so + the value swallowed the following JSON text up to the next quote + (tests/native/test_json_unescape.cpp pins the alignment itself); the + refusal here must hold with the quote intact. + """ + _open(sink, tmp_path / "spool") + valid = _row_meta(0, dtype="uint8", shape=[64]) + assert '"hook_name": "h"' in valid, valid + # A single backslash in the metadata TEXT: json.dumps doubles it on the + # wire, and the driver's outer FindString undoes exactly that doubling. + metadata = valid.replace('"hook_name": "h"', f'"hook_name": "{bad_name}"') + with pytest.raises(ValueError): + json.loads(metadata) + + assert sink.call(op="parse_metadata", metadata_json=metadata)["ok"] is False + assert _submit_row(sink, metadata, bytes(64), "uint8", [64])["ok"] is False + assert sink.call(op="snapshot")["snapshot"]["submitted_records"] == 0 @pytest.mark.parametrize("field, value, reason", [ diff --git a/tests/test_native_reader_parity_live.py b/tests/test_native_reader_parity_live.py index 8ad306dcd..83d3e2aab 100644 --- a/tests/test_native_reader_parity_live.py +++ b/tests/test_native_reader_parity_live.py @@ -2410,8 +2410,14 @@ def _native_read(driver, op, sel, fake_s3, **limits): insecure=True, **limits) +# The list is the set of bytes ClickHouse actually escapes when it renders a +# tuple, measured against this server with `SELECT tuple(concat('block', +# char(N), 'resid')) FORMAT TSV` across the control range: \b \t \n \f \r \' +# and \\ (plus \0, left out here because a NUL in an object key is a separate +# question). 0x07 and 0x0B travel RAW inside a tuple and so need no entry. @pytest.mark.parametrize("hook_name", [ "block\\resid", "block'quoted", "block\tresid", + "block\nresid", "block\rresid", "block\bresid", "block\fresid", ]) def test_hydrate_accepts_metadata_the_sql_escaper_rewrites(fake_s3, hook_name): """The footer binding compares DECODED values, not escaped footer text. @@ -2421,6 +2427,12 @@ def test_hydrate_accepts_metadata_the_sql_escaper_rewrites(fake_s3, hook_name): two representations refused these three valid captures with "catalog descriptor does not match the pack footer: field 12". Python hydrates them; native must too, and hand back the same 16 bytes. + + Decoding is not enough on its own: the two decoders have to reach the + same BYTE. \\b and \\f decoded to the letters "b" and "f" on the catalog + side while the footer side decoded them to 0x08 and 0x0C, so a hook_name + carrying either staged, uploaded and indexed and then failed that same + field-12 binding. """ import base64 as _base64 diff --git a/tests/test_native_sink_ring_e2e.py b/tests/test_native_sink_ring_e2e.py index d5c866712..a44f14f46 100644 --- a/tests/test_native_sink_ring_e2e.py +++ b/tests/test_native_sink_ring_e2e.py @@ -50,6 +50,13 @@ class _CaptureHookRuntime: def __init__(self, runtime) -> None: self._runtime = runtime self.metadata = None + # RecordRuntime.emit_output's return value is the ONLY observable + # that says which transport carried each record: it is OVERSIZED + # exactly on the CPU-direct branch (records.py:333) and a ring + # reservation otherwise. Persisted counts and payload bytes are + # identical either way, so without this the test cannot tell a + # record that went through the Ring from one that bypassed it. + self.reservations = [] def should_emit(self, hook): return True @@ -62,7 +69,9 @@ def prepare_output(self, *, hook, output_index, output_id, output_spec, entry = ProducerPlanBuilder().record_output( output_id=output_id, output_spec=output_spec, output=output, ) - return self._runtime.emit_output(entry, self.metadata, output) + reservation = self._runtime.emit_output(entry, self.metadata, output) + self.reservations.append(reservation) + return reservation def _metadata(capture_id: str, tensor: torch.Tensor, *, step: int): @@ -125,7 +134,14 @@ def test_native_pack_sink_is_the_engines_record_sink(tmp_path: Path): def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path): - """The explicit entry point: record_sink=create_native_pack_sink(...).""" + """The explicit entry point: record_sink=create_native_pack_sink(...). + + Three records, and the point is that they do NOT all take the same + route: two fit the 64 KiB payload ring and one does not. The persisted + count and the payload bytes are the same whichever transport ran, so + the per-record StepReservation is what pins the split. + """ + from dmi.adapters.base import StepReservation from dmi.api.v1 import ( HookPointV1, HookSpecV1, MonitoringEngine, TransportSpec, ) @@ -152,7 +168,8 @@ def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path): hook_runtime.metadata = _metadata("capture-ring", first, step=0) hook(first.cuda()) - # Above the 64 KiB Ring: the CPU-direct fallback, same sink. + # 128 KiB, above the 64 KiB Ring: the CPU-direct fallback, same + # sink. Asserted, not just asserted-in-a-comment, below. second = torch.arange(32 * 1024, dtype=torch.float32) hook_runtime.metadata = _metadata("capture-direct", second, step=1) hook(second.cuda()) @@ -167,6 +184,15 @@ def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path): snapshot = handle.native_sink.snapshot() assert snapshot["persisted_records"] == 3, snapshot assert snapshot["failures"] == 0, snapshot + # 48 bytes, then 128 KiB, then 4 bytes against a 64 KiB payload + # ring: reserved in the ring, CPU-direct, reserved in the ring. + # RESERVED and not FLUSHED because neither ring record comes near + # filling the ring, so nothing forces a drain. + assert hook_runtime.reservations == [ + StepReservation.RESERVED, + StepReservation.OVERSIZED, + StepReservation.RESERVED, + ], hook_runtime.reservations finally: engine.close()