From bc7f453aa4a7e2a11deff1004813abbf8437291d Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 8 Oct 2026 10:27:30 +0200 Subject: [PATCH 1/3] Date a GC mount-token sighting after the read that returned it The GC credited the first sighting of a mount token with a clock sample taken before the LIST of gc/server-roots/ and the slot read. A slow walk then counted as time the token was watched, and a later round could write gc_fenced over a holder whose last renewal still gave it authority. Both callers of computeHeartbeatFloor (the regular round and the rebuild) now pass the observer clock as a function; the sighting is sampled after the read, including the re-decision after a refused fence-out. The stability test keeps its entry sample, and a sample older than the sighting never counts as time held, so the elapsed time cannot wrap. Closes backlog task CAS-342. Related: https://github.com/Altinity/ClickHouse/issues/2431 Co-Authored-By: Claude Fable 5.1 --- .../ContentAddressed/Gc/CasGc.cpp | 4 +- .../ContentAddressed/Gc/CasGc.h | 2 +- .../ContentAddressed/Pool/CasServerRoot.cpp | 16 +- .../ContentAddressed/Pool/CasServerRoot.h | 22 +- src/Disks/tests/gtest_cas_gc_ack_floor.cpp | 106 +++++++++ src/Disks/tests/gtest_cas_mount.cpp | 219 ++++++++++++++++-- 6 files changed, 340 insertions(+), 29 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index d610228e5b7e7..16534edebb2b8 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -712,7 +712,7 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al /// PUT per newly-fenced mount. { GcPhaseTimer t(phase_sink, "heartbeat_floor"); - const HeartbeatFloor floor = computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn(), + const HeartbeatFloor floor = computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn, stable_threshold_ms, mount_obs); report.fence_outs = floor.fenced_now; if (floor.fenced_now > 0) @@ -4763,7 +4763,7 @@ RebuildReport Gc::rebuildBaseline(bool force) /// `mountObservationThresholdMs` -- see its doc comment (CasServerRoot.h). const uint64_t stable_threshold_ms = mountObservationThresholdMs( ttl_ms, static_cast(store->poolConfig().mount_renew_period.count())); - computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn(), stable_threshold_ms, mount_obs); + computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn, stable_threshold_ms, mount_obs); /// Retired-in-snapshot: the rebuilt seal's `condemned_summary` must be TOTAL over gc_shards so a /// subsequent regular round reads graduation/carry decisions zero-I/O off it (and its `carryParentRefs` diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index 48ddf5ef07862..fa6215beb1292 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -427,7 +427,7 @@ class Gc /// OBSERVATION clock: monotonic on this /// process, injected for tests, defaults to `Pool::bootMs()`, and the ONLY clock the heartbeat /// gate's own fence-out threshold is measured against (mirrors `claimMountAwaitingExpiry`'s - /// `mono_ms_fn`, but at heartbeat-gate granularity — one GC round is one observation tick). + /// `mono_ms_fn`). /// Everything else in the round stays deterministic/clock-free. /// /// `log_` is the logger every round-engine log line is emitted through; pass a disk/srid-scoped diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index 10de8e758b3c0..d50a9fb3f85b6 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -1017,11 +1017,18 @@ MountClaimResult claimMountAwaitingExpiry( } } +bool MountIncarnationObservation::heldFor(uint64_t threshold_ms, uint64_t sample_ms) const +{ + return sample_ms >= first_seen_mono_ms && sample_ms - first_seen_mono_ms >= threshold_ms; +} + HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64_t now_ms, - uint64_t mono_now_ms, uint64_t stable_threshold_ms, - MountObservationMap & obs) + const std::function & mono_ms_fn, + uint64_t stable_threshold_ms, MountObservationMap & obs) { HeartbeatFloor floor; + /// Taken before any read: it can confirm that a token held, never date a sighting. + const uint64_t round_start_ms = mono_ms_fn(); /// `obs` is keyed by every srid this leader has EVER observed, but a /// srid removed from the LIST entirely (its `/mount` key gone -- e.g. `SYSTEM CAS @@ -1083,12 +1090,13 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 /// counts as `live` this call. const auto it = obs.find(srid); const bool stable = it != obs.end() && it->second.etag == observed->etag - && mono_now_ms - it->second.first_seen_mono_ms >= stable_threshold_ms; + && it->second.heldFor(stable_threshold_ms, round_start_ms); if (!stable) { + /// A sample from before this read would count the walk to the slot as time watched. if (it == obs.end() || it->second.etag != observed->etag) - obs.insert_or_assign(srid, MountIncarnationObservation{observed->etag, mono_now_ms}); + obs.insert_or_assign(srid, MountIncarnationObservation{observed->etag, mono_ms_fn()}); ++floor.live; return std::nullopt; } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index 1c1fd8928f971..f583ca2af83ca 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -346,13 +346,16 @@ MountClaimResult claimMountAwaitingExpiry( /// One `server_root_id`'s cross-round incarnation-stability observation, /// owned by the GC leader instance (`Cas::Gc::mount_obs`) and threaded through consecutive -/// `computeHeartbeatFloor` calls — one GC round is one observation tick. Mirrors -/// `claimMountAwaitingExpiry`'s observation loop, but at heartbeat-gate granularity rather than a -/// tight poll loop. +/// `computeHeartbeatFloor` calls. `first_seen_mono_ms` is the reading taken when the slot read +/// that first returned `etag` came back. struct MountIncarnationObservation { Etag etag; uint64_t first_seen_mono_ms = 0; + + /// `sample_ms` is taken before the read that confirmed `etag`. A sample older than the sighting + /// says nothing about how long the token held, so it never counts as held. + bool heldFor(uint64_t threshold_ms, uint64_t sample_ms) const; }; /// Keyed by `server_root_id`. In-memory only: a fresh leader (after a steal, or a process restart) @@ -371,8 +374,8 @@ using MountObservationMap = std::map; /// terminated marker; /// - otherwise, observation-based liveness (the same /// principle `claimMountAwaitingExpiry` uses for a mount's OWN reopen, applied here to the GC's -/// fence-out): `obs` remembers, per srid, the incarnation last seen and the leader's OWN -/// monotonic clock reading (`mono_now_ms`) at the moment it first saw it. A body whose +/// fence-out): `obs` remembers, per srid, the incarnation last seen and the reading of the +/// leader's OWN monotonic clock (`mono_ms_fn`) when it first saw it. A body whose /// CURRENT incarnation differs from (or is absent from) `obs` is (re)started fresh — counted `live`, /// never fenced this call, regardless of what its stamped `expires_at_ms` claims (a bare /// wall-clock stamp is never trusted — see `claimMount`'s "certificate of death" doc). Only once @@ -386,8 +389,9 @@ using MountObservationMap = std::map; /// /// `now_ms` is WALL clock, used only for the audit/diagnostic log line — it never participates in the /// fence decision (mirrors `claimMountAwaitingExpiry`'s `now_ms_fn` vs `mono_ms_fn` split). -/// `mono_now_ms` is the OBSERVATION clock: the caller's OWN monotonic reading, never compared against -/// any other node's clock. `obs` is owned by the caller and threaded across consecutive calls (one GC +/// `mono_ms_fn` is the OBSERVATION clock: the caller's OWN monotonic clock, never compared against +/// any other node's clock. It is sampled once at entry for the stability test and after the read for +/// each sighting. `obs` is owned by the caller and threaded across consecutive calls (one GC /// leader instance, `Cas::Gc::mount_obs`) — a fresh leader starts with an empty map (safe: delays /// fencing one round, never fences early). /// @@ -407,8 +411,8 @@ struct HeartbeatFloor }; HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64_t now_ms, - uint64_t mono_now_ms, uint64_t stable_threshold_ms, - MountObservationMap & obs); + const std::function & mono_ms_fn, + uint64_t stable_threshold_ms, MountObservationMap & obs); /// One `gc/server-roots//mount` slot whose holder is not provably finished with the prefix. struct NonTerminalMountSlot diff --git a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp index ba4259709f740..8bd8f8b6ca344 100644 --- a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp +++ b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp @@ -1017,6 +1017,102 @@ void runExpiredMountFenceOutScenario(const PoolConfig & config) EXPECT_TRUE(runRoundsUntilAbsent(store, gc, *backend, layout, blob)); } +/// Advances the GC's observer clock by `read_cost_ms` on every read of `watched_key`, so a GC call +/// spends that long between its entry sample and the read it decides on. Atomics because the round's +/// read-ahead pool reads other keys through this backend on other threads. +class SlotReadClockBackend : public InMemoryBackend +{ +public: + String watched_key; + std::atomic mono{0}; + std::atomic read_cost_ms{0}; + std::atomic last_read_done_at{0}; + std::atomic watched_reads{0}; + + std::optional read(const String & key, TransportAccess & access) override + { + if (key == watched_key) + { + last_read_done_at = mono += read_cost_ms.load(); + ++watched_reads; + } + return InMemoryBackend::read(key, access); + } +}; + +enum class SightingCall : uint8_t +{ + Round, + Rebuild, +}; + +void runSlowWalkSightingScenario(SightingCall sighting_call) +{ + auto backend = std::make_shared(); + auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const Layout & layout = store->layout(); + const String srid2 = "stale-server"; + backend->watched_key = layout.mountKey(srid2); + + const uint64_t ttl_ms = static_cast(store->poolConfig().mount_lease_ttl_ms.count()); + const uint64_t threshold_ms = mountObservationThresholdMs( + ttl_ms, static_cast(store->poolConfig().mount_renew_period.count())); + /// Longer than `threshold - TTL`: a sighting credited from before this walk fences a holder + /// whose last renewal still gives it authority. + const uint64_t walk_ms = threshold_ms - ttl_ms + 1'000; + + Gc gc(store, kGc, [] { return uint64_t{1'000'000}; }, [backend] { return backend->mono.load(); }); + + const RootNamespace ns{"00/aa@cas@"}; + const ManifestRef r = ref("srv-a:1", 1, 0xAA); + writeBlobBody(*backend, layout, DB::UInt128(7)); + writeManifestRaw(*backend, layout, ns, r, {blobEntryFor("a", DB::UInt128(7))}); + publishCommittedTransition(*backend, layout, ns, "tbl", std::nullopt, r); + + /// The rebuild needs a state to rebuild; one round before the slot exists leaves the rebuild as + /// the call that first sees it. + if (sighting_call == SightingCall::Rebuild) + gc.runRegularRound(); + + /// Claims once and never renews: a live-shaped body whose token never changes. + CasRequests renewer_requests = openRequestsForTest(backend); + MountLeaseRenewer srid2_renewer(renewer_requests, renewer_requests, layout, srid2, DB::UInt128(0x2222), + /*writer_epoch=*/1, + std::chrono::milliseconds(100), [] { return 1000u; }, [] { return 0u; }, {}, + std::chrono::milliseconds(0), [] { return 0u; }); + srid2_renewer.start(); + backend->watched_reads = 0; + + backend->read_cost_ms = walk_ms; + if (sighting_call == SightingCall::Round) + { + const RoundReport first = gc.runRegularRound(); + ASSERT_EQ(first.fence_outs, 0u); + } + else + { + const RebuildReport rebuilt = gc.rebuildBaseline(/*force*/ true); + ASSERT_TRUE(rebuilt.performed) << rebuilt.refusal; + } + ASSERT_EQ(backend->watched_reads.load(), 1u); + const uint64_t sighted_at = backend->last_read_done_at.load(); + ASSERT_EQ(sighted_at, walk_ms); + backend->read_cost_ms = 0; + + /// One threshold after the sighting call started, less than one after its read. The store's own + /// mount renews before each round so only srid2 can be fenced. + store->renewWatermarkOnce(); + backend->mono = threshold_ms; + const RoundReport early = gc.runRegularRound(); + EXPECT_EQ(early.fence_outs, 0u) << "fenced a token watched for less than the threshold"; + + store->renewWatermarkOnce(); + backend->mono = sighted_at + threshold_ms; + const RoundReport due = gc.runRegularRound(); + EXPECT_EQ(due.fence_outs, 1u); + EXPECT_TRUE(decodeMountLease(readObj(*backend, backend->watched_key)->bytes).gc_fenced); +} + } TEST(CASGCAckFloor, ExpiredMountFencedOutAndExcluded) @@ -1036,6 +1132,16 @@ TEST(CASGcFenceOut, ThresholdUnchangedByUnsafeKnob) PoolConfig{.pool_prefix = "p", .server_root_id = "test", .unsafe_remount_no_delay = true}); } +TEST(CASGcFenceOut, RoundCountsASightingFromAfterItsRead) +{ + runSlowWalkSightingScenario(SightingCall::Round); +} + +TEST(CASGcFenceOut, RebuildCountsASightingFromAfterItsRead) +{ + runSlowWalkSightingScenario(SightingCall::Rebuild); +} + /// fix-round F6 (author-review: `Gc`'s own `mono_ms_fn` used to default to the RAW static `Pool:: /// bootMs()`, bypassing the Pool's own injectable `config.boot_ms_fn` -- a time-controlled test can /// desync the mount side's fake clock from the GC side's real one). This mirrors diff --git a/src/Disks/tests/gtest_cas_mount.cpp b/src/Disks/tests/gtest_cas_mount.cpp index 783bd94b1dae1..ae0afa6f1a048 100644 --- a/src/Disks/tests/gtest_cas_mount.cpp +++ b/src/Disks/tests/gtest_cas_mount.cpp @@ -1502,7 +1502,7 @@ namespace /// Rev.6 §token-stability observation removed the wall clock from the fence DECISION; `kNowMs` below /// is threaded through only as `computeHeartbeatFloor`'s audit-only `now_ms`. constexpr uint64_t kNowMs = 1'000'000; -/// The fence-out threshold measured on the LEADER's OWN monotonic clock (`mono_now_ms`), independent +/// The fence-out threshold measured on the LEADER's OWN monotonic clock (`mono_ms_fn`), independent /// of any lease's stamped `expires_at_ms`. constexpr uint64_t kStableThresholdMs = 10'000; @@ -1540,6 +1540,13 @@ void renewMount(CasOperation & op, const Layout & l, const String & srid) mustCommit(op.replace(l.mountKey(srid), encodeMountLease(m), got->etag, Retry::standard()), "renewed mount " + srid); } + +/// An observer clock that does not move during a call: the entry sample and every sighting read the +/// same value. +std::function frozenClock(uint64_t ms) +{ + return [ms] { return ms; }; +} } TEST(CASHeartbeatFloor, FirstSightNeverFencesEvenIfStampLooksExpired) @@ -1553,7 +1560,7 @@ TEST(CASHeartbeatFloor, FirstSightNeverFencesEvenIfStampLooksExpired) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, /*now_ms*/ kNowMs, /*mono_now_ms*/ 0, + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, /*now_ms*/ kNowMs, frozenClock(0), kStableThresholdMs, obs); EXPECT_EQ(floor.fenced_now, 0u); @@ -1570,14 +1577,14 @@ TEST(CASHeartbeatFloor, StableIncarnationPastThresholdIsFenced) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); EXPECT_EQ(floor_before.fenced_now, 0u); const MountLease before = decodeMountLease(ops.op.read(l.mountKey("s1"), Retry::standard())->bytes); /// No renewal in between: the SAME incarnation, observed since mono 0, is now stable for the full /// threshold on the leader's own clock. - const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ kStableThresholdMs, + const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); EXPECT_EQ(floor2.fenced_now, 1u); @@ -1595,7 +1602,7 @@ TEST(CASHeartbeatFloor, RenewalBetweenRoundsRestartsObservation) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); ASSERT_TRUE(obs.contains("s1")); const Etag first_etag = obs.at("s1").etag; @@ -1603,7 +1610,7 @@ TEST(CASHeartbeatFloor, RenewalBetweenRoundsRestartsObservation) const Etag renewed_etag = currentEtag(ops.op, l.mountKey("s1")); EXPECT_NE(renewed_etag, first_etag); - const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ kStableThresholdMs, + const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); EXPECT_EQ(floor2.fenced_now, 0u); @@ -1626,7 +1633,7 @@ TEST(CASHeartbeatFloor, UnseenSridPrunedFromObservationMap) seedMount(ops.op, l, "s2", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); ASSERT_TRUE(obs.contains("s1")); ASSERT_TRUE(obs.contains("s2")); @@ -1638,7 +1645,7 @@ TEST(CASHeartbeatFloor, UnseenSridPrunedFromObservationMap) renewMount(ops.op, l, "s1"); ASSERT_EQ(ops.op.removeCurrent(l.mountKey("s2"), Retry::standard()), Removal::Removed); - computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ kStableThresholdMs, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); EXPECT_TRUE(obs.contains("s1")); EXPECT_FALSE(obs.contains("s2")) << "a srid removed from the LIST entirely must be pruned from obs, not linger forever"; @@ -1665,7 +1672,7 @@ TEST(CASHeartbeatFloor, ClassifiesAndFencesOut) MountObservationMap obs; /// Round 1 (mono 0): first sight of every non-terminal mount — nothing is fence-eligible yet. - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); EXPECT_EQ(floor_before.live, 3u); // s1, s2, s3: observation just started EXPECT_EQ(floor_before.terminated, 1u); // s5 EXPECT_EQ(floor_before.fenced_now, 0u); @@ -1682,7 +1689,7 @@ TEST(CASHeartbeatFloor, ClassifiesAndFencesOut) /// Round 2 (mono == threshold): s1/s2's renewed incarnations restart their observation (still /// live); s3's original incarnation has now held stable for the full threshold -> fenced. - const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ kStableThresholdMs, + const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); EXPECT_EQ(floor2.live, 2u); // s1, s2: renewed, observation restarted @@ -1760,14 +1767,14 @@ TEST(CASHeartbeatFloor, FenceOutLosesTheIncarnationRaceAndReclassifiesLive) MountObservationMap obs; /// Round 1: first sight, observation starts — never reaches the fence-out path (the race /// decorator stays armed for round 2). - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); EXPECT_EQ(floor_before.fenced_now, 0u); /// Round 2: the incarnation has been stable past threshold, so the function attempts the /// fence-out. The decorator renews concurrently under the real incarnation, the write is refused, /// and the re-decision reclassifies the slot as live (observation restarted on the new /// incarnation) — never fenced. - const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ kStableThresholdMs, + const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); EXPECT_EQ(floor2.fenced_now, 0u); @@ -1778,6 +1785,192 @@ TEST(CASHeartbeatFloor, FenceOutLosesTheIncarnationRaceAndReclassifiesLive) EXPECT_FALSE(decodeMountLease(after->bytes).gc_fenced); } +namespace +{ +/// Moves the observer clock by `read_cost_ms` on every read of a mount slot, so a call spends that +/// long between its entry sample and the read it decides on. `renew_before_next_fence` makes the +/// holder renew just before the next guarded write of a slot lands, so the fence-out is refused and +/// decided again on the new token. `fail_slot_reads` makes every slot read time out. +class SlotClockBackend : public InMemoryBackend +{ +public: + explicit SlotClockBackend(uint64_t & mono_) : mono(mono_) {} + + uint64_t read_cost_ms = 0; + bool renew_before_next_fence = false; + bool fail_slot_reads = false; + + std::optional read(const String & key, TransportAccess & access) override + { + if (key.ends_with("/mount")) + { + if (fail_slot_reads) + throw Poco::TimeoutException("injected: mount slot read timed out"); + mono += read_cost_ms; + } + return InMemoryBackend::read(key, access); + } + + std::expected write(const String & key, const String & bytes, + const std::optional & expected_value, + TransportAccess & access) override + { + if (renew_before_next_fence && expected_value && key.ends_with("/mount")) + { + renew_before_next_fence = false; + const auto got = InMemoryBackend::read(key, access); + MountLease m = decodeMountLease(got->bytes); + m.seq += 1; + EXPECT_TRUE(InMemoryBackend::write(key, encodeMountLease(m), got->value, access).has_value()); + } + return InMemoryBackend::write(key, bytes, expected_value, access); + } + +private: + uint64_t & mono; +}; +} + +TEST(CASHeartbeatFloor, ConfirmingCallDecidesOnItsEntrySample) +{ + uint64_t mono = 0; + auto b = std::make_shared(mono); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); + const auto clock = [&mono] { return mono; }; + MountObservationMap obs; + + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + ASSERT_EQ(obs.at("s1").first_seen_mono_ms, 0u); + + /// The call starts one millisecond short of the threshold; its walk passes the threshold many times. + mono = kStableThresholdMs - 1; + b->read_cost_ms = 10 * kStableThresholdMs; + const HeartbeatFloor slow = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + EXPECT_EQ(slow.fenced_now, 0u); + EXPECT_EQ(slow.live, 1u); + EXPECT_EQ(obs.at("s1").first_seen_mono_ms, 0u) << "an unchanged token keeps its first sighting"; + + b->read_cost_ms = 0; + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); +} + +TEST(CASHeartbeatFloor, FailedSlotReadKeepsTheWatch) +{ + uint64_t mono = 0; + auto b = std::make_shared(mono); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); + const auto clock = [&mono] { return mono; }; + MountObservationMap obs; + + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + ASSERT_EQ(obs.at("s1").first_seen_mono_ms, 0u); + + mono = kStableThresholdMs / 2; + b->fail_slot_reads = true; + EXPECT_ANY_THROW(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs)); + ASSERT_TRUE(obs.contains("s1")); + EXPECT_EQ(obs.at("s1").first_seen_mono_ms, 0u); + + b->fail_slot_reads = false; + mono = kStableThresholdMs; + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u) + << "a failed read restarted the observation"; +} + +TEST(CASHeartbeatFloor, SampleOlderThanTheSightingNeverFences) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); + MountObservationMap obs; + + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); + ASSERT_EQ(obs.at("s1").first_seen_mono_ms, kStableThresholdMs); + + /// A clock that went back: the unsigned difference would wrap past any threshold. + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs - 1), + kStableThresholdMs, obs); + EXPECT_EQ(floor.fenced_now, 0u); + EXPECT_EQ(floor.live, 1u); +} + +namespace +{ +constexpr uint64_t kWalkMs = 4'000; +} + +TEST(CASHeartbeatFloor, FirstSightingCountsFromAfterItsRead) +{ + uint64_t mono = 0; + auto b = std::make_shared(mono); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); + const auto clock = [&mono] { return mono; }; + MountObservationMap obs; + + /// The call starts at 0 and reaches the slot at `kWalkMs`. + b->read_cost_ms = kWalkMs; + ASSERT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 0u); + ASSERT_TRUE(obs.contains("s1")); + EXPECT_EQ(obs.at("s1").first_seen_mono_ms, kWalkMs); + + /// One threshold after the first call started, less than one after its read. + b->read_cost_ms = 0; + mono = kStableThresholdMs; + const HeartbeatFloor early = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + EXPECT_EQ(early.fenced_now, 0u) << "fenced a token watched for less than the threshold"; + EXPECT_EQ(early.live, 1u); + + mono = kWalkMs + kStableThresholdMs; + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); +} + +TEST(CASHeartbeatFloor, RedecisionAfterARefusedFenceCountsFromItsRead) +{ + uint64_t mono = 0; + auto b = std::make_shared(mono); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); + const auto clock = [&mono] { return mono; }; + MountObservationMap obs; + + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + ASSERT_TRUE(obs.contains("s1")); + const Etag first = obs.at("s1").etag; + + /// Stable at this call's start, so it tries the fence-out. The holder renews first, the write is + /// refused, and the call decides again on the new token after reading it. + mono = kStableThresholdMs; + b->read_cost_ms = kWalkMs; + b->renew_before_next_fence = true; + const HeartbeatFloor refused = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + ASSERT_FALSE(b->renew_before_next_fence) << "the call never attempted the fence-out"; + ASSERT_EQ(refused.fenced_now, 0u); + ASSERT_EQ(refused.live, 1u); + ASSERT_NE(obs.at("s1").etag, first); + + /// Nothing reads after the re-decision, so the clock holds the value of the last read. + const uint64_t reread_at = mono; + ASSERT_GT(reread_at, kStableThresholdMs); + EXPECT_EQ(obs.at("s1").first_seen_mono_ms, reread_at); + + /// One threshold after that call started, less than one after its re-read. + b->read_cost_ms = 0; + mono = 2 * kStableThresholdMs; + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 0u) + << "fenced the renewed token before it was watched for the threshold"; + + mono = reread_at + kStableThresholdMs; + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); +} + TEST(CASHeartbeatFloor, EmptyPrefixYieldsNoLiveMounts) { auto b = std::make_shared(); @@ -1785,7 +1978,7 @@ TEST(CASHeartbeatFloor, EmptyPrefixYieldsNoLiveMounts) Ops ops(b); MountObservationMap obs; - const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, /*mono*/ 0, kStableThresholdMs, obs); + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); EXPECT_EQ(floor.live, 0u); EXPECT_EQ(floor.terminated, 0u); From 732155f15e7853b46fc32c5afb6886ddba5b53c8 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 8 Oct 2026 17:05:54 +0200 Subject: [PATCH 2/3] Wait for the lease a CAS mount holder declares before fencing or reclaiming its slot Every observer waited its own cas_mount_lease_ttl_ms before treating an unchanged slot token as a dead holder: the GC floor, the same-server restart and remount reclaim. A holder configured with a longer lease was fenced inside its own right to write when any observer ran with a shorter value, and nothing in the protocol could tell. Observers now wait max(own lease, declared lease) + 5 % + their own poll step, where the declared lease is expires_at_ms - started_at_ms of the body they observe (0 when the stamps are missing, equal, or reversed, so the wait is never shorter than before). The formula stays in one place, mountObservationThresholdMs, and saturates instead of wrapping. The setting is per server: how long this server may write after its last confirmed renewal; every other server waits at least that long. Mixed values are safe in any order. The GC leader logs one WARNING per slot whose declared lease differs from its own; fenced and departed slots are not warned about. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Fable 5.1 --- .../cas/architecture/garbage-collection.md | 4 +- .../cas/architecture/mounts-and-leases.md | 31 +- docs/en/antalya/cas/configuration.md | 22 +- .../ContentAddressed/Gc/CasGc.cpp | 24 +- .../ContentAddressed/Gc/CasGc.h | 3 + .../ContentAddressed/Pool/CasServerRoot.cpp | 70 +++- .../ContentAddressed/Pool/CasServerRoot.h | 34 +- src/Disks/tests/cas_test_helpers.h | 48 +++ src/Disks/tests/gtest_cas_gc_ack_floor.cpp | 105 ++++- src/Disks/tests/gtest_cas_mount.cpp | 380 +++++++++++++++--- src/Disks/tests/gtest_cas_pool.cpp | 8 +- 11 files changed, 595 insertions(+), 134 deletions(-) diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index 46b34600296d5..e232103a56778 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -193,8 +193,8 @@ token of each `mount` object. - **Writes / deletes:** a token-guarded `PUT` per fenced mount (`gc_fenced = true`, `seq + 1`); no deletes - **Safety:** fences only after this leader's *own* monotonic clock has watched the mount's - write-token hold unchanged for `cas_mount_lease_ttl_ms + 5% + cas_mount_renew_period_ms` - (defaults 30 s and 10 s; every server sharing the pool must run the same values); the write is + write-token hold unchanged for `L + 5% + cas_mount_renew_period_ms`, where `L` is the longer of this server's + `cas_mount_lease_ttl_ms` and the lease the slot's body declares (defaults 30 s and 10 s); the write is guarded by that exact token. A completed fence-out stays valid even if this GC later loses leadership. `expires_at_ms` (another host's wall clock) is never trusted. - **Fails the round if:** nothing — a per-mount `PUT` conflict is re-observed and re-classified diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index eb852b070da0d..581ea65aafcd4 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -119,22 +119,23 @@ inside authority already proved by the last confirmed lease. GC's own view of a dead server is symmetric and clock-skew-immune: a slot becomes fence-eligible only after the leader observes the *same* renewal token hold stable, on its own monotonic clock, -for `TTL + floor(TTL/20) + period` — close to, but not identical to, the threshold a re-mounting -server uses to wait out a predecessor, which observes `TTL + floor(TTL/20) + max(1, -floor(period/2))`. Both thresholds are evaluated purely on the observer's own clock and its own -configured `TTL`/`period`; nothing about the writer's timing travels on the wire. The stamped -`expires_at_ms` never participates in either decision — it is a writer-stamped diagnostic used by -`system.cas_mounts` and by the non-authoritative decommission epoch-recovery precheck, never an -authorization; local fencing is derived instead from the confirmed request's pre-I/O `BOOTTIME` +for `L + floor(L/20) + period`. `L` is the longer of the leader's own `cas_mount_lease_ttl_ms` and the +lease the slot's body declares (`expires_at_ms − started_at_ms`). A body without stamps, or with a +non-positive interval (a farewell), declares nothing. A re-mounting server waits out a predecessor the +same way with the step `max(1, floor(period/2))`. `period` is always the observer's own. The two stamps +are compared only with each other, never with the observer's clock. `expires_at_ms` alone is a +writer-stamped diagnostic used by `system.cas_mounts` and the non-authoritative decommission +epoch-recovery precheck. Local fencing is derived from the confirmed request's pre-I/O `BOOTTIME` anchor plus the TTL, and wall-clock `now` stays audit-only. -Every server sharing a pool must therefore run the identical `cas_mount_lease_ttl_ms` and -`cas_mount_renew_period_ms`: a member or GC leader configured with a shorter threshold than its -peers can fence out a healthy peer whose token-update gap merely exceeds that shorter threshold — -a peer renewing frequently stays live, one that missed a renewal does not. Change these values only -with every member of the pool stopped; a graceful restart removes only that member's own startup -observation and does not make mixed thresholds safe. With the defaults (TTL 30 s, period 10 s, -margin 2 s), `TTL − margin − period − 2 × envelope = 4 s` is the scheduling-lateness budget before +`cas_mount_lease_ttl_ms` is local to each server. It is how long this server may write after its last +confirmed renewal, and every other server waits at least this long before fencing or reclaiming its +slot. Mixed values are safe, and so is changing the value one server at a time in any order. A longer +lease on one server delays recovery after that server's crash, and the GC leader logs one `WARNING` per +slot whose declared lease differs from its own. A server built without this rule waits only its own +lease: keep the values equal until every server of the pool runs a build with the rule. + +With the defaults (TTL 30 s, period 10 s, margin 2 s), `TTL − margin − period − 2 × envelope = 4 s` is the scheduling-lateness budget before the first renewal attempt of a period can begin, where `envelope = attempt_timeout + 2 × cap` and `cap` is `attempt_timeout` when the disk's `connect_timeout_ms` is `0`, else `min(connect_timeout_ms, attempt_timeout)` (7 s with defaults). @@ -175,7 +176,7 @@ a `MountClaimResult::Kind` together with a `MountPriorState` describing which ce | `None` | no reclaim needed (fresh claim or same-epoch refresh) | | `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 | +| `UncleanObserved` | this claimant's own token-stability observation held for the full observation threshold (the longer of the two leases, plus drift and step) | | `UncleanUnsafe` | the operator's explicit `cas_unsafe_remount_no_delay` authorization carried the slot's exact token — not a certificate of death | ## Behavioral mount-slot model {#mount-state-machines} diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index a490428b30321..f411ea47fb3ff 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -96,7 +96,7 @@ entirely before release. Treat this table as a snapshot of the current build, no | `cas_blob_hash` | `cityhash128` | Pool blob content-hash function (`cityhash128` \| `xxh3-128` \| `sha256`). Recorded in the pool at creation; a mismatching config is refused at mount | | `cas_blob_hash_allow_new` | `false` | Explicit opt-in to admit a new hash algorithm into an existing pool. One-way: once admitted, the pool carries both algorithms permanently | | `skip_access_check` | `false` | Skip the boot-time capability probe (start now, fix later). Only the preflight probe is skipped — the conditional-write correctness check still runs on every writable mount. **Not available on a writable generation-token (GCS) disk**, which refuses to mount with it: there, the probe battery is the only proof that a token-exact delete carries its generation precondition. Mount such a disk read-only if you need to defer the check | -| `cas_mount_lease_ttl_ms` | `30000` | Milliseconds for which a mount lease remains valid after a successful claim or renewal (≥ 1). Lower values shorten stale-mount recovery but reduce tolerance for object-storage and scheduling delays | +| `cas_mount_lease_ttl_ms` | `30000` | How long this server may write after its last confirmed renewal; every other server waits at least this long before fencing or reclaiming its slot (≥ 1). Lower values shorten recovery after this server's crash but reduce tolerance for object-storage and scheduling delays | | `cas_mount_renew_period_ms` | `10000` | Milliseconds between background mount-lease renewals (≥ 1). It must leave enough time for two attempt envelopes (a renewal write and its settlement read) and the lease safety margin before the TTL expires: `period + 2 × envelope + margin < TTL` | | `cas_gc_snapshot_generations_to_keep` | `3` | GC snapshot generations retained | | `cas_gc_shards` | `1` | Blob-hash-prefix reducer shards (≥ 1). Recorded in the pool at creation; a mismatching config is refused at mount | @@ -112,17 +112,15 @@ entirely before release. Treat this table as a snapshot of the current build, no | `cas_unsafe_remount_no_delay` | `0` | Reclaim a mount slot that carries this server's own uuid at once after a hard restart, without observing the slot's token for the lease TTL. Unsafe whenever two processes can hold the same `server_uuid` (a copied uuid file, a stalled predecessor). After such a reclaim the predecessor can still start conditional writes until its own cutoff (`confirmed deadline − cas_lease_safety_margin_ms − 2 × envelope`) or until its next renewal meets the token guard, and a request it already sent may still materialize later. That is not a data hazard: ref-log keys carry `(writer_epoch, sequence)` and creates are conditional, so two writers can never commit different bodies to one key, and recovery's epoch seal settles any straggler (recovery fails closed after 64 successive seal-create attempts displaced by newly materializing old-epoch transactions). The exposure is availability, not data. Intended for test stands and deployments that guarantee one process per uuid | | `cas_staging_backend` | `local` | Blob staging backend (`local` \| `s3`); `s3` is opt-in and requires native same-store copy on writable mount | -All servers sharing a pool must run the same `cas_mount_lease_ttl_ms` and `cas_mount_renew_period_ms`. -Startup reclaim and GC's fence-out both judge liveness by the mount slot's write token holding stable -on the observer's own `CLOCK_BOOTTIME`, using the observer's own threshold — nothing about a writer's -timing travels on the wire. Startup observes `cas_mount_lease_ttl_ms + floor(cas_mount_lease_ttl_ms / -20) + max(1, floor(cas_mount_renew_period_ms / 2))`; GC observes `cas_mount_lease_ttl_ms + -floor(cas_mount_lease_ttl_ms / 20) + cas_mount_renew_period_ms`. A pool member or GC leader -configured with a shorter threshold than its peers can therefore fence out a healthy peer whose -token-update gap exceeds that shorter threshold — a peer renewing frequently stays live, one that -missed a renewal does not. Change these values only with every member of the pool stopped: a -graceful restart removes only that member's own startup observation and does not make mixed -thresholds safe. +Both settings are local. Startup reclaim and GC's fence-out judge liveness by the mount slot's write +token holding stable on the observer's own `CLOCK_BOOTTIME` for `L + floor(L / 20) + step`. `L` is the +longer of the observer's `cas_mount_lease_ttl_ms` and the lease the holder's body declares +(`expires_at_ms − started_at_ms`; a body without stamps or with a non-positive interval declares +nothing). Startup's step is `max(1, floor(cas_mount_renew_period_ms / 2))` and GC's is +`cas_mount_renew_period_ms`, both the observer's own. Mixed values are therefore safe in any order; +a longer lease on one server delays recovery after that server's crash. A server built without this +rule waits only its own lease, so keep the values equal until every server of the pool runs a build +with it. A shorter TTL reduces the tolerance for object-storage delays; a shorter renewal period increases it (renewal starts earlier) at the cost of more background traffic. With the defaults, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 16534edebb2b8..6433c83fb6b4b 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -696,24 +696,19 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al } /// Etag-guarded fence-out of dead mounts (liveness only — graduation itself paces on GC - /// rounds via `new_round`, not on heartbeat acks). Fencing no longer trusts a predecessor's stamped - /// `expires_at_ms` against our wall clock — it - /// fences ONLY once `mount_obs` has watched the mount's write-token hold unchanged for the full - /// threshold on THIS leader's own monotonic clock (shares `claimMountAwaitingExpiry`'s - /// `TTL + Drift` formula for a mount's own reopen wait, but with the full renewal period as the - /// cadence term below instead of half of it, so the two thresholds are close, not identical). + /// rounds via `new_round`, not on heartbeat acks). A slot is fenced only after `mount_obs` has + /// watched its token hold unchanged on this leader's own monotonic clock for + /// `mountObservationThresholdMs` over the longer of this server's lease and the one the slot + /// declares, with the full renewal period as the step. const uint64_t ttl_ms = static_cast(store->poolConfig().mount_lease_ttl_ms.count()); - /// The formula is shared with `claimMountAwaitingExpiry` via - /// `mountObservationThresholdMs` -- see its doc comment (CasServerRoot.h). - const uint64_t stable_threshold_ms = mountObservationThresholdMs( - ttl_ms, static_cast(store->poolConfig().mount_renew_period.count())); + const uint64_t renew_period_ms = static_cast(store->poolConfig().mount_renew_period.count()); /// PHASE 3/18 `heartbeat_floor`: one LIST of `gc/server-roots/`, one GET per mount slot, and a fence /// PUT per newly-fenced mount. { GcPhaseTimer t(phase_sink, "heartbeat_floor"); const HeartbeatFloor floor = computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn, - stable_threshold_ms, mount_obs); + ttl_ms, renew_period_ms, mount_obs, lease_mismatch_warned); report.fence_outs = floor.fenced_now; if (floor.fenced_now > 0) ProfileEvents::increment(ProfileEvents::CASGCHeartbeatFenceOuts, floor.fenced_now); @@ -4759,11 +4754,8 @@ RebuildReport Gc::rebuildBaseline(bool force) /// returned classification counts are not needed for the round mint — graduation paces on rounds). /// Use the same threshold/`mount_obs` as the regular round. const uint64_t ttl_ms = static_cast(store->poolConfig().mount_lease_ttl_ms.count()); - /// Share the identical formula with `claimMountAwaitingExpiry` via - /// `mountObservationThresholdMs` -- see its doc comment (CasServerRoot.h). - const uint64_t stable_threshold_ms = mountObservationThresholdMs( - ttl_ms, static_cast(store->poolConfig().mount_renew_period.count())); - computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn, stable_threshold_ms, mount_obs); + const uint64_t renew_period_ms = static_cast(store->poolConfig().mount_renew_period.count()); + computeHeartbeatFloor(op, layout, now_ms_fn(), mono_ms_fn, ttl_ms, renew_period_ms, mount_obs, lease_mismatch_warned); /// Retired-in-snapshot: the rebuilt seal's `condemned_summary` must be TOTAL over gc_shards so a /// subsequent regular round reads graduation/carry decisions zero-I/O off it (and its `carryParentRefs` diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index fa6215beb1292..e17cfa7e13c94 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -1009,6 +1009,9 @@ class Gc /// leader (after a steal, or a process restart) starts empty, which only delays fencing an /// already-dead mount by one extra round (safe: never fences early). MountObservationMap mount_obs; + /// Srids `computeHeartbeatFloor` already warned about for a lease length different from ours. + /// In-memory: a new leader names each mismatched holder once more. + std::set lease_mismatch_warned; /// Throttle for the orphan sweep's retention rollup. The premise retains on any pass whose epochs are /// not closed-and-folded, so an unconditional `LOG_INFO` would repeat the same sentence for as long as diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index d50a9fb3f85b6..32bb5efcf1edc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -918,9 +919,8 @@ String mountDoubleStartMessage(const String & srid, const std::optional::max(); + return threshold_ms; } MountClaimResult claimMountAwaitingExpiry( @@ -949,13 +962,9 @@ MountClaimResult claimMountAwaitingExpiry( /// A zero poll interval would spin; a single-ms floor keeps the loop a real (bounded) wait. const uint64_t poll = poll_interval_ms == 0 ? 1 : poll_interval_ms; - /// Rate-bound observation threshold: the full lease TTL, plus a 5% allowance for clock-rate - /// mismatch between the holder's and our own local clock, plus one poll interval for observation - /// discreteness. It is measured only with OUR OWN clock (`mono_ms_fn`); no cross-node wall-clock - /// comparison participates in this loop. `poll` here is half the renewal period (the caller's own - /// poll cadence), so this threshold is close to, but not identical to, GC's heartbeat fence-out - /// threshold, which passes the full renewal period into the same shared helper. - const uint64_t threshold_ms = mountObservationThresholdMs(ttl_ms, poll); + /// Set when an observation starts, from the observed body; one token is one body, so the threshold + /// of an observation never changes. + uint64_t threshold_ms = 0; std::optional observed; uint64_t observed_since = 0; @@ -1005,7 +1014,10 @@ MountClaimResult claimMountAwaitingExpiry( return r; observed = *current_etag; observed_since = mono_ms_fn(); - if (on_wait_start && r.body) + if (!r.body) + throw Exception(ErrorCodes::LOGICAL_ERROR, "CAS mount '{}': observed a mount token without its body", srid); + threshold_ms = mountObservationThresholdMs(ttl_ms, declaredMountLeaseMs(*r.body), poll); + if (on_wait_start) on_wait_start(*r.body, threshold_ms); LOG_INFO(getLogger("CasMountLease"), "Attempting to mount content-addressed server root {} after node change or hard " @@ -1022,9 +1034,28 @@ bool MountIncarnationObservation::heldFor(uint64_t threshold_ms, uint64_t sample return sample_ms >= first_seen_mono_ms && sample_ms - first_seen_mono_ms >= threshold_ms; } +namespace +{ +/// The setting is local, so different values are legal; a longer one slows recovery after its holder's +/// crash, which the operator should know about. +void warnOnceIfHolderLeaseDiffers( + const String & srid, const MountLease & holder, uint64_t observer_lease_ms, std::set & warned) +{ + const uint64_t declared_ms = declaredMountLeaseMs(holder); + if (declared_ms == 0 || declared_ms == observer_lease_ms || !warned.insert(srid).second) + return; + LOG_WARNING(getLogger("CasHeartbeatFloor"), + "This server is configured with cas_mount_lease_ttl_ms={} but the holder of slot {} (hostname={}, pid={}) " + "declares a lease of {} ms. Observers wait for the longer of the two; a longer lease on one server " + "delays recovery after that server's crash.", + observer_lease_ms, srid, holder.hostname, holder.pid, declared_ms); +} +} + HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64_t now_ms, const std::function & mono_ms_fn, - uint64_t stable_threshold_ms, MountObservationMap & obs) + uint64_t observer_lease_ms, uint64_t cadence_ms, MountObservationMap & obs, + std::set & lease_mismatch_warned) { HeartbeatFloor floor; /// Taken before any read: it can confirm that a token held, never date a sighting. @@ -1059,6 +1090,7 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 /// holder's own renewal installed, and the observation check below then sees the new /// incarnation and restarts the window -- which counts the slot `live` and declines the write. /// That is why no arm that counts a slot ever also asks for a fence-out body. + uint64_t threshold_ms = 0; WriteResult fenced_out = op.readModifyWrite(key, [&](const std::optional & observed) -> std::optional { @@ -1083,6 +1115,9 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 return std::nullopt; } + warnOnceIfHolderLeaseDiffers(srid, m, observer_lease_ms, lease_mismatch_warned); + threshold_ms = mountObservationThresholdMs(observer_lease_ms, declaredMountLeaseMs(m), cadence_ms); + /// Observation-based liveness: stable ONLY if the SAME incarnation was already being /// watched and has now held for the full threshold on our OWN monotonic clock. Anything /// else — no prior observation, or a changed incarnation (a live renewal, including one @@ -1090,7 +1125,7 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 /// counts as `live` this call. const auto it = obs.find(srid); const bool stable = it != obs.end() && it->second.etag == observed->etag - && it->second.heldFor(stable_threshold_ms, round_start_ms); + && it->second.heldFor(threshold_ms, round_start_ms); if (!stable) { @@ -1119,7 +1154,7 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 "CAS GC fenced out mount lease for content-addressed server root {} at " "wall-clock ms {}: its write incarnation held unchanged for >= {} ms on the GC " "leader's own monotonic clock (token-stability observation)", - srid, now_ms, stable_threshold_ms); + srid, now_ms, threshold_ms); return true; } /// Declined: the decision above already classified and counted this slot, and asked for no @@ -1130,9 +1165,10 @@ HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64 return true; }, Retry::standard()); - /// Prune every `obs` entry for a srid this pass's walk never saw at all. + /// Prune every `obs` and `lease_mismatch_warned` entry for a srid this pass's walk never saw at all. for (auto it = obs.begin(); it != obs.end(); ) it = seen_srids.contains(it->first) ? std::next(it) : obs.erase(it); + std::erase_if(lease_mismatch_warned, [&](const String & s) { return !seen_srids.contains(s); }); return floor; } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index f583ca2af83ca..847e50dfd520f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -304,9 +305,9 @@ String mountDoubleStartMessage(const String & srid, const std::optional; /// CURRENT incarnation differs from (or is absent from) `obs` is (re)started fresh — counted `live`, /// never fenced this call, regardless of what its stamped `expires_at_ms` claims (a bare /// wall-clock stamp is never trusted — see `claimMount`'s "certificate of death" doc). Only once -/// the SAME incarnation has held for `>= stable_threshold_ms` OF THE LEADER'S OWN CLOCK does the body +/// the SAME incarnation has held, on the leader's own clock, for +/// `mountObservationThresholdMs(observer_lease_ms, declaredMountLeaseMs(body), cadence_ms)` does the body /// become FENCE-eligible; /// - FENCE-eligible → one incarnation-guarded `replace` preserving the WHOLE body, setting /// `gc_fenced = true` and `seq + 1`. On `Committed` → excluded (`fenced_now`); on `Conflict` @@ -394,6 +398,9 @@ using MountObservationMap = std::map; /// each sighting. `obs` is owned by the caller and threaded across consecutive calls (one GC /// leader instance, `Cas::Gc::mount_obs`) — a fresh leader starts with an empty map (safe: delays /// fencing one round, never fences early). +/// A body whose declared lease is non-zero and differs from `observer_lease_ms` is named in one `WARNING` +/// per srid; `lease_mismatch_warned` holds the srids already named and is pruned like `obs`. The check +/// never changes a classification or a write. /// /// The fence-out is BOTH safety and liveness. Safety: a sleeper's later renewal permanently fails /// (its `replace` now mismatches the fenced incarnation → `tripMountLost`), so it can never re-arm @@ -412,7 +419,8 @@ struct HeartbeatFloor HeartbeatFloor computeHeartbeatFloor(CasOperation & op, const Layout & l, uint64_t now_ms, const std::function & mono_ms_fn, - uint64_t stable_threshold_ms, MountObservationMap & obs); + uint64_t observer_lease_ms, uint64_t cadence_ms, MountObservationMap & obs, + std::set & lease_mismatch_warned); /// One `gc/server-roots//mount` slot whose holder is not provably finished with the prefix. struct NonTerminalMountSlot diff --git a/src/Disks/tests/cas_test_helpers.h b/src/Disks/tests/cas_test_helpers.h index 5b02caedc9dff..d87e16d3b15df 100644 --- a/src/Disks/tests/cas_test_helpers.h +++ b/src/Disks/tests/cas_test_helpers.h @@ -27,6 +27,7 @@ #include #include #include +#include #include #include @@ -40,7 +41,9 @@ /// For `ChunkFaultBackend`'s `DefiniteFailure` mode, which needs a real S3-classified error, and for /// the ambiguity it raises otherwise. #include +#include #include +#include #include #include @@ -53,8 +56,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -94,6 +99,49 @@ inline DB::Cas::CasRequests openRequestsForTest(DB::Cas::Backend & backend) return openRequestsForTest(DB::Cas::BackendPtr(std::shared_ptr(), &backend)); } +/// Routes one logger into memory at `level` for the scope, then restores its channel and level. +class ScopedLogCapture +{ +public: + ScopedLogCapture(const String & logger_name, const String & level) + : logger(getLogger(logger_name)) + , channel(new Poco::StreamChannel(stream)) + , old_channel(logger->getChannel(), /*shared=*/true) + , old_level(logger->getLevel()) + { + logger->setChannel(channel.get()); + logger->setLevel(level); + } + + ~ScopedLogCapture() + { + logger->setChannel(old_channel); + logger->setLevel(old_level); + } + + ScopedLogCapture(const ScopedLogCapture &) = delete; + ScopedLogCapture & operator=(const ScopedLogCapture &) = delete; + + String captured() const { return stream.str(); } + + size_t count(std::string_view needle) const + { + const String haystack = captured(); + size_t n = 0; + for (size_t pos = 0; (pos = haystack.find(needle, pos)) != String::npos; pos += needle.size()) + ++n; + return n; + } + +private: + LoggerPtr logger; + std::ostringstream stream; // STYLE_CHECK_ALLOW_STD_STRING_STREAM + Poco::AutoPtr channel; + /// `shared=true`: the parked channel is not ours, so take a reference instead of stealing one. + Poco::AutoPtr old_channel; + int old_level; +}; + /// An open-fence operation together with the `CasRequests` it refers to, for a fixture that holds a /// backend and needs to call a production entry point taking a `CasOperation &`. Neither copyable nor /// movable: the operation points at the member beside it. diff --git a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp index 8bd8f8b6ca344..7627c4f1968ba 100644 --- a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp +++ b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp @@ -940,8 +940,8 @@ void runExpiredMountFenceOutScenario(const PoolConfig & config) srid2_renewer.start(); ASSERT_FALSE(decodeMountLease(readObj(*backend, layout.mountKey(srid2))->bytes).gc_fenced); - // The fence-out threshold on the GC leader's OWN monotonic clock — mirrors the production formula - // in `Gc::runRegularRound` (ttl + 5% drift allowance + one round's worth of renewal slack). + // The fence-out threshold on the GC leader's OWN monotonic clock. The holder declares 100 ms, shorter + // than the observer's lease, so L is the observer's own: `L + L/20 + step`, step the renewal period. const uint64_t ttl_ms = static_cast(store->poolConfig().mount_lease_ttl_ms.count()); const uint64_t threshold_ms = ttl_ms + ttl_ms / 20 + static_cast(store->poolConfig().mount_renew_period.count()); @@ -1056,7 +1056,7 @@ void runSlowWalkSightingScenario(SightingCall sighting_call) const uint64_t ttl_ms = static_cast(store->poolConfig().mount_lease_ttl_ms.count()); const uint64_t threshold_ms = mountObservationThresholdMs( - ttl_ms, static_cast(store->poolConfig().mount_renew_period.count())); + ttl_ms, /*declared_lease_ms=*/0, static_cast(store->poolConfig().mount_renew_period.count())); /// Longer than `threshold - TTL`: a sighting credited from before this walk fences a holder /// whose last renewal still gives it authority. const uint64_t walk_ms = threshold_ms - ttl_ms + 1'000; @@ -1113,6 +1113,52 @@ void runSlowWalkSightingScenario(SightingCall sighting_call) EXPECT_TRUE(decodeMountLease(readObj(*backend, backend->watched_key)->bytes).gc_fenced); } +/// Defaults give the observer 30 s + 1.5 s + 10 s = 41.5 s; the holder's 60 s gives 73 s. +void runLongerDeclaredLeaseScenario(SightingCall early_call) +{ + uint64_t mono = 0; + auto backend = std::make_shared(); + auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const Layout & layout = store->layout(); + ASSERT_EQ(store->poolConfig().mount_lease_ttl_ms.count(), 30'000); + ASSERT_EQ(store->poolConfig().mount_renew_period.count(), 10'000); + Gc gc(store, kGc, [] { return uint64_t{1'000'000}; }, [&mono] { return mono; }); + + const RootNamespace ns{"00/aa@cas@"}; + const ManifestRef r = ref("srv-a:1", 1, 0xAA); + writeBlobBody(*backend, layout, DB::UInt128(7)); + writeManifestRaw(*backend, layout, ns, r, {blobEntryFor("a", DB::UInt128(7))}); + publishCommittedTransition(*backend, layout, ns, "tbl", std::nullopt, r); + + /// Claims once with a 60 s lease and never renews. + const String holder_srid = "long-lease-server"; + CasRequests renewer_requests = openRequestsForTest(backend); + MountLeaseRenewer holder(renewer_requests, renewer_requests, layout, holder_srid, DB::UInt128(0x3333), + /*writer_epoch=*/1, std::chrono::milliseconds(60'000), [] { return 1000u; }, [] { return 0u; }, {}, + std::chrono::milliseconds(0), [] { return 0u; }); + holder.start(); + const auto holder_fenced = [&] + { return decodeMountLease(readObj(*backend, layout.mountKey(holder_srid))->bytes).gc_fenced; }; + + ASSERT_EQ(gc.runRegularRound().fence_outs, 0u); + /// The store's own slot renews before each decision so only the holder can be fenced. + store->renewWatermarkOnce(); + mono = 41'500; + if (early_call == SightingCall::Round) + EXPECT_EQ(gc.runRegularRound().fence_outs, 0u); + else + ASSERT_TRUE(gc.rebuildBaseline(/*force*/ true).performed); + EXPECT_FALSE(holder_fenced()) << "fenced after the observer's own lease while the declared lease still ran"; + + store->renewWatermarkOnce(); + mono = 72'999; + EXPECT_EQ(gc.runRegularRound().fence_outs, 0u); + store->renewWatermarkOnce(); + mono = 73'000; + EXPECT_EQ(gc.runRegularRound().fence_outs, 1u); + EXPECT_TRUE(holder_fenced()); +} + } TEST(CASGCAckFloor, ExpiredMountFencedOutAndExcluded) @@ -1120,8 +1166,9 @@ TEST(CASGCAckFloor, ExpiredMountFencedOutAndExcluded) runExpiredMountFenceOutScenario(PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); } -/// GC's fence-out threshold (`ttl + 5% drift allowance + one round's worth of renewal slack`, computed in -/// `Gc::runRegularRound`) never reads `PoolConfig::unsafe_remount_no_delay` -- that knob is consulted only +/// GC's fence-out threshold (`L + L/20 + step`; the holder's 100 ms lease is shorter than the observer's, so L +/// is the observer's own and the step is the renewal period) never reads `PoolConfig::unsafe_remount_no_delay` +/// -- that knob is consulted only /// by `Pool::mountWritable`'s own reclaim decision, never by GC's heartbeat-floor observation. Runs the /// IDENTICAL three-round scenario as `ExpiredMountFencedOutAndExcluded` with the knob turned on, and /// asserts the SAME round-by-round fence-out counts -- including the one-millisecond-short discriminator @@ -1142,6 +1189,16 @@ TEST(CASGcFenceOut, RebuildCountsASightingFromAfterItsRead) runSlowWalkSightingScenario(SightingCall::Rebuild); } +TEST(CASGcFenceOut, RoundWaitsForTheHoldersLongerLease) +{ + runLongerDeclaredLeaseScenario(SightingCall::Round); +} + +TEST(CASGcFenceOut, RebuildWaitsForTheHoldersLongerLease) +{ + runLongerDeclaredLeaseScenario(SightingCall::Rebuild); +} + /// fix-round F6 (author-review: `Gc`'s own `mono_ms_fn` used to default to the RAW static `Pool:: /// bootMs()`, bypassing the Pool's own injectable `config.boot_ms_fn` -- a time-controlled test can /// desync the mount side's fake clock from the GC side's real one). This mirrors @@ -1650,3 +1707,41 @@ TEST(CASGCRetire, OutcomeLogUnobservedConflictDoesNotReportItVanished) EXPECT_TRUE(refusal_reached) << "no round ever wrote an outcome log, so the arm was never reached"; } #endif + +/// The round passes the pool's own lease: the store's slot, renewed with that lease, is not named; +/// a holder renewing with 100 ms is named once across two rounds. +TEST(CASGCLeaseMismatch, RoundWarnsOnceAboutAHolderWithADifferentLease) +{ + auto backend = std::make_shared(); + auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + const Layout & layout = store->layout(); + + const String holder_srid = "short-lease-server"; + CasRequests renewer_requests = openRequestsForTest(backend); + MountLeaseRenewer holder(renewer_requests, renewer_requests, layout, holder_srid, DB::UInt128(0x2222), + /*writer_epoch=*/1, + std::chrono::milliseconds(100), [] { return 1000u; }, [] { return 0u; }, {}, + std::chrono::milliseconds(0), [] { return 0u; }); + holder.start(); + + const RootNamespace ns{"00/aa@cas@"}; + const ManifestRef r = ref("srv-a:1", 1, 0xAA); + const UInt128 blob = DB::UInt128(7); + writeBlobBody(*backend, layout, blob); + writeManifestRaw(*backend, layout, ns, r, {blobEntryFor("a", blob)}); + publishCommittedTransition(*backend, layout, ns, "tbl", std::nullopt, r); + + uint64_t gc_now = 1'000'000; + uint64_t gc_mono = 0; + Gc gc(store, kGc, [&] { return gc_now; }, [&] { return gc_mono; }); + + ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + gc.runRegularRound(); + gc_mono = 1; + gc.runRegularRound(); + + EXPECT_EQ(capture.count("declares a lease"), 1u) << capture.captured(); + EXPECT_EQ(capture.count("This server is configured with cas_mount_lease_ttl_ms=30000 but the holder of slot " + "short-lease-server (hostname="), 1u) << capture.captured(); + EXPECT_EQ(capture.count("declares a lease of 100 ms."), 1u) << capture.captured(); +} diff --git a/src/Disks/tests/gtest_cas_mount.cpp b/src/Disks/tests/gtest_cas_mount.cpp index ae0afa6f1a048..2e7735b95ad6a 100644 --- a/src/Disks/tests/gtest_cas_mount.cpp +++ b/src/Disks/tests/gtest_cas_mount.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -1082,29 +1083,71 @@ TEST(CASMountAwaitExpiry, ForeignUuidFailsClosedImmediately) EXPECT_EQ(sleeps, 0); // never waits across UUIDs } -/// rev.6: the predecessor's own stamped `expires_at_ms` (however skewed) is NEVER consulted for the -/// reclaim decision any more — the wait is bounded purely by OUR OWN `ttl_ms`-derived threshold. A -/// prior incarnation minted with an absurdly large `ttl` (so its own stamp claims aliveness for -/// ~100000ms) still reclaims within the SAME small threshold as any other case, because that stamp is -/// never read for timing. -TEST(CASMountAwaitExpiry, SkewedFarFutureExpiryHasNoEffectOnObservationThreshold) +namespace +{ +/// Restart reclaim as this server with a 30 s lease and a 1 s poll. Returns the observer clock at the +/// reclaim and records every threshold `on_wait_start` reports. +uint64_t reclaimAsShortLeaseObserver(Ops & ops, const Layout & l, std::vector & thresholds) { - auto b = std::make_shared(); - Layout l("p"); - Ops ops(b); - ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, /*now*/ 1000, /*ttl*/ 100000).kind, MountClaimResult::Claimed); - - uint64_t wall = 1000; uint64_t mono = 0; - auto now_fn = [&] { return wall; }; - auto mono_fn = [&] { return mono; }; - auto sleep_fn = [&](uint64_t ms) { wall += ms; mono += ms; }; - const auto r = claimMountAwaitingExpiry( - ops.op, l, "r", UInt128(1), /*our_epoch*/ 8, now_fn, mono_fn, /*ttl*/ 100, /*poll*/ 20, sleep_fn); + ops.op, l, "r", UInt128(1), /*our_epoch*/ 8, [] { return uint64_t{1000}; }, [&] { return mono; }, + /*ttl*/ 30'000, /*poll*/ 1'000, [&](uint64_t ms) { mono += ms; }, + [&](const MountLease &, uint64_t threshold_ms) { thresholds.push_back(threshold_ms); }); EXPECT_EQ(r.kind, MountClaimResult::Claimed); - EXPECT_LE(mono, 100u + 100u / 20 + 20u + 20u); // bounded by OUR threshold, not the predecessor's stamp - EXPECT_EQ(decodeMountLease(ops.op.read(l.mountKey("r"), Retry::standard())->bytes).writer_epoch, 8u); // reclaimed + EXPECT_EQ(r.prior, MountPriorState::UncleanObserved); + return mono; +} +} + +TEST(CASMountAwaitExpiry, HolderDeclaringALongerLeaseIsWaitedOutInFull) +{ + Ops ops(std::make_shared()); + Layout l("p"); + ASSERT_EQ(claimMount(ops.op, l, "r", UInt128(1), 7, /*now*/ 1000, /*ttl*/ 60'000).kind, MountClaimResult::Claimed); + std::vector thresholds; + EXPECT_EQ(reclaimAsShortLeaseObserver(ops, l, thresholds), 64'000u) + << "reclaimed before the predecessor's declared lease ran out"; + EXPECT_EQ(thresholds, std::vector{64'000}); +} + +/// A shorter declared lease (10 s), a missing start stamp, equal stamps, an expiry before the start. +TEST(CASMountAwaitExpiry, ShorterOrUnusableDeclaredLeaseWaitsTheObserversOwn) +{ + const std::vector> stamps{{1'000, 11'000}, {0, 60'000}, {5'000, 5'000}, {70'000, 10'000}}; + for (const auto & [started, expires] : stamps) + { + SCOPED_TRACE(fmt::format("started_at_ms={} expires_at_ms={}", started, expires)); + Ops ops(std::make_shared()); + Layout l("p"); + const MountLease m{.server_uuid = UInt128(1), .writer_epoch = 7, .hostname = "h", .pid = 1, + .started_at_ms = started, .seq = 1, .expires_at_ms = expires, .write_attempt_id = UInt128(1)}; + mustCommit(ops.op.create(l.mountKey("r"), encodeMountLease(m), Retry::standard()), "seeded mount"); + std::vector thresholds; + EXPECT_EQ(reclaimAsShortLeaseObserver(ops, l, thresholds), 33'000u); + EXPECT_EQ(thresholds, std::vector{32'500}); + } +} + +/// The certificates of death need no wait, whatever lease the body declares. +TEST(CASMountAwaitExpiry, FencedAndFarewellSlotsReclaimWithoutWaitingWhateverTheyDeclare) +{ + for (const bool fenced : {true, false}) + { + Ops ops(std::make_shared()); + Layout l("p"); + MountLease m{.server_uuid = UInt128(1), .writer_epoch = 7, .hostname = "h", .pid = 1, + .started_at_ms = 1000, .seq = 1, .expires_at_ms = 61'000, .write_attempt_id = UInt128(1)}; + m.gc_fenced = fenced; + m.min_active_build_sequence = fenced ? 0 : std::numeric_limits::max(); + mustCommit(ops.op.create(l.mountKey("r"), encodeMountLease(m), Retry::standard()), "seeded mount"); + int sleeps = 0; + const auto r = claimMountAwaitingExpiry(ops.op, l, "r", UInt128(1), 8, [] { return uint64_t{1000}; }, + [] { return uint64_t{0}; }, 30'000, 1'000, [&](uint64_t) { ++sleeps; }); + EXPECT_EQ(r.kind, MountClaimResult::Claimed); + EXPECT_EQ(r.prior, fenced ? MountPriorState::Fenced : MountPriorState::Clean); + EXPECT_EQ(sleeps, 0); + } } TEST(CASMountClaim, UnsafeAuthorizationIsTokenExact) @@ -1502,22 +1545,25 @@ namespace /// Rev.6 §token-stability observation removed the wall clock from the fence DECISION; `kNowMs` below /// is threaded through only as `computeHeartbeatFloor`'s audit-only `now_ms`. constexpr uint64_t kNowMs = 1'000'000; -/// The fence-out threshold measured on the LEADER's OWN monotonic clock (`mono_ms_fn`), independent -/// of any lease's stamped `expires_at_ms`. -constexpr uint64_t kStableThresholdMs = 10'000; +/// The observing leader's own lease and step, and the threshold they give a body that declares nothing. +constexpr uint64_t kObserverLeaseMs = 30'000; +constexpr uint64_t kCadenceMs = 1'000; +constexpr uint64_t kStableThresholdMs = 32'500; +constexpr std::string_view kLeaseMismatchMarker = "declares a lease"; /// Seed one mount body under mountKey(srid) via the on-storage codec — the same interface the renewer /// writes through. MountLease seedMount( CasOperation & op, const Layout & l, const String & srid, - uint64_t expires_at_ms, bool gc_fenced, uint64_t min_active_build_sequence, uint64_t seq = 1) + uint64_t expires_at_ms, bool gc_fenced, uint64_t min_active_build_sequence, uint64_t seq = 1, + uint64_t started_at_ms = kNowMs) { MountLease m; m.server_uuid = UInt128(srid.back()); // distinct per srid; content is irrelevant to the gate m.writer_epoch = 1; m.hostname = "h-" + srid; m.pid = 100; - m.started_at_ms = kNowMs; + m.started_at_ms = started_at_ms; m.seq = seq; m.expires_at_ms = expires_at_ms; m.min_active_build_sequence = min_active_build_sequence; @@ -1560,8 +1606,9 @@ TEST(CASHeartbeatFloor, FirstSightNeverFencesEvenIfStampLooksExpired) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; + std::set warned; const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, /*now_ms*/ kNowMs, frozenClock(0), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor.fenced_now, 0u); EXPECT_EQ(floor.live, 1u); @@ -1577,7 +1624,8 @@ TEST(CASHeartbeatFloor, StableIncarnationPastThresholdIsFenced) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + std::set warned; + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor_before.fenced_now, 0u); const MountLease before = decodeMountLease(ops.op.read(l.mountKey("s1"), Retry::standard())->bytes); @@ -1585,7 +1633,7 @@ TEST(CASHeartbeatFloor, StableIncarnationPastThresholdIsFenced) /// No renewal in between: the SAME incarnation, observed since mono 0, is now stable for the full /// threshold on the leader's own clock. const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor2.fenced_now, 1u); EXPECT_EQ(floor2.fenced_srids, std::vector{"s1"}); @@ -1602,7 +1650,8 @@ TEST(CASHeartbeatFloor, RenewalBetweenRoundsRestartsObservation) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_TRUE(obs.contains("s1")); const Etag first_etag = obs.at("s1").etag; @@ -1611,7 +1660,7 @@ TEST(CASHeartbeatFloor, RenewalBetweenRoundsRestartsObservation) EXPECT_NE(renewed_etag, first_etag); const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor2.fenced_now, 0u); ASSERT_TRUE(obs.contains("s1")); @@ -1633,7 +1682,8 @@ TEST(CASHeartbeatFloor, UnseenSridPrunedFromObservationMap) seedMount(ops.op, l, "s2", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; - computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_TRUE(obs.contains("s1")); ASSERT_TRUE(obs.contains("s2")); @@ -1645,7 +1695,7 @@ TEST(CASHeartbeatFloor, UnseenSridPrunedFromObservationMap) renewMount(ops.op, l, "s1"); ASSERT_EQ(ops.op.removeCurrent(l.mountKey("s2"), Retry::standard()), Removal::Removed); - computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_TRUE(obs.contains("s1")); EXPECT_FALSE(obs.contains("s2")) << "a srid removed from the LIST entirely must be pruned from obs, not linger forever"; @@ -1670,9 +1720,10 @@ TEST(CASHeartbeatFloor, ClassifiesAndFencesOut) /*min_active_build_sequence*/ std::numeric_limits::max()); MountObservationMap obs; + std::set warned; /// Round 1 (mono 0): first sight of every non-terminal mount — nothing is fence-eligible yet. - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor_before.live, 3u); // s1, s2, s3: observation just started EXPECT_EQ(floor_before.terminated, 1u); // s5 EXPECT_EQ(floor_before.fenced_now, 0u); @@ -1690,7 +1741,7 @@ TEST(CASHeartbeatFloor, ClassifiesAndFencesOut) /// Round 2 (mono == threshold): s1/s2's renewed incarnations restart their observation (still /// live); s3's original incarnation has now held stable for the full threshold -> fenced. const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor2.live, 2u); // s1, s2: renewed, observation restarted EXPECT_EQ(floor2.terminated, 1u); // s5 @@ -1765,9 +1816,10 @@ TEST(CASHeartbeatFloor, FenceOutLosesTheIncarnationRaceAndReclassifiesLive) seedMount(ops.op, l, "s1", /*expires*/ kNowMs - 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; + std::set warned; /// Round 1: first sight, observation starts — never reaches the fence-out path (the race /// decorator stays armed for round 2). - const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + const HeartbeatFloor floor_before = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor_before.fenced_now, 0u); /// Round 2: the incarnation has been stable past threshold, so the function attempts the @@ -1775,7 +1827,7 @@ TEST(CASHeartbeatFloor, FenceOutLosesTheIncarnationRaceAndReclassifiesLive) /// and the re-decision reclassifies the slot as live (observation restarted on the new /// incarnation) — never fenced. const HeartbeatFloor floor2 = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor2.fenced_now, 0u); EXPECT_EQ(floor2.live, 1u); @@ -1840,20 +1892,21 @@ TEST(CASHeartbeatFloor, ConfirmingCallDecidesOnItsEntrySample) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); const auto clock = [&mono] { return mono; }; MountObservationMap obs; + std::set warned; - computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_EQ(obs.at("s1").first_seen_mono_ms, 0u); /// The call starts one millisecond short of the threshold; its walk passes the threshold many times. mono = kStableThresholdMs - 1; b->read_cost_ms = 10 * kStableThresholdMs; - const HeartbeatFloor slow = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + const HeartbeatFloor slow = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(slow.fenced_now, 0u); EXPECT_EQ(slow.live, 1u); EXPECT_EQ(obs.at("s1").first_seen_mono_ms, 0u) << "an unchanged token keeps its first sighting"; b->read_cost_ms = 0; - EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 1u); } TEST(CASHeartbeatFloor, FailedSlotReadKeepsTheWatch) @@ -1865,19 +1918,20 @@ TEST(CASHeartbeatFloor, FailedSlotReadKeepsTheWatch) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); const auto clock = [&mono] { return mono; }; MountObservationMap obs; + std::set warned; - computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_EQ(obs.at("s1").first_seen_mono_ms, 0u); mono = kStableThresholdMs / 2; b->fail_slot_reads = true; - EXPECT_ANY_THROW(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs)); + EXPECT_ANY_THROW(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned)); ASSERT_TRUE(obs.contains("s1")); EXPECT_EQ(obs.at("s1").first_seen_mono_ms, 0u); b->fail_slot_reads = false; mono = kStableThresholdMs; - EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u) + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 1u) << "a failed read restarted the observation"; } @@ -1888,13 +1942,14 @@ TEST(CASHeartbeatFloor, SampleOlderThanTheSightingNeverFences) Ops ops(b); seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); MountObservationMap obs; + std::set warned; - computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs), kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_EQ(obs.at("s1").first_seen_mono_ms, kStableThresholdMs); /// A clock that went back: the unsigned difference would wrap past any threshold. const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(kStableThresholdMs - 1), - kStableThresholdMs, obs); + kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor.fenced_now, 0u); EXPECT_EQ(floor.live, 1u); } @@ -1913,22 +1968,23 @@ TEST(CASHeartbeatFloor, FirstSightingCountsFromAfterItsRead) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); const auto clock = [&mono] { return mono; }; MountObservationMap obs; + std::set warned; /// The call starts at 0 and reaches the slot at `kWalkMs`. b->read_cost_ms = kWalkMs; - ASSERT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 0u); + ASSERT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 0u); ASSERT_TRUE(obs.contains("s1")); EXPECT_EQ(obs.at("s1").first_seen_mono_ms, kWalkMs); /// One threshold after the first call started, less than one after its read. b->read_cost_ms = 0; mono = kStableThresholdMs; - const HeartbeatFloor early = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + const HeartbeatFloor early = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(early.fenced_now, 0u) << "fenced a token watched for less than the threshold"; EXPECT_EQ(early.live, 1u); mono = kWalkMs + kStableThresholdMs; - EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 1u); } TEST(CASHeartbeatFloor, RedecisionAfterARefusedFenceCountsFromItsRead) @@ -1940,8 +1996,9 @@ TEST(CASHeartbeatFloor, RedecisionAfterARefusedFenceCountsFromItsRead) seedMount(ops.op, l, "s1", /*expires*/ 10, /*fenced*/ false, /*min_active_build_sequence*/ 0); const auto clock = [&mono] { return mono; }; MountObservationMap obs; + std::set warned; - computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_TRUE(obs.contains("s1")); const Etag first = obs.at("s1").etag; @@ -1950,7 +2007,7 @@ TEST(CASHeartbeatFloor, RedecisionAfterARefusedFenceCountsFromItsRead) mono = kStableThresholdMs; b->read_cost_ms = kWalkMs; b->renew_before_next_fence = true; - const HeartbeatFloor refused = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs); + const HeartbeatFloor refused = computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned); ASSERT_FALSE(b->renew_before_next_fence) << "the call never attempted the fence-out"; ASSERT_EQ(refused.fenced_now, 0u); ASSERT_EQ(refused.live, 1u); @@ -1964,11 +2021,11 @@ TEST(CASHeartbeatFloor, RedecisionAfterARefusedFenceCountsFromItsRead) /// One threshold after that call started, less than one after its re-read. b->read_cost_ms = 0; mono = 2 * kStableThresholdMs; - EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 0u) + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 0u) << "fenced the renewed token before it was watched for the threshold"; mono = reread_at + kStableThresholdMs; - EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kStableThresholdMs, obs).fenced_now, 1u); + EXPECT_EQ(computeHeartbeatFloor(ops.op, l, kNowMs, clock, kObserverLeaseMs, kCadenceMs, obs, warned).fenced_now, 1u); } TEST(CASHeartbeatFloor, EmptyPrefixYieldsNoLiveMounts) @@ -1978,7 +2035,8 @@ TEST(CASHeartbeatFloor, EmptyPrefixYieldsNoLiveMounts) Ops ops(b); MountObservationMap obs; - const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kStableThresholdMs, obs); + std::set warned; + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); EXPECT_EQ(floor.live, 0u); EXPECT_EQ(floor.terminated, 0u); @@ -1986,6 +2044,226 @@ TEST(CASHeartbeatFloor, EmptyPrefixYieldsNoLiveMounts) EXPECT_EQ(floor.already_fenced, 0u); } +namespace +{ +HeartbeatFloor floorAt(Ops & ops, const Layout & l, uint64_t mono_ms, MountObservationMap & obs, std::set & warned) +{ + return computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(mono_ms), kObserverLeaseMs, kCadenceMs, obs, warned); +} +} + +/// The body declares 60 s against the observer's 30 s: 60 s + 3 s + the 1 s step. +TEST(CASHeartbeatFloor, HolderDeclaringALongerLeaseIsWatchedForThatLease) +{ + Ops ops(std::make_shared()); + Layout l("p"); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + MountObservationMap obs; + std::set warned; + floorAt(ops, l, 0, obs, warned); + EXPECT_EQ(floorAt(ops, l, kStableThresholdMs, obs, warned).fenced_now, 0u) + << "fenced after the observer's own lease while the declared lease still ran"; + EXPECT_EQ(floorAt(ops, l, 63'999, obs, warned).fenced_now, 0u); + EXPECT_EQ(floorAt(ops, l, 64'000, obs, warned).fenced_now, 1u); +} + +/// A shorter declared lease, a missing start stamp, equal stamps, an expiry before the start. +TEST(CASHeartbeatFloor, ShorterOrUnusableDeclaredLeaseIsWatchedForTheObserversOwn) +{ + Ops ops(std::make_shared()); + Layout l("p"); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 10'000, false, 0); + seedMount(ops.op, l, "s2", /*expires*/ 60'000, false, 0, /*seq*/ 1, /*started_at_ms*/ 0); + seedMount(ops.op, l, "s3", /*expires*/ kNowMs, false, 0); + seedMount(ops.op, l, "s4", /*expires*/ kNowMs - 1, false, 0); + MountObservationMap obs; + std::set warned; + floorAt(ops, l, 0, obs, warned); + EXPECT_EQ(floorAt(ops, l, kStableThresholdMs - 1, obs, warned).fenced_now, 0u); + EXPECT_EQ(floorAt(ops, l, kStableThresholdMs, obs, warned).fenced_now, 4u); +} + +/// `gc_fenced` and farewell slots stay terminal whatever lease they declare: no observation, no write. +TEST(CASHeartbeatFloor, TerminalSlotsAreExcludedWhateverTheyDeclare) +{ + Ops ops(std::make_shared()); + Layout l("p"); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ true, 0); + seedMount(ops.op, l, "s2", /*expires*/ kNowMs + 60'000, false, std::numeric_limits::max()); + MountObservationMap obs; + std::set warned; + const HeartbeatFloor floor = floorAt(ops, l, 0, obs, warned); + EXPECT_EQ(floor.already_fenced, 1u); + EXPECT_EQ(floor.terminated, 1u); + EXPECT_TRUE(obs.empty()); +} + +TEST(CASMountObservationThreshold, LongerOfOwnAndDeclaredLeasePlusDriftAndStep) +{ + EXPECT_EQ(mountObservationThresholdMs(30'000, 0, 1'000), 32'500u); + EXPECT_EQ(mountObservationThresholdMs(30'000, 10'000, 1'000), 32'500u); + EXPECT_EQ(mountObservationThresholdMs(30'000, 60'000, 1'000), 64'000u); +} + +TEST(CASMountObservationThreshold, SaturatesInsteadOfWrapping) +{ + constexpr uint64_t max = std::numeric_limits::max(); + /// `20 * (max / 21)` plus its twentieth is `max - 15`: the step alone decides the overflow. + EXPECT_EQ(mountObservationThresholdMs(0, 20 * (max / 21), 14), max - 1); + EXPECT_EQ(mountObservationThresholdMs(0, 20 * (max / 21), 16), max); + EXPECT_EQ(mountObservationThresholdMs(30'000, max - 10, 0), max); +} + +TEST(CASMountObservationThreshold, DeclaredLeaseIsThePositiveStampInterval) +{ + MountLease m; + m.started_at_ms = 1'000; + m.expires_at_ms = 61'000; + EXPECT_EQ(declaredMountLeaseMs(m), 60'000u); + m.started_at_ms = 0; + EXPECT_EQ(declaredMountLeaseMs(m), 0u) << "a body without the start stamp"; + m.started_at_ms = 61'000; + EXPECT_EQ(declaredMountLeaseMs(m), 0u) << "a farewell stamps both equal"; + m.started_at_ms = 70'000; + EXPECT_EQ(declaredMountLeaseMs(m), 0u) << "expiry before start"; +} + +TEST(CASHeartbeatFloor, HolderWithADifferentLeaseIsWarnedAboutWithoutChangingTheClassification) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + const String before = ops.op.read(l.mountKey("s1"), Retry::standard())->bytes; + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(capture.count( + "This server is configured with cas_mount_lease_ttl_ms=30000 but the holder of slot s1 " + "(hostname=h-s1, pid=100) declares a lease of 60000 ms. Observers wait for the longer of the two; " + "a longer lease on one server delays recovery after that server's crash."), 1u) + << capture.captured(); + EXPECT_EQ(floor.live, 1u); + EXPECT_EQ(floor.fenced_now, 0u); + EXPECT_EQ(ops.op.read(l.mountKey("s1"), Retry::standard())->bytes, before); + EXPECT_EQ(warned, std::set{"s1"}); +} + +TEST(CASHeartbeatFloor, HolderWithTheSameLeaseIsNotWarnedAbout) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + kObserverLeaseMs, /*fenced*/ false, /*min_active_build_sequence*/ 0); + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(capture.count(kLeaseMismatchMarker), 0u) << capture.captured(); + EXPECT_TRUE(warned.empty()); +} + +TEST(CASHeartbeatFloor, FarewellBodyIsNotWarnedAbout) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + /// What `MountLeaseRenewer::terminate` writes: both stamps equal, farewell sentinel. + seedMount(ops.op, l, "s1", /*expires*/ kNowMs, /*fenced*/ false, + /*min_active_build_sequence*/ std::numeric_limits::max()); + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + const HeartbeatFloor floor = computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(floor.terminated, 1u); + EXPECT_EQ(capture.count(kLeaseMismatchMarker), 0u) << capture.captured(); +} + +/// Both slots keep a 60 s body that differs from the observer's 30 s, but a fenced or departed holder +/// cannot delay anything, so naming it would only alarm the operator. +TEST(CASHeartbeatFloor, TerminalSlotsWithADifferentLeaseAreNotWarnedAbout) +{ + Ops ops(std::make_shared()); + Layout l("p"); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ true, /*min_active_build_sequence*/ 0); + seedMount(ops.op, l, "s2", /*expires*/ kNowMs + 60'000, /*fenced*/ false, + /*min_active_build_sequence*/ std::numeric_limits::max()); + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + const HeartbeatFloor floor = floorAt(ops, l, 0, obs, warned); + + EXPECT_EQ(floor.already_fenced, 1u); + EXPECT_EQ(floor.terminated, 1u); + EXPECT_EQ(capture.count(kLeaseMismatchMarker), 0u) << capture.captured(); + EXPECT_TRUE(warned.empty()); +} + +TEST(CASHeartbeatFloor, MissingStartStampIsNotWarnedAbout) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + /// A body without `started_at_ms` decodes it as 0; the interval to `expires_at_ms` means nothing. + seedMount(ops.op, l, "s1", /*expires*/ 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0, + /*seq*/ 1, /*started_at_ms*/ 0); + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(capture.count(kLeaseMismatchMarker), 0u) << capture.captured(); +} + +TEST(CASHeartbeatFloor, LeaseMismatchIsWarnedOncePerServerRootId) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + seedMount(ops.op, l, "s2", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + + DB::Cas::tests::ScopedLogCapture capture("CasHeartbeatFloor", "warning"); + MountObservationMap obs; + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + /// A renewal gives s1 a new incarnation; the next round reads both bodies again. + renewMount(ops.op, l, "s1"); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(1), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(capture.count(kLeaseMismatchMarker), 2u) << capture.captured(); + EXPECT_EQ(capture.count("holder of slot s1 ("), 1u) << capture.captured(); + EXPECT_EQ(capture.count("holder of slot s2 ("), 1u) << capture.captured(); +} + +TEST(CASHeartbeatFloor, UnseenSridPrunedFromLeaseMismatchWarned) +{ + auto b = std::make_shared(); + Layout l("p"); + Ops ops(b); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + seedMount(ops.op, l, "s2", /*expires*/ kNowMs + 60'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + + MountObservationMap obs; + std::set warned; + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(0), kObserverLeaseMs, kCadenceMs, obs, warned); + ASSERT_EQ(warned, (std::set{"s1", "s2"})); + + ASSERT_EQ(ops.op.removeCurrent(l.mountKey("s2"), Retry::standard()), Removal::Removed); + computeHeartbeatFloor(ops.op, l, kNowMs, frozenClock(1), kObserverLeaseMs, kCadenceMs, obs, warned); + + EXPECT_EQ(warned, std::set{"s1"}); +} + /// ---- Task 1 (Phase 2): `listMounts` — read-only mount-slot enumeration for introspection ---- TEST(CASListMounts, ClassifiesEveryStateReadOnly) diff --git a/src/Disks/tests/gtest_cas_pool.cpp b/src/Disks/tests/gtest_cas_pool.cpp index 9825cae00a865..d082ace5c556d 100644 --- a/src/Disks/tests/gtest_cas_pool.cpp +++ b/src/Disks/tests/gtest_cas_pool.cpp @@ -2521,9 +2521,11 @@ TEST(CASMountOpenWaits, UncleanOpenPaysOnlyTheObservationWindow) })); ASSERT_TRUE(store); - /// The token-stability observation window is paid in full, pinned to the exact configured - /// formula (`mountObservationThresholdMs`): threshold_ms = ttl_ms + ttl_ms/20 + poll_interval_ms - /// = 500 + 25 + 50 = 575 ms, where poll_interval_ms = max(1, mount_renew_period/2) = 50 ms. The + /// The token-stability observation window is paid in full, pinned to the exact + /// formula (`mountObservationThresholdMs`). The predecessor declares the same lease, so own and + /// declared coincide: + /// threshold_ms = ttl_ms + ttl_ms/20 + poll_interval_ms = 500 + 25 + 50 = 575 ms, where + /// poll_interval_ms = max(1, mount_renew_period/2) = 50 ms. The /// loop only re-checks the threshold between polls, so the observed wait rounds UP to the next /// whole poll: ceil(575 / 50) * 50 = 600 ms, i.e. exactly 12 polls of 50 ms each -- because this /// predecessor's death was never certified, only observed. From b7e424247ab997ae2b0d3c06853e10a2c445f2a5 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 15 Oct 2026 00:22:27 +0200 Subject: [PATCH 3/3] Cap the CAS mount lease a holder declares at 10 minutes Observers wait for the lease a holder declares, so an hour-long `cas_mount_lease_ttl_ms` would block reclaim of a silently dead member for an hour. Clamp the configured value to 600000 ms when the settings load, with one warning naming the configured and effective values. The renewal deadline, the slot's `expires_at_ms` and the observer's own threshold all read the clamped setting. The observer side does not clamp what a holder declares. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../cas/architecture/mounts-and-leases.md | 2 +- docs/en/antalya/cas/configuration.md | 2 +- .../ContentAddressedSettings.cpp | 12 ++++++++++ src/Disks/tests/gtest_ca_wiring.cpp | 21 ++++++++++++++++++ src/Disks/tests/gtest_cas_mount.cpp | 13 +++++++++++ src/Disks/tests/gtest_cas_settings.cpp | 22 +++++++++++++++++++ 6 files changed, 70 insertions(+), 2 deletions(-) diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index 581ea65aafcd4..50177395653d4 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -130,7 +130,7 @@ anchor plus the TTL, and wall-clock `now` stays audit-only. `cas_mount_lease_ttl_ms` is local to each server. It is how long this server may write after its last confirmed renewal, and every other server waits at least this long before fencing or reclaiming its -slot. Mixed values are safe, and so is changing the value one server at a time in any order. A longer +slot. A configured value above 10 minutes (600000 ms) is clamped to 10 minutes with a warning at startup, so no holder declares a longer lease. Mixed values are safe, and so is changing the value one server at a time in any order. A longer lease on one server delays recovery after that server's crash, and the GC leader logs one `WARNING` per slot whose declared lease differs from its own. A server built without this rule waits only its own lease: keep the values equal until every server of the pool runs a build with the rule. diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index f411ea47fb3ff..7119a9f790da5 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -96,7 +96,7 @@ entirely before release. Treat this table as a snapshot of the current build, no | `cas_blob_hash` | `cityhash128` | Pool blob content-hash function (`cityhash128` \| `xxh3-128` \| `sha256`). Recorded in the pool at creation; a mismatching config is refused at mount | | `cas_blob_hash_allow_new` | `false` | Explicit opt-in to admit a new hash algorithm into an existing pool. One-way: once admitted, the pool carries both algorithms permanently | | `skip_access_check` | `false` | Skip the boot-time capability probe (start now, fix later). Only the preflight probe is skipped — the conditional-write correctness check still runs on every writable mount. **Not available on a writable generation-token (GCS) disk**, which refuses to mount with it: there, the probe battery is the only proof that a token-exact delete carries its generation precondition. Mount such a disk read-only if you need to defer the check | -| `cas_mount_lease_ttl_ms` | `30000` | How long this server may write after its last confirmed renewal; every other server waits at least this long before fencing or reclaiming its slot (≥ 1). Lower values shorten recovery after this server's crash but reduce tolerance for object-storage and scheduling delays | +| `cas_mount_lease_ttl_ms` | `30000` | How long this server may write after its last confirmed renewal; every other server waits at least this long before fencing or reclaiming its slot (≥ 1, at most 600000 — a larger value is clamped to 10 minutes with a warning at startup, so no holder declares a longer lease). Lower values shorten recovery after this server's crash but reduce tolerance for object-storage and scheduling delays | | `cas_mount_renew_period_ms` | `10000` | Milliseconds between background mount-lease renewals (≥ 1). It must leave enough time for two attempt envelopes (a renewal write and its settlement read) and the lease safety margin before the TTL expires: `period + 2 × envelope + margin < TTL` | | `cas_gc_snapshot_generations_to_keep` | `3` | GC snapshot generations retained | | `cas_gc_shards` | `1` | Blob-hash-prefix reducer shards (≥ 1). Recorded in the pool at creation; a mismatching config is refused at mount | diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp index 63e3fbd70b146..81200593fe09f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp @@ -28,6 +28,10 @@ namespace ErrorCodes namespace { +/// Observers wait for the lease a holder declares, so a long lease would block reclaim of a silently +/// dead member for that long. The configured TTL is clamped here, before anything derives from it. +constexpr UInt64 max_mount_lease_ttl_ms = 600'000; + /// Poco renders repeated XML elements as `name`, `name[1]`, `name[2]` -- the convention /// `StorageURL` and `HTTPDictionarySource` both handle when reading repeated `
` elements. /// A key of ours that appears twice must be recognised by its base name rather than passed over as @@ -252,6 +256,14 @@ void ContentAddressedSettings::validate() "content_addressed disk: cas_mount_lease_ttl_ms must be >= 1 (got {})", settings[ContentAddressedSetting::mount_lease_ttl_ms].value); + if (settings[ContentAddressedSetting::mount_lease_ttl_ms] > max_mount_lease_ttl_ms) + { + LOG_WARNING(getLogger("ContentAddressedSettings"), + "content_addressed disk: cas_mount_lease_ttl_ms={} exceeds the maximum of {} ms; using {} ms", + settings[ContentAddressedSetting::mount_lease_ttl_ms].value, max_mount_lease_ttl_ms, max_mount_lease_ttl_ms); + settings[ContentAddressedSetting::mount_lease_ttl_ms] = max_mount_lease_ttl_ms; + } + if (settings[ContentAddressedSetting::mount_renew_period_ms] == 0) throw Exception(ErrorCodes::BAD_ARGUMENTS, "content_addressed disk: cas_mount_renew_period_ms must be >= 1 (got {})", diff --git a/src/Disks/tests/gtest_ca_wiring.cpp b/src/Disks/tests/gtest_ca_wiring.cpp index 098be83ff5ceb..d3ad411760c44 100644 --- a/src/Disks/tests/gtest_ca_wiring.cpp +++ b/src/Disks/tests/gtest_ca_wiring.cpp @@ -347,6 +347,7 @@ namespace DB::ErrorCodes namespace DB::ContentAddressedSetting { extern const ContentAddressedSettingsString staging_backend; + extern const ContentAddressedSettingsUInt64 mount_lease_ttl_ms; } namespace @@ -3081,3 +3082,23 @@ TEST(CASWiring, TryFromDiskOnLocalDiskIsExceptionFreeAndCountsNoError) "a NOT_IMPLEMENTED exception — it runs per disk on every asynchronous-metrics tick"; std::filesystem::remove_all(tmp); } + +/// A pool mounted from a 1-hour `cas_mount_lease_ttl_ms` declares the clamped 10 minutes in its slot. +TEST(CASWiringMount, OverlongConfiguredLeaseIsDeclaredClamped) +{ + auto settings = DB::Cas::tests::makeSettingsForTest( + "test", std::filesystem::temp_directory_path() / "ca_wiring_clamp_scratch"); + settings[DB::ContentAddressedSetting::mount_lease_ttl_ms] = 3'600'000; + settings.validate(); + auto storage = std::make_shared( + DB::Cas::tests::makeLocalObjectStorageForTest(), "pool", "srv1", "", nullptr, settings); + storage->startup(); + + const auto pool = storage->store(); + EXPECT_EQ(pool->poolConfig().mount_lease_ttl_ms.count(), 600'000); + DB::Cas::tests::OperationForTest op(*pool->poolBackendPtr()); + const auto mounts = DB::Cas::listMounts(*op, pool->layout(), /*now_ms=*/0, /*skew_margin_ms=*/0); + ASSERT_EQ(mounts.size(), 1u); + EXPECT_GT(mounts[0].lease.started_at_ms, 0u); + EXPECT_EQ(mounts[0].lease.expires_at_ms - mounts[0].lease.started_at_ms, 600'000u); +} diff --git a/src/Disks/tests/gtest_cas_mount.cpp b/src/Disks/tests/gtest_cas_mount.cpp index 2e7735b95ad6a..5c248e3b8ee70 100644 --- a/src/Disks/tests/gtest_cas_mount.cpp +++ b/src/Disks/tests/gtest_cas_mount.cpp @@ -2067,6 +2067,19 @@ TEST(CASHeartbeatFloor, HolderDeclaringALongerLeaseIsWatchedForThatLease) EXPECT_EQ(floorAt(ops, l, 64'000, obs, warned).fenced_now, 1u); } +/// A 10 minute declared lease (the longest a new holder declares) against the observer's 30 s: 600 s + 30 s + the 1 s step. +TEST(CASHeartbeatFloor, HolderDeclaringTheMaximumLeaseIsFencedAtItsThresholdNotLater) +{ + Ops ops(std::make_shared()); + Layout l("p"); + seedMount(ops.op, l, "s1", /*expires*/ kNowMs + 600'000, /*fenced*/ false, /*min_active_build_sequence*/ 0); + MountObservationMap obs; + std::set warned; + floorAt(ops, l, 0, obs, warned); + EXPECT_EQ(floorAt(ops, l, 630'999, obs, warned).fenced_now, 0u); + EXPECT_EQ(floorAt(ops, l, 631'000, obs, warned).fenced_now, 1u); +} + /// A shorter declared lease, a missing start stamp, equal stamps, an expiry before the start. TEST(CASHeartbeatFloor, ShorterOrUnusableDeclaredLeaseIsWatchedForTheObserversOwn) { diff --git a/src/Disks/tests/gtest_cas_settings.cpp b/src/Disks/tests/gtest_cas_settings.cpp index a362d96e67cf8..b94d18c4af825 100644 --- a/src/Disks/tests/gtest_cas_settings.cpp +++ b/src/Disks/tests/gtest_cas_settings.cpp @@ -26,6 +26,7 @@ namespace DB::ContentAddressedSetting extern const ContentAddressedSettingsUInt64 gc_shards; extern const ContentAddressedSettingsUInt64 gc_interval_sec; extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; + extern const ContentAddressedSettingsUInt64 mount_lease_ttl_ms; extern const ContentAddressedSettingsString scratch_path; extern const ContentAddressedSettingsBool unsafe_remount_no_delay; } @@ -611,3 +612,24 @@ TEST(CASContentAddressedSettings, UnsafeRemountNoDelayIsOffByDefault) EXPECT_TRUE(s[ContentAddressedSetting::unsafe_remount_no_delay].value); } } + +TEST(CASContentAddressedSettings, MountLeaseTtlAboveTheMaximumIsClampedWithOneWarning) +{ + ScopedCasSettingsLogCapture capture; + auto cfg = makeConfig("srv13600000"); + ContentAddressedSettings s; + s.loadFromConfig(*cfg, "disk", "/data", "/data/scratch", identity_macros); + EXPECT_EQ(s[ContentAddressedSetting::mount_lease_ttl_ms].value, 600'000u); + const auto log = capture.captured(); + EXPECT_EQ(countOccurrences(log, "cas_mount_lease_ttl_ms=3600000 exceeds the maximum of 600000 ms; using 600000 ms"), 1u) << log; +} + +TEST(CASContentAddressedSettings, MountLeaseTtlAtTheMaximumIsKeptSilently) +{ + ScopedCasSettingsLogCapture capture; + auto cfg = makeConfig("srv1600000"); + ContentAddressedSettings s; + s.loadFromConfig(*cfg, "disk", "/data", "/data/scratch", identity_macros); + EXPECT_EQ(s[ContentAddressedSetting::mount_lease_ttl_ms].value, 600'000u); + EXPECT_EQ(countOccurrences(capture.captured(), "exceeds the maximum"), 0u) << capture.captured(); +}