diff --git a/docs/en/antalya/cas/architecture/manifests-and-refs.md b/docs/en/antalya/cas/architecture/manifests-and-refs.md index 96f2b9114cd96..efa2a6708bef4 100644 --- a/docs/en/antalya/cas/architecture/manifests-and-refs.md +++ b/docs/en/antalya/cas/architecture/manifests-and-refs.md @@ -55,6 +55,23 @@ before any owner transition names it. Rewriting a part therefore writes a **new* the **same** blobs and moves the ref in one ref-log record — a repoint, covered in full on the [part-lifecycle page](/antalya/cas/architecture/part-lifecycle#repoint). +The writer retains a compact record after its own successful `stageManifest`: the confirmed object +incarnation and the blob references (`blob_refs`) in the order the caller passed the entries. +The encoded body sorts entries by path, so this order can differ from a decoded body. Only the +order of dependency checks and `BlobReuseAdopt` events depends on it. The matching `promote` requires this record and +never reads the immutable body back. A manifest not staged by that `PartWriteTxn` is an invariant +violation: `promote` raises `LOGICAL_ERROR` before the ref-log append, even if another transaction +already committed that manifest. Repeating `promote` on the same `PartWriteTxn` for its own +already committed manifest remains an idempotent no-op. + +The record only shows that the body existed when the conditional create was confirmed. It is not +a fresh existence check or proof of GC protection. If the body is removed after staging, `promote` +does not detect that removal and can still commit the ref. This also applies to external deletion +by a bucket lifecycle rule, expiry rule, or manual operation. A backend read of the missing body +raises `FILE_DOESNT_EXIST`; a manifest decode cache hit can delay detection. `fsck` reports a +dangling reference when it confirms that the body is absent and the ref still names it. +Do not put lifecycle or expiry rules on the pool prefix. + ## Manifest lifecycle and the orphan sweep {#manifest-lifecycle} ```mermaid diff --git a/docs/en/antalya/cas/architecture/part-lifecycle.md b/docs/en/antalya/cas/architecture/part-lifecycle.md index f5ad05dc5f0fb..c98077d6bcc06 100644 --- a/docs/en/antalya/cas/architecture/part-lifecycle.md +++ b/docs/en/antalya/cas/architecture/part-lifecycle.md @@ -53,7 +53,7 @@ sequenceDiagram end PW->>PW: merge upload results on the owning thread, one no-throw swap TX->>PW: promote - PW->>S3: GET and validate the precommit manifest body + PW->>PW: validate the staged manifest record and blob dependency proofs PW->>S3: append ref-log txn: retire old committed, Precommit to Committed, SetPublishedAt Note over PW: commit durable, then retire the build sequence end @@ -95,9 +95,12 @@ explicit `Materialized` proof. A trusted source manifest instead yields `Trusted blob I/O. Pool size is the server setting `cas_blob_upload_pool_size` (default 16). -**Step 4 — `promote`.** Reads and revalidates the precommit manifest body once; sets the commit -state to Uncertain before the append — past that point, failure is no longer proof of the negative -— then checks that the precommit is still the live owner and validates explicit dependency proofs. +**Step 4 — `promote`.** Uses the in-memory record produced by this `PartWriteTxn`'s successful +conditional create: the confirmed object incarnation and blob references (`blob_refs`). A missing +record is an invariant violation and raises `LOGICAL_ERROR` before the append. There is no manifest `GET`. +Promotion then sets the commit state to `Uncertain` before the append — past that point, failure +is no longer proof of the negative — checks that the precommit is still the live owner, and +validates explicit dependency proofs. `Materialized` leaves are edge-protected; `TrustedManifest` leaves are trusted through the durable source-manifest edge with no per-file `HEAD`; anything else is a `LOGICAL_ERROR`. The whole thing lands as one ref-log record: optional retirement of the old committed binding, the pure diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasPartManifestFormat.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasPartManifestFormat.cpp index 7c5109c0e4b7d..563ff7b7585c3 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasPartManifestFormat.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasPartManifestFormat.cpp @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -14,6 +15,7 @@ namespace DB namespace ErrorCodes { extern const int CORRUPTED_DATA; + extern const int LIMIT_EXCEEDED; } } @@ -75,6 +77,20 @@ String bannerFor(std::string_view path, uint64_t n) return std::move(w).take(); } +void validateManifestEntryPath(std::string_view path) +{ + if (path.empty() || path.front() == '/') + throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS part manifest: invalid entry path '{}'", path); + for (std::string_view rest = path; !rest.empty();) + { + const size_t slash = rest.find('/'); + const std::string_view segment = rest.substr(0, slash); + if (segment.empty() || segment == "." || segment == "..") + throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS part manifest: invalid entry path '{}'", path); + rest = slash == std::string_view::npos ? std::string_view{} : rest.substr(slash + 1); + } +} + } std::string_view entryPlacementToWireWord(EntryPlacement placement) @@ -94,31 +110,52 @@ String encodePartManifest(const PartManifest & m) std::vector sorted; sorted.reserve(m.entries.size()); for (const auto & e : m.entries) + { + validateManifestEntryPath(e.path); sorted.push_back(&e); + } std::sort(sorted.begin(), sorted.end(), [](const ManifestEntry * a, const ManifestEntry * b) { return a->path < b->path; }); for (size_t i = 1; i < sorted.size(); ++i) if (sorted[i]->path == sorted[i - 1]->path) throw Exception(ErrorCodes::CORRUPTED_DATA, "PartManifest: duplicate path '{}'", sorted[i]->path); + const uint64_t line_cap = traitsFor(FormatId::PartManifest).line_cap; + const auto check_line_bytes = [&](size_t bytes, std::string_view what) + { + if (!fitsLineCap(bytes, line_cap)) + throw Exception(ErrorCodes::LIMIT_EXCEEDED, + "CAS part manifest: the {} line encodes to {} bytes, over the {}-byte line cap", + what, bytes, line_cap); + }; + CasJsonWriter out(256); writeHeaderLine(out, FormatId::PartManifest); + check_line_bytes(out.size() - 1, "header"); /// descriptor meta line: ManifestRef (epoch/build/ord, shared rendering with refsnaplog) + root /// namespace + payload digest. { + const size_t line_start = out.size(); bool first = true; writeManifestRefFields(out, first, kBareManifestRefKeys, m.ref); writeStringField(out, PartManifestWire::ns, m.root_namespace_id.string(), first); writeHex128Field(out, PartManifestWire::payload_digest, m.payload_digest, first); closeObject(out, first); + check_line_bytes(out.size() - line_start, "descriptor"); writeChar('\n', out); } for (const ManifestEntry * e : sorted) + { + const size_t line_start = out.size(); writeEntryRecord(out, *e); + check_line_bytes(out.size() - line_start - 1, "entry record"); + } + const size_t trailer_start = out.size(); writeTrailerLine(out, sorted.size()); + check_line_bytes(out.size() - trailer_start - 1, "trailer"); /// payload zone: one banner + raw bytes + '\n' per Inline entry, in path order. Blob entries /// carry no payload-zone bytes (their bytes live in a separately addressed CAS blob). @@ -127,6 +164,7 @@ String encodePartManifest(const PartManifest & m) if (e->placement != EntryPlacement::Inline) continue; const String banner = bannerFor(e->path, e->inline_bytes.size()); + check_line_bytes(banner.size(), "payload banner"); out.append(banner); writeChar('\n', out); out.append(e->inline_bytes); @@ -208,20 +246,7 @@ PartManifest decodePartManifest(std::string_view data) throw Exception(ErrorCodes::CORRUPTED_DATA, "PartManifest: record must start with \"path\""); ManifestEntry e; e.path = r.readString(); - - /// Manifest bytes arrive over the interserver relink channel: enforce the same path hygiene - /// as CasLayout::checkNamespace so no future consumer can inherit a traversal. Relative, - /// no empty/'.'/'..' segments. (Syntactic only — legal projection subdirs pass.) - if (e.path.empty() || e.path.front() == '/') - throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS part manifest: invalid entry path '{}'", e.path); - for (std::string_view rest = e.path; !rest.empty();) - { - const size_t slash = rest.find('/'); - const std::string_view seg = rest.substr(0, slash); - if (seg.empty() || seg == "." || seg == "..") - throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS part manifest: invalid entry path '{}'", e.path); - rest = (slash == std::string_view::npos) ? std::string_view{} : rest.substr(slash + 1); - } + validateManifestEntryPath(e.path); std::optional pm; BlobRefFields blob_ref; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp index 51ff507c141f4..0b3fd79a50edd 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp @@ -616,9 +616,9 @@ ManifestId PartWriteTxn::stageManifest(std::vector entries) "stageManifest: manifest ordinal cap {} exceeded for build_seq {}", kMaxManifestOrdinal, build_seq); const ManifestRef ref{epoch, build_seq, next_manifest_ordinal++}; - /// Build the body. payload_digest is integrity/debug only — never a key, never dedup, never - /// in-degree. The body repeats its own ref + namespace for fail-closed RefMatchesBody / - /// ManifestNamespaceMatches at read/fold/promote time. + /// Build the body. `payload_digest` is integrity/debug only: never a key, dedup input, or in-degree. + /// Repeating `ref` and `root_namespace_id` lets readers and GC folds validate `refMatchesBody` and + /// `manifestNamespaceMatches`. const RootNamespace owning_ns = manifestNamespace(); PartManifest body; body.ref = ref; @@ -633,6 +633,12 @@ ManifestId PartWriteTxn::stageManifest(std::vector entries) "stageManifest: encoded manifest {} bytes exceeds cap {}", encoded_text.size(), kMaxManifestEncodedBytes); const String encoded = sealObject(FormatId::PartManifest, encoded_text); + std::vector blob_refs; + blob_refs.reserve(body.entries.size()); + for (const ManifestEntry & entry : body.entries) + if (entry.placement == EntryPlacement::Blob) + blob_refs.push_back(entry.ref); + const ManifestId id{owning_ns, ref}; const String key = store->layout().manifestKey(id); @@ -690,20 +696,22 @@ ManifestId PartWriteTxn::stageManifest(std::vector entries) }); staged_manifests.push_back(id); - staged_manifest_ids.insert(id); /// A3 mint-tightening: `precommitAdd`'s only legal fresh-ownership source + staged_manifest_ids.emplace( + id, + StagedManifest{manifest_incarnation, std::move(blob_refs)}); return id; } -void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & final_ref_name, const ManifestId & id) +void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & final_ref_name, const ManifestId & manifest_id) { requireAlive(); /// ManifestNamespaceMatches at the source: the precommit's manifest must belong to the target - /// namespace (its key is built from id.root_namespace). A cross-namespace precommit is a bug. - if (id.root_namespace != target_ns) + /// namespace (its key is built from manifest_id.root_namespace). A cross-namespace precommit is a bug. + if (manifest_id.root_namespace != target_ns) throw Exception(ErrorCodes::LOGICAL_ERROR, "precommitAdd: manifest namespace '{}' != target namespace '{}'", - id.root_namespace.string(), target_ns.string()); + manifest_id.root_namespace.string(), target_ns.string()); /// A3 mint-tightening (ABA barrier for the relink confirm's exact-`ManifestRef` equality): an /// unowned `ManifestId` may enter ownership only from the transaction that freshly staged it. @@ -713,11 +721,11 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & /// the closure -- consistent with every other capture there -- rather than capturing `this`, /// which would let a closure that outlives this stack (see the capture comment below) dereference /// a dead transaction. Enforcement, however, is NOT done here: see the closure for why. - const bool id_staged_by_this_txn = staged_manifest_ids.contains(id); + const bool id_staged_by_this_txn = staged_manifest_ids.contains(manifest_id); /// One `owner_transition` create-precommit op. /// No body HEAD — a missing body is a legal fail-closed, non-activating intent, unchanged from the - /// old protocol. Precommit ownership carries NO separate build token: `id.ref` (writer_epoch, + /// old protocol. Precommit ownership carries NO separate build token: `manifest_id.ref` (writer_epoch, /// build_sequence, manifest_ordinal) IS the build identity (the tuple /// §Transaction Log Format: "there is no second build token"). `build_ops` is invoked from inside /// the per-namespace flush, so it sees the table's CURRENT state including any earlier item of the @@ -727,7 +735,7 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & /// /// THE INTENT IS RECORDED BEFORE THE APPEND, and that ordering is the whole point: an `Unresolved` /// append MAY HAVE LANDED (`CasRefLedger.cpp`, the `Unresolved` arm says so explicitly), so a - /// `precommitAdd` that throws can still have made `id.ref` a live precommit owner. Setting the + /// `precommitAdd` that throws can still have made `manifest_id.ref` a live precommit owner. Setting the /// fields afterwards left that case with an object which believed it had never precommitted: /// `abandon` queued no removal and `cleanupStagedManifestDebrisBestEffort` -- deciding from the same /// unset state -- writer-DELETED the body, so a wedge that later resolved as committed installed a @@ -736,7 +744,7 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & /// values, so this cannot throw between the record and the append. precommit_target_ns = target_ns; precommit_final_ref = final_ref_name; - precommit_manifest = id.ref; + precommit_manifest = manifest_id.ref; precommit_state = PrecommitState::Uncertain; store->appendRefOps(target_ns, MutationScope::ref(final_ref_name), @@ -744,7 +752,7 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & /// and manifest id) rather than `[&]`: defense-in-depth so a closure that ever outlives this /// stack frame (e.g. if the append lane stranded its item) never dereferences a dead stack. The /// ref-lane leadership guard is the real fix; this makes the closure self-contained regardless. - [pool = store, target_ns, final_ref_name, id, id_staged_by_this_txn](const RefTableState & state) -> std::vector + [pool = store, target_ns, final_ref_name, manifest_id, id_staged_by_this_txn](const RefTableState & state) -> std::vector { /// Idempotent re-add: the target ref is ALREADY committed to this EXACT manifest_ref (a /// legitimate re-drive calling precommitAdd+promote again for content that is already @@ -752,26 +760,18 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & /// for an already-owned manifest would violate "no conflicting owner may name the same /// manifest", which the state machine enforces for every OTHER case. /// - /// A3 mint-tightening's enforcement lives HERE, gated on this SAME live-state read, rather - /// than unconditionally before the closure: a legitimate re-drive can be handed an id that - /// this exact `PartWriteTxn` object never staged (e.g. a fresh build re-precommitting - /// content a PRIOR, since-destroyed build already promoted -- `CasPromoteRepublish. - /// PromoteSameManifestIsIdempotent`). That case is not a fresh ownership grant: the manifest - /// is already `final_ref_name`'s live, currently-protected binding, so re-affirming it - /// creates no new ABA surface. Checking membership-or-committed-match atomically under the - /// SAME state read (rather than a separate pre-check against a possibly-stale resolve) is - /// what keeps this race-free: a snapshot taken before the closure could see "already - /// committed to id.ref" and let a later closure invocation append a fresh owner_transition - /// after a concurrent repoint moved the ref away from id.ref in between -- exactly the - /// re-ownership of a dropped identity this check exists to prevent. + /// Keep the committed-match check and the staging-membership guard in this closure. + /// A separate resolve could observe a committed match, then race with a repoint or drop + /// before the append. Both decisions must use the same `state` so a stale observation + /// cannot grant fresh ownership to a foreign manifest id. if (const auto it = state.getCommitted().find(final_ref_name); - it != state.getCommitted().end() && it->second.manifest_ref == id.ref) + it != state.getCommitted().end() && it->second.manifest_ref == manifest_id.ref) return {}; if (!id_staged_by_this_txn) throw Exception(ErrorCodes::LOGICAL_ERROR, "precommitAdd: manifest '{}' was not staged by this transaction and is not the current " "committed manifest of ref '{}' -- refusing to re-own a foreign or previously-dropped identity", - manifestRefDebugString(id.ref), final_ref_name); + manifestRefDebugString(manifest_id.ref), final_ref_name); std::vector ops; if (state.getLifecycle() != RefLifecycle::Live) @@ -785,7 +785,7 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & } RefOp add; add.kind = RefOpKind::OwnerTransition; - add.new_binding = RefOwnerBinding{RefOwnerKind::Precommit, final_ref_name, id.ref}; + add.new_binding = RefOwnerBinding{RefOwnerKind::Precommit, final_ref_name, manifest_id.ref}; ops.push_back(add); return ops; }, @@ -807,28 +807,26 @@ void PartWriteTxn::precommitAdd(const RootNamespace & target_ns, const String & }); } -bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final_ref_name, UInt128 promote_build_id, const ManifestId & id, +bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final_ref_name, UInt128 promote_build_id, const ManifestId & manifest_id, PublishPolicy policy) { requireAlive(); - if (id.root_namespace != target_ns) + if (manifest_id.root_namespace != target_ns) throw Exception(ErrorCodes::LOGICAL_ERROR, "promote: manifest namespace '{}' != target namespace '{}'", - id.root_namespace.string(), target_ns.string()); + manifest_id.root_namespace.string(), target_ns.string()); - /// Read + validate the manifest body ONCE (O(manifest entries), one streaming read). Absent or - /// invalid ⇒ fail closed: a committed ref must never name a missing/mismatched manifest. - const String manifest_key = store->layout().manifestKey(id); - CasOperation op = store->mountRequests().admit(); - const auto body_got = op.read(manifest_key, Retry::standard()); - if (!body_got) - throwCasWriteRetryLater(fmt::format( - "promote: manifest body absent at {} — failing closed (retry with a fresh ManifestId)", manifest_key)); - const PartManifest body = decodePartManifest(openObject(FormatId::PartManifest, body_got->bytes)); - if (!refMatchesBody(id.ref, body)) - throwCasWriteRetryLater(fmt::format("promote: RefMatchesBody failed for {}", manifest_key)); - if (!manifestNamespaceMatches(target_ns, body)) - throwCasWriteRetryLater(fmt::format("promote: ManifestNamespaceMatches failed for {}", manifest_key)); + const auto staged = staged_manifest_ids.find(manifest_id); + if (staged == staged_manifest_ids.end()) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "promote: manifest '{}' was not staged by this transaction", manifestRefDebugString(manifest_id.ref)); + + const String manifest_key = store->layout().manifestKey(manifest_id); + if (staged->second.incarnation.key() != manifest_key) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "promote: staged manifest record names key '{}', expected '{}'", + staged->second.incarnation.key(), manifest_key); + std::vector manifest_blob_refs = staged->second.blob_refs; /// Blob readiness is checked inside the append closure below, after owner liveness. It consumes /// only this build's explicit dependency proofs and performs no backend copy-forward. @@ -841,7 +839,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final std::optional repoint_old; /// `created`: whether `final_ref_name` has NO committed row as of THIS builder's read of `state` /// -- set as the very first statement of the closure (before any other branch), so it is correct - /// on every path: idempotent re-promote no-op (a committed row for `id.ref` already existed -> + /// on every path: idempotent re-promote no-op (a committed row for `manifest_id.ref` already existed -> /// false), an intended repoint (a committed row for a DIFFERENT manifest already existed -> false), /// or a genuine first-time bind (no committed row -> true). Same in-closure-output pattern as /// `repoint_old` immediately below; read by the caller only after `appendRefOps` returns. @@ -853,14 +851,12 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final /// may not. commit_state = CommitState::Uncertain; store->appendRefOps(target_ns, MutationScope::ref(final_ref_name), - /// Capture the closure's INPUTS by value (ref name, manifest id, promote build id, repoint flag, - /// and the manifest `body` it revalidates) rather than `[&]`, as defense-in-depth against a - /// closure outliving this stack. Unlike `precommitAdd`, this closure cannot be made fully - /// self-contained: `dependencyProof` reads this build's `deps` member (so it must - /// keep `this`), and `repoint_old`/`created` are OUTPUTS read after the call returns (so they stay - /// references). The ref-lane leadership guard is the real fix; both residual by-reference captures - /// are safe because the guard guarantees the closure is never invoked after this frame unwinds. - [this, final_ref_name, id, promote_build_id, policy, body, &repoint_old, &created] + /// Own the inputs and `blob_refs` in the closure. `this` supplies dependency proofs; + /// `repoint_old` and `created` are outputs read after `appendRefOps` returns. + /// The ref-lane leadership guard completes or removes queued items on every exit, so these + /// borrowed references cannot be used after this frame unwinds. + [this, final_ref_name, manifest_id, promote_build_id, policy, + blob_refs = std::move(manifest_blob_refs), &repoint_old, &created] (const RefTableState & state) -> std::vector { created = !state.getCommitted().contains(final_ref_name); @@ -871,7 +867,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final /// precommit that a first successful call already consumed. Mirrors precommitAdd's matching /// guard; the DIFFERENT-manifest case below remains the BUG-1a fail-closed leak guard. if (const auto it = state.getCommitted().find(final_ref_name); - it != state.getCommitted().end() && it->second.manifest_ref == id.ref) + it != state.getCommitted().end() && it->second.manifest_ref == manifest_id.ref) return {}; if (policy.expected) @@ -893,9 +889,9 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final /// a removal of the precommit binding, the blobs' in-degree was already decremented and a Δ=0 /// move would re-publish a committed ref over to-be-deleted blobs ⇒ a dangling committed /// manifest (INV_NO_DANGLE). `RefTableState::getPrecommits` materializes live ownership directly: - /// `id.ref` alone identifies the build (there is no second + /// `manifest_id.ref` alone identifies the build (there is no second /// build token"), so an exact-binding lookup answers the same question directly. - if (!state.getPrecommits().contains({final_ref_name, id.ref})) + if (!state.getPrecommits().contains({final_ref_name, manifest_id.ref})) throwCasWriteRetryLater(fmt::format( "promote: precommit owner binding for ref '{}' (build {}) was removed (abandon or GC " "reclaim) and is no longer the live owner — failing closed; the build must restart " @@ -912,16 +908,14 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final /// build's precommit edge is durable — matching the relink trust model (ordinary /// ReplicatedMergeTree interserver trust). A genuinely-absent adopted blob is an invariant /// violation caught by fsck, not here. - for (const ManifestEntry & e : body.entries) + for (const BlobRef & blob_ref : blob_refs) { - if (e.placement != EntryPlacement::Blob) - continue; - const auto proof = dependencyProof(e.ref); + const auto proof = dependencyProof(blob_ref); if (!proof) throw Exception(ErrorCodes::LOGICAL_ERROR, "promote: blob leaf {} has no dependency proof at commit — a pending upload " "never completed; failing closed", - store->layout().blobKey(e.ref)); + store->layout().blobKey(blob_ref)); switch (*proof) { @@ -943,7 +937,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final { ev.type = CasEventType::BlobReuseAdopt; ev.object_kind = CasEventObjectKind::Blob; - ev.object_hash = blobIdOf(e.ref); + ev.object_hash = blobIdOf(blob_ref); ev.outcome = "adopt"; ev.reason = "manifest-trust"; /// distinguishable trusted-adopt class (empty token) }); @@ -952,7 +946,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final throw Exception(ErrorCodes::LOGICAL_ERROR, "promote: blob leaf {} has an unnamed dependency proof at commit; failing closed", - store->layout().blobKey(e.ref)); + store->layout().blobKey(blob_ref)); } /// Refuse to overwrite a live committed ref that names a DIFFERENT manifest: that would orphan the @@ -965,7 +959,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final /// write-retry-later, not LOGICAL_ERROR: a conflicting durable state the caller handles, never a /// must-not-happen invariant. if (const auto it = state.getCommitted().find(final_ref_name); - it != state.getCommitted().end() && !(it->second.manifest_ref == id.ref)) + it != state.getCommitted().end() && !(it->second.manifest_ref == manifest_id.ref)) { if (!policy.expected) throwCasWriteRetryLater(fmt::format( @@ -1001,11 +995,11 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final } RefOp transition; transition.kind = RefOpKind::OwnerTransition; - transition.old_binding = RefOwnerBinding{RefOwnerKind::Precommit, final_ref_name, id.ref}; - transition.new_binding = RefOwnerBinding{RefOwnerKind::Committed, final_ref_name, id.ref}; + transition.old_binding = RefOwnerBinding{RefOwnerKind::Precommit, final_ref_name, manifest_id.ref}; + transition.new_binding = RefOwnerBinding{RefOwnerKind::Committed, final_ref_name, manifest_id.ref}; ops.push_back(transition); - ops.push_back(makeSetPublishedAtOp(final_ref_name, id.ref, Cas::nowMs())); + ops.push_back(makeSetPublishedAtOp(final_ref_name, manifest_id.ref, Cas::nowMs())); return ops; }, RootMutationOrigin::Writer, RootMutationKind::Promote); @@ -1022,7 +1016,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final e.type = CasEventType::BuildPublish; e.namespace_ = target_ns.string(); e.ref_name = final_ref_name; - e.object_hash = manifestRefDebugString(id.ref); + e.object_hash = manifestRefDebugString(manifest_id.ref); e.token = u128ToHex(promote_build_id); e.outcome = "promoted"; e.reason = "promote: atomic owner move precommit(build_id) -> ref(final_ref_name) after fail-closed reval"; @@ -1049,7 +1043,7 @@ bool PartWriteTxn::promote(const RootNamespace & target_ns, const String & final e.namespace_ = target_ns.string(); e.ref_name = final_ref_name; e.object_kind = CasEventObjectKind::Manifest; - e.object_hash = manifestRefDebugString(id.ref); + e.object_hash = manifestRefDebugString(manifest_id.ref); e.outcome = "repointed"; e.reason = "promote: committed ref retargeted from an old manifest to a " "new one in one ref-log record -- standalone write/remove on a committed part"; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.h index 9cd10ed46e055..c3ebc23ad8b0d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.h @@ -1,4 +1,5 @@ #pragma once +#include #include #include #include @@ -7,7 +8,6 @@ #include #include #include -#include #include #include @@ -202,42 +202,36 @@ class PartWriteTxn /// Mint a root-local part `ManifestId`, write its body under /// `cas/manifests////000001.zst` via the pool's shared request /// controller. It uses budgeted attempts with resolve-before-reissue and performs no preliminary HEAD, - /// because `manifest_ordinal` is monotone within this build. It enforces manifest-size caps before the body - /// write returns and therefore before any owner transition is published. The body is not retained after a - /// successful write; on retry the caller re-stages from source. Every call uses a fresh manifest ordinal. + /// because `manifest_ordinal` is monotone within this build. Entry paths and manifest-size caps are + /// validated before the body write. After a successful write, the confirmed incarnation and `blob_refs` + /// are retained for `promote`; the body itself is not retained. On retry the caller re-stages from source. + /// Every call uses a fresh manifest ordinal. /// The id is recorded for best-effort `abandon` cleanup. ManifestId stageManifest(std::vector entries); - /// Add this transaction's precommit owner intent; there is no `_precommits` namespace. One - /// `appendRefOps` call appending an OwnerTransition `RefOp` (new_binding = {Precommit, - /// final_ref_name, id.ref}) to final_ref_name's ref-log entry, so the later promote is an atomic - /// owner move over that same entry. Needs NO body-exists HEAD as a safety authority: GC and - /// promotion handle a missing precommit manifest body by failing closed (a missing-body precommit - /// is a non-activating, non-promotable intent). + /// Add the precommit owner intent with one `appendRefOps` call. Its `OwnerTransition` binds + /// `(final_ref_name, manifest_id.ref)` as `Precommit` in the ref log. There is no separate + /// `_precommits` namespace and no body `HEAD`. /// - /// A3 mint-tightening: `id` must either be one THIS transaction minted via `stageManifest`, or + /// A3 mint-tightening: `manifest_id` must either be one THIS transaction minted via `stageManifest`, or /// already be `final_ref_name`'s current committed manifest (the idempotent re-drive -- see the /// closure in the .cpp). Any other id names a manifest this transaction never staged and does not /// currently own -- most commonly one a DIFFERENT, since-abandoned/dropped transaction once staged /// -- and granting it fresh ownership here would let a later exact-`ManifestRef` equality check /// (the relink confirm) compare true against a token whose blobs may already be reclaimed (an ABA). /// Rejected with `LOGICAL_ERROR`: this is a programming-invariant violation, never an operational one. - void precommitAdd(const RootNamespace & target_ns, const String & final_ref_name, const ManifestId & id); + void precommitAdd(const RootNamespace & target_ns, const String & final_ref_name, const ManifestId & manifest_id); /// Atomically promote the precommit to the committed ref with one `appendRefOps` call on the target ref's /// ref-log entry. - /// 1. `Materialized` leaves are already protected by the durable precommit edge, so no writer-side retired-view - /// refresh is needed; - /// 2. stream-read the precommit manifest body; validate RefMatchesBody / ManifestNamespaceMatches; - /// 3. `TrustedManifest` leaves use the durable source-manifest edge with no per-file HEAD/loadMeta - /// probe; a genuinely absent trusted blob is an invariant violation caught by fsck, not here; - /// 4. a body-absent precommit or lost owner-liveness ⇒ ABORTED; a missing dependency proof ⇒ - /// LOGICAL_ERROR (fail closed); - /// 5. atomically replace precommit(build_id) owner with committed(final_ref_name) owner by appending - /// ONE pure-move `RefOp` (old_binding={Precommit,final_ref_name,T}, new_binding={Committed,final_ref_name,T}, - /// same manifest_ref T) and setting refs[final_ref_name]; - /// 6. promotion never emits blob deltas. A missing-body precommit - /// is non-activating and was rejected at step 4 (the writer re-stages with a fresh ManifestId). + /// `manifest_id` must have been staged by this `PartWriteTxn`. Its retained `StagedManifest` supplies + /// `blob_refs` without a body `GET`; missing staging evidence throws `LOGICAL_ERROR` before the append. + /// Inside the append closure, an already committed own manifest is a no-op. Otherwise, check the + /// expected committed binding and live precommit, then require a dependency proof for every `BlobRef`. + /// `Materialized` leaves rely on the precommit edge; `TrustedManifest` leaves rely on the source + /// manifest edge. Neither needs a per-blob probe here. Missing ownership throws `NETWORK_ERROR`; + /// missing or invalid dependency proofs throw `LOGICAL_ERROR`. + /// Commit one `Precommit` to `Committed` owner move with `SetPublishedAt`; promotion emits no blob deltas. /// /// PROCESS-RESTART INVARIANT: a `PartWriteTxn` is a plain in-memory C++ object owned by the wiring's /// `ContentAddressedTransaction` — it is NEVER persisted and NEVER resumed across a process @@ -272,7 +266,7 @@ class PartWriteTxn /// manifest -- an absent row means the ref was moved or dropped under the writer, a different manifest /// means a concurrent repoint won. Either way the stale writer must re-resolve instead of recreating or /// overwriting the binding. - bool promote(const RootNamespace & target_ns, const String & final_ref_name, UInt128 build_id, const ManifestId & id, + bool promote(const RootNamespace & target_ns, const String & final_ref_name, UInt128 build_id, const ManifestId & manifest_id, PublishPolicy policy = {}); /// Retire the build sequence so the GC watermark floor can advance; staged manifest debris is best-effort @@ -336,6 +330,12 @@ class PartWriteTxn /// dependencies are blob-only, so `ObjectKind` is not part of the key. using DepKey = BlobRef; + struct StagedManifest + { + Etag incarnation; + std::vector blob_refs; + }; + /// Own the complete bounded `HEAD`/metadata/publication/reconciliation state machine. Build-neutral: /// returns the materialized dependency and diagnostics without folding into `deps`. BlobUploadResult ensureBlobPresent(const BlobUploadRequest & req) const; @@ -384,12 +384,11 @@ class PartWriteTxn std::vector staged_manifests; /// for best-effort abandon cleanup - /// Every `ManifestId` this transaction has minted via `stageManifest`, checked by `precommitAdd`: - /// an unowned id may enter ownership only from the transaction that freshly staged it. Kept separate from - /// `staged_manifests` above -- that vector's role (best-effort abandon cleanup) is unrelated and - /// could change independently; this set exists purely as the ABA barrier's identity check, so it - /// stays correct even if the cleanup vector's contents or lifetime ever change. - std::set staged_manifest_ids; + /// Confirmed staging records, inserted by `stageManifest` before it returns the manifest id. + /// Membership is the source of fresh ownership in `precommitAdd` and is required by `promote`. + /// Retain records for this build's lifetime, including after `promote`, so repeated promotion stays + /// idempotent. Keep this ownership evidence separate from the `staged_manifests` cleanup list. + std::map staged_manifest_ids; std::map deps; /// dependencies recorded by this build (blobs only) diff --git a/src/Disks/tests/gtest_cas_part_manifest_format.cpp b/src/Disks/tests/gtest_cas_part_manifest_format.cpp index dd6eab02c0030..df416c561a28b 100644 --- a/src/Disks/tests/gtest_cas_part_manifest_format.cpp +++ b/src/Disks/tests/gtest_cas_part_manifest_format.cpp @@ -9,6 +9,11 @@ using namespace DB::Cas; +namespace DB::ErrorCodes +{ + extern const int LIMIT_EXCEEDED; +} + namespace { @@ -316,9 +321,6 @@ TEST(CASPartManifestFormat, MixedAlgoEntriesRoundTrip) EXPECT_EQ(got.entries[1].blob_size, 200u); } -/// Builds a single-Blob-entry manifest whose entry path is exactly `path` -- `encodePartManifest` -/// itself does not validate path shape (only ordering/duplicates), so this lets the negative cases -/// below reach `decodePartManifest`'s shape check unobstructed. static PartManifest manifestWithSinglePath(std::string_view path) { PartManifest m; @@ -336,17 +338,101 @@ static PartManifest manifestWithSinglePath(std::string_view path) return m; } -/// T11: manifest bytes arrive over the interserver relink channel, so decode enforces the same path -/// hygiene as CasLayout::checkNamespace -- relative, no empty/'.'/'..' segments, no leading '/'. -/// `encodePartManifest` does not itself reject these (see `manifestWithSinglePath`), so each case -/// must fail closed at decode time instead. +TEST(CASPartManifestFormat, EncodeRejectsMalformedEntryPaths) +{ + for (const char * path : {"../evil", "/abs", "", "a//b", "a/./b"}) + { + SCOPED_TRACE(path); + PartManifest manifest = manifestWithSinglePath("safe"); + manifest.entries[0].path = path; + expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] { encodePartManifest(manifest); }); + } +} + +TEST(CASPartManifestFormat, EncodeEnforcesEntryRecordLineCap) +{ + const size_t line_cap = traitsFor(FormatId::PartManifest).line_cap; + for (const EntryPlacement placement : {EntryPlacement::Inline, EntryPlacement::Blob}) + { + const size_t record_bytes = placement == EntryPlacement::Inline + ? std::string_view{R"({"path":"","place":"inline","size":1})"}.size() + : std::string_view{R"({"path":"","place":"blob","algo":"ch128","digest":"00112233445566778899aabbccddeeff","size":10})"}.size(); + for (const char path_char : {'a', '"', '\x01'}) + { + const size_t escaped_char_bytes = path_char == 'a' ? 1 : path_char == '"' ? 2 : 6; + for (const size_t record_size : {line_cap - 1, line_cap, line_cap + 1}) + { + SCOPED_TRACE(static_cast(placement)); + SCOPED_TRACE(static_cast(path_char)); + SCOPED_TRACE(record_size); + PartManifest manifest = manifestWithSinglePath("safe"); + auto & entry = manifest.entries[0]; + entry.placement = placement; + entry.inline_bytes = "v"; + const size_t path_bytes = record_size - record_bytes; + entry.path = String(path_bytes / escaped_char_bytes, path_char) + String(path_bytes % escaped_char_bytes, 'a'); + if (record_size > line_cap) + expectThrowsCode(DB::ErrorCodes::LIMIT_EXCEEDED, [&] { encodePartManifest(manifest); }); + else + { + manifest.payload_digest = computePayloadDigest(manifest); + const auto decoded = decodePartManifest(encodePartManifest(manifest)); + ASSERT_EQ(decoded.entries.size(), 1u); + EXPECT_EQ(decoded.entries[0].path, entry.path); + EXPECT_EQ(decoded.entries[0].size(), entry.size()); + } + } + } + } +} + +TEST(CASPartManifestFormat, EncodeEnforcesDescriptorLineCap) +{ + const size_t line_cap = traitsFor(FormatId::PartManifest).line_cap; + const size_t descriptor_bytes = std::string_view{ + R"({"epoch":"17","build":"66","ord":7,"namespace":"","payload_digest":"00000000000000000000000000000000"})"}.size(); + for (const size_t record_size : {line_cap, line_cap + 1}) + { + SCOPED_TRACE(record_size); + PartManifest manifest = manifestWithSinglePath("safe"); + manifest.entries.clear(); + manifest.root_namespace_id = RootNamespace(String(record_size - descriptor_bytes, 'a')); + if (record_size > line_cap) + expectThrowsCode(DB::ErrorCodes::LIMIT_EXCEEDED, [&] { encodePartManifest(manifest); }); + else + { + manifest.payload_digest = computePayloadDigest(manifest); + const auto decoded = decodePartManifest(encodePartManifest(manifest)); + EXPECT_EQ(decoded.root_namespace_id, manifest.root_namespace_id); + } + } +} + +TEST(CASPartManifestFormat, InlinePayloadMayExceedLineCap) +{ + PartManifest manifest = manifestWithSinglePath("safe"); + auto & entry = manifest.entries[0]; + entry.placement = EntryPlacement::Inline; + entry.inline_bytes = String(traitsFor(FormatId::PartManifest).line_cap + 1, '\x01'); + manifest.payload_digest = computePayloadDigest(manifest); + const auto decoded = decodePartManifest(encodePartManifest(manifest)); + ASSERT_EQ(decoded.entries.size(), 1u); + EXPECT_EQ(decoded.entries[0].inline_bytes, entry.inline_bytes); +} + TEST(CASPartManifestFormat, DecodeRejectsMalformedEntryPaths) { + const String valid = encodePartManifest(manifestWithSinglePath("safe")); + const String needle = R"("path":"safe")"; + const size_t pos = valid.find(needle); + ASSERT_NE(pos, String::npos); + for (const char * path : {"../evil", "/abs", "", "a//b", "a/./b"}) { SCOPED_TRACE(path); - const String encoded = encodePartManifest(manifestWithSinglePath(path)); - expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] { decodePartManifest(encoded); }); + String forged = valid; + forged.replace(pos, needle.size(), "\"path\":\"" + String(path) + "\""); + expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] { decodePartManifest(forged); }); } } diff --git a/src/Disks/tests/gtest_cas_part_write.cpp b/src/Disks/tests/gtest_cas_part_write.cpp index a60f5b45b11ad..a6b43351ad32a 100644 --- a/src/Disks/tests/gtest_cas_part_write.cpp +++ b/src/Disks/tests/gtest_cas_part_write.cpp @@ -1,6 +1,5 @@ #include #include -#include #include #include #include @@ -438,6 +437,129 @@ TEST(CASPartWriteTxn, StageManifestUsesPerBuildOrdinals) EXPECT_EQ(next.ref.manifest_ordinal, 1u); } +TEST(CASPartWriteTxn, PromoteOwnStagedManifestWithoutEntriesDoesNotReadBody) +{ + auto backend = std::make_shared(); + auto counting = std::make_shared(backend); + auto store = Pool::open(counting, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const RootNamespace ns{"srv1/no-promote-manifest-get"}; + auto build = startBuildFor(store, ns, "part"); + + const ManifestId id = build->stageManifest({}); + build->precommitAdd(ns, "part", id); + + const String manifest_key = store->layout().manifestKey(id); + const size_t gets_before = counting->getCountFor(manifest_key); + EXPECT_TRUE(build->promote(ns, "part", build->buildId(), id)); + EXPECT_EQ(counting->getCountFor(manifest_key) - gets_before, 0u); +} + +TEST(CASPartWriteTxn, PromoteOwnStagedManifestWithInlineAndBlobsDoesNotReadBody) +{ + auto backend = std::make_shared(); + auto counting = std::make_shared(backend); + auto store = Pool::open(counting, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const RootNamespace ns{"srv1/promote-staged-deps"}; + auto build = startBuildFor(store, ns, "part"); + + ManifestEntry inline_entry; + inline_entry.path = "small"; + inline_entry.placement = EntryPlacement::Inline; + inline_entry.inline_bytes = "abc"; + const ManifestId id = build->stageManifest( + {inline_entry, blobManifestEntry("a.bin", "blob-a"), blobManifestEntry("b.bin", "blob-b")}); + build->precommitAdd(ns, "part", id); + build->putBlob(idOf("blob-a"), BlobSource::fromString("blob-a")); + build->putBlob(idOf("blob-b"), BlobSource::fromString("blob-b")); + const String manifest_key = store->layout().manifestKey(id); + const size_t gets_before = counting->getCountFor(manifest_key); + EXPECT_TRUE(build->promote(ns, "part", build->buildId(), id)); + EXPECT_EQ(counting->getCountFor(manifest_key) - gets_before, 0u); + const auto resolved = store->resolveRef(ns, "part"); + ASSERT_TRUE(resolved); + EXPECT_EQ(resolved->manifest_id, id); +} + +TEST(CASPartWriteTxn, PromoteRejectsManifestNotStagedByThisTransaction) +{ + for (const bool already_committed : {false, true}) + { + SCOPED_TRACE(already_committed); + auto backend = std::make_shared(); + auto counting = std::make_shared(backend); + auto store = Pool::open(counting, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const RootNamespace ns{"srv1/promote-foreign"}; + auto owner = startBuildFor(store, ns, "part"); + const ManifestId id = owner->stageManifest({}); + owner->precommitAdd(ns, "part", id); + if (already_committed) + ASSERT_TRUE(owner->promote(ns, "part", owner->buildId(), id)); + + auto publisher = startBuildFor(store, ns, "part"); + const String manifest_key = store->layout().manifestKey(id); + const size_t gets_before = counting->getCountFor(manifest_key); + const size_t tail_before = store->tailSinceSnapshotCountForTest(ns); +#if defined(DEBUG_OR_SANITIZER_BUILD) + EXPECT_DEATH(publisher->promote(ns, "part", publisher->buildId(), id), ""); +#else + DB::Cas::tests::expectThrowsCodeWithMessage(DB::ErrorCodes::LOGICAL_ERROR, "was not staged by this transaction", [&] + { + publisher->promote(ns, "part", publisher->buildId(), id); + }); +#endif + EXPECT_EQ(counting->getCountFor(manifest_key) - gets_before, 0u); + EXPECT_EQ(publisher->commitState(), PartWriteTxn::CommitState::NotAttempted); + EXPECT_EQ(store->tailSinceSnapshotCountForTest(ns), tail_before); + const auto resolved = store->resolveRef(ns, "part"); + EXPECT_EQ(resolved.has_value(), already_committed); + if (resolved) + EXPECT_EQ(resolved->manifest_id, id); + } +} + +TEST(CASPartWriteTxn, StageManifestRejectsTraversalPathBeforeBodyWrite) +{ + auto backend = std::make_shared(); + auto store = openPool(backend); + const RootNamespace ns{"srv1/traversal-path"}; + auto build = startBuildFor(store, ns, "part"); + + ManifestEntry entry; + entry.path = "../evil"; + entry.placement = EntryPlacement::Inline; + entry.inline_bytes = "payload"; + + expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] + { + build->stageManifest({entry}); + }); + + OperationForTest op(*backend); + EXPECT_TRUE((*op).list(store->layout().manifestNamespacePrefix(ns), "", 10, Retry::once()).keys.empty()); +} + +TEST(CASPartWriteTxn, StageManifestRejectsOversizedEntryRecordBeforeBodyWrite) +{ + for (const char path_char : {'a', '"', '\x01'}) + { + SCOPED_TRACE(static_cast(path_char)); + auto backend = std::make_shared(); + auto store = openPool(backend); + const RootNamespace ns{"srv1/oversized-entry-record"}; + auto build = startBuildFor(store, ns, "part"); + + ManifestEntry entry; + entry.path = String(path_char == 'a' ? 65536 : path_char == '"' ? 32768 : 10923, path_char); + entry.placement = EntryPlacement::Inline; + entry.inline_bytes = "payload"; + expectThrowsCode(DB::ErrorCodes::LIMIT_EXCEEDED, [&] { build->stageManifest({entry}); }); + + OperationForTest op(*backend); + EXPECT_TRUE((*op).list(store->layout().manifestNamespacePrefix(ns), "", 10, Retry::once()).keys.empty()); + EXPECT_FALSE(store->resolveRef(ns, "part")); + } +} + /// B171: the `cas_owner` owner-triple stamping (`PartWriteTxn::ownerMeta`) was DELETED — protection is now /// the build-root precommit edge (reachability), not revocable object metadata GC reads per-candidate. /// The old `CASPartWriteTxn.BlobCarriesOwnerTripleInMetadata` asserted that stamping; its coverage is replaced @@ -2232,62 +2354,33 @@ namespace /// cap ever changes, update this one to match. constexpr uint64_t kExpectedManifestEncodedCap = 256ULL << 20; -/// The exact encoded size `PartWriteTxn::stageManifest` would compute for a single Blob-placement entry whose -/// path is `path_len` bytes long, staged under `ns` — measured through the SAME `encodePartManifest` -/// codec `stageManifest` calls, so this is an exact reproduction rather than a hand-derived estimate. -/// `ref` and `payload_digest` are fixed-width fields (20 and 16 bytes respectively): their VALUES don't -/// affect the encoded size, only their presence does, so the zero-valued placeholders here reproduce -/// the exact same byte count `stageManifest` would produce with its real (non-zero) values. -size_t manifestEncodedSizeForPathLen(const RootNamespace & ns, size_t path_len) -{ - PartManifest probe; - probe.root_namespace_id = ns; - ManifestEntry e; - e.path = String(path_len, 'a'); - e.placement = EntryPlacement::Blob; - e.ref = DB::Cas::BlobRef{DB::Cas::BlobHashAlgo::CityHash128, DB::Cas::BlobDigest::fromU128(UInt128{})}; - - e.blob_size = 12345; - probe.entries = {std::move(e)}; - return encodePartManifest(probe).size(); -} - -/// Finds the exact boundary: the SMALLEST `path_len` whose single-entry manifest encodes to MORE than -/// `kExpectedManifestEncodedCap` bytes under `ns`. `path_len - 1` therefore encodes to AT MOST the cap -/// (the encoding is monotonic in `path_len` — a longer path can only grow the encoded size). Starts -/// from a linear estimate (the encoding is affine in `path_len`: fixed framing overhead plus a constant -/// number of bytes per path byte) and walks to the exact crossing, so this stays correct even if the -/// framing overhead changes, without needing a full binary search over a ~256 MiB range. -size_t findManifestEncodedCapBoundaryPathLen(const RootNamespace & ns) -{ - constexpr size_t probe_lo = 1000; - constexpr size_t probe_hi = 2'000'000; - const size_t size_lo = manifestEncodedSizeForPathLen(ns, probe_lo); - const size_t size_hi = manifestEncodedSizeForPathLen(ns, probe_hi); - const double slope = static_cast(size_hi - size_lo) / static_cast(probe_hi - probe_lo); - const double intercept = static_cast(size_lo) - slope * static_cast(probe_lo); - - size_t path_len = static_cast(std::ceil( - (static_cast(kExpectedManifestEncodedCap) - intercept) / slope)) + 1; - - while (manifestEncodedSizeForPathLen(ns, path_len) <= kExpectedManifestEncodedCap) - ++path_len; - while (path_len > 1 && manifestEncodedSizeForPathLen(ns, path_len - 1) > kExpectedManifestEncodedCap) - --path_len; - return path_len; -} - -/// A one-entry Blob ManifestEntry with a synthetic `path_len`-byte path (used only to inflate the -/// encoded manifest size towards the OQ7 cap). -ManifestEntry wideBlobManifestEntry(size_t path_len) +PartManifest manifestWithEncodedSize(const RootNamespace & ns, const ManifestRef & ref, size_t encoded_size) { - ManifestEntry e; - e.path = String(path_len, 'a'); - e.placement = EntryPlacement::Blob; - e.ref = DB::Cas::BlobRef{DB::Cas::BlobHashAlgo::CityHash128, DB::Cas::BlobDigest::fromU128(UInt128{0x42})}; + PartManifest manifest; + manifest.root_namespace_id = ns; + manifest.ref = ref; + const size_t empty_size = encodePartManifest(manifest).size(); - e.blob_size = 12345; - return e; + ManifestEntry entry; + entry.path = "0/"; + entry.placement = EntryPlacement::Blob; + entry.ref = BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(UInt128{0x42})}; + entry.blob_size = 12345; + manifest.entries = {entry}; + const size_t entry_overhead = encodePartManifest(manifest).size() - empty_size - entry.path.size(); + const size_t entry_count = 2 * kExpectedManifestEncodedCap / traitsFor(FormatId::PartManifest).line_cap; + const size_t framing_size = empty_size + std::to_string(entry_count).size() - 1; + const size_t path_bytes = encoded_size - framing_size - entry_count * entry_overhead; + + manifest.entries.clear(); + manifest.entries.reserve(entry_count); + for (size_t i = 0; i < entry_count; ++i) + { + entry.path = std::to_string(i) + "/"; + entry.path.resize(path_bytes / entry_count + (i < path_bytes % entry_count), 'a'); + manifest.entries.push_back(entry); + } + return manifest; } } @@ -2310,13 +2403,11 @@ TEST(CASPartWriteTxn, ManifestCapEncodedBytesJustUnderStagesSuccessfully) auto s = Pool::open(b, PoolConfig{.pool_prefix = "p", .server_root_id = "test", .boot_ms_fn = [] { return uint64_t{0}; }}); const RootNamespace ns{"srv1/tbl"}; - const size_t path_len_over = findManifestEncodedCapBoundaryPathLen(ns); - ASSERT_GT(path_len_over, 1u); - const size_t path_len_under = path_len_over - 1; - ASSERT_LE(manifestEncodedSizeForPathLen(ns, path_len_under), kExpectedManifestEncodedCap); - auto build = startBuildFor(s, ns, "wide_part"); - const ManifestId id = build->stageManifest({wideBlobManifestEntry(path_len_under)}); + PartManifest manifest = manifestWithEncodedSize( + ns, ManifestRef{s->liveWriterEpoch(), build->buildSeq(), 1}, kExpectedManifestEncodedCap); + ASSERT_EQ(encodePartManifest(manifest).size(), kExpectedManifestEncodedCap); + const ManifestId id = build->stageManifest(std::move(manifest.entries)); EXPECT_EQ(id.root_namespace, ns); OperationForTest op(*b); EXPECT_TRUE((*op).head(s->layout().manifestKey(id), Retry::once()).has_value()) @@ -2332,17 +2423,17 @@ TEST(CASPartWriteTxn, ManifestCapEncodedBytesOverThrowsBeforeBodyWrite) auto s = openPool(b); const RootNamespace ns{"srv1/tbl"}; - const size_t path_len_over = findManifestEncodedCapBoundaryPathLen(ns); - ASSERT_GT(manifestEncodedSizeForPathLen(ns, path_len_over), kExpectedManifestEncodedCap); - auto build = startBuildFor(s, ns, "wide_part"); + PartManifest manifest = manifestWithEncodedSize( + ns, ManifestRef{s->liveWriterEpoch(), build->buildSeq(), 1}, kExpectedManifestEncodedCap + 1); + ASSERT_EQ(encodePartManifest(manifest).size(), kExpectedManifestEncodedCap + 1); OperationForTest op(*b); const size_t keys_before = (*op).list("", "", 100, Retry::once()).keys.size(); bool threw = false; try { - build->stageManifest({wideBlobManifestEntry(path_len_over)}); + build->stageManifest(std::move(manifest.entries)); } catch (const DB::Exception & e) { diff --git a/src/Disks/tests/gtest_cas_promote_republish.cpp b/src/Disks/tests/gtest_cas_promote_republish.cpp index 46eec55753341..323b7fc9b6473 100644 --- a/src/Disks/tests/gtest_cas_promote_republish.cpp +++ b/src/Disks/tests/gtest_cas_promote_republish.cpp @@ -124,12 +124,18 @@ TEST(CASPromoteRepublish, PromoteSameManifestIsIdempotent) auto s = openPool(b); const RootNamespace ns{"srv/tbl@cas@"}; const String ref = "all_0_0_0"; - const ManifestId id = publishCommitted(s, ns, ref, inlineEntries("f", "AAA")); + auto build = s->beginPartWrite(PartWriteInfo{.intended_ref = ns.string() + "/" + ref, .intended_namespace = ns}); + const ManifestId id = build->stageManifest(inlineEntries("f", "AAA")); + build->precommitAdd(ns, ref, id); + EXPECT_TRUE(build->promote(ns, ref, build->buildId(), id)); - /// Re-precommit + re-promote the SAME id onto the same ref: allowed (same manifest_ref). - auto build2 = s->beginPartWrite(PartWriteInfo{.intended_ref = ns.string() + "/" + ref, .intended_namespace = ns}); - build2->precommitAdd(ns, ref, id); - EXPECT_NO_THROW(build2->promote(ns, ref, build2->buildId(), id)); + const size_t tail_before = s->tailSinceSnapshotCountForTest(ns); + build->precommitAdd(ns, ref, id); + EXPECT_FALSE(build->promote(ns, ref, build->buildId(), id)); + EXPECT_EQ(s->tailSinceSnapshotCountForTest(ns), tail_before); + const auto resolved = s->resolveRef(ns, ref); + ASSERT_TRUE(resolved); + EXPECT_EQ(resolved->manifest_id, id); } /// Sanity companion to BUG 1a: promote over an ABSENT ref (the normal insert path) must succeed