From bb13294c5d1db66cf779409f5dd8eab4e83fe856 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 6 Nov 2026 03:02:55 +0100 Subject: [PATCH 01/23] Accept only a strictly greater epoch when claiming a CAS mount slot Port of https://github.com/Altinity/ClickHouse/pull/2512 (antalya-26.6) to antalya-26.8. (cherry picked from commit 2a034cdb3743800197eaaf0c9e6298c0cf238ce3) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Mikhail Filimonov --- .../cas/architecture/mounts-and-leases.md | 14 +- .../Formats/CasServerRootFormats.cpp | 4 +- .../Formats/CasServerRootFormats.h | 2 +- .../ContentAddressed/Pool/CasPool.cpp | 6 +- .../ContentAddressed/Pool/CasServerRoot.cpp | 44 ++- .../ContentAddressed/Pool/CasServerRoot.h | 30 +- src/Disks/tests/gtest_cas_mount.cpp | 334 +++++++++++++++++- .../tests/gtest_cas_mount_claim_conflicts.cpp | 27 +- .../tests/gtest_cas_server_root_format.cpp | 14 + 9 files changed, 419 insertions(+), 56 deletions(-) diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index 71bd26132d2f5..d6e53da9ce99b 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -165,14 +165,18 @@ a `MountClaimResult::Kind` together with a `MountPriorState` describing which ce | `Kind` | Meaning | |---|---| -| `Claimed` | fresh claim (absent slot), same-`(uuid, epoch)` refresh, or a certified reclaim | -| `LiveDoubleStart` | same `server_uuid`, different `writer_epoch`, and no certificate of death yet — a live twin, wait it out | +| `Claimed` | fresh claim (absent slot), or a certified reclaim of an older `writer_epoch` | +| `LiveDoubleStart` | same `server_uuid` with an older `writer_epoch` and no certificate of death yet — a live twin, wait it out; or the same `(uuid, epoch)` already in the slot — two starts got one epoch, refused at once | | `ForeignOwner` | different `server_uuid` — refused unconditionally | | `FencedSelf` | same `(uuid, epoch)`, but `gc_fenced` — terminal for *this* epoch; the caller must mint a fresh one | +A claim never installs a `writer_epoch` at or below the one in the slot. When the slot holds a newer +`writer_epoch` of the same `server_uuid`, the claim fails with `ABORTED` and writes nothing: another +start or a decommission of this `server_root_id` allocated after this one. + | `MountPriorState` | Certificate that justified the reclaim | |---|---| -| `None` | no reclaim needed (fresh claim or same-epoch refresh) | +| `None` | no reclaim needed (fresh claim) | | `Clean` | the predecessor's own graceful farewell (`min_active_build_sequence == UINT64_MAX`) | | `Fenced` | GC's own threshold-gated fence-out (`gc_fenced`) | | `UncleanObserved` | this claimant's own token-stability observation held for the full `TTL + drift` window | @@ -190,8 +194,8 @@ stateDiagram-v2 Live --> Live: renewer beat, putOverwrite seq+1 Live --> Fenced: GC observes a stable token past threshold, gc_fenced=1, body preserved Live --> Terminated: certified drain, terminal farewell (expires_at=now, min_active_build_sequence=MAX) - Fenced --> Live: same-uuid claim with a fresh writer_epoch, instant reclaim - Terminated --> Live: same-uuid claim with a fresh writer_epoch, instant reclaim + Fenced --> Live: same-uuid claim with a higher writer_epoch, instant reclaim + Terminated --> Live: same-uuid claim with a higher writer_epoch, instant reclaim Live --> Live: same-uuid claim, proven-dead token via UncleanObserved Live --> Live: same-uuid claim under cas_unsafe_remount_no_delay, no observation Fenced --> Fenced: same uuid and epoch claim, FencedSelf, no write diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.cpp index f430d2e4d612a..b692a82c5811f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.cpp @@ -129,8 +129,8 @@ ServerEpoch decodeServerEpoch(std::string_view data) else r.skipUnknown(key); } - if (!saw) - throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS server-epoch: missing next_writer_epoch"); + if (!saw || e.next_writer_epoch == 0) + throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS server-epoch: missing or zero next_writer_epoch"); if (!body_in.eof() || !in.eof()) throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS server-epoch: trailing bytes"); return e; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.h index 8e0bd8118a1e8..776288c96906b 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasServerRootFormats.h @@ -78,7 +78,7 @@ OwnerObject decodeOwner(std::string_view data); String encodeServerEpoch(const ServerEpoch & e); /// Decode the epoch counter, requiring its `next_writer_epoch` field and rejecting bytes after the body line. -/// Unknown JSON fields are skipped for forward-compatible reads; malformed input, a missing `next_writer_epoch`, +/// Unknown JSON fields are skipped for forward-compatible reads; malformed input, a missing or zero `next_writer_epoch`, /// and trailing data throw `CORRUPTED_DATA`. ServerEpoch decodeServerEpoch(std::string_view data); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index 794d412bf984d..bf96e4296487a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -718,10 +718,10 @@ void Pool::mountWritable(PoolPtr & store, UInt128 our_uuid, MountClaimPolicy pol const bool unsafe = store->config.unsafe_remount_no_delay; if (unsafe) { - /// One bare attempt first: a slot held by OUR uuid under another epoch is reclaimed at + /// One bare attempt first: a slot held by OUR uuid under an older epoch is reclaimed at /// once under the operator's authorization, carrying the exact token this read saw so a /// slot that moves in between is refused. Every outcome but a claim (an absent slot - /// freshly minted, or a same-epoch refresh) falls through to the ordinary observed path + /// freshly minted) falls through to the ordinary observed path /// below. claim = claimMount(claim_op, store->pool_layout, srid, our_uuid, writer_epoch, now_ms(), ttl_ms, /*proven_dead_incarnation=*/{}, emit_mount_event); @@ -820,7 +820,7 @@ void Pool::mountWritable(PoolPtr & store, UInt128 our_uuid, MountClaimPolicy pol /// the three `MountPriorState`s with no such proof (`UncleanUnsafe` has no proof at all, not merely /// no proof of a CLEAN death: it is the operator's explicit `cas_unsafe_remount_no_delay` /// acceptance of that risk, with no observation behind it). `Clean`, drained farewell, and `None`, a - /// fresh mount / same-epoch refresh with nothing to hand over, are the proven ones. + /// fresh mount with nothing to hand over, are the proven ones. /// An EXHAUSTIVE switch, not a positive allowlist -- a future `MountPriorState` /// enumerator with no proof of clean death must fail the BUILD (a missing `-Wswitch` case), never /// silently fall through to "clean". diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index d50a9fb3f85b6..b88c02dd4af1c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -817,10 +817,23 @@ MountClaimResult claimMount( return {.kind = MountClaimResult::ForeignOwner, .body = existing, .etag = std::nullopt}; } - /// Same uuid + same epoch: it is OUR OWN claim — but a FENCED body is terminal for this - /// (uuid, epoch): the GC dropped its ack from the floor when it fenced. Refreshing it in place - /// would reactivate a fenced incarnation — a fence permanently consumes this `(server_uuid, - /// writer_epoch)` pair, so the caller must re-open with a fresh `writer_epoch`. + /// An older allocation that reached the slot late must not replace a newer incarnation of this + /// server: recovery seals only epochs below the live one, so a late write of the newer epoch would + /// never be sealed out. + if (existing.writer_epoch > our_epoch) + { + emitMountEvent(sink, CasEventType::MountConflict, srid, "newer_epoch", &existing, + "same server_uuid, newer writer_epoch in the mount slot — refusing to replace it"); + throw Exception(ErrorCodes::ABORTED, + "CAS mount slot '{}' holds writer_epoch {} of this server_uuid, newer than the writer_epoch {} " + "this claim allocated: another start or a decommission of server_root_id '{}' allocated after " + "this one. Refusing to replace the newer incarnation ({})", + key, existing.writer_epoch, our_epoch, srid, describeMountHolder(existing)); + } + + /// Our own `(uuid, epoch)` is in the slot, but this call has written nothing: the engine settles a + /// sent attempt inside the write itself. Another process was allocated the same epoch, and its body + /// is never replaced. A fenced one is terminal for the pair; the caller takes a fresh epoch. if (existing.writer_epoch == our_epoch) { if (existing.gc_fenced) @@ -830,20 +843,12 @@ MountClaimResult claimMount( "recover with a fresh writer_epoch"); return {.kind = MountClaimResult::FencedSelf, .body = existing, .etag = std::nullopt}; } - const MountLease body = makeMountBody(our_uuid, our_epoch, existing.seq + 1, now_ms, ttl_ms); - if (const std::optional raced - = conflictOrThrow(op.replace(key, encodeMountLease(body), got->etag, Retry::standard()), - fmt::format("CAS mount slot refresh of '{}'", key))) - /// The mount changed under us between the read and the write, so `got->etag` is KNOWN - /// STALE -- that mismatch is exactly why the write was refused. What the write's resolve - /// read observed is current, and it is that pair that is reported. - return racedDoubleStart(*raced); - emitMountEvent(sink, CasEventType::MountClaim, srid, "refresh", &existing, - "own claim replayed — refreshed seq + expiry"); - return {.kind = MountClaimResult::Claimed, .body = body, .etag = std::nullopt}; + emitMountEvent(sink, CasEventType::MountConflict, srid, "same_epoch_occupied", &existing, + "own (uuid, epoch) is already in the mount slot, written by another process — refusing to write"); + return {.kind = MountClaimResult::LiveDoubleStart, .body = existing, .etag = got->etag}; } - /// Same uuid, DIFFERENT epoch: reclaim ONLY on a certificate of death that needs no fresh + /// Same uuid, OLDER epoch: reclaim ONLY on a certificate of death that needs no fresh /// wall-clock trust — never by comparing `expires_at_ms` against `now_ms`: /// - `gc_fenced` → the fence-out is terminal for that incarnation by construction (its renewer's /// every renewal fails the token guard forever, so it can never write again) — there is no @@ -968,6 +973,13 @@ MountClaimResult claimMountAwaitingExpiry( threshold_met ? observed : std::nullopt, sink, /*unsafe_reclaim_authorization=*/{}); if (r.kind != MountClaimResult::LiveDoubleStart) return r; + /// Observation certifies only an older epoch dead; a body of our own pair is never replaced, + /// so waiting on it could not end. + if (r.body && r.body->server_uuid == our_uuid && r.body->writer_epoch == our_epoch) + return r; + /// A raced write observed a newer epoch of ours; the re-read refuses it with the newer-epoch error. + if (r.body && r.body->server_uuid == our_uuid && r.body->writer_epoch > our_epoch) + return claimMount(op, l, srid, our_uuid, our_epoch, now_ms_fn(), ttl_ms, std::nullopt, sink, {}); /// `claimMount` already read the current body, and a raced write reports whatever its own /// resolve read observed. Reuse `r.etag` whenever it is set instead of re-reading the SAME key diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index f583ca2af83ca..5e3aecc4b9194 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -185,9 +185,8 @@ uint64_t allocateWriterEpoch(CasOperation & op, const Layout & l, const String & const ObserveRefCatalog & observe_catalog); /// Which certificate of death, or the operator's explicit unsafe authorization, justified a -/// same-uuid, different-epoch mount reclaim. `None` when no -/// reclaim of that kind happened (a fresh claim, a -/// same-epoch refresh, `LiveDoubleStart`, `ForeignOwner`, `FencedSelf`). +/// same-uuid, older-epoch mount reclaim. `None` when no reclaim of that kind happened (a fresh +/// claim, `LiveDoubleStart`, `ForeignOwner`, `FencedSelf`). enum class MountPriorState { None, @@ -202,12 +201,13 @@ enum class MountPriorState /// already settled who may write; the lease settles whether a live incarnation currently holds the /// slot. Decision over `get(mountKey)`: /// - absent → write our body via `create` → `Claimed`; -/// - same `server_uuid` AND same `writer_epoch` as (our_uuid, our_epoch) → it is OUR OWN claim -/// (a replay / the renewer adopting it): -/// - `gc_fenced` → terminal for THIS (uuid, epoch) — a fence costs an epoch, so refreshing it -/// in place would reactivate a fenced incarnation → `FencedSelf` (no write); -/// - otherwise → refresh (`replace` to bump seq + fresh `expires_at_ms`) → `Claimed`; -/// - same `server_uuid`, DIFFERENT `writer_epoch` → reclaimed ONLY on a certificate of death, or the +/// - same `server_uuid` and a NEWER `writer_epoch` than `our_epoch` → throw `ABORTED`, no write: an +/// older allocation never replaces a newer incarnation, whatever certifies its end; +/// - same `server_uuid` AND same `writer_epoch` → another process was allocated our epoch; never +/// refreshed: +/// - `gc_fenced` → terminal for THIS (uuid, epoch) → `FencedSelf` (no write); +/// - otherwise → `LiveDoubleStart` (no write); +/// - same `server_uuid`, OLDER `writer_epoch` → reclaimed ONLY on a certificate of death, or the /// operator's explicit unsafe authorization, that /// needs no fresh wall-clock trust (see /// `claimMountAwaitingExpiry` below for how a plain "looks expired" reading is turned into one): @@ -248,8 +248,8 @@ struct MountClaimResult /// rather than the proposal, because a caller cannot check a convention it cannot see. std::optional body; /// Which certificate of death, or the operator's explicit unsafe authorization, justified a - /// same-uuid, different-epoch `Claimed` reclaim (`None` for - /// every other `Kind`, and for the absent-slot / same-epoch-refresh `Claimed` cases). + /// same-uuid, older-epoch `Claimed` reclaim (`None` for + /// every other `Kind`, and for the absent-slot `Claimed` case). MountPriorState prior = MountPriorState::None; /// The incarnation of the body this result observed, for /// `LiveDoubleStart` only (a fresh `Claimed`/`FencedSelf`/`ForeignOwner` write/observe has no @@ -273,6 +273,8 @@ class MountFencedException : public DB::Exception : DB::Exception(msg, DB::ErrorCodes::ABORTED) {} }; +/// A caller never claims twice with one `writer_epoch`; a retry allocates a new one. That is why a +/// body of our own `(uuid, epoch)` found in the slot belongs to another process. /// `proven_dead_incarnation`: the incarnation of a same-uuid, different-epoch lease that the CALLER /// already proved dead by observation (see `claimMountAwaitingExpiry`) — matching it against the /// CURRENTLY observed incarnation is the ONLY way (besides `gc_fenced` / the clean marker) a @@ -298,9 +300,9 @@ String mountDoubleStartMessage(const String & srid, const std::optional newer_epoch; + + std::expected write( + const String & key, const String & bytes, + const std::optional & expected_value, TransportAccess & access) override + { + auto result = InMemoryBackend::write(key, bytes, expected_value, access); + if (arm_on_next_epoch_write && key == epoch_key && result) + { + arm_on_next_epoch_write = false; + install_on_next_mount_read = true; + } + return result; + } + + std::optional read(const String & key, TransportAccess & access) override + { + if (install_on_next_mount_read && key == mount_key) + { + install_on_next_mount_read = false; + installNewerStart(access); + } + return InMemoryBackend::read(key, access); + } + +private: + bool install_on_next_mount_read = false; + + void installNewerStart(TransportAccess & access) + { + const auto epoch = InMemoryBackend::read(epoch_key, access); + const auto mount = InMemoryBackend::read(mount_key, access); + if (!epoch || !mount) + throw std::runtime_error("NewerStartBackend: a control object is missing"); + ServerEpoch counter = decodeServerEpoch(epoch->bytes); + newer_epoch = counter.next_writer_epoch; + ++counter.next_writer_epoch; + if (!InMemoryBackend::write(epoch_key, encodeServerEpoch(counter), epoch->value, access)) + throw std::runtime_error("NewerStartBackend: the epoch bump conflicted"); + const MountLease newer{ + .server_uuid = UInt128(1), + .writer_epoch = *newer_epoch, + .hostname = "newer-start", + .pid = 2, + .started_at_ms = 1, + .seq = 1, + .expires_at_ms = 4'000'000'000'000, + .write_attempt_id = UInt128{0xB}}; + if (!InMemoryBackend::write(mount_key, encodeMountLease(newer), mount->value, access)) + throw std::runtime_error("NewerStartBackend: the mount install conflicted"); + } +}; + +/// A competitor's conditional bump of the epoch counter lands with the very bytes ours sends, and ours +/// is lost unapplied. Our resolve read then finds our bytes under a moved precondition and reports the +/// allocation committed, so two allocators hold one writer_epoch. +class EpochCollisionBackend : public InMemoryBackend +{ +public: + String epoch_key; + bool armed = false; + + std::expected write( + const String & key, const String & bytes, + const std::optional & expected_value, TransportAccess & access) override + { + if (armed && expected_value && key == epoch_key) + { + armed = false; + if (!InMemoryBackend::write(key, bytes, expected_value, access)) + throw std::runtime_error("EpochCollisionBackend: the competitor's bump conflicted"); + throw Poco::TimeoutException("EpochCollisionBackend: response lost, our attempt did not apply"); + } + return InMemoryBackend::write(key, bytes, expected_value, access); + } +}; + +/// Applies the first conditional write of `key` after arming, then loses its response. +class LandedReplaceAckLostOnceBackend : public InMemoryBackend +{ +public: + String key; + bool armed = false; + + std::expected write( + const String & k, const String & bytes, + const std::optional & expected_value, TransportAccess & access) override + { + if (armed && expected_value && k == key) + { + armed = false; + (void)InMemoryBackend::write(k, bytes, expected_value, access); + throw Poco::TimeoutException("LandedReplaceAckLostOnceBackend: simulated lost PUT response"); + } + return InMemoryBackend::write(k, bytes, expected_value, access); + } +}; + /// Counts every request that reaches the store, per primitive, so a test can pin how many a protocol /// step costs rather than only what it produced. class RequestCountingBackend final : public InMemoryBackend @@ -504,6 +611,24 @@ TEST(CASServerRootEpoch, AllocatorIsMonotoneAndSurvivesMountConcept) EXPECT_GT(allocateWriterEpoch(ops.op, l, "r", EpochMintPolicy::NormalMount, 0, emptyCatalogObservation()), e2); } +/// No writer emits a zero counter; an allocation over one would hand out writer_epoch 0. +TEST(CASServerRootEpoch, ZeroCounterFailsAllocationWithoutWriting) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + mustCommit(ops.op.create(l.epochKey("r"), encodeServerEpoch(ServerEpoch{.next_writer_epoch = 0}), Retry::standard()), + "zero counter"); + const auto before = ops.op.read(l.epochKey("r"), Retry::standard()); + ASSERT_TRUE(before.has_value()); + + DB::Cas::tests::expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] + { + (void)allocateWriterEpoch(ops.op, l, "r", EpochMintPolicy::NormalMount, 0, emptyCatalogObservation()); + }); + EXPECT_EQ(ops.op.read(l.epochKey("r"), Retry::standard())->bytes, before->bytes); +} + /// Phase C (spec rev.4): an ABSENT epoch object over a PRESENT mount object means durable epoch /// state was lost while a mount is live/recent — re-minting epoch 1 there is how a same-(uuid, /// epoch) twin is born. Refuse. @@ -976,7 +1101,13 @@ TEST(CASMountAwaitExpiry, LiveRenewingTwinTimesOutAsDoubleStart) { wall += ms; mono += ms; - ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, wall, 100).kind, MountClaimResult::Claimed); + const auto got = ops.op.read(l.mountKey("r"), Retry::standard()); + ASSERT_TRUE(got.has_value()); + MountLease renewed = decodeMountLease(got->bytes); + ++renewed.seq; + renewed.expires_at_ms = wall + 100; + mustCommit(ops.op.replace(l.mountKey("r"), encodeMountLease(renewed), got->etag, Retry::standard()), + "twin renewal"); }; const auto r = claimMountAwaitingExpiry( @@ -1116,10 +1247,16 @@ TEST(CASMountClaim, UnsafeAuthorizationIsTokenExact) const Etag stale = ops.op.read(l.mountKey("r"), Retry::standard())->etag; /// `Etag` equality compares (key, value), and it is minted only through the request planes -- there - /// is no cross-key comparison to exercise here. Build the stale token by refreshing the SAME slot a - /// second time (same uuid, same epoch): `stale`, read before this refresh, is then a genuinely stale - /// token for the slot's CURRENT value, without touching an unrelated key. - ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, /*now*/ 1000, /*ttl*/ 30000).kind, MountClaimResult::Claimed); + /// is no cross-key comparison to exercise here. Build the stale token by renewing the SAME slot once: + /// `stale`, read before this renewal, is then a genuinely stale token for the slot's CURRENT value. + { + const auto got = ops.op.read(l.mountKey("r"), Retry::standard()); + ASSERT_TRUE(got.has_value()); + MountLease renewed = decodeMountLease(got->bytes); + ++renewed.seq; + mustCommit(ops.op.replace(l.mountKey("r"), encodeMountLease(renewed), got->etag, Retry::standard()), + "holder renewal"); + } const Etag current = ops.op.read(l.mountKey("r"), Retry::standard())->etag; /// A stale token is refused: nothing authorizes a reclaim over a slot that moved. @@ -1211,6 +1348,40 @@ TEST(CASMountStartup, WriterEpochStrictlyIncreasesAcrossReopen) EXPECT_GT(e2, e1); } +TEST(CASMountStartup, DelayedStartOverANewerIncarnationFailsWithoutWaiting) +{ + auto b = std::make_shared(); + const Layout l("p"); + b->epoch_key = l.epochKey("r"); + b->mount_key = l.mountKey("r"); + { + /// Bootstrap: owner, counter and a clean farewell of epoch 1. + auto first = Pool::open(b, PoolConfig{.pool_prefix = "p", .server_id = UInt128(1), .server_root_id = "r"}); + } + b->arm_on_next_epoch_write = true; + + /// Shared, heap-owned: the Pool may outlive this frame through a background publish. + auto fake_boot = std::make_shared>(0); + auto waits = std::make_shared>(0); + DB::Cas::tests::expectThrowsCodeWithMessage(DB::ErrorCodes::ABORTED, "newer than the writer_epoch", [&] + { + (void)Pool::open(b, PoolConfig{ + .pool_prefix = "p", .server_id = UInt128(1), .server_root_id = "r", + .boot_ms_fn = [fake_boot] { return fake_boot->load(); }, + .wait_sleep_fn = [fake_boot, waits](uint64_t ms) + { + *fake_boot += ms; + ++(*waits); + }}); + }); + ASSERT_TRUE(b->newer_epoch.has_value()) << "the newer start was never installed"; + EXPECT_EQ(waits->load(), 0u) << "a newer incarnation is refused, not observed"; + Ops ops(b); + const MountLease slot = decodeMountLease(ops.op.read(l.mountKey("r"), Retry::standard())->bytes); + EXPECT_EQ(slot.writer_epoch, *b->newer_epoch); + EXPECT_EQ(slot.write_attempt_id, UInt128{0xB}); +} + TEST(CASMountStartup, FreshWritablePoolBootstrapsAnExplicitEmptyCatalog) { auto backend = std::make_shared(); @@ -2075,6 +2246,159 @@ TEST(CASClaimMount, SameEpochFencedIsNotRefreshable) MountClaimResult::Claimed); } +/// An older allocation that reaches the slot late must not replace a newer incarnation of this server, +/// whatever certifies that incarnation's end: recovery seals only epochs below the live one. +TEST(CASClaimMountMonotonicity, LowerEpochOverANewerBodyFailsWithoutWriting) +{ + enum class Shape { Live, GcFenced, CleanFarewell, ProvenDead, UnsafeAuthorized }; + for (const Shape shape : {Shape::Live, Shape::GcFenced, Shape::CleanFarewell, Shape::ProvenDead, + Shape::UnsafeAuthorized}) + { + SCOPED_TRACE(static_cast(shape)); + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + const String key = l.mountKey("r"); + ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), /*our_epoch=*/8, 1000, 30000).kind, MountClaimResult::Claimed); + if (shape == Shape::GcFenced || shape == Shape::CleanFarewell) + { + const auto got = ops.op.read(key, Retry::standard()); + ASSERT_TRUE(got.has_value()); + MountLease body = decodeMountLease(got->bytes); + ++body.seq; + if (shape == Shape::GcFenced) + body.gc_fenced = true; + else + body.min_active_build_sequence = std::numeric_limits::max(); + mustCommit(ops.op.replace(key, encodeMountLease(body), got->etag, Retry::standard()), "terminal newer body"); + } + const auto before = ops.op.read(key, Retry::standard()); + ASSERT_TRUE(before.has_value()); + const std::optional token = before->etag; + + DB::Cas::tests::expectThrowsCodeWithMessage(DB::ErrorCodes::ABORTED, "newer than the writer_epoch", [&] + { + (void)claimMount(ops.op, l, "r", UInt128(1), /*our_epoch=*/7, 2000, 30000, + shape == Shape::ProvenDead ? token : std::nullopt, /*sink=*/{}, + shape == Shape::UnsafeAuthorized ? token : std::nullopt); + }); + EXPECT_EQ(ops.op.read(key, Retry::standard())->bytes, before->bytes) << "the newer body was replaced"; + } +} + +TEST(CASClaimMountMonotonicity, TwoAllocatorsSharingAnEpochDoNotShareTheSlot) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + b->epoch_key = l.epochKey("r"); + ASSERT_EQ(allocateWriterEpoch(ops.op, l, "r", EpochMintPolicy::NormalMount, 0, emptyCatalogObservation()), 1u); + + b->armed = true; + const uint64_t shared_epoch + = allocateWriterEpoch(ops.op, l, "r", EpochMintPolicy::NormalMount, 0, emptyCatalogObservation()); + ASSERT_FALSE(b->armed) << "the collision was never injected"; + ASSERT_EQ(shared_epoch, 2u); + /// One bump past the shared epoch: the competitor's, which our resolve read took for ours. + ASSERT_EQ(decodeServerEpoch(ops.op.read(l.epochKey("r"), Retry::standard())->bytes).next_writer_epoch, 3u); + + /// The competitor claims first. + ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), shared_epoch, 1000, 30000).kind, MountClaimResult::Claimed); + const auto competitor = ops.op.read(l.mountKey("r"), Retry::standard()); + ASSERT_TRUE(competitor.has_value()); + + const MountClaimResult ours = claimMount(ops.op, l, "r", UInt128(1), shared_epoch, 2000, 30000); + EXPECT_EQ(ours.kind, MountClaimResult::LiveDoubleStart); + ASSERT_TRUE(ours.body.has_value()); + EXPECT_EQ(ours.body->write_attempt_id, decodeMountLease(competitor->bytes).write_attempt_id); + ASSERT_TRUE(ours.etag.has_value()); + EXPECT_TRUE(*ours.etag == competitor->etag); + EXPECT_EQ(ops.op.read(l.mountKey("r"), Retry::standard())->bytes, competitor->bytes) << "the slot was written"; +} + +TEST(CASClaimMountMonotonicity, AwaitingExpiryReturnsAtOnceOnABodyOfOurOwnPair) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + /// Another process allocated epoch 7 as well; its body is in the slot. + ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, 1000, 100).kind, MountClaimResult::Claimed); + const auto before = ops.op.read(l.mountKey("r"), Retry::standard()); + ASSERT_TRUE(before.has_value()); + + uint64_t mono = 0; + int sleeps = 0; + const auto r = claimMountAwaitingExpiry(ops.op, l, "r", UInt128(1), /*our_epoch=*/7, + [] { return uint64_t{1000}; }, [&] { return mono; }, /*ttl_ms=*/100, /*poll_interval_ms=*/20, + [&](uint64_t ms) + { + mono += ms; + if (++sleeps > 100) + throw std::runtime_error("the observation of our own (uuid, epoch) never ended"); + }); + EXPECT_EQ(r.kind, MountClaimResult::LiveDoubleStart); + EXPECT_EQ(sleeps, 0) << "no observation can authorize replacing our own pair"; + EXPECT_EQ(ops.op.read(l.mountKey("r"), Retry::standard())->bytes, before->bytes); +} + +TEST(CASClaimMountMonotonicity, AwaitingExpiryFailsAtOnceWhenARacedWriteMeetsANewerEpoch) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + /// Absent at our read; a newer start of this server mints the slot before our create lands. + b->before_put_if_absent = [&] + { + DB::Cas::tests::OperationForTest racer(*b); + ASSERT_EQ(claimMount(*racer, l, "r", UInt128(1), /*our_epoch=*/8, 1000, 100).kind, MountClaimResult::Claimed); + }; + + uint64_t mono = 0; + int sleeps = 0; + DB::Cas::tests::expectThrowsCodeWithMessage(DB::ErrorCodes::ABORTED, "newer than the writer_epoch", [&] + { + (void)claimMountAwaitingExpiry(ops.op, l, "r", UInt128(1), /*our_epoch=*/7, + [] { return uint64_t{1000}; }, [&] { return mono; }, /*ttl_ms=*/100, /*poll_interval_ms=*/20, + [&](uint64_t ms) + { + mono += ms; + ++sleeps; + }); + }); + EXPECT_FALSE(b->before_put_if_absent) << "the newer start was never installed"; + EXPECT_EQ(sleeps, 0) << "a newer incarnation is refused, not observed"; + EXPECT_EQ(decodeMountLease(ops.op.read(l.mountKey("r"), Retry::standard())->bytes).writer_epoch, 8u); +} + +/// The engine settles our own landed attempt inside the write call, so a claim never meets its own body +/// on a later call. +TEST(CASClaimMountMonotonicity, AConfirmedLandedReclaimIsStillClaimed) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + b->key = l.mountKey("r"); + ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, 1000, 30000).kind, MountClaimResult::Claimed); + { + const auto got = ops.op.read(b->key, Retry::standard()); + ASSERT_TRUE(got.has_value()); + MountLease fenced = decodeMountLease(got->bytes); + fenced.gc_fenced = true; + ++fenced.seq; + mustCommit(ops.op.replace(b->key, encodeMountLease(fenced), got->etag, Retry::standard()), "fence-out"); + } + + b->armed = true; + const MountClaimResult r = claimMount(ops.op, l, "r", UInt128(1), 8, 2000, 30000); + ASSERT_FALSE(b->armed) << "the lost response was never injected"; + EXPECT_EQ(r.kind, MountClaimResult::Claimed); + EXPECT_EQ(r.prior, MountPriorState::Fenced); + ASSERT_TRUE(r.body.has_value()); + const MountLease slot = decodeMountLease(ops.op.read(b->key, Retry::standard())->bytes); + EXPECT_EQ(slot.writer_epoch, 8u); + EXPECT_EQ(slot.write_attempt_id, r.body->write_attempt_id); +} + /// ---- rev.6 Task 4: observation-based lease reclaim (no cross-node wall-clock trust) ---- /// A same-uuid, different-epoch lease whose STAMPED `expires_at_ms` looks long expired on OUR wall diff --git a/src/Disks/tests/gtest_cas_mount_claim_conflicts.cpp b/src/Disks/tests/gtest_cas_mount_claim_conflicts.cpp index 6eb8f9c9b7b85..d5aa0235b6be6 100644 --- a/src/Disks/tests/gtest_cas_mount_claim_conflicts.cpp +++ b/src/Disks/tests/gtest_cas_mount_claim_conflicts.cpp @@ -116,8 +116,14 @@ TEST(CASMountClaimConflicts, SlotChangedInsideAdoptionWindow) /// Rewrite the slot under a NEW incarnation after our read, so our adoption write is refused. backend->before_put_overwrite = [&] { - CasOperation racer = requests.admit(); - claimMount(racer, layout, "r", DB::UInt128(1), 7, now + 1, /*ttl_ms=*/100); + OperationForTest racer(*backend); + const String key = layout.mountKey("r"); + const auto got = (*racer).read(key, Retry::standard()); + ASSERT_TRUE(got.has_value()); + MountLease rewritten = decodeMountLease(got->bytes); + ++rewritten.seq; + ASSERT_TRUE(std::holds_alternative( + (*racer).replace(key, encodeMountLease(rewritten), got->etag, Retry::standard()))); }; auto renewer = makeRenewer(requests, now); expectThrowsCodeWithMessage( @@ -216,9 +222,8 @@ TEST(CASMountClaimConflicts, ALostCreateReportsTheOccupantNotTheProposer) << "the operator message must name the foreign holder"; } -/// The same for the refresh branch: a body that changed under our own adoption is the one the message -/// must name. The reclaim branch reaches the identical helper, so it is not repeated here. -TEST(CASMountClaimConflicts, ALostRefreshReportsTheObservedBodyNotTheProposer) +/// The same for the reclaim branch: a body that changed under our own reclaim is the one the message must name. +TEST(CASMountClaimConflicts, ALostReclaimReportsTheObservedBodyNotTheProposer) { auto backend = std::make_shared(); Layout layout("p"); @@ -227,7 +232,8 @@ TEST(CASMountClaimConflicts, ALostRefreshReportsTheObservedBodyNotTheProposer) CasOperation seed = requests.admit(); ASSERT_EQ(claimMount(seed, layout, "r", DB::UInt128(1), 7, now, /*ttl_ms=*/100).kind, MountClaimResult::Claimed); - /// A distinguishable body lands under our own refresh, so a result carrying the proposal cannot + markMountGcFenced(seed, layout, "r"); + /// A distinguishable body lands under our own reclaim, so a result carrying the proposal cannot /// pass by accident: our proposal would carry `seq` 2 and this process's pid. backend->before_put_overwrite = [&] { @@ -242,7 +248,7 @@ TEST(CASMountClaimConflicts, ALostRefreshReportsTheObservedBodyNotTheProposer) (*racer).replace(key, encodeMountLease(raced), got->etag, Retry::standard()))); }; CasOperation op = requests.admit(); - const MountClaimResult claim = claimMount(op, layout, "r", DB::UInt128(1), 7, now, /*ttl_ms=*/100); + const MountClaimResult claim = claimMount(op, layout, "r", DB::UInt128(1), 8, now, /*ttl_ms=*/100); EXPECT_EQ(claim.kind, MountClaimResult::LiveDoubleStart); ASSERT_TRUE(claim.body.has_value()); @@ -256,7 +262,7 @@ TEST(CASMountClaimConflicts, ALostRefreshReportsTheObservedBodyNotTheProposer) /// process in the "Existing mount" line of an operator message -- the same defect as naming it after a /// conflict that did observe someone. `ProvenAbsent` is the reachable half; `NotObserved` (the resolve /// read itself failed) leaves through the same branch. -TEST(CASMountClaimConflicts, ARacedRefreshThatObservedNothingNamesNoHolder) +TEST(CASMountClaimConflicts, ARacedReclaimThatObservedNothingNamesNoHolder) { auto backend = std::make_shared(); Layout layout("p"); @@ -265,7 +271,8 @@ TEST(CASMountClaimConflicts, ARacedRefreshThatObservedNothingNamesNoHolder) CasOperation seed = requests.admit(); ASSERT_EQ(claimMount(seed, layout, "r", DB::UInt128(1), 7, now, /*ttl_ms=*/100).kind, MountClaimResult::Claimed); - /// The slot is removed under our own refresh, so the refused precondition resolves to a proven + markMountGcFenced(seed, layout, "r"); + /// The slot is removed under our own reclaim, so the refused precondition resolves to a proven /// absence rather than to an occupant. backend->before_put_overwrite = [&] { @@ -276,7 +283,7 @@ TEST(CASMountClaimConflicts, ARacedRefreshThatObservedNothingNamesNoHolder) ASSERT_EQ((*racer).remove(key, got->etag, Retry::standard()), Removal::Removed); }; CasOperation op = requests.admit(); - const MountClaimResult claim = claimMount(op, layout, "r", DB::UInt128(1), 7, now, /*ttl_ms=*/100); + const MountClaimResult claim = claimMount(op, layout, "r", DB::UInt128(1), 8, now, /*ttl_ms=*/100); EXPECT_EQ(claim.kind, MountClaimResult::LiveDoubleStart); EXPECT_FALSE(claim.etag.has_value()); diff --git a/src/Disks/tests/gtest_cas_server_root_format.cpp b/src/Disks/tests/gtest_cas_server_root_format.cpp index bf04753c9b9f1..42fcf33e484f0 100644 --- a/src/Disks/tests/gtest_cas_server_root_format.cpp +++ b/src/Disks/tests/gtest_cas_server_root_format.cpp @@ -52,6 +52,20 @@ TEST(CASFormatBattery, ServerEpoch) currentFormatHeader("cas_epoch") + "{\"next_writer_epoch\":\"7\"}\n"}); } +TEST(CASServerEpochFormat, ZeroNextWriterEpochIsRejected) +{ + const String data = currentFormatHeader("cas_epoch") + "{\"next_writer_epoch\":\"0\"}\n"; + try + { + decodeServerEpoch(data); + FAIL() << "expected CORRUPTED_DATA"; + } + catch (const DB::Exception & e) + { + EXPECT_EQ(e.code(), DB::ErrorCodes::CORRUPTED_DATA); + } +} + CAS_BATTERY_COVERS(MountLease); TEST(CASFormatBattery, MountLease) From ba448a8e2c81afca24c1a27e25734ade720fdc11 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 6 Nov 2026 03:05:00 +0100 Subject: [PATCH 02/23] Adopt a CAS blob only when its marker reads Clean Port of https://github.com/Altinity/ClickHouse/pull/2552 (antalya-26.6) to antalya-26.8. Conflict: `src/Disks/tests/gtest_cas_part_write.cpp`, counter reads without `.load()` as on antalya-26.8. (cherry picked from commit fef96d50e696f9d7ba56dbacfef1d690a135ae02) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Mikhail Filimonov --- docs/en/antalya/cas/architecture/backend.md | 2 +- .../antalya/cas/architecture/blob-protocol.md | 33 +-- .../cas/architecture/design-history.md | 2 +- .../cas/architecture/garbage-collection.md | 12 +- .../cas/architecture/part-lifecycle.md | 2 +- docs/en/antalya/metrics_and_events.md | 4 +- src/Common/ProfileEvents.cpp | 4 +- .../ContentAddressed/Gc/CasGc.cpp | 5 +- .../ContentAddressed/Gc/CasGcMetaWriter.cpp | 15 +- .../ContentAddressed/Gc/CasGcMetaWriter.h | 5 +- .../ContentAddressed/Pool/CasPartWriteTxn.cpp | 29 +-- .../ContentAddressed/README.md | 6 +- src/Disks/tests/gtest_cas_gc_ack_floor.cpp | 12 +- src/Disks/tests/gtest_cas_gc_meta_writer.cpp | 35 ++- src/Disks/tests/gtest_cas_part_write.cpp | 222 +++++++++++++----- .../tests/gtest_cas_protocol_scenarios.cpp | 15 +- src/Disks/tests/gtest_cas_upload_detached.cpp | 28 +-- src/Disks/tests/gtest_cas_upload_fanout.cpp | 14 +- 18 files changed, 280 insertions(+), 165 deletions(-) diff --git a/docs/en/antalya/cas/architecture/backend.md b/docs/en/antalya/cas/architecture/backend.md index cc0740ea73da7..8f889a3bfe038 100644 --- a/docs/en/antalya/cas/architecture/backend.md +++ b/docs/en/antalya/cas/architecture/backend.md @@ -34,7 +34,7 @@ different semantics: `PartWriteTxn::ensureBlobPresent` owns `HEAD`, freshness me `publishBlob` only moves the selected bytes. Writer readiness is represented by `BlobDependencyProof`, not by token presence. `Materialized` -means the writer observed a present non-condemned body or completed publication and metadata +means the writer observed a present body with a `Clean` marker or completed publication and metadata reconciliation. `TrustedManifest` means a durable source manifest proves the blob and requires no blob I/O. Pending state and writer tokens are not stored in the dependency record. diff --git a/docs/en/antalya/cas/architecture/blob-protocol.md b/docs/en/antalya/cas/architecture/blob-protocol.md index 54f6399d073a7..bb72311d07c8c 100644 --- a/docs/en/antalya/cas/architecture/blob-protocol.md +++ b/docs/en/antalya/cas/architecture/blob-protocol.md @@ -36,11 +36,11 @@ sequenceDiagram alt body present S3-->>Writer: present, size, backend token t1 Writer->>S3: GET .meta (point read, body never streamed) - alt meta Clean or absent + alt meta Clean Writer->>Writer: record token-free BlobDependencyProof::Materialized (never adopt t1) - else meta Condemned + else meta Condemned or absent Writer->>S3: unconditional publish with fresh envelope - Writer->>S3: reconcile .meta to Clean + Writer->>S3: create or reconcile .meta to Clean end else body absent S3-->>Writer: absent @@ -56,9 +56,10 @@ Ordered steps in `PartWriteTxn::ensureBlobPresent`: still live. 2. **Mandatory observation.** Every physical materialization begins with one blob `HEAD`, regardless of size, provider, staging backend, or whether another writer probably uploaded the same hash. -3. On a hit, the writer reads `.meta`. `Clean` or absent metadata permits adoption; the writer - records a `Materialized` dependency proof without retaining the observed token. `Condemned` - requires a new publication. +3. On a hit, the writer reads `.meta`. Only `Clean` permits adoption; the writer records a + `Materialized` dependency proof without retaining the observed token. `Condemned` or absent + metadata requires a new publication. `GC` deletes `.meta` right after the body, so an absent + marker can follow a delete that landed between the `HEAD` and the read. 4. On a miss, the writer does not read `.meta` before publication. It publishes its own payload unconditionally, then creates or reconciles `.meta` to `Clean`. 5. Streaming publication mints a fresh `incarnation_tag` and can use ordinary multipart. The first @@ -67,7 +68,8 @@ Ordered steps in `PartWriteTxn::ensureBlobPresent`: 6. `BlobSource::beginPublication` consumes the shared, monotonic `publication_attempted` state before backend I/O. A lost response cannot re-enable verbatim copy on a later attempt. 7. Retryable or ambiguous failures restart at `HEAD`. No dependency proof is recorded until a - present non-condemned body was observed, or publication and metadata reconciliation completed. + present body with a `Clean` marker was observed, or publication and metadata reconciliation + completed. **Two writers uploading identical content** may both observe absence and both publish. The last physical incarnation wins, but the key proves that both payloads have the same logical identity and @@ -151,7 +153,7 @@ sequenceDiagram alt writer republished S3-->>GC: TokenMismatch -- nothing deleted, blob is live at t2 else genuinely dead - S3-->>GC: Deleted -- then drop the .meta only while it is still Condemned at round n or older + S3-->>GC: Deleted -- then drop the .meta only while it is still Condemned at round n+1 or older end ``` @@ -163,7 +165,7 @@ and a writer never trusts a body it did not itself just write. Why this closes the race in both directions: -- A writer that **adopts** a present body must have read a non-`Condemned` marker, and its precommit +- A writer that **adopts** a present body must have read a `Clean` marker, and its precommit edge was durable *before* that read. The next fold therefore sees in-degree ≥ 1 and spares the blob. - A writer that **replaces a condemned incarnation** changes the token. A stale `deleteExact(t1)` then returns @@ -174,9 +176,11 @@ Why this closes the race in both directions: at the entry's round or newer. A `Clean`, absent, older-round or unreadable marker never graduates the entry: `GC` does not throw, it carries the entry and rewrites the marker at the current round. -- The delete job removes the `.meta` only while it is still `Condemned` at the entry's round or - older, and conditionally on the version it read. A marker rewritten `Clean` by a writer, or - condemned again at a later round, stays. +- The delete job removes the `.meta` only while it is still `Condemned` at a round before the + deleting round, and conditionally on the version it read. It also removes a marker the graduation + gate re-stamped. A marker rewritten `Clean` by a writer, or condemned in the deleting round, + stays. The delete job runs best effort: if it fails, the `Condemned` marker outlives its body + until a writer of that content replaces it, which costs storage, not safety. Both directions degrade to a spurious re-upload or a no-op delete. Neither can lose data or leave a dangling manifest entry. @@ -189,8 +193,9 @@ in RAM at a time. Native object storage streams and can use multipart. `.meta` has exactly two states: `Clean` (body present, may be referenced) and `Condemned` (`GC` observed zero in-degree; the body is still present and a writer may replace it). An -*absent* `.meta` reads exactly like `Clean` — there is no third "unaccounted" state in the -stored format; `unaccounted` is a `cas-fsck` classification, not something `GC` ever writes. +*absent* `.meta` never permits adoption: the writer republishes as for an absent body. There is +no third "unaccounted" state in the stored format; `unaccounted` is a `cas-fsck` classification, +not something `GC` ever writes. The record carries `state`, `condemn_round`, and `size`, and deliberately carries **no token**: it is a per-hash hint, not a per-incarnation fact. All safety comes from the body's in-envelope diff --git a/docs/en/antalya/cas/architecture/design-history.md b/docs/en/antalya/cas/architecture/design-history.md index 6b5c371dd86f2..a6d38058b1e74 100644 --- a/docs/en/antalya/cas/architecture/design-history.md +++ b/docs/en/antalya/cas/architecture/design-history.md @@ -21,7 +21,7 @@ the sequence of pivots that produced the architecture described elsewhere in thi | **Integer in-degree refcount** (a mutable counter, incremented per reference, decremented per release) | The decide-to-reference-then-not-yet-durable window let the fold observe in-degree 0 for a still-live blob; a mutable counter also costs a `CAS` round-trip proportional to write volume | A derived count: `GC` folds a multiset of `+`/`-` source-edge deltas, so losing or duplicating a record can only delay reclamation, never accelerate one | | **Extending zero-copy replication instead of a new mechanism** | Zero-copy's structural costs (a commit spanning local disk, S3, and `Keeper`; a mutable refcount) are inherent to its design, not a bug to patch | `CAS` is an alternative to zero-copy, not a replacement: both remain available, `metadata_type = cas` is opt-in per disk, and no existing deployment needs to migrate | | **Per-incarnation body keys** (`blobs/xx/.`, an alternative to the in-body incarnation tag) | A resurrect reusing the condemned incarnation instead of minting a fresh one reintroduced the shared-key race; structurally this was generation-in-the-key again | The in-body `incarnation_tag` plus exact-token body delete, which keeps the generation out of every object key | -| **Meta as the lifecycle linearizer** (a per-hash `.meta` object whose presence/absence *was* the authority for a blob's lifetime) | The marker is a point-read hint only, never consulted by reads; treating it as the linearizer would assert a guarantee the design does not make | The meta stays advisory: the in-body incarnation tag and exact-token delete are the real authority, and an absent meta reads identically to `Clean` | +| **Meta as the lifecycle linearizer** (a per-hash `.meta` object whose presence/absence *was* the authority for a blob's lifetime) | The marker is a point-read hint only, never consulted by reads; treating it as the linearizer would assert a guarantee the design does not make | The meta stays advisory: the in-body incarnation tag and exact-token delete are the real authority, and an absent meta never permits adoption | | **Raw immutable bodies with a three-state tombstone meta** | A resurrect displacing the body forced a terminal-tombstone handshake — a writer↔`GC` liveness coupling that could re-enable data loss | The settled one-key-per-hash design with an in-body incarnation tag | | **Conditional blob creation and conditional staged copy** | Coupled immutable payload publication to provider-specific ETag/generation responses, forced generation-token GCS blobs into a single-`PUT` size cliff, and split the writer into provider/source-specific branches | One provider-neutral state machine: durable precommit, mandatory blob `HEAD`, unconditional publication for absent or condemned bodies, freshness-metadata reconciliation, and explicit `Materialized`/`TrustedManifest` proof | | **A persistent, append-only namespace registry for `GC` discovery** | Never deregistered on drop, so it grew monotonically forever; its fence cost scaled with namespaces ever created, not namespaces live | Discovery from the ref data itself, made safe by two independent coordinates: a durable per-shard incarnation plus a pool-global round | diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index 24c90c6dd670e..0d5d2951a421a 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -691,7 +691,7 @@ flowchart LR D -->|"still zero, .meta absent, Clean, older or unreadable"| C2["carried, .meta rewritten at this round, never throw"] D -->|"current token not equal to t"| SUP["SUPERSEDED -- a writer resurrected, re-condemn the CURRENT token"] G --> E["round n+2, pre-CAS: exact-token DELETE of blob at t"] - E -->|"Deleted or Absent"| F["then drop the .meta if it is still Condemned at round n or older"] + E -->|"Deleted or Absent"| F["then drop the .meta if it is still Condemned from a round before this one"] E -->|TokenMismatch| H["nothing deleted -- live at a newer token, leave the .meta alone"] ``` @@ -703,9 +703,11 @@ round that issues it and replaces an absent marker, `Clean`, or an older-round ` Graduation reads the marker in the round that graduates and accepts only `Condemned` at the entry's round or newer: an older round may belong to another incarnation of the same content. The marker delete after a redelete -removes only `Condemned` at the entry's round or older, conditional on the version it read, so a delayed delete -never removes the marker of a later incarnation. It can leave an orphan `Condemned` marker over an absent body; -the next writer of that hash republishes and clears it, and `cas-fsck` counts it in `meta_without_body`. +removes only `Condemned` from a round before the deleting round, conditional on the version it read. This covers +a marker the graduation gate re-stamped at a round later than the entry's. It can also remove the marker of a +later incarnation; that is safe because a writer adopts only on `Clean` and republishes over an absent marker. +A failed delete job can leave an orphan `Condemned` marker over an absent body; the next writer of that hash +republishes and clears it, and `cas-fsck` counts it in `meta_without_body`. A blob whose in-degree reaches zero in round `n` is retired with `condemn_round = n`; it can graduate to `delete_pending` in round `n+1` at the earliest and be deleted in round `n+2`, so a minimum of two full rounds @@ -764,7 +766,7 @@ logs: | `PUT` run segments | 1 per non-pure-carry shard, plus 1 fold seal | | `HEAD` blobs | 1 per newly condemned | | `GET` blob `.meta` at graduation | 1 per entry the graduation budget admits, plus 1 `GET` and 1 `PUT` per refusal (the rewrite job re-reads the marker before its conditional `PUT`) | -| `GET` + conditional `DELETE` blob `.meta` after a redelete | 1 `GET` per `Deleted` or `Absent` outcome; the `DELETE` only while the marker is `Condemned` at the entry's round or older | +| `GET` + conditional `DELETE` blob `.meta` after a redelete | 1 `GET` per `Deleted` or `Absent` outcome; the `DELETE` only while the marker is `Condemned` from a round before the deleting round | | Blob `HEAD` + conditional `DELETE` | 1 `HEAD` per `redelete` entry — an entry that graduated in an *earlier* round, not the current one — up to `cas_gc_round_redelete_budget`; a `DELETE` only when the body is present at the condemned token | | Successful lease `CAS gc/state` | 1 | | Commit `CAS gc/state` | 1 | diff --git a/docs/en/antalya/cas/architecture/part-lifecycle.md b/docs/en/antalya/cas/architecture/part-lifecycle.md index f5ad05dc5f0fb..9d93c7af7b875 100644 --- a/docs/en/antalya/cas/architecture/part-lifecycle.md +++ b/docs/en/antalya/cas/architecture/part-lifecycle.md @@ -90,7 +90,7 @@ calling thread only submits and joins, never occupies a pool slot, so a pool of to a correct serial run and can never deadlock. The contract is merge-nothing: if any task threw, nothing is merged and the first error in dispatch order is rethrown. Results are folded into the dependency set on the owning thread, into a copy, committed by one no-throw swap. Every physical -task starts with blob `HEAD`; a present non-condemned body or a completed publication yields an +task starts with blob `HEAD`; a present body with a `Clean` marker or a completed publication yields an explicit `Materialized` proof. A trusted source manifest instead yields `TrustedManifest` without blob I/O. Pool size is the server setting `cas_blob_upload_pool_size` (default 16). diff --git a/docs/en/antalya/metrics_and_events.md b/docs/en/antalya/metrics_and_events.md index 7e968baf16c66..11f835d70a3c3 100644 --- a/docs/en/antalya/metrics_and_events.md +++ b/docs/en/antalya/metrics_and_events.md @@ -270,9 +270,9 @@ These counters cover per-blob metadata creates, rewrites, and deletes. | Event | Type | Description | | --- | --- | --- | -| `CASMetaAdoptBackfill` | `Number` | Number of blob-adoption paths that attempted a missing-metadata Clean backfill. Counts the attempt, not a guaranteed metadata creation. | +| `CASMetaAbsentRepublish` | `Number` | Number of blob publication attempts started because the body was present but its freshness marker was absent; such a body is republished, never adopted. Counts attempts, including ones that fail. | | `CASMetaCompareSwap` | `Number` | Number of conditional CAS blob-metadata rewrites, including lost races. Growing values indicate metadata updates or contention. | -| `CASMetaCreateClean` | `Number` | Number of absent-body publication paths that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata creation. | +| `CASMetaCreateClean` | `Number` | Number of publications after an absent body or an absent marker that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata creation. | | `CASMetaDelete` | `Number` | Number of exact-token CAS blob-metadata delete attempts. Growth indicates GC or cleanup activity. | | `CASMetaPut` | `Number` | Number of conditional CAS blob-metadata create attempts, including lost races. Growth indicates metadata reconciliation or write contention. | | `CASMetaResurrectClean` | `Number` | Number of condemned-body replacement paths that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata reset. | diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 21a47a3f7b1c8..0535d36449a04 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -954,8 +954,8 @@ The server successfully detected this situation and will download merged part fr M(CASMetaPut, "Number of conditional CAS blob-metadata create attempts, including lost races. Growth indicates metadata reconciliation or write contention.", ValueType::Number) \ M(CASMetaCompareSwap, "Number of conditional CAS blob-metadata rewrites, including lost races. Growing values indicate metadata updates or contention.", ValueType::Number) \ M(CASMetaDelete, "Number of exact-token CAS blob-metadata delete attempts. Growth indicates GC or cleanup activity.", ValueType::Number) \ - M(CASMetaCreateClean, "Number of absent-body publication paths that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata creation.", ValueType::Number) \ - M(CASMetaAdoptBackfill, "Number of blob-adoption paths that attempted a missing-metadata Clean backfill. Counts the attempt, not a guaranteed metadata creation.", ValueType::Number) \ + M(CASMetaCreateClean, "Number of publications after an absent body or an absent marker that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata creation.", ValueType::Number) \ + M(CASMetaAbsentRepublish, "Number of blob publication attempts started because the body was present but its freshness marker was absent; such a body is republished, never adopted. Counts attempts, including ones that fail.", ValueType::Number) \ M(CASMetaResurrectClean, "Number of condemned-body replacement paths that entered Clean metadata reconciliation. Counts the reason entry, not a guaranteed metadata reset.", ValueType::Number) \ M(CASGCMetaOps, "Number of per-hash metadata operations executed by CAS GC. Growing values indicate more GC candidates or metadata work.", ValueType::Number) \ M(CASGCEnumerationPages, "Number of CAS GC LIST pages fetched while enumerating the object universe. Growing values indicate a larger universe or more frequent scans.", ValueType::Number) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 368c2cf1379d2..9e535c3026b92 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -629,7 +629,10 @@ void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & e /// no reason (the meta is advisory, but there is no reason to touch it on that path at all). if (io.del == Removal::Removed || io.del == Removal::Gone) { - meta_writer->scheduleConfirmedMetaDelete(entry.ref, entry.condemn_round); + /// The round before this one, not `entry.condemn_round`: the graduation gate re-stamps a marker it + /// refused at its own, later round. Deleting a marker is safe for writers that adopt only on `Clean`; + /// a 26.6.4 writer still backfills an absent marker and adopts. + meta_writer->scheduleConfirmedMetaDelete(entry.ref, ctx.new_round - 1); } } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp index 62f49d74df883..e5be358883fbd 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp @@ -33,9 +33,8 @@ namespace /// - every GC write stamps the round of the attempt that issues it, and replaces an absent marker, `Clean` /// or an older-round `Condemned`; /// - graduation accepts only `Condemned` at the entry's round or newer, read in that round (`Gc::fold`); -/// - a delete job removes only `Condemned` at its entry's round or older, `If-Match` the etag it read. -/// While `gc/state` is readable, a later incarnation that can own a job is condemned at a larger round, so no job -/// removes its marker; a rebuild that regresses rounds breaks this. +/// - a delete job removes only `Condemned` from a round before the deleting one, `If-Match` the etag it read. +/// Such a job may remove a later incarnation's marker; that is safe because a writer adopts only on `Clean`. /// /// GC never writes `Clean`, not even on a spare: a deposed leader that cleared a spare's marker and then lost /// its round CAS would leave `Clean` over a still-condemned body, and a stale exact-token redelete would delete @@ -55,10 +54,10 @@ void writeCondemnedMeta(CasOperation & op, const Layout & layout, const BlobRef /// Rounds only grow while `gc/state` is readable, so a marker at a newer round guards a later incarnation /// and `Clean` means a writer displaced the body: both are left alone. `If-Match` the etag read here, so a /// marker changed since the read is left alone too. -void deleteConfirmedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, uint64_t condemn_round) +void deleteConfirmedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, uint64_t newest_round) { const auto lm = loadMeta(op, layout, ref); - if (!lm || lm->meta.state != MetaState::Condemned || lm->meta.condemn_round > condemn_round) + if (!lm || lm->meta.state != MetaState::Condemned || lm->meta.condemn_round > newest_round) return; deleteMetaExact(op, layout, ref, lm->etag); } @@ -127,12 +126,12 @@ void GcMetaWriter::scheduleCondemnMarkerWrite(const BlobRef & ref, uint64_t cond }); } -void GcMetaWriter::scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t condemn_round) +void GcMetaWriter::scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t newest_round) { - submit([st = state, ref, condemn_round]() + submit([st = state, ref, newest_round]() { CasOperation op = st->store->openRequests().admit(); - deleteConfirmedMeta(op, st->store->layout(), ref, condemn_round); + deleteConfirmedMeta(op, st->store->layout(), ref, newest_round); }); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h index b57fbe46427b3..4e374b4291301 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h @@ -35,8 +35,9 @@ class GcMetaWriter void scheduleCondemnMarkerWrite(const BlobRef & ref, uint64_t condemn_round, uint64_t size); /// Drop the marker of a blob whose body is confirmed deleted or absent, but only while it is `Condemned` - /// at `condemn_round` or older: a newer marker, or `Clean`, belongs to a later incarnation. - void scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t condemn_round); + /// at `newest_round` or older; a newer marker and `Clean` stay. The marker removed may be a later + /// incarnation's, which is safe because a writer adopts only on `Clean`. + void scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t newest_round); /// Successful-path protocol barrier. Wait for every job scheduled so far, propagating any /// pool/framework exception recorded by `ThreadPool`; per-hash operation exceptions are caught diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp index 51ff507c141f4..61bceefc1478d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp @@ -26,7 +26,7 @@ namespace ProfileEvents extern const Event CASMetaPut; extern const Event CASMetaCompareSwap; extern const Event CASMetaCreateClean; - extern const Event CASMetaAdoptBackfill; + extern const Event CASMetaAbsentRepublish; extern const Event CASMetaResurrectClean; } @@ -420,22 +420,10 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) if (loaded) validateMetaSize(loaded->meta); - if (!loaded || loaded->meta.state == MetaState::Clean) + /// Only a `Clean` marker read after the `HEAD` permits adoption. GC deletes the marker right + /// after the body, so an absent marker may mean the body this `HEAD` saw is already gone. + if (loaded && loaded->meta.state == MetaState::Clean) { - /// Observation can produce durable metadata and dependency readiness too. - requireAdmitted("after the mandatory `HEAD`"); - if (!loaded) - { - /// A backfilled marker is a point-read hint for the next observer, so a competing - /// writer that got there first settles the same question: its outcome is not read. - ProfileEvents::increment(ProfileEvents::CASMetaAdoptBackfill); - putMetaIfAbsent( - op, - store->layout(), - ref, - BlobMeta{.state = MetaState::Clean, .condemn_round = 0, .size = logical_size}, - policy); - } requireAdmitted("before the body-put-avoided observation is recorded"); ProfileEvents::increment(ProfileEvents::CASBlobBodyPutAvoided); EventEmitter{*store}.emit([&](CasEvent & event) @@ -445,7 +433,7 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) event.object_hash = blobIdOf(ref); event.token = present->etag.render(); event.outcome = "observed"; - event.reason = "a present non-condemned blob was observed after mandatory `HEAD`"; + event.reason = "a present blob with a `Clean` marker was observed after mandatory `HEAD`"; event.detail = {{"action", "observed"}, {"size", std::to_string(source.size)}}; }); requireAdmitted("before the observed dependency proof is returned"); @@ -454,7 +442,10 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) BlobDepRecord{ObjectKind::Blob, BlobDependencyProof::Materialized, source.size}, BlobUploadDiagnostics{BlobMaterializationAction::Observed, std::nullopt, std::nullopt}}; } - reason = BlobPublicationReason::Condemned; + if (loaded) + reason = BlobPublicationReason::Condemned; + else + ProfileEvents::increment(ProfileEvents::CASMetaAbsentRepublish); } const bool first_publication = source.beginPublication(); @@ -526,7 +517,7 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) event.object_hash = blobIdOf(ref); event.outcome = "published"; event.reason = reason == BlobPublicationReason::Absent - ? "blob was absent after mandatory `HEAD`" + ? "blob or its marker was absent after mandatory `HEAD`" : "blob was present but `Condemned` after mandatory `HEAD`"; event.detail = { {"action", "published"}, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md index 44430e021ddbb..3c4ffaf4274b2 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md @@ -53,9 +53,9 @@ Everything persisted is a text format with a versioned header line; see - **Write**: `ContentAddressedTransaction` buffers or spills file writes (`scratch_path`, or opt-in S3 staging), hashes the payload, and hands it to - `Pool/CasPartWriteTxn`: start every materialization with blob `HEAD`, adopt a present - non-condemned body or publish the writer's source unconditionally, reconcile `.meta` to - `Clean`, and record explicit `Materialized` proof; then stage the manifest, `precommitAdd` the ref, + `Pool/CasPartWriteTxn`: start every materialization with blob `HEAD`, adopt only on a `Clean` + marker read after the `HEAD`, otherwise publish the writer's source unconditionally and reconcile + `.meta` to `Clean`, and record explicit `Materialized` proof; then stage the manifest, `precommitAdd` the ref, then promote it to committed in the ref log. - **Read**: disk path → `Parts/PartPathParser` (namespace + part + file) → ref table resolves the manifest → `Pool/CasManifestReader` locates the diff --git a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp index f3bd6da9095a8..b9001dfbb26a0 100644 --- a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp +++ b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp @@ -1466,17 +1466,13 @@ TEST(CASGCCondemnMarker, SwallowedMarkerWriteCarriesEntryInsteadOfDeleting) - carries_before, 4u) << "every refused graduation must count one unconfirmed carry"; - /// Heal the backend: the carry-time retry publishes the marker, the entry confirms + graduates, and - /// the pipeline reclaims the blob. The retry stamped a round above the entry's, so the entry's marker - /// delete leaves it: an orphan over the absent body. - const uint64_t entry_round = e->condemn_round; + /// Heal the backend: the carry-time retry publishes the marker at a round above the entry's, the entry + /// confirms and graduates, and the pipeline reclaims the blob together with that re-stamped marker. backend->fail_meta_writes.store(false); EXPECT_TRUE(runRoundsUntilAbsent(store, gc, *backend, store->layout(), blob)); EXPECT_FALSE(currentEntryFor(*backend, store->layout(), blob).has_value()); - const auto orphan = loadMetaForTest(*backend, store->layout(), blob); - ASSERT_TRUE(orphan.has_value()); - EXPECT_EQ(orphan->meta.state, MetaState::Condemned); - EXPECT_GT(orphan->meta.condemn_round, entry_round); + EXPECT_FALSE(loadMetaForTest(*backend, store->layout(), blob).has_value()) + << "the re-stamped marker outlived its body"; } /// The healthy-path counterpart: with the condemn-time marker write landing normally, the gate must not diff --git a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp index 2c1f23ceaacd4..546bbfdf67116 100644 --- a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp +++ b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp @@ -78,7 +78,7 @@ TEST(CASGcMetaWriter, RealConfirmedMetaDeleteCompletesAcrossOwnerDestruction) auto gc = std::make_unique(store, DB::Cas::tests::u128Of(kGcId)); backend->arm(); - gc->metaWriterForTest().scheduleConfirmedMetaDelete(ref, /*condemn_round=*/1); + gc->metaWriterForTest().scheduleConfirmedMetaDelete(ref, /*newest_round=*/1); awaitLatchEntered(*backend); @@ -179,9 +179,9 @@ namespace { /// The delete job, scheduled the way `Gc::applyRedeleteOutcome` schedules it, and drained. -void runConfirmedMetaDelete(Gc & gc, const BlobRef & ref, uint64_t condemn_round) +void runConfirmedMetaDelete(Gc & gc, const BlobRef & ref, uint64_t newest_round) { - gc.metaWriterForTest().scheduleConfirmedMetaDelete(ref, condemn_round); + gc.metaWriterForTest().scheduleConfirmedMetaDelete(ref, newest_round); gc.metaWriterForTest().drain(); } @@ -739,9 +739,7 @@ TEST(CASGcMetaWriter, GraduationRefusesAnOlderRoundMarker) EXPECT_EQ(runGateRound(store, gc).graduated, 1u); EXPECT_EQ(runGateRound(store, gc).redeleted, 1u); EXPECT_FALSE(blobPresent(*backend, layout, x)); - const auto orphan = markerOf(*backend, layout, x); - ASSERT_TRUE(orphan.has_value()) << "the job for round " << c << " removed a newer marker"; - EXPECT_EQ(orphan->meta.condemn_round, c + 1); + EXPECT_FALSE(markerOf(*backend, layout, x).has_value()) << "the re-stamped marker outlived its body"; } TEST(CASGcMetaWriter, SparedThenRecondemnedBlobGraduates) @@ -775,3 +773,28 @@ TEST(CASGcMetaWriter, SparedThenRecondemnedBlobGraduates) return entry && entry->delete_pending; }, /*max_rounds*/ 2)) << "a spared, re-condemned blob never graduates"; } + +/// The condemn-marker write is lost, so the gate refuses and re-stamps the marker at the refusing round. +/// The redelete must remove that marker with the body; nothing else ever would. +TEST(CASGcMarker, RedeleteRemovesMarkerRestampedByGate) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x3604); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + ASSERT_NO_FATAL_FAILURE(removeMarker(*backend, layout, x)); + + ASSERT_EQ(runGateRound(store, gc).graduated, 0u); + const auto restamped = markerOf(*backend, layout, x); + ASSERT_TRUE(restamped.has_value()); + ASSERT_EQ(restamped->meta.condemn_round, c + 1); + ASSERT_EQ(runGateRound(store, gc).graduated, 1u); + + ASSERT_EQ(runGateRound(store, gc).redeleted, 1u); + EXPECT_FALSE(blobPresent(*backend, layout, x)); + EXPECT_FALSE(markerOf(*backend, layout, x).has_value()) << "a Condemned marker outlived its body"; +} diff --git a/src/Disks/tests/gtest_cas_part_write.cpp b/src/Disks/tests/gtest_cas_part_write.cpp index a60f5b45b11ad..1916584042ced 100644 --- a/src/Disks/tests/gtest_cas_part_write.cpp +++ b/src/Disks/tests/gtest_cas_part_write.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -31,7 +32,7 @@ namespace ProfileEvents extern const Event CASMetaPut; extern const Event CASMetaCompareSwap; extern const Event CASMetaCreateClean; -extern const Event CASMetaAdoptBackfill; +extern const Event CASMetaAbsentRepublish; extern const Event CASMetaResurrectClean; extern const Event CASBlobAdoptTrusted; } @@ -512,44 +513,32 @@ TEST(CASPartWriteTxnMetaCounters, CreateCleanAndChokePointCountOnFreshBody) EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaCreateClean] - reason_before, 1); } -/// §0 introspection: an adopt of a pre-existing body that has NO meta at all (a pre-protocol blob, or a -/// lost race with a concurrent fresh-uploader's own meta write) backfills a Clean meta through the -/// `putMetaIfAbsent` choke point (`CASMetaPut`), tagged with its reason (`CASMetaAdoptBackfill`). No -/// existing test elsewhere in the suite drives this branch: every other pre-seeded raw body in this file -/// pairs `writeRawBlobBody` with `writeMetaClean`, which skips the `!lm` backfill branch entirely. -TEST(CASPartWriteTxnMetaCounters, AdoptBackfillCountsChokePointAndReason) +/// A present body without a marker is republished, and the marker is created through the same choke +/// point (`CASMetaPut`) and reason (`CASMetaCreateClean`) as for an absent body. +TEST(CASPartWriteTxnMetaCounters, AbsentMarkerRepublishCountsChokePointAndReason) { const auto put_before = ProfileEvents::global_counters[ProfileEvents::CASMetaPut]; - const auto reason_before = ProfileEvents::global_counters[ProfileEvents::CASMetaAdoptBackfill]; + const auto create_before = ProfileEvents::global_counters[ProfileEvents::CASMetaCreateClean]; + const auto republish_before = ProfileEvents::global_counters[ProfileEvents::CASMetaAbsentRepublish]; auto b = std::make_shared(); auto s = openPool(b); - const String payload = "adopt-backfill-payload-counters"; + const String payload = "absent-marker-republish-counters"; const UInt128 hash = u128Of(payload); const BlobRef id = idOf(payload); - - /// Pre-seed a present body big enough that `ensureBlobPresent`'s logical-size guard does not underflow — - /// deliberately WITHOUT any meta (unlike PutBlobAdoptsWhenMetaCleanNoRetireView), so the adopt reaches - /// the `!lm` backfill branch. - const uint64_t header_len = s->poolMeta().blob_header_len; - String raw_body(header_len, '\0'); + String raw_body(s->poolMeta().blob_header_len, '\0'); raw_body += payload; writeRawBlobBody(*b, s->layout(), hash, raw_body); - /// Adopt must happen under a durable precommit edge (EDGE-BEFORE-OBSERVE). - const RootNamespace ns{"srv/tbl"}; - auto build = startBuildFor(s, ns, "ref_adopt_backfill"); - const ManifestId manifest_id = build->stageManifest({blobManifestEntry("data.bin", payload)}); - build->precommitAdd(ns, "ref_adopt_backfill", manifest_id); - auto ref = build->putBlob(id, BlobSource::fromString(payload)); - EXPECT_EQ(ref.ref, id); + auto build = precommittedBuildForPayload(s, RootNamespace{"srv1/absent-marker-counters"}, "part", payload); + EXPECT_EQ(build->putBlob(id, BlobSource::fromString(payload)).ref, id); EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaPut] - put_before, 1); - EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaAdoptBackfill] - reason_before, 1); - + EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaCreateClean] - create_before, 1); + EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaAbsentRepublish] - republish_before, 1); const auto lm = loadMetaForTest(*b, s->layout(), hash); - ASSERT_TRUE(lm.has_value()) << "the adopt-backfill must leave a Clean meta for future point-readers"; + ASSERT_TRUE(lm.has_value()); EXPECT_EQ(lm->meta.state, MetaState::Clean); } @@ -2712,11 +2701,10 @@ TEST(CASPartWrite, AmbiguousTimeoutsThenCommitRestreamsFromSource) EXPECT_TRUE((*op).head(s->layout().blobKey(idOf(payload)), Retry::once()).has_value()) << "the blob body must be durable"; } -/// Ambiguous-but-landed: the FIRST attempt's response is lost AFTER the write actually landed -/// server-side. The occupancy resolve observes the key present and the existing 412 machinery takes -/// over — the occupant is ADOPTED (content-addressed identity: any occupant of this key IS the -/// content), with NO reissue and NO second body upload. -TEST(CASPartWrite, AmbiguousLandedWriteAdoptsOccupantWithoutReupload) +/// Ambiguous-but-landed: the first publication's response is lost after the body landed, so the +/// marker was never created. A present body without a marker is never adopted, so the landed attempt +/// costs one more publication, after which the marker is `Clean`. +TEST(CASPartWrite, AmbiguousLandedWriteRepublishesOnceThenClean) { auto b = std::make_shared(); /// Heap-owned, not a plain local: `~Pool` emits terminate events into the sink, and a background @@ -2742,24 +2730,19 @@ TEST(CASPartWrite, AmbiguousLandedWriteAdoptsOccupantWithoutReupload) const PutBlobResult res = build->putBlob(idOf(payload), countingSource(payload, payload_streams)); EXPECT_EQ(res.size, payload.size()); - EXPECT_EQ(b->publish_stream_attempts, 1) << "a landed ambiguous attempt must be observed, never reissued"; - EXPECT_EQ(b->blob_head_attempts, 2) << "the ambiguity is resolved by restarting from HEAD"; - EXPECT_EQ(payload_streams, 1); + EXPECT_EQ(b->publish_stream_attempts, 2) << "the landed attempt and exactly one republication"; + EXPECT_EQ(b->blob_head_attempts, 2); + EXPECT_EQ(payload_streams, 2); - const String key = s->layout().blobKey(idOf(payload)); + const auto marker = loadMetaForTest(*b, s->layout(), u128Of(payload)); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.state, MetaState::Clean); const std::vector observed_events = events->snapshot(); - const auto adopt = std::find_if(observed_events.begin(), observed_events.end(), - [](const CasEvent & e) { return e.type == CasEventType::BlobReuseAdopt; }); - ASSERT_NE(adopt, observed_events.end()) << "the landed occupant must be ADOPTED (the standard dedup leg)"; - CasRequests probe(b, Fence::open()); - CasOperation probe_op = probe.admit(); - const auto landed = probe_op.head(key, Retry::standard()); - ASSERT_TRUE(landed.has_value()); - EXPECT_EQ(adopt->token, landed->etag.render()) - << "the adopted token must be the landed incarnation, rendered"; EXPECT_EQ(std::count_if(observed_events.begin(), observed_events.end(), - [](const CasEvent & e) { return e.type == CasEventType::BlobPut; }), 0) - << "no fresh-upload event: the body was never re-uploaded"; + [](const CasEvent & e) { return e.type == CasEventType::BlobReuseAdopt; }), 0) + << "a body without a marker was adopted"; + EXPECT_EQ(std::count_if(observed_events.begin(), observed_events.end(), + [](const CasEvent & e) { return e.type == CasEventType::BlobPut; }), 1); } /// Every physical publication mints its own envelope, and the request engine never reissues one. Two @@ -2925,16 +2908,12 @@ TEST(CASPartWrite, AmbiguousNonLandingPublicationStopsAtOuterBound) EXPECT_EQ(payload_streams, 8); } -/// A server-side copy publication is ambiguous-but-landed: its response is lost after the -/// destination was created. The occupancy resolve observes the destination present and the occupant -/// is adopted — Committed-in-effect WITHOUT a re-copy. -TEST(CASPartWrite, AmbiguousCopyLandedAdoptsDestinationWithoutRecopy) +/// A server-side copy publication is ambiguous-but-landed: its response is lost after the destination +/// was created, so the marker was never created. The verbatim copy is spent, so the one republication +/// streams the payload under a fresh envelope, after which the marker is `Clean`. +TEST(CASPartWrite, AmbiguousCopyLandedRepublishesOnceThenClean) { auto b = std::make_shared(); - /// Heap-owned, not a plain local: `~Pool` emits terminate events into the sink, and a background - /// publish can hold an extra `shared_from_this()` past this frame's return regardless of - /// declaration order relative to the Pool, so a by-reference capture of a local would dangle. - auto events = std::make_shared(); auto s = openBlobFaultPool(b); const RootNamespace ns{"srv/tbl"}; const String payload = "staged-payload-A"; @@ -2946,11 +2925,6 @@ TEST(CASPartWrite, AmbiguousCopyLandedAdoptsDestinationWithoutRecopy) ASSERT_TRUE(std::holds_alternative((*seed_op).create(staging_key, staging_bytes, Retry::once()))); } - s->setEventSink([events](const CasEvent & e) - { - events->push(e); - }); - auto build = startBuildFor(s, ns, "part_copy_landed"); const ManifestId id = build->stageManifest({blobManifestEntry("a.bin", payload)}); build->precommitAdd(ns, "part_copy_landed", id); @@ -2958,23 +2932,27 @@ TEST(CASPartWrite, AmbiguousCopyLandedAdoptsDestinationWithoutRecopy) BlobSource source; source.size = payload.size(); source.server_side_copy_from = staging_key; + source.open = [payload]() -> std::unique_ptr + { + return std::make_unique(payload); + }; b->fault_count = 1; b->land_despite_fault = true; const PutBlobResult res = build->putBlob(idOf(payload), std::move(source)); EXPECT_EQ(res.size, payload.size()); - EXPECT_EQ(b->publish_copy_attempts, 1) << "a landed ambiguous copy must be resolved, never re-copied"; - EXPECT_EQ(b->publish_stream_attempts, 0); + EXPECT_EQ(b->publish_copy_attempts, 1) << "the verbatim copy is a once-only privilege"; + EXPECT_EQ(b->publish_stream_attempts, 1) << "one republication after the landed copy"; EXPECT_EQ(b->blob_head_attempts, 2); const String key = s->layout().blobKey(idOf(payload)); OperationForTest op(*b); const auto got = (*op).read(key, Retry::once()); ASSERT_TRUE(got.has_value()); - EXPECT_EQ(got->bytes, staging_bytes) << "the destination is the staging object's verbatim copy"; - const std::vector observed_events = events->snapshot(); - EXPECT_NE(std::find_if(observed_events.begin(), observed_events.end(), - [](const CasEvent & e) { return e.type == CasEventType::BlobReuseAdopt; }), - observed_events.end()) << "the landed destination must be ADOPTED"; + EXPECT_NE(got->bytes, staging_bytes) << "the landed copy was adopted"; + EXPECT_EQ(got->bytes.substr(s->poolMeta().blob_header_len), payload); + const auto marker = loadMetaForTest(*b, s->layout(), u128Of(payload)); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.state, MetaState::Clean); } /// A server-side copy publication is ambiguous-and-absent: the first copy attempt times out with @@ -3018,3 +2996,117 @@ TEST(CASPartWrite, AmbiguousCopyAbsentReattemptsAndCommits) EXPECT_NE(got->bytes, staging_bytes); EXPECT_EQ(got->bytes.substr(s->poolMeta().blob_header_len), payload); } + +namespace +{ + +/// Runs a side effect once, on the first read of the watched key, before that read is served. +class BeforeMetaReadBackend final : public InMemoryBackend +{ +public: + String watched_key; + std::function trigger; + + std::optional read(const String & key, TransportAccess & access) override + { + if (key == watched_key && trigger) + std::exchange(trigger, {})(); + return InMemoryBackend::read(key, access); + } +}; + +/// A present body and no marker, as a writer that crashed between its publication and its marker +/// create leaves them. +void seedBodyWithoutMarker(Backend & backend, const PoolPtr & store, const String & payload) +{ + String raw_body(store->poolMeta().blob_header_len, '\0'); + raw_body += payload; + writeRawBlobBody(backend, store->layout(), u128Of(payload), raw_body); +} + +BlobUploadResult uploadOnce(const PartWriteTxnPtr & build, const BlobRef & ref, const String & payload) +{ + return build->uploadBlobDetached(BlobUploadRequest{ref, BlobSource::fromString(payload), payload.size()}); +} + +} + +/// GC deletes the body and then its marker between the writer's `HEAD` and its marker read. The writer +/// saw a present body and an absent marker; adopting would name a deleted body. +TEST(CASPartWriteTxnMarkerRace, AbsentMarkerAfterHeadRepublishes) +{ + auto b = std::make_shared(); + auto s = openPool(b); + const String payload = "absent-marker-after-head"; + const BlobRef ref = idOf(payload); + const String blob_key = s->layout().blobKey(ref); + seedBodyWithoutMarker(*b, s, payload); + writeMetaClean(*b, s->layout(), u128Of(payload), payload.size()); + condemnMeta(*b, s->layout(), u128Of(payload), /*condemn_round*/ 1); + const Etag condemned_token = [&] + { + OperationForTest op(*b); + return (*op).head(blob_key, Retry::once()).value().etag; + }(); + + auto build = precommittedBuildForPayload(s, RootNamespace{"srv1/marker-race"}, "part", payload); + bool fired = false; + b->watched_key = s->layout().blobMetaKey(ref); + b->trigger = [&] + { + fired = true; + OperationForTest op(*b); + ASSERT_EQ((*op).remove(blob_key, condemned_token, Retry::once()), Removal::Removed); + const auto marker = loadMeta(*op, s->layout(), ref); + ASSERT_TRUE(marker.has_value()); + ASSERT_EQ(deleteMetaExact(*op, s->layout(), ref, marker->etag), Removal::Removed); + }; + + const BlobUploadResult result = uploadOnce(build, ref, payload); + ASSERT_TRUE(fired) << "the writer never read the marker after its `HEAD`"; + + EXPECT_EQ(result.diagnostics.action, BlobMaterializationAction::Published) << "adopted a deleted body"; + EXPECT_EQ(result.diagnostics.reason, std::optional(BlobPublicationReason::Absent)); + OperationForTest op(*b); + const auto head = (*op).head(blob_key, Retry::once()); + ASSERT_TRUE(head.has_value()) << "the writer's dependency names an absent body"; + EXPECT_NE(head->etag, condemned_token); + const auto marker = loadMetaForTest(*b, s->layout(), u128Of(payload)); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.state, MetaState::Clean); + + /// GC's exact-token delete of the condemned incarnation, replayed late, misses the fresh body. + EXPECT_EQ((*op).remove(blob_key, condemned_token, Retry::once()), Removal::Mismatch); + EXPECT_TRUE((*op).head(blob_key, Retry::once()).has_value()); +} + +/// A crash leftover costs one upload: the first writer republishes and creates the marker, the next one +/// adopts on that `Clean` marker. +TEST(CASPartWriteTxnMarkerRace, PresentBodyWithoutMarkerRepublishesOnce) +{ + auto b = std::make_shared(); + auto s = openPool(b); + const String payload = "present-body-without-marker"; + const BlobRef ref = idOf(payload); + const String blob_key = s->layout().blobKey(ref); + seedBodyWithoutMarker(*b, s, payload); + OperationForTest op(*b); + const Etag leftover = (*op).head(blob_key, Retry::once())->etag; + + const auto creates_before = ProfileEvents::global_counters[ProfileEvents::CASMetaCreateClean].load(); + auto first = precommittedBuildForPayload(s, RootNamespace{"srv1/leftover-a"}, "part", payload); + const BlobUploadResult first_result = uploadOnce(first, ref, payload); + EXPECT_EQ(first_result.diagnostics.action, BlobMaterializationAction::Published); + EXPECT_EQ(first_result.diagnostics.reason, std::optional(BlobPublicationReason::Absent)); + EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASMetaCreateClean].load() - creates_before, 1); + const auto republished = (*op).head(blob_key, Retry::once()); + ASSERT_TRUE(republished.has_value()); + EXPECT_NE(republished->etag, leftover); + const auto marker = loadMetaForTest(*b, s->layout(), u128Of(payload)); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.state, MetaState::Clean); + + auto second = precommittedBuildForPayload(s, RootNamespace{"srv1/leftover-b"}, "part", payload); + EXPECT_EQ(uploadOnce(second, ref, payload).diagnostics.action, BlobMaterializationAction::Observed); + EXPECT_EQ((*op).head(blob_key, Retry::once())->etag, republished->etag); +} diff --git a/src/Disks/tests/gtest_cas_protocol_scenarios.cpp b/src/Disks/tests/gtest_cas_protocol_scenarios.cpp index f33018415c0af..73c73b243ecb6 100644 --- a/src/Disks/tests/gtest_cas_protocol_scenarios.cpp +++ b/src/Disks/tests/gtest_cas_protocol_scenarios.cpp @@ -50,6 +50,7 @@ using DB::Cas::tests::loadMetaForTest; using DB::Cas::tests::streamingHexOf; using DB::Cas::tests::u128Of; using DB::Cas::tests::writeBlobRaw; +using DB::Cas::tests::writeMetaClean; namespace { @@ -190,8 +191,9 @@ TEST(CASProtocol, RevalidateReObservesStaleTokenKeepsWhenUnchanged) auto s = openPool(b); const RootNamespace ns{"srv1/tbl"}; - /// X pre-exists out-of-band; the build dedup-adopts it via putBlob (records the current token t0). + /// X and its `Clean` marker pre-exist out-of-band; the build dedup-adopts X via putBlob (token t0). writeBlobRaw(*b, s->layout(), "payload-X", s->poolMeta().blob_header_len, s->poolMeta().pool_id); + writeMetaClean(*b, s->layout(), u128Of("payload-X"), 9); const String blob_key = s->layout().blobKey(idOf("payload-X")); const Etag t0 = currentIncarnation(*b, blob_key); @@ -199,7 +201,7 @@ TEST(CASProtocol, RevalidateReObservesStaleTokenKeepsWhenUnchanged) auto build = startBuildFor(s, ns, "part_1"); const ManifestId id = build->stageManifest({blobEntry("data.bin", "payload-X")}); build->precommitAdd(ns, "part_1", id); - build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// dedup → adopts t0 + build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// adopts t0 on its `Clean` marker /// GC advanced the round to 1 with an EMPTY retired set; fence to 1. X is NOT condemned and its /// token is unchanged. @@ -224,6 +226,7 @@ TEST(CASProtocol, RevalidateReObservesStaleTokenAdoptsWhenDisplaced) const RootNamespace ns{"srv1/tbl"}; writeBlobRaw(*b, s->layout(), "payload-X", s->poolMeta().blob_header_len, s->poolMeta().pool_id); + writeMetaClean(*b, s->layout(), u128Of("payload-X"), 9); const String blob_key = s->layout().blobKey(idOf("payload-X")); const Etag t0 = currentIncarnation(*b, blob_key); @@ -231,7 +234,7 @@ TEST(CASProtocol, RevalidateReObservesStaleTokenAdoptsWhenDisplaced) /// Wiring order (EDGE-BEFORE-OBSERVE): stageManifest -> precommitAdd -> putBlob. const ManifestId id = build->stageManifest({blobEntry("data.bin", "payload-X")}); build->precommitAdd(ns, "part_1", id); - build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// dedup → adopts t0 + build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// adopts t0 on its `Clean` marker /// Another writer displaces X out-of-band ⇒ a new current token t1 (same payload, fresh tag). const Etag t1 = displaceBlobToken(*b, s->layout(), idOf("payload-X")); @@ -269,6 +272,7 @@ TEST(CASProtocol, RevalidateAdoptsLiveTokenWhenOnlyPhantomCondemnedAtDifferentTo { auto s0 = Pool::open(b, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); writeBlobRaw(*b, s0->layout(), "payload-X", s0->poolMeta().blob_header_len, s0->poolMeta().pool_id); + writeMetaClean(*b, s0->layout(), u128Of("payload-X"), 9); } const String blob_key = layout.blobKey(idOf("payload-X")); const Etag t0 = currentIncarnation(*b, blob_key); @@ -285,7 +289,7 @@ TEST(CASProtocol, RevalidateAdoptsLiveTokenWhenOnlyPhantomCondemnedAtDifferentTo auto build = startBuildFor(s, ns, "part_1"); const ManifestId id = build->stageManifest({blobEntry("data.bin", "payload-X")}); build->precommitAdd(ns, "part_1", id); - build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// dedup → adopts t0 + build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// adopts t0 on its `Clean` marker /// promote: the materialized leaf is edge-protected (not re-observed) ⇒ commit. Lands. t0 untouched. build->promote(ns, "part_1", build->buildId(), id); @@ -480,6 +484,7 @@ TEST(CASProtocol, DisplacedToLiveTokenCommitsAtCurrentIncarnation) const RootNamespace ns{"srv1/tbl"}; writeBlobRaw(*b, s->layout(), "payload-X", s->poolMeta().blob_header_len, s->poolMeta().pool_id); + writeMetaClean(*b, s->layout(), u128Of("payload-X"), 9); const String blob_key = s->layout().blobKey(idOf("payload-X")); const Etag t0 = currentIncarnation(*b, blob_key); @@ -487,7 +492,7 @@ TEST(CASProtocol, DisplacedToLiveTokenCommitsAtCurrentIncarnation) /// Wiring order (EDGE-BEFORE-OBSERVE): stageManifest -> precommitAdd -> putBlob. const ManifestId id = build->stageManifest({blobEntry("data.bin", "payload-X")}); build->precommitAdd(ns, "part_1", id); - build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// dedup → adopts t0 + build->putBlob(idOf("payload-X"), BlobSource::fromString("payload-X")); /// adopts t0 on its `Clean` marker /// Another writer displaces X to t1 (uncondemned) before our gate runs. const Etag t1 = displaceBlobToken(*b, s->layout(), idOf("payload-X")); diff --git a/src/Disks/tests/gtest_cas_upload_detached.cpp b/src/Disks/tests/gtest_cas_upload_detached.cpp index 21c8ca69e5d92..358d1da58e51e 100644 --- a/src/Disks/tests/gtest_cas_upload_detached.cpp +++ b/src/Disks/tests/gtest_cas_upload_detached.cpp @@ -230,25 +230,23 @@ TEST(CASUploadDetached, ExistingCleanHeadsAndObservesWithoutPublication) EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASBlobBodyPutAvoided], avoided_before + 1); } -TEST(CASUploadDetached, ExistingBodyWithoutMetadataBackfillsWithoutPublication) +/// A body without a marker is never adopted: it is republished once and the marker is created `Clean`. +TEST(CASUploadDetached, ExistingBodyWithoutMetadataRepublishesOnceThenClean) { - const String payload = "mandatory-head-metadata-backfill"; + const String payload = "mandatory-head-metadata-absent"; const BlobRef ref = idOf(payload); auto backend = std::make_shared(); auto store = openUploadPool(backend); seedPresentBody(*backend, store->layout(), store->poolMeta(), ref, payload); - auto build = precommitBuildFor(store, RootNamespace{"srv1/protocol-backfill"}, "part", payload); + auto build = precommitBuildFor(store, RootNamespace{"srv1/protocol-absent-meta"}, "part", payload); backend->watch(store->layout().blobKey(ref), store->layout().blobMetaKey(ref)); const BlobUploadResult result = build->uploadBlobDetached( BlobUploadRequest{ref, BlobSource::fromString(payload), payload.size()}); EXPECT_EQ(result.dep.proof, BlobDependencyProof::Materialized); - ASSERT_FALSE(backend->operations.empty()); - EXPECT_EQ(backend->operations.front(), "head"); - EXPECT_EQ(backend->blob_heads, 1u); - EXPECT_EQ(backend->meta_gets, 1u); - EXPECT_EQ(backend->publish_calls, 0u); + EXPECT_EQ(result.diagnostics.action, BlobMaterializationAction::Published); + EXPECT_EQ(backend->operations, (std::vector{"head", "meta-get", "publish"})); const auto meta = loadMetaForTest(*backend, store->layout(), u128Of(payload)); ASSERT_TRUE(meta.has_value()); EXPECT_EQ(meta->meta.state, MetaState::Clean); @@ -312,8 +310,8 @@ TEST(CASUploadDetached, PresentCondemnedPublishesFreshAndQueuedOldDeleteMisses) } } -/// A present body with absent metadata is observed and backfilled `Clean` without publication. -TEST(CASUploadDetached, PresentBodyWithoutMetadataBackfills) +/// A present body with absent metadata is republished as if absent, and the marker is created `Clean`. +TEST(CASUploadDetached, PresentBodyWithoutMetadataRepublishesThenClean) { const RootNamespace ns{"srv1/nsAdopt"}; const String ref_name = "part"; @@ -324,7 +322,7 @@ TEST(CASUploadDetached, PresentBodyWithoutMetadataBackfills) { b = std::make_shared(); s = openUploadPool(b); - seedPresentBody(*b, s->layout(), s->poolMeta(), blob, payload); /// body present, no meta: backfill + seedPresentBody(*b, s->layout(), s->poolMeta(), blob, payload); /// body present, no meta build = precommitBuildFor(s, ns, ref_name, payload); }; @@ -340,14 +338,14 @@ TEST(CASUploadDetached, PresentBodyWithoutMetadataBackfills) const BlobUploadResult r = build1->uploadBlobDetached( BlobUploadRequest{blob, BlobSource::fromString(payload), payload.size()}); - EXPECT_EQ(r.diagnostics.action, BlobMaterializationAction::Observed); - EXPECT_EQ(r.diagnostics.reason, std::nullopt); - EXPECT_EQ(r.diagnostics.transport, std::nullopt); + EXPECT_EQ(r.diagnostics.action, BlobMaterializationAction::Published); + EXPECT_EQ(r.diagnostics.reason, std::optional(BlobPublicationReason::Absent)); + EXPECT_EQ(r.diagnostics.transport, std::optional(BlobPublicationTransport::Streaming)); EXPECT_EQ(r.dep.proof, BlobDependencyProof::Materialized); EXPECT_EQ(r.dep.size, payload.size()); EXPECT_EQ(build1->dependencyProof(blob), std::nullopt); - /// The point-read backfilled a Clean meta. + /// The publication's reconcile created a Clean meta. EXPECT_EQ(metaStateAt(*b1, s1->layout(), payload), std::optional(MetaState::Clean)); std::shared_ptr b2; diff --git a/src/Disks/tests/gtest_cas_upload_fanout.cpp b/src/Disks/tests/gtest_cas_upload_fanout.cpp index 5989ea44ccd30..8728fe38b5e02 100644 --- a/src/Disks/tests/gtest_cas_upload_fanout.cpp +++ b/src/Disks/tests/gtest_cas_upload_fanout.cpp @@ -320,8 +320,8 @@ TEST(CASUploadFanout, CopiedAndMovedRequestsSharePublicationAttemptedState) /// Test 1 (spec §1 "serial-vs-parallel equivalence for successful runs"): a multi-blob part that /// exercises every branch of `uploadBlobDetached` produces IDENTICAL recorded deps and IDENTICAL backend /// end state whether the fan-out runs serially (pool size 1) or in parallel (pool size 4). It covers -/// present-clean observation, metadata backfill, fresh local publication, staging copy, and local and -/// staged condemned-body republication. +/// present-clean observation, republication of a body without metadata, fresh local publication, +/// staging copy, and local and staged condemned-body republication. namespace { @@ -337,7 +337,7 @@ struct WorldA }; const char * const kObserved = "fanoutA-observed-clean"; -const char * const kAdopt = "fanoutA-head-miss-adopt"; +const char * const kAbsentMarker = "fanoutA-absent-marker"; const char * const kFresh = "fanoutA-fresh-local"; const char * const kStaging = "fanoutA-s3-staging"; const char * const kResLocal = "fanoutA-condemned-local"; @@ -355,8 +355,8 @@ WorldA arrangeWorldA() seedPresentBody(*w.b, w.s->layout(), w.s->poolMeta(), kObserved); writeMetaClean(*w.b, w.s->layout(), u128Of(kObserved), std::string(kObserved).size()); - /// HEAD-miss then 412-path live adopt with meta backfill: present body, NO meta, not cached. - seedPresentBody(*w.b, w.s->layout(), w.s->poolMeta(), kAdopt); + /// Present body, NO meta: republished like an absent body. + seedPresentBody(*w.b, w.s->layout(), w.s->poolMeta(), kAbsentMarker); /// fresh local streaming: nothing present. @@ -389,13 +389,13 @@ WorldA arrangeWorldA() w.requests = { localRequest(kObserved), - localRequest(kAdopt), + localRequest(kAbsentMarker), localRequest(kFresh), s3Request(kStaging, "p/staging/mount1/A-staging.tmp"), localRequest(kResLocal), s3Request(kResS3, "p/staging/mount1/A-republish.tmp"), }; - w.payloads = {kObserved, kAdopt, kFresh, kStaging, kResLocal, kResS3}; + w.payloads = {kObserved, kAbsentMarker, kFresh, kStaging, kResLocal, kResS3}; return w; } From f4dc9e42729f19e4490482d67be975b8c608dab1 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 6 Nov 2026 03:05:00 +0100 Subject: [PATCH 03/23] Show CAS mount renewal progress in system.cas_mounts Port of https://github.com/Altinity/ClickHouse/pull/2513 (antalya-26.6) to antalya-26.8. (cherry picked from commit c147de3f55eeda5905e5be974d530d595cd2e223) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Mikhail Filimonov --- .../cas/architecture/mounts-and-leases.md | 1 + .../en/operations/system-tables/cas_mounts.md | 10 +- .../ContentAddressed/Backend/CasRequests.cpp | 28 +++++ .../ContentAddressed/Backend/CasRequests.h | 17 +++ .../ContentAddressed/Pool/CasMountRuntime.cpp | 51 +++++++-- .../ContentAddressed/Pool/CasMountRuntime.h | 14 +++ .../ContentAddressed/Pool/CasPool.h | 1 + .../ContentAddressed/Pool/CasServerRoot.cpp | 1 + .../ContentAddressed/Pool/CasServerRoot.h | 1 + src/Disks/tests/gtest_cas_heartbeat.cpp | 1 + .../gtest_cas_mount_renewal_progress.cpp | 107 ++++++++++++++++++ .../tests/gtest_cas_request_observer.cpp | 83 ++++++++++++++ .../StorageSystemContentAddressedMounts.cpp | 21 ++++ .../test_cas_mount_renewal_retry/test.py | 50 ++++++++ 14 files changed, 373 insertions(+), 13 deletions(-) create mode 100644 src/Disks/tests/gtest_cas_mount_renewal_progress.cpp create mode 100644 src/Disks/tests/gtest_cas_request_observer.cpp diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index d6e53da9ce99b..6ee68f6ac3ebe 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -278,6 +278,7 @@ becomes `state = 'corrupt'`, never an exception). Shows every `server_root_id` i | `state` | one of `live`, `expired`, `terminated`, `fenced`, `corrupt` | | `is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count` | GC health, process-local; **`NULL` on every peer row** — a process-local fact must never be stamped onto another server's row | | `lifecycle`, `lifecycle_reason`, `lifecycle_detail`, `lifecycle_since` | the SQL surface for the in-process `PoolLifecycle` runtime above: `lifecycle` is one of `live`, `not_live`, `identity_lost`, `vanished`, `constructing`, `shutdown`; `lifecycle_reason` distinguishes `replaced` from `forgotten` for a `vanished` disk; `lifecycle_detail` carries the full diagnosis text; `lifecycle_since` is when the current non-live state began (`NULL` while live) | +| `renewal_in_progress`, `renewal_attempts`, `last_renewal_error` | the running (or last) lease renewal: conditional `PUT`s sent so far and the last failed request, updated per attempt; **`NULL` on every peer row** | The lifecycle snapshot is I/O-free and ungated, so a not-live, never-started, or vanished disk still produces a row instead of silently disappearing from the table. diff --git a/docs/en/operations/system-tables/cas_mounts.md b/docs/en/operations/system-tables/cas_mounts.md index 5a9465db04af6..4a61e48f80f04 100644 --- a/docs/en/operations/system-tables/cas_mounts.md +++ b/docs/en/operations/system-tables/cas_mounts.md @@ -41,6 +41,9 @@ rows. - `lifecycle_reason` ([String](/sql-reference/data-types/string)) — The enum-clean sub-state word for a `vanished` disk: `replaced` or `forgotten`. Empty for every other lifecycle, so `lifecycle || '(' || lifecycle_reason || ')'` reads e.g. `vanished(forgotten)`. - `lifecycle_detail` ([String](/sql-reference/data-types/string)) — The full typed reason text naming the actual cause when not live: the vanish diagnosis (a data root replaced by a foreign pool, or decommissioned by `SYSTEM CAS FORGET` at a given time) or the identity-loss message. Empty when live. - `lifecycle_since` ([Nullable(DateTime)](/sql-reference/data-types/nullable)) — When this server entered the current non-live lifecycle state. `NULL` when live, or when the state has no backing pool to date from. +- `renewal_in_progress` ([Nullable(UInt8)](/sql-reference/data-types/nullable)) — `1` while this server renews its mount lease. +- `renewal_attempts` ([Nullable(UInt64)](/sql-reference/data-types/nullable)) — Conditional `PUT`s sent by the current renewal, or by the last one once it ended. Updated per attempt, so a renewal stuck in retries is visible before it ends. +- `last_renewal_error` ([Nullable(String)](/sql-reference/data-types/nullable)) — The last failed request of that renewal, as ` : `. `class` is `refused` when the store definitely refused the request (access denied, malformed request, entity too large), `transport` otherwise; `code` is the store's error name, such as `SlowDown`. `NULL` when no request failed. `lifecycle`/`lifecycle_reason`/`lifecycle_detail`/`lifecycle_since` are the SQL surface for diagnosing an identity-lost or forgotten disk without reading server logs — see the @@ -49,9 +52,10 @@ diagnosing an identity-lost or forgotten disk without reading server logs — se what each lifecycle value means, and [`SYSTEM CAS FORGET`](/antalya/cas/operations/debugging#sql-forget) for the command that produces `vanished(forgotten)`. -:::note Local-only GC-health columns -`is_leader`, `pending_reclaim`, `last_success_age_seconds`, and `wedged_namespace_count` are process-local -facts about *this* server's own GC scheduler. They are populated **only** on the row whose `server_root_id` matches +:::note Local-only columns +`is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count`, +`renewal_in_progress`, `renewal_attempts` and `last_renewal_error` are process-local facts about +*this* server. They are populated **only** on the row whose `server_root_id` matches this server's own mount, and are `NULL` on every row describing another server's mount — stamping a local health fact onto a peer's row would misread as "the peer is the GC leader" during an incident. To see the peer's own view of these columns, query `system.cas_mounts` on that server. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp index a1366b6d61cd6..6ea453eb1005e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp @@ -1,5 +1,6 @@ #include +#include #include #include #include @@ -278,6 +279,27 @@ bool isFirstAttemptFuseTimeout([[maybe_unused]] const std::exception & e, [[mayb #endif } +bool isStoreAnswer(const std::exception & e) +{ + const auto * db = dynamic_cast(&e); + return dynamic_cast(&e) && !(db && isDeterministicLocalFailure(db->code())); +} + +String describeRequestFailure(std::string_view verb, const String & key, const std::exception & e) +{ + const auto & poco = dynamic_cast(e); + String code; +#if USE_AWS_S3 + if (const auto * s3 = dynamic_cast(&e)) + code = s3->getExceptionName(); +#endif + if (const auto * db = dynamic_cast(&e); code.empty() && db) + code = String(ErrorCodes::getName(db->code())); + if (code.empty()) + code = poco.name(); + return fmt::format("{} {}: {} {} {}", verb, key, isDefinitelyRefusedWrite(e) ? "refused" : "transport", code, poco.message()); +} + CasRequests::CasRequests(BackendPtr backend_, Fence fence_, std::function now_ms_, std::function sleep_ms_, CasHotKeys * hot_keys_) @@ -924,6 +946,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co detail::recordAttempt(); ++state.attempts_sent; state.sent_any = true; + if (observer.on_write_attempt) + observer.on_write_attempt(state.attempts_sent); /// Disengaged means the attempt threw: its fate is unproven, and nothing may be read out of it. std::optional> outcome; @@ -952,6 +976,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co { if (isDeterministicLocalFailure(e.code())) throw; + if (observer.on_failure) + observer.on_failure("PUT", key, e); /// ONE refresh per call, only for the class a credential could explain, and only when a /// reissue could sign with what it installs. So an oversized entity never triggers a /// re-acquisition, a denial fresh credentials do not fix is refused on the second look @@ -1008,6 +1034,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co /// still have landed, so `outcome` stays disengaged and the read below settles it. if (!dynamic_cast(&e)) throw; + if (observer.on_failure) + observer.on_failure("PUT", key, e); } if (outcome && outcome->has_value()) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h index 3195d35bb8965..1c979c5d122b8 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h @@ -60,6 +60,10 @@ bool isFirstAttemptFuseTimeout(const std::exception & e, size_t attempt_no); /// same failure and buries the root cause behind a retryable exception. The set is `LOGICAL_ERROR`, /// `NOT_IMPLEMENTED`, `BAD_ARGUMENTS` and `CORRUPTED_DATA`. bool isDeterministicLocalFailure(int code); +/// An exception the store or the transport raised, as opposed to a local fault. +bool isStoreAnswer(const std::exception & e); +/// `VERB key: refused|transport code message`; `code` is the store's error name when it sent one. +String describeRequestFailure(std::string_view verb, const String & key, const std::exception & e); /// Throw the recoverable "CAS write could not be committed, retry later" condition. /// @@ -157,6 +161,15 @@ struct ListPage /// The walk's callback: FALSE stops the walk. using ListedKeyFn = std::function; +/// Live progress of one operation, called on its thread inside the engine's retry loops; must not throw. +struct RequestObserver +{ + /// A conditional write is about to send its `attempts_sent`-th physical attempt. + std::function on_write_attempt; + /// A conditional `PUT` or a body `GET` failed with a store answer (`isStoreAnswer`). + std::function on_failure; +}; + namespace detail { /// The engine's per-attempt counters, behind functions so the header need not declare the events. @@ -281,6 +294,7 @@ class CasOperation /// The verdict point: is this operation still admitted? For the sites that guard a decision rather /// than a request. bool admitted() const { return gate(0) == Gate::Ok; } + void observeRequests(RequestObserver observer_) { observer = std::move(observer_); } /// The write lane for keys several writers of this pool share. CasHotKeys & hotKeys() const { return *owner.hot_keys; } @@ -470,6 +484,7 @@ class CasOperation /// Written immediately before a read-class give-up throws, cleared and read only by the resolve /// read that swallows it. Every other caller lets the exception carry the verdict. std::optional last_read_stop; + RequestObserver observer; }; template @@ -501,6 +516,8 @@ auto CasOperation::readLoop(std::string_view verb, const String & subject, const } catch (const std::exception & e) { + if (observer.on_failure && verb == "read" && isStoreAnswer(e)) + observer.on_failure("GET", subject, e); bool refreshed = false; if (refreshAndClassifyReadFault(e, refresh_attempted, refreshed)) throw; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp index 95c6fb7a6550e..637eda1978e8c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -421,6 +422,12 @@ uint64_t CasMountRuntime::startRenewer() return anchor; } +RenewalProgress CasMountRuntime::renewalProgress() const +{ + std::lock_guard lock(renewal_progress_mutex); + return renewal_progress; +} + MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_call) { return MountRenewOperationEnvironment{ @@ -430,6 +437,29 @@ MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_c return config.renewal_live_for_test ? config.renewal_live_for_test() : renewalLive(worker_call); }, .cancelled = [this] { return renewalCancelled(); }, + .requests = RequestObserver{ + .on_write_attempt = [this](uint32_t attempts_sent) + { + ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalAttempts); + if (attempts_sent > 1) + ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalRetries); + std::lock_guard lock(renewal_progress_mutex); + renewal_progress.attempts_sent = attempts_sent; + }, + .on_failure = [this](std::string_view verb, const String & key, const std::exception & e) + { + /// Runs inside the engine's catch; a formatting failure must not change how the renewal ends. + try + { + String error = describeRequestFailure(verb, key, e); + std::lock_guard lock(renewal_progress_mutex); + renewal_progress.last_error = std::move(error); + } + catch (...) + { + } + }, + }, }; } @@ -465,13 +495,6 @@ void CasMountRuntime::consumeRenewResult( { /// Driver ownership has already been restored by `DriverLease::finish`; this is the single logical /// consumption boundary and it runs without `driver_mutex` or renewer access. - /// The physical counters come off the result rather than off a per-attempt callback, so they count - /// the same on every ending: a renewal that gave up still sent what it sent. - if (result.attempts_sent > 0) - { - ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalAttempts, result.attempts_sent); - ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalRetries, result.attempts_sent - 1); - } if (result.resolved_by_read) ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalResolved); @@ -535,9 +558,17 @@ uint64_t CasMountRuntime::renewRenewerOnce( /// so it renews on the renewer's open plane: admitted under the mount fence it could only ever give /// up, and every remount would fail at this step. `RemountCall` is reached from /// `renewRenewerForRemountOnce` alone. - const MountRenewResult result = active == RenewalDriverState::RemountCall - ? call.renewer->renewForRemount(renewalEnvironment(worker_call)) - : call.renewer->renew(renewalEnvironment(worker_call)); + { + std::lock_guard lock(renewal_progress_mutex); + renewal_progress = RenewalProgress{.in_progress = true, .attempts_sent = 0, .last_error = std::nullopt}; + } + MountRenewResult result; + { + SCOPE_EXIT({ std::lock_guard lock(renewal_progress_mutex); renewal_progress.in_progress = false; }); + result = active == RenewalDriverState::RemountCall + ? call.renewer->renewForRemount(renewalEnvironment(worker_call)) + : call.renewer->renew(renewalEnvironment(worker_call)); + } const RenewalDriverState destination = active == RenewalDriverState::WorkerCall ? RenewalDriverState::WorkerIdle : (active == RenewalDriverState::RemountCall ? RenewalDriverState::Parked : RenewalDriverState::Dormant); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h index 7f3ff1d76f36d..04ec8491f84b7 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -134,6 +135,14 @@ struct MountFence std::atomic lost{false}; }; +/// The current logical renewal of the mount lease so far, or the last one once it ended. +struct RenewalProgress +{ + bool in_progress = false; + uint32_t attempts_sent = 0; + std::optional last_error; +}; + /// Owns the live writer-incarnation mechanics shared by the pool's mount and recovery orchestration: /// the `MountLeaseRenewer`, local `MountFence`, build watermark and in-flight build registry, /// `live_writer_epoch`, unclean-boundary marker, and both persistent workers. `Pool` retains the higher-level @@ -172,6 +181,8 @@ class CasMountRuntime /// Renew the merged mount heartbeat once, including its build-watermark floor. A read-only runtime /// has no renewer and fails with a logical exception rather than fabricating a heartbeat. void renewWatermarkOnce(); + /// Takes only its own lock, so a renewal stuck in a request never holds up the reader. + RenewalProgress renewalProgress() const; /// ---- local write fence ---- /// Return whether a mutable operation may start under the locally observed lease state. @@ -508,6 +519,9 @@ class CasMountRuntime uint64_t remount_handled_generation = 0; ThreadFromGlobalPool renewal_worker; ThreadFromGlobalPool remount_worker; + /// A leaf lock, never held across I/O. + mutable std::mutex renewal_progress_mutex; + RenewalProgress renewal_progress TSA_GUARDED_BY(renewal_progress_mutex); /// Counted entries into `scheduleRemount`; retained as a test-only observability seam. std::atomic schedule_remount_calls_for_test{0}; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index 0f565d93f3de6..ad87acc020980 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -545,6 +545,7 @@ class Pool : public std::enable_shared_from_this time_t since = 0; }; LifecycleSnapshot lifecycleSnapshot() const; + RenewalProgress renewalProgress() const { return mount_runtime.renewalProgress(); } /// `SYSTEM CAS FORGET` — the operator force-Vanish (spec §5). Drives THIS pool to /// `Vanished(forgotten)` with the fence-first protocol, node-locally, regardless of the current diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index b88c02dd4af1c..74693e8e4d9b1 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -1629,6 +1629,7 @@ MountRenewResult MountLeaseRenewer::renewOn( result.attempt_start_boot_ms = attempt_start_boot_ms; CasOperation op = plane.admit(environment.live); + op.observeRequests(environment.requests); std::optional written; try { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index 5e3aecc4b9194..9cc34bf54efbc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -71,6 +71,7 @@ struct MountRenewOperationEnvironment /// `NotAttempted` rather than terminal only when this node had already been asked to stop -- /// sampling it afterwards would read a flag that the refusal itself may have set. std::function cancelled; + RequestObserver requests; }; /// Validate a `server_root_id` — the explicit, configured identity of the content-addressed layout diff --git a/src/Disks/tests/gtest_cas_heartbeat.cpp b/src/Disks/tests/gtest_cas_heartbeat.cpp index c39935d7ca7f7..db8ded0afd6d8 100644 --- a/src/Disks/tests/gtest_cas_heartbeat.cpp +++ b/src/Disks/tests/gtest_cas_heartbeat.cpp @@ -181,6 +181,7 @@ MountRenewOperationEnvironment renewalEnvironment( .boot_ms = [&boot_ms] { return boot_ms; }, .live = live, .cancelled = cancelled, + .requests = {}, }; } diff --git a/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp b/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp new file mode 100644 index 0000000000000..805f94a15bfa7 --- /dev/null +++ b/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp @@ -0,0 +1,107 @@ +#include + +#include +#include +#include +#include + +#include + +#include "config.h" + +#if USE_AWS_S3 +#include +#endif + +namespace ProfileEvents +{ +extern const Event CASMountRenewalAttempts; +} + +using namespace DB::Cas; + +namespace +{ + +uint64_t attemptsCounter() +{ + return ProfileEvents::global_counters[ProfileEvents::CASMountRenewalAttempts].load(); +} + +/// Fixed boot clock and no-op sleep: every armed fault is finite, so retries end without time passing. +struct Fixture +{ + std::shared_ptr backend = std::make_shared(); + PoolPtr store; + String mount_key; + + explicit Fixture(const String & prefix) + { + backend->setAttemptTimeoutMs(10); + store = Pool::open(backend, PoolConfig{ + .pool_prefix = prefix, + .server_root_id = "test", + .mount_lease_ttl_ms = std::chrono::milliseconds(1000), + .cas_request_budget = CasRequestBudget{.attempt_timeout_ms = 10, .lease_safety_margin_ms = 20, .connect_timeout_cap_ms = std::nullopt}, + .boot_ms_fn = [] { return uint64_t{100}; }, + .retry_sleep_fn = [](uint64_t) {}, + }); + mount_key = store->layout().mountKey("test"); + } +}; + +} + +TEST(CASMountRenewalProgress, AttemptsAndLastErrorAreVisibleWhileTheRenewalRuns) +{ + Fixture f("renewal-progress-in-flight"); + const auto timeout = std::make_exception_ptr(Poco::TimeoutException("injected renewal timeout")); + f.backend->failNextWriteWith(f.mount_key, timeout); + f.backend->failNextWriteWith(f.mount_key, timeout); + const uint64_t attempts_before = attemptsCounter(); + /// Read at the third attempt, the first that reaches the store. Heap state and a raw pointer: the + /// hook must not keep the pool alive, and it is removed before the pool's farewell write. + auto in_flight = std::make_shared>>(); + Pool * pool = f.store.get(); + f.backend->onBeforeWrite(f.mount_key, [in_flight, pool] + { + if (!*in_flight) + *in_flight = std::make_pair(pool->renewalProgress(), attemptsCounter()); + }); + + f.store->renewWatermarkOnce(); + f.backend->onBeforeWrite(f.mount_key, {}); + + ASSERT_TRUE(in_flight->has_value()); + const auto & [seen, attempts_counted] = **in_flight; + EXPECT_TRUE(seen.in_progress); + EXPECT_EQ(seen.attempts_sent, 3u); + EXPECT_EQ(seen.last_error, "PUT " + f.mount_key + ": transport Timeout injected renewal timeout"); + EXPECT_EQ(attempts_counted - attempts_before, 3u) << "attempts are counted as they are sent"; + + const RenewalProgress ended = f.store->renewalProgress(); + EXPECT_FALSE(ended.in_progress); + EXPECT_EQ(ended.attempts_sent, 3u); + EXPECT_EQ(ended.last_error, seen.last_error) << "an ended renewal keeps its progress"; + + f.store->renewWatermarkOnce(); + const RenewalProgress next = f.store->renewalProgress(); + EXPECT_EQ(next.attempts_sent, 1u); + EXPECT_FALSE(next.last_error.has_value()) << "the next renewal starts clean"; +} + +#if USE_AWS_S3 +TEST(CASMountRenewalProgress, AStoreRefusalIsClassedRefused) +{ + Fixture f("renewal-progress-refused"); + f.backend->failNextWriteWith(f.mount_key, std::make_exception_ptr( + DB::S3Exception("the store refused the renewal", Aws::S3::S3Errors::ACCESS_DENIED, "AccessDenied"))); + + EXPECT_THROW(f.store->renewWatermarkOnce(), DB::Exception); + + const RenewalProgress ended = f.store->renewalProgress(); + EXPECT_FALSE(ended.in_progress) << "a renewal that threw has still ended"; + EXPECT_EQ(ended.attempts_sent, 1u); + EXPECT_EQ(ended.last_error, "PUT " + f.mount_key + ": refused AccessDenied the store refused the renewal"); +} +#endif diff --git a/src/Disks/tests/gtest_cas_request_observer.cpp b/src/Disks/tests/gtest_cas_request_observer.cpp new file mode 100644 index 0000000000000..3aa6565803278 --- /dev/null +++ b/src/Disks/tests/gtest_cas_request_observer.cpp @@ -0,0 +1,83 @@ +#include + +#include +#include +#include +#include + +#include + +#include + +#include +#include + +namespace DB::ErrorCodes +{ +extern const int BAD_ARGUMENTS; +extern const int NETWORK_ERROR; +} + +using namespace DB::Cas; + +namespace +{ + +/// One seeded key and an operation whose observer records every call as a line. +struct Fixture +{ + std::shared_ptr backend = std::make_shared(); + CasRequests requests{backend, Fence::open(), [] { return uint64_t{100}; }, [](uint64_t) {}}; + Etag etag = std::get(requests.admit().create("k", "v1", Retry::standard())).etag; + std::vector events; + CasOperation op = requests.admit(); + + Fixture() + { + op.observeRequests(RequestObserver{ + .on_write_attempt = [this](uint32_t attempts_sent) { events.push_back(fmt::format("attempt {}", attempts_sent)); }, + .on_failure = [this](std::string_view verb, const String & key, const std::exception & e) + { + events.push_back(describeRequestFailure(verb, key, e)); + }, + }); + } +}; + +} + +TEST(CASRequestObserver, WriteAttemptsAndFailuresAreReportedAsTheyHappen) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(Poco::TimeoutException("injected write timeout"))); + f.backend->failNextReadWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::NETWORK_ERROR, "injected read failure"))); + + ASSERT_TRUE(std::holds_alternative(f.op.replace("k", "v2", f.etag, Retry::standard()))); + EXPECT_EQ(f.events, (std::vector{ + "attempt 1", + "PUT k: transport Timeout injected write timeout", + "GET k: transport NETWORK_ERROR injected read failure", + "attempt 2", + })); +} + +/// A local bug is not an answer from the store, so it is not published as the request's error. +TEST(CASRequestObserver, ALocalFailureIsNotReportedAsAStoreAnswer) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "injected local failure"))); + + EXPECT_THROW((void)f.op.replace("k", "v2", f.etag, Retry::standard()), DB::Exception); + EXPECT_EQ(f.events, (std::vector{"attempt 1"})); +} + +/// A local failure of the resolve read is not published either, only the store answer before it. +TEST(CASRequestObserver, ALocalFailureOfTheResolveReadIsNotReported) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(Poco::TimeoutException("injected write timeout"))); + f.backend->failNextReadWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "injected local failure"))); + + EXPECT_THROW((void)f.op.replace("k", "v2", f.etag, Retry::standard()), DB::Exception); + EXPECT_EQ(f.events, (std::vector{"attempt 1", "PUT k: transport Timeout injected write timeout"})); +} diff --git a/src/Storages/System/StorageSystemContentAddressedMounts.cpp b/src/Storages/System/StorageSystemContentAddressedMounts.cpp index ba8b8670cc105..79285618518c0 100644 --- a/src/Storages/System/StorageSystemContentAddressedMounts.cpp +++ b/src/Storages/System/StorageSystemContentAddressedMounts.cpp @@ -57,6 +57,9 @@ StorageSystemContentAddressedMounts::StorageSystemContentAddressedMounts(const S {"lifecycle_reason", std::make_shared(), "The enum-clean sub-state word for a vanished disk: replaced or forgotten. Empty for every other lifecycle (so lifecycle || '(' || lifecycle_reason || ')' reads e.g. vanished(forgotten))."}, {"lifecycle_detail", std::make_shared(), "The full typed reason text naming the actual cause when not live: the vanish diagnosis (data root replaced by a foreign pool / decommissioned by SYSTEM CAS FORGET at