From 22f20e5c639e4daf6afebc60bb87224d22e89ceb Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 16 Oct 2026 02:35:09 +0200 Subject: [PATCH 1/7] Rename the CAS farewell request plane to the lease plane The plane already carries the claim and the remount re-anchor, and the background renewal moves onto it next; "farewell" names one of its users. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../ContentAddressed/Pool/CasMountRuntime.cpp | 6 +++--- .../ContentAddressed/Pool/CasMountRuntime.h | 9 +++++---- .../ContentAddressed/Pool/CasPool.cpp | 10 +++++----- .../ContentAddressed/Pool/CasPool.h | 16 +++++++--------- .../ContentAddressed/Tools/CasDecommission.cpp | 2 +- 5 files changed, 21 insertions(+), 22 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp index b073de81f7d2c..0a6667b728e65 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp @@ -55,7 +55,7 @@ int64_t wallClockNowSeconds() CasMountRuntime::CasMountRuntime( BackendPtr backend_ptr_, CasRequests & mount_requests_, - CasRequests & farewell_requests_, + CasRequests & lease_requests_, const Layout & layout_, MountConfig config_, String server_root_id_, @@ -64,7 +64,7 @@ CasMountRuntime::CasMountRuntime( std::function remount_attempt_) : backend_ptr(std::move(backend_ptr_)) , mount_requests(mount_requests_) - , farewell_requests(farewell_requests_) + , lease_requests(lease_requests_) , layout(layout_) , config(std::move(config_)) , server_root_id(std::move(server_root_id_)) @@ -369,7 +369,7 @@ void CasMountRuntime::installRenewer( const std::function & now_ms) { auto replacement = std::make_unique( - mount_requests, farewell_requests, layout, server_root_id, our_uuid, writer_epoch, + mount_requests, lease_requests, layout, server_root_id, our_uuid, writer_epoch, config.mount_lease_ttl_ms, now_ms, [this] { return minActive(); }, [this](CasEvent e) { emitEvent(std::move(e)); }, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h index 04ec8491f84b7..b0a6d25fb8e7c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h @@ -155,10 +155,11 @@ class CasMountRuntime public: CasMountRuntime( BackendPtr backend_ptr_, - /// The two planes the `MountLeaseRenewer` runs on: renewals under the mount fence, the farewell - /// on an open one. Owned by `Pool` and outliving this runtime. + /// The two planes the `MountLeaseRenewer` runs on: bounded renewals under the mount fence; the + /// claim, the remount re-anchor and the farewell on the open-fence lease plane. Owned by `Pool` + /// and outliving this runtime. CasRequests & mount_requests_, - CasRequests & farewell_requests_, + CasRequests & lease_requests_, const Layout & layout_, MountConfig config_, String server_root_id_, @@ -468,7 +469,7 @@ class CasMountRuntime /// ---- injected environment (no `Pool` back-reference); initialized first, in this order ---- BackendPtr backend_ptr; CasRequests & mount_requests; - CasRequests & farewell_requests; + CasRequests & lease_requests; const Layout & layout; MountConfig config; String server_root_id; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index 83d21a39500d7..e65c8f56d9fc0 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -183,7 +183,7 @@ Pool::Pool(BackendPtr backend_, PoolConfig config_, PoolMeta meta_) config.boot_ms_fn, config.retry_sleep_fn ? config.retry_sleep_fn : mountPlaneSleepFn(), &hot_keys) - , farewell_requests(pool_backend, Fence::open(), config.boot_ms_fn, config.retry_sleep_fn, &hot_keys) + , lease_requests(pool_backend, Fence::open(), config.boot_ms_fn, config.retry_sleep_fn, &hot_keys) /// The open plane's fence is the pool's teardown flag: generation 0 forever, exactly like /// `Fence::open`, but `admit` refuses once `beginTeardown` ran. A GC round, an FSCK or a probe in /// flight is then refused at its next request instead of running to completion under a disk that @@ -242,13 +242,13 @@ Pool::Pool(BackendPtr backend_, PoolConfig config_, PoolMeta meta_) [this] (const RootNamespace & ns) { cancelInflightBuildsForNamespace(ns); }, config.recovery_pre_first_request_hook_for_test) /// Mount / write-fence / build-watermark / self-remount runtime. Injected with - /// backend/layout + the mount and farewell planes + the `MountConfig` slice + `server_root_id` + the event-sink reference + the pool + /// backend/layout + the mount and lease planes + the `MountConfig` slice + `server_root_id` + the event-sink reference + the pool /// `cas_request_budget` + the `remount_attempt` callback (== `Pool::tryRemountOnce`, whose claim/ /// recovery ORCHESTRATION stays on Pool). The callback captures `this`; it is invoked only at runtime /// (post-construction). Declared/constructed AFTER `ref_ledger`, preserving the original member order /// verbatim (mount destroyed first, ledger last; both orders proven safe -- see the header note). , mount_runtime( - pool_backend, mount_requests, farewell_requests, + pool_backend, mount_requests, lease_requests, pool_layout, config.mountConfig(), config.server_root_id, event_sink_, config.cas_request_budget, [this] { return tryRemountOnce(); }) @@ -1948,7 +1948,7 @@ void Pool::setCasRetrySleepForTest(std::function sleep_fn) { /// All three planes, not just the ledger's: a test that replaces the retry sleep must not be left /// with a real one on the plane the site under test happens to use. - farewell_requests.setSleepFnForTest(sleep_fn); + lease_requests.setSleepFnForTest(sleep_fn); ref_ledger.setCasRetrySleepForTest(sleep_fn); /// `CasRequests` falls back to the engine's plain sleep for an empty argument -- which is neither /// the mount plane's nor the open plane's default. Re-install both, so clearing the seam cannot @@ -1960,7 +1960,7 @@ void Pool::setCasRetrySleepForTest(std::function sleep_fn) void Pool::setCasRequestNowFnForTest(std::function now_fn) { mount_requests.setNowFnForTest(now_fn); - farewell_requests.setNowFnForTest(now_fn); + lease_requests.setNowFnForTest(now_fn); gc_requests.setNowFnForTest(std::move(now_fn)); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index 0a678e067e2cc..d95e675510c38 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -256,10 +256,10 @@ struct PoolConfig /// boot clock (`Pool::bootMs`); injected by tests to drive the fence deadline deterministically. std::function boot_ms_fn = {}; - /// The inter-attempt sleep for the mount, farewell and GC request planes (`mount_requests`, - /// `farewell_requests`, `gc_requests`), installed at their CONSTRUCTION -- before this `Pool` has + /// The inter-attempt sleep for the mount, lease and GC request planes (`mount_requests`, + /// `lease_requests`, `gc_requests`), installed at their CONSTRUCTION -- before this `Pool` has /// claimed or read anything. Empty = each plane's own production default (an interruptible real - /// sleep for the mount and GC planes, `CasRequests`'s own real sleep for the farewell plane). A test + /// sleep for the mount and GC planes, `CasRequests`'s own real sleep for the lease plane). A test /// that also freezes `boot_ms_fn` must supply a matching sleep here: a retry loop bound to a clock /// that only moves when this function is called would otherwise retry forever against a REAL sleep /// that never calls it, because the deadline it measures against never appears to elapse. @@ -760,11 +760,9 @@ class Pool : public std::enable_shared_from_this /// operation admitted here is refused the moment the fence trips, is re-armed under a fresh lease /// incarnation, or runs out of room before the lease expires. CasRequests & mountRequests() { return mount_requests; } - /// The farewell plane, on an open fence -- shared with the mount-lease renewer's own claim/adopt, - /// not only its release: a self-remount claims with the fence already latched lost, so gating the - /// claim on the fence could never reclaim, and refusing the farewell because the mount fence has - /// already run down would leave the slot looking live until GC fences it out. - CasRequests & farewellRequests() { return farewell_requests; } + /// The lease plane, on an open fence: the renewer's claim and adoption, the remount re-anchor and + /// the farewell. None of them may be refused for want of the lease they establish or hand back. + CasRequests & leaseRequests() { return lease_requests; } /// The open-fence plane: GC, the offline tools, this pool's own reads, and the bootstrap-control /// claims. None of them hold a mount lease -- the claims are what ESTABLISHES one, so gating them /// on the fence would make a self-remount, which runs with the fence latched lost, unable ever to @@ -1208,7 +1206,7 @@ class Pool : public std::enable_shared_from_this /// config they take their clock from. `mutable` because issuing a request is not a change to the /// pool: a `const` observer still has to read the store. mutable CasRequests mount_requests; - mutable CasRequests farewell_requests; + mutable CasRequests lease_requests; mutable CasRequests gc_requests; std::shared_ptr detached_work = std::make_shared(); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp index eef2ba04d9bf4..19339638ecc59 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp @@ -173,7 +173,7 @@ DecommissionReport decommissionPoolMember(BackendPtr backend, PoolConfig config, if (drain_now_fn && drain_sleep_fn) { /// Re-affirms the same values `config` above already installed on `mount_requests`/ - /// `farewell_requests`/`gc_requests` at construction, and additionally wires `ref_ledger`'s own + /// `lease_requests`/`gc_requests` at construction, and additionally wires `ref_ledger`'s own /// retry sleep, which has no construction-time seam of its own. `sweepNamespace` below issues /// its deletes on `admin`'s own GC plane. admin->setCasRequestNowFnForTest(drain_now_fn); From c057500ccf3ff08db9df4a6e4a36bd925e7ec1d7 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 16 Oct 2026 02:35:09 +0200 Subject: [PATCH 2/7] Add a CAS retry policy with no deadline and spaced reissues `Retry::untilDefinitive` has no window and no lease bound. Its reissues wait a draw from [0.8, 1.2] of 1000 ms minus what the failed attempt took (`Retry::pauseAfterAttempt`) instead of the growing backoff, so the first good second after an outage ends a retry series. Existing policies are unchanged. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../ContentAddressed/Backend/CasRequests.cpp | 17 +- .../ContentAddressed/Backend/CasRequests.h | 20 ++- .../ContentAddressed/Backend/CasRetry.cpp | 16 ++ .../ContentAddressed/Backend/CasRetry.h | 24 +++ src/Disks/tests/gtest_cas_requests.cpp | 150 ++++++++++++++++++ 5 files changed, 222 insertions(+), 5 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp index 6ea453eb1005e..9e96b1a33efba 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp @@ -893,14 +893,23 @@ std::optional CasOperation::gatedPause(uint64_t pause_ms, uint32_t return std::nullopt; } +uint64_t CasOperation::reissuePause(const WriteState & state, uint64_t unspaced_ms) const +{ + if (!state.spacing) + return unspaced_ms; + return Retry::pauseAfterAttempt(state.spacing->spacing_ms, state.spacing->attempt_started_ms, owner.now_ms()); +} + std::optional CasOperation::pauseAndReissue(WriteState & state, const Retry::Bound & bound) { - return gatedPause(Retry::backoff(++state.reissues), 2, state, bound, detail::recordReissue, /*should_sleep=*/true); + const uint64_t pause_ms = reissuePause(state, Retry::backoff(++state.reissues)); + return gatedPause(pause_ms, 2, state, bound, detail::recordReissue, /*should_sleep=*/true); } std::optional CasOperation::pauseForConflict(WriteState & state, const Retry::Bound & bound) { - return gatedPause(Retry::conflictBackoff(), 2, state, bound, detail::recordConflictPause, /*should_sleep=*/true); + return gatedPause(reissuePause(state, Retry::conflictBackoff()), 2, state, bound, detail::recordConflictPause, + /*should_sleep=*/true); } /// A flat pause before reissuing an attempt whose failure text named a failed connection. @@ -908,7 +917,7 @@ static constexpr uint64_t kConnectHintPauseMs = 50; std::optional CasOperation::pauseFlat(WriteState & state, const Retry::Bound & bound) { - return gatedPause(kConnectHintPauseMs, 2, state, bound, detail::recordReissue, /*should_sleep=*/true); + return gatedPause(reissuePause(state, kConnectHintPauseMs), 2, state, bound, detail::recordReissue, /*should_sleep=*/true); } std::optional CasOperation::reissueAtOnce(WriteState & state, const Retry::Bound & bound) @@ -943,6 +952,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co if (!fits(reservation, bound)) return gaveUp(GaveUp::Why::Deadline, sourceFor(bound), state); + if (policy.attempt_spacing_ms) + state.spacing = WriteState::Spacing{*policy.attempt_spacing_ms, owner.now_ms()}; detail::recordAttempt(); ++state.attempts_sent; state.sent_any = true; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h index 2509077b47907..732e1c8fd554d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h @@ -354,6 +354,13 @@ class CasOperation Observation last_seen = NotObserved{}; uint32_t reissues = 0; bool refresh_attempted = false; + /// Under `Retry::attempt_spacing_ms`: that spacing, and when the latest attempt was sent. + struct Spacing + { + uint64_t spacing_ms = 0; + uint64_t attempt_started_ms = 0; + }; + std::optional spacing; }; /// Why a read-class request stopped without an answer. Every give-up below throws the same @@ -419,7 +426,8 @@ class CasOperation std::optional gatedPause(uint64_t pause_ms, uint32_t envelopes, WriteState & state, const Retry::Bound & bound, void (*record)(), bool should_sleep); /// Admission, then the jittered sleep. A value means the call ended during it; nullopt means the - /// caller may send another attempt. + /// caller may send another attempt. Under `WriteState::spacing` this and the two siblings below + /// sleep `reissuePause` instead. std::optional pauseAndReissue(WriteState & state, const Retry::Bound & bound); /// The sibling for a clean lost race: the same admission and the same reservation, a flat /// `Retry::conflictBackoff` sleep, and `state.reissues` untouched, so a transport fault that follows @@ -432,6 +440,8 @@ class CasOperation /// sleep at all, and `state.reissues` untouched -- the fuse is a connection-quality answer about a /// fresh connection, not a store fault, so nothing here is paced against it. std::optional reissueAtOnce(WriteState & state, const Retry::Bound & bound); + /// The pause before a write reissue: `unspaced_ms`, or `Retry::pauseAfterAttempt` under `state.spacing`. + uint64_t reissuePause(const WriteState & state, uint64_t unspaced_ms) const; /// `sleep_ms` plus `envelopes` attempt reservations, saturating. uint64_t reservedFor(uint64_t sleep_ms, uint32_t envelopes) const; @@ -467,6 +477,7 @@ auto CasOperation::readLoop(std::string_view verb, const String & subject, const const Retry::Bound & bound, Fn && once) { bool refresh_attempted = false; + uint64_t attempt_started_ms = 0; /// Two counters, deliberately kept separate: `attempt_no` is the PHYSICAL attempt count handed to /// the transport (so a reissue is seen as attempt >= 2); `ordinary_reissues` is the /// exponential-backoff index. They advance together on an ordinary failure, but the first-attempt @@ -484,6 +495,8 @@ auto CasOperation::readLoop(std::string_view verb, const String & subject, const if (!fits(reservation, bound)) giveUpReadDeadline(verb, subject, bound, attempt_no - 1); + if (policy.attempt_spacing_ms) + attempt_started_ms = owner.now_ms(); detail::recordAttempt(); try { @@ -526,7 +539,10 @@ auto CasOperation::readLoop(std::string_view verb, const String & subject, const } } - const uint64_t pause_ms = Retry::backoff(++ordinary_reissues); + ++ordinary_reissues; + const uint64_t pause_ms = policy.attempt_spacing_ms + ? Retry::pauseAfterAttempt(*policy.attempt_spacing_ms, attempt_started_ms, owner.now_ms()) + : Retry::backoff(ordinary_reissues); const uint64_t needed = reservedFor(pause_ms, 1); switch (gate(needed)) { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.cpp index c7b4c27215b53..f7dfde7d3c905 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.cpp @@ -17,6 +17,22 @@ uint64_t Retry::backoff(uint32_t attempt) return thread_local_rng() % (ceiling + 1); /// full jitter: uniform(0, ceiling) } +uint64_t Retry::spacedPause(uint64_t spacing_draw_ms, uint64_t attempt_started_ms, uint64_t now_ms) +{ + const uint64_t taken_ms = now_ms > attempt_started_ms ? now_ms - attempt_started_ms : 0; + return taken_ms >= spacing_draw_ms ? 0 : spacing_draw_ms - taken_ms; +} + +uint64_t Retry::drawSpacing(uint64_t attempt_spacing_ms_) +{ + const uint64_t spread = attempt_spacing_ms_ / 5; + const uint64_t low = attempt_spacing_ms_ - spread; + const uint64_t high = attempt_spacing_ms_ > std::numeric_limits::max() - spread + ? std::numeric_limits::max() + : attempt_spacing_ms_ + spread; + return low + thread_local_rng() % (high - low + 1); +} + Retry::Bound Retry::bind(uint64_t now_ms) const { const uint64_t own_deadline_ms = policy_deadline_ms diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.h index de235e8fd3faf..86336201c53e2 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.h @@ -1,5 +1,6 @@ #pragma once #include +#include #include namespace DB::Cas @@ -23,6 +24,12 @@ struct Retry /// iterations a fresh window. Empty for a single verb, which gets its window from where it is /// called. std::optional policy_deadline_ms = std::nullopt; + /// When set, a reissue in `CasOperation`'s write and read loops waits until a draw from [0.8, 1.2] + /// of this has passed since the failed attempt started, instead of the growing backoff. A + /// first-attempt fuse timeout is still reissued at once. + std::optional attempt_spacing_ms = std::nullopt; + + static constexpr uint64_t kDefinitiveAttemptSpacingMs = 1'000; /// Full jitter: uniform(0, min(5000, 200 << (attempt-1))) milliseconds. `attempt` is 1-based; /// `attempt == 0` returns 0. @@ -53,6 +60,23 @@ struct Retry } /// The standard policy, but at most one attempt is ever sent. static Retry once() { return {.window_ms = 90'000, .lease_deadline_ms = std::nullopt, .single_attempt = true}; } + /// No window and no lease bound: the operation ends on a definitive answer, or when the fence or the + /// caller's liveness refuses. Reissues are spaced by `kDefinitiveAttemptSpacingMs`. + static Retry untilDefinitive() + { + return {.window_ms = std::numeric_limits::max(), .lease_deadline_ms = std::nullopt, + .single_attempt = false, .attempt_spacing_ms = kDefinitiveAttemptSpacingMs}; + } + /// `spacing_draw_ms` minus what the failed attempt took, not below 0. A sample before the start + /// counts as no time taken. + static uint64_t spacedPause(uint64_t spacing_draw_ms, uint64_t attempt_started_ms, uint64_t now_ms); + /// A uniform draw in [0.8, 1.2] of `attempt_spacing_ms_`, saturating at the top of the range. + static uint64_t drawSpacing(uint64_t attempt_spacing_ms_); + /// The pause before the next attempt under `attempt_spacing_ms`: `spacedPause` of a fresh draw. + static uint64_t pauseAfterAttempt(uint64_t attempt_spacing_ms_, uint64_t attempt_started_ms, uint64_t now_ms) + { + return spacedPause(drawSpacing(attempt_spacing_ms_), attempt_started_ms, now_ms); + } /// This policy, made single-attempt. A frozen loop policy keeps its absolute deadline through it, /// which is what lets a loop send an unrepeatable request under the same bound as the rest. diff --git a/src/Disks/tests/gtest_cas_requests.cpp b/src/Disks/tests/gtest_cas_requests.cpp index 31383f78c117b..73a5cc2c5b45f 100644 --- a/src/Disks/tests/gtest_cas_requests.cpp +++ b/src/Disks/tests/gtest_cas_requests.cpp @@ -34,6 +34,7 @@ #include +#include #include #include #include @@ -610,6 +611,51 @@ TEST(CASRetry, BindSaturatesAndLeavesAnEqualLeaseOffTheLeaseSource) EXPECT_TRUE(lease.lease_bound); } +TEST(CASRetrySpacing, UntilDefinitiveHasNoDeadlineAndSpacesItsAttempts) +{ + constexpr uint64_t largest = std::numeric_limits::max(); + const Retry policy = Retry::untilDefinitive(); + EXPECT_FALSE(policy.lease_deadline_ms.has_value()); + EXPECT_FALSE(policy.single_attempt); + EXPECT_EQ(policy.attempt_spacing_ms, std::optional(1'000)); + for (const uint64_t now : {uint64_t{0}, uint64_t{1'000'000}, largest - 1, largest}) + { + const Retry::Bound bound = policy.bind(now); + EXPECT_EQ(bound.deadline_ms, largest) << "now " << now; + EXPECT_FALSE(bound.lease_bound) << "now " << now; + } + EXPECT_FALSE(Retry::standard().attempt_spacing_ms.has_value()); + EXPECT_FALSE(Retry::untilLeaseSafe(2'000'000, 2'000).attempt_spacing_ms.has_value()); +} + +TEST(CASRetrySpacing, SpacedPauseWaitsOutTheRestOfTheDraw) +{ + EXPECT_EQ(Retry::spacedPause(1'000, 5'000, 5'000), 1'000u); + EXPECT_EQ(Retry::spacedPause(1'000, 5'000, 5'300), 700u); + EXPECT_EQ(Retry::spacedPause(1'000, 5'000, 6'000), 0u); + EXPECT_EQ(Retry::spacedPause(1'000, 5'000, 10'000), 0u); + EXPECT_EQ(Retry::spacedPause(1'000, 5'000, 4'000), 1'000u); +} + +TEST(CASRetrySpacing, DrawIsWithinAFifthOfTheSpacing) +{ + bool low = false; + bool high = false; + for (int i = 0; i < 2'000; ++i) + { + const uint64_t draw = Retry::drawSpacing(1'000); + ASSERT_GE(draw, 800u); + ASSERT_LE(draw, 1'200u); + low = low || draw < 850; + high = high || draw > 1'150; + } + /// The bounds alone cannot tell a draw from a constant. + EXPECT_TRUE(low); + EXPECT_TRUE(high); + constexpr uint64_t largest = std::numeric_limits::max(); + EXPECT_GE(Retry::drawSpacing(largest), largest - largest / 5) << "the top of the range must not wrap"; +} + TEST(CASRequests, CreateThenReplaceThenRemove) { FakeClock clock; @@ -723,6 +769,110 @@ TEST(CASRequests, AmbiguousCreateThatNeverLandedIsReissued) EXPECT_EQ(clock.sleeps.size(), 1u); } +namespace +{ + +/// Each guarded write takes `write_ms` on the injected clock and, while `failing` is positive, throws a +/// transport timeout after it. +class SlowFailingWrites final : public CountingBackend +{ +public: + explicit SlowFailingWrites(FakeClock & clock_) : clock(clock_) {} + + std::expected write(const String & key, const String & bytes, + const std::optional & expected_value, TransportAccess & access) override + { + if (expected_value) + { + clock.now += write_ms; + if (failing > 0) + { + --failing; + throw Poco::TimeoutException("injected write timeout"); + } + } + return CountingBackend::write(key, bytes, expected_value, access); + } + + uint64_t write_ms = 0; + int failing = 0; + +private: + FakeClock & clock; +}; + +} + +/// Unclear writes are reissued past the standard 90 s window, each a spacing draw after the previous +/// attempt started, and the read that settles each one is sent at once. +TEST(CASRequestsSpacing, UnclearWritesAreReissuedPastAnyWindowOneSpacingApart) +{ + FakeClock clock; + auto backend = std::make_shared(clock); + auto requests = makeRequests(backend, clock); + auto op = requests.admit(); + const Etag seen = *orThrow(op.create("k", "v1", Retry::standard()), "seed"); + constexpr int failures = 120; + backend->write_ms = 300; + backend->failing = failures; + const uint64_t gets_before = backend->getTotal(); + const uint64_t start = clock.now; + + const WriteResult result = op.replace("k", "v2", seen, Retry::untilDefinitive()); + + const auto * committed = std::get_if(&result); + ASSERT_NE(committed, nullptr); + EXPECT_EQ(committed->attempts_sent, failures + 1u); + EXPECT_GT(clock.now - start, 90'000u); + ASSERT_EQ(clock.sleeps.size(), static_cast(failures)); + for (const uint64_t pause : clock.sleeps) + { + EXPECT_GE(pause, 500u); + EXPECT_LE(pause, 900u); + } + EXPECT_EQ(backend->getTotal() - gets_before, static_cast(failures)) << "one settling read per unclear attempt"; +} + +TEST(CASRequestsSpacing, ReadRetriesAreSpaced) +{ + FakeClock clock; + auto backend = std::make_shared(); + auto requests = makeRequests(backend, clock); + auto op = requests.admit(); + (void)orThrow(op.create("k", "v", Retry::standard()), "seed"); + backend->failNextReadWith("k", std::make_exception_ptr(Poco::TimeoutException("injected read failure"))); + backend->failNextReadWith("k", std::make_exception_ptr(Poco::TimeoutException("injected read failure"))); + + const auto object = op.read("k", Retry::untilDefinitive()); + + ASSERT_TRUE(object.has_value()); + ASSERT_EQ(clock.sleeps.size(), 2u); + for (const uint64_t pause : clock.sleeps) + { + EXPECT_GE(pause, 800u); + EXPECT_LE(pause, 1'200u); + } +} + +/// Golden: a policy without spacing keeps the growing backoff. +TEST(CASRequestsSpacing, AStandardPolicyKeepsItsGrowingBackoff) +{ + FakeClock clock; + auto backend = std::make_shared(clock); + auto requests = makeRequests(backend, clock); + auto op = requests.admit(); + const Etag seen = *orThrow(op.create("k", "v1", Retry::standard()), "seed"); + backend->failing = 3; + + const WriteResult result = op.replace("k", "v2", seen, Retry::standard()); + + ASSERT_TRUE(std::holds_alternative(result)); + ASSERT_EQ(clock.sleeps.size(), 3u); + EXPECT_LE(clock.sleeps[0], 200u); + EXPECT_LE(clock.sleeps[1], 400u); + EXPECT_LE(clock.sleeps[2], 800u); +} + /// The engine's own attempt number reaches the transport through `TransportAccess::attemptNo()`, for /// every primitive -- write, read (the resolve read is its own call, with its own attempt count) and /// list. From fb57c5e7d9c4fa77ac229e7211f9c4abfe4d906d Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 16 Oct 2026 02:36:09 +0200 Subject: [PATCH 3/7] Renew the CAS mount lease until a definitive answer The background renewal gave up at the lease deadline, fenced the mount and remounted under a fresh epoch, although the store had not answered and nothing had touched the slot. It now keeps one tuple and retries it on the lease plane under `Retry::untilDefinitive`: - a refusal waits one renewal period and sends the same body again; - a refused precondition whose read failed repeats only the read; - a commit, a slot that moved, a local failure, a stop or a park ends the series. A stop between engine calls, after a refusal or a read that found the slot unchanged, leaves the renewer active so the farewell still runs; a stop during a call that sent an attempt, or while the reads repeat, is terminal and classified as cancelled. A committed deadline is published in `DriverLease::finish`, under the driver mutex, so a remount cannot arm a new epoch between a commit and its deadline. The lease plane sleeps interruptibly on a stop or a park, like the mount plane. The startup and remount re-anchors stay bounded by the lease, and so does a runtime with `bounded_background_renewal`. Closes: https://github.com/Altinity/ClickHouse/issues/2431 Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../ContentAddressed/Pool/CasMountRuntime.cpp | 49 ++++-- .../ContentAddressed/Pool/CasMountRuntime.h | 17 ++- .../ContentAddressed/Pool/CasPool.cpp | 15 +- .../ContentAddressed/Pool/CasPool.h | 10 +- .../ContentAddressed/Pool/CasServerRoot.cpp | 144 +++++++++++++++--- .../ContentAddressed/Pool/CasServerRoot.h | 37 +++-- 6 files changed, 208 insertions(+), 64 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp index 0a6667b728e65..3e0d2ad5d4c40 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp @@ -305,6 +305,11 @@ RenewalDriverState CasMountRuntime::DriverLease::finish( const MountRenewResult * result) { std::lock_guard lock(runtime.driver_mutex); + /// Published before the driver leaves the call, so a remount cannot arm a new epoch between this + /// renewal's commit and its deadline. The renewer cannot change meanwhile: `installRenewer` requires + /// a dormant or parked driver. + if (result && result->outcome == MountRenewOutcome::Committed) + runtime.publishRenewedDeadline(result->attempt_start_boot_ms); if (runtime.workers_stop_requested) runtime.renewal_driver_state = RenewalDriverState::Stopping; else if (active == RenewalDriverState::WorkerCall @@ -429,6 +434,12 @@ RenewalProgress CasMountRuntime::renewalProgress() const MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_call) { + uint64_t refusal_pause_ms = 0; + if (worker_call) + { + std::lock_guard lock(driver_mutex); + refusal_pause_ms = static_cast(std::max(0, renewal_period.count())); + } return MountRenewOperationEnvironment{ .boot_ms = [this] { return bootMsNow(); }, .live = [this, worker_call] @@ -459,6 +470,7 @@ MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_c } }, }, + .refusal_pause_ms = refusal_pause_ms, }; } @@ -483,7 +495,10 @@ bool CasMountRuntime::renewalCancelled() const void CasMountRuntime::sleepInterruptibly(uint64_t ms) { std::unique_lock lock(driver_mutex); - driver_cv.wait_for(lock, std::chrono::milliseconds(ms), [this] { return workers_stop_requested; }); + driver_cv.wait_for(lock, std::chrono::milliseconds(ms), [this] + { + return renewal_driver_state == RenewalDriverState::Stopping || renewal_driver_state == RenewalDriverState::ParkRequested; + }); } void CasMountRuntime::consumeRenewResult( @@ -503,17 +518,7 @@ void CasMountRuntime::consumeRenewResult( && result.deadline_source == GaveUp::Source::Lease) ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalDeadlineExceeded); - if (result.outcome == MountRenewOutcome::Committed) - { - const uint64_t ttl_ms = static_cast(config.mount_lease_ttl_ms.count()); - setMountDeadline( - result.attempt_start_boot_ms > std::numeric_limits::max() - ttl_ms - ? std::numeric_limits::max() - : result.attempt_start_boot_ms + ttl_ms); - reportMountRenewCompletion(result); - return; - } - if (result.outcome == MountRenewOutcome::NotAttempted) + if (result.outcome != MountRenewOutcome::Terminal) { reportMountRenewCompletion(result); return; @@ -543,6 +548,15 @@ void CasMountRuntime::consumeRenewResult( std::rethrow_exception(result.failure); } +void CasMountRuntime::publishRenewedDeadline(uint64_t attempt_start_boot_ms) +{ + const uint64_t ttl_ms = static_cast(config.mount_lease_ttl_ms.count()); + const uint64_t deadline = attempt_start_boot_ms > std::numeric_limits::max() - ttl_ms + ? std::numeric_limits::max() + : attempt_start_boot_ms + ttl_ms; + setMountDeadline(deadline); +} + uint64_t CasMountRuntime::renewRenewerOnce( AdmittedRenewerCall call, RenewalDriverState active, @@ -564,14 +578,19 @@ uint64_t CasMountRuntime::renewRenewerOnce( MountRenewResult result; { SCOPE_EXIT({ std::lock_guard lock(renewal_progress_mutex); renewal_progress.in_progress = false; }); - result = active == RenewalDriverState::RemountCall - ? call.renewer->renewForRemount(renewalEnvironment(worker_call)) - : call.renewer->renew(renewalEnvironment(worker_call)); + if (active == RenewalDriverState::RemountCall) + result = call.renewer->renewForRemount(renewalEnvironment(worker_call)); + else if (worker_call && !config.bounded_background_renewal) + result = call.renewer->renewUntilDefinitive(renewalEnvironment(worker_call)); + else + result = call.renewer->renew(renewalEnvironment(worker_call)); } const RenewalDriverState destination = active == RenewalDriverState::WorkerCall ? RenewalDriverState::WorkerIdle : (active == RenewalDriverState::RemountCall ? RenewalDriverState::Parked : RenewalDriverState::Dormant); const RenewalDriverState returned_state = call.lease->finish(destination, &result); + if (config.renewal_finished_hook_for_test) + config.renewal_finished_hook_for_test(result); if (result.outcome == MountRenewOutcome::Terminal && config.renewal_terminal_deposited_hook_for_test) config.renewal_terminal_deposited_hook_for_test(); consumeRenewResult(result, active, returned_state, propagate_failure); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h index b0a6d25fb8e7c..d6c3ee0f11309 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h @@ -109,6 +109,12 @@ struct MountConfig /// Test-only override of the renewal's liveness predicate, for exact pre/post-send gate /// interleavings. FALSE ends the renewal exactly as a lost fence does. std::function renewal_live_for_test = {}; + /// The decommission tool's pool: the background renewal stays bounded by the lease, as the startup + /// and remount re-anchors are, instead of retrying until a definitive answer. + bool bounded_background_renewal = false; + /// Deterministic test interposition on the renewing thread right after `DriverLease::finish` + /// returned driver ownership, before the result is consumed. Runs without `driver_mutex`. + std::function renewal_finished_hook_for_test = {}; }; /// Local, in-memory write fence. It is deliberately not checked by reading the object store for every @@ -334,10 +340,11 @@ class CasMountRuntime || lifecycle() == PoolLifecycle::IdentityLost; } - /// The inter-attempt sleep the mount plane runs on. A plain sleep would hold a parked or stopping - /// renewal for the whole capped backoff; this one wakes on the same stop signal the workers watch. - /// It shortens a stop, not a fence loss: the fence cannot see a stop request, so a woken operation - /// still reissues unless its own liveness predicate refuses. + /// The inter-attempt sleep the mount and lease planes run on. A plain sleep would hold a parked or + /// stopping renewal for the whole pause; this one wakes while the workers are stopping or a park is + /// requested. It shortens a stop, not a fence loss: the fence cannot see a stop request, so a woken + /// operation still reissues unless its own liveness predicate refuses. Once the workers have + /// stopped it sleeps in full, so the farewell that follows is paced like any other write. void sleepInterruptibly(uint64_t ms); /// The mount fence's admission verdict, as `Fence::admit` expects it: may a request admitted under @@ -454,6 +461,8 @@ class CasMountRuntime RenewalDriverState active_state, RenewalDriverState returned_state, bool propagate_failure); + /// Publish a committed renewal's deadline, its start plus the lease. Requires `driver_mutex`. + void publishRenewedDeadline(uint64_t attempt_start_boot_ms); void renewalLoop(); void remountLoop(); ThreadFromGlobalPool makeWorker(std::function body); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index e65c8f56d9fc0..698681efb3cf6 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -183,7 +183,12 @@ Pool::Pool(BackendPtr backend_, PoolConfig config_, PoolMeta meta_) config.boot_ms_fn, config.retry_sleep_fn ? config.retry_sleep_fn : mountPlaneSleepFn(), &hot_keys) - , lease_requests(pool_backend, Fence::open(), config.boot_ms_fn, config.retry_sleep_fn, &hot_keys) + , lease_requests( + pool_backend, + Fence::open(), + config.boot_ms_fn, + config.retry_sleep_fn ? config.retry_sleep_fn : mountPlaneSleepFn(), + &hot_keys) /// The open plane's fence is the pool's teardown flag: generation 0 forever, exactly like /// `Fence::open`, but `admit` refuses once `beginTeardown` ran. A GC round, an FSCK or a probe in /// flight is then refused at its next request instead of running to completion under a disk that @@ -1948,12 +1953,12 @@ void Pool::setCasRetrySleepForTest(std::function sleep_fn) { /// All three planes, not just the ledger's: a test that replaces the retry sleep must not be left /// with a real one on the plane the site under test happens to use. - lease_requests.setSleepFnForTest(sleep_fn); ref_ledger.setCasRetrySleepForTest(sleep_fn); - /// `CasRequests` falls back to the engine's plain sleep for an empty argument -- which is neither - /// the mount plane's nor the open plane's default. Re-install both, so clearing the seam cannot - /// leave a parked renewal held for a whole capped backoff, or the open plane deaf to a teardown. + /// `CasRequests` falls back to the engine's plain sleep for an empty argument -- which is none of + /// the planes' defaults. Re-install them, so clearing the seam cannot leave a parked renewal held + /// for a whole pause, or the open plane deaf to a teardown. gc_requests.setSleepFnForTest(sleep_fn ? sleep_fn : openPlaneSleepFn()); + lease_requests.setSleepFnForTest(sleep_fn ? sleep_fn : mountPlaneSleepFn()); mount_requests.setSleepFnForTest(sleep_fn ? std::move(sleep_fn) : mountPlaneSleepFn()); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index d95e675510c38..34ee0316c3435 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -258,8 +258,8 @@ struct PoolConfig /// The inter-attempt sleep for the mount, lease and GC request planes (`mount_requests`, /// `lease_requests`, `gc_requests`), installed at their CONSTRUCTION -- before this `Pool` has - /// claimed or read anything. Empty = each plane's own production default (an interruptible real - /// sleep for the mount and GC planes, `CasRequests`'s own real sleep for the lease plane). A test + /// claimed or read anything. Empty = each plane's own production default, an interruptible real + /// sleep. A test /// that also freezes `boot_ms_fn` must supply a matching sleep here: a retry loop bound to a clock /// that only moves when this function is called would otherwise retry forever against a REAL sleep /// that never calls it, because the deadline it measures against never appears to elapse. @@ -1171,9 +1171,9 @@ class Pool : public std::enable_shared_from_this return std::forward(mutation)(); } - /// The mount plane's inter-attempt sleep: interruptible, so a parked or stopping renewal is not - /// held for a whole capped backoff. Named rather than inlined because the test seam has to be able - /// to put it back. + /// The mount and lease planes' inter-attempt sleep: interruptible, so a parked or stopping renewal is + /// not held for a whole pause. Named rather than inlined because the test seam has to be able to put + /// it back. std::function mountPlaneSleepFn() { return [this](uint64_t ms) { mount_runtime.sleepInterruptibly(ms); }; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index 74693e8e4d9b1..e5ac3fa361dcf 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -294,7 +294,7 @@ void emitMountRenewEvent( event.type = CasEventType::WatermarkRenew; event.outcome = String{outcome}; event.reason = outcome == "recovered" - ? "CAS mount renewal recovered before its confirmed lease-safety deadline" + ? "CAS mount renewal committed after a retry or a settling read" : "CAS mount renewal ended without retained authority and fenced the mount"; event.detail = { {"server_root_id", *context.server_root_id}, @@ -381,12 +381,13 @@ void deliverMountRenewObservability( LOG_INFO( getLogger("CasMountLeaseRenewer"), "CAS mount renewal '{}' recovered after {} physical attempts in {} ms " - "(classification={}, confirmed_deadline_boot_ms={})", + "(classification={}, confirmed_deadline_boot_ms={}, remaining_confirmed_budget_ms={})", *context.server_root_id, context.attempts_sent, elapsedSince(context.observability_start_boot_ms, now_boot_ms), classification, - context.confirmed_deadline_boot_ms); + context.confirmed_deadline_boot_ms, + remainingConfirmedBudget(context, now_boot_ms)); } catch (...) { @@ -408,12 +409,13 @@ void deliverMountRenewObservability( LOG_WARNING( getLogger("CasMountLeaseRenewer"), "CAS mount renewal '{}' fenced after {} physical attempts in {} ms " - "(classification={}, confirmed_deadline_boot_ms={})", + "(classification={}, confirmed_deadline_boot_ms={}, remaining_confirmed_budget_ms={})", *context.server_root_id, context.attempts_sent, elapsedSince(context.observability_start_boot_ms, now_boot_ms), classification, - context.confirmed_deadline_boot_ms); + context.confirmed_deadline_boot_ms, + remainingConfirmedBudget(context, now_boot_ms)); } catch (...) { @@ -1474,6 +1476,7 @@ uint64_t MountLeaseRenewer::start(Liveness liveness) seq = 1; last_etag = etag; + last_body = body; last_committed_attempt_start_boot_ms = attempt_start_boot_ms; const uint64_t ttl_ms = static_cast(ttl.count()); confirmed_deadline_boot_ms = attempt_start_boot_ms > std::numeric_limits::max() - ttl_ms @@ -1580,16 +1583,49 @@ MountRenewResult MountLeaseRenewer::terminalResult(MountRenewResult result) MountRenewResult MountLeaseRenewer::renew(const MountRenewOperationEnvironment & environment) { - return renewOn(mount_requests, environment); + return renewOn(mount_requests, environment, /*until_definitive=*/false); } MountRenewResult MountLeaseRenewer::renewForRemount(const MountRenewOperationEnvironment & environment) { - return renewOn(open_requests, environment); + return renewOn(open_requests, environment, /*until_definitive=*/false); +} + +MountRenewResult MountLeaseRenewer::renewUntilDefinitive(const MountRenewOperationEnvironment & environment) +{ + return renewOn(open_requests, environment, /*until_definitive=*/true); +} + +Observation MountLeaseRenewer::readUntilObserved( + CasOperation & op, const Retry & policy, const MountRenewOperationEnvironment & environment, + uint64_t write_started_boot_ms) const +{ + const auto boot_clock = environment.boot_ms ? environment.boot_ms : boot_ms_fn; + uint64_t pause_ms = Retry::pauseAfterAttempt(*policy.attempt_spacing_ms, write_started_boot_ms, boot_clock()); + while (op.admitted()) + { + op.pause(pause_ms); + try + { + std::optional got = op.read(key, policy); + if (!got) + return ProvenAbsent{}; + return std::move(*got); + } + catch (const Exception & e) + { + if (isDeterministicLocalFailure(e.code())) + throw; + /// Without a deadline the read repeats a transport fault itself, so what reaches here is + /// the store's refusal or a stop, which `op.admitted` tells apart. + pause_ms = environment.refusal_pause_ms; + } + } + return NotObserved{}; } MountRenewResult MountLeaseRenewer::renewOn( - CasRequests & plane, const MountRenewOperationEnvironment & environment) + CasRequests & plane, const MountRenewOperationEnvironment & environment, bool until_definitive) { const MountRenewObservabilityRegistration observability_registration = beginMountRenewObservabilityCall(); const MountRenewObservabilityCallGuard observability_guard(observability_registration); @@ -1602,10 +1638,11 @@ MountRenewResult MountLeaseRenewer::renewOn( static_cast(renewer_state)); const auto boot_clock = environment.boot_ms ? environment.boot_ms : boot_ms_fn; - /// Sampled BEFORE the write. A refused admission is reported as "never attempted" only when this - /// node had already been asked to stop, and reading the flag afterwards could not tell that apart - /// from a flag the refusal itself set. - const bool cancelled = environment.cancelled && environment.cancelled(); + /// Sampled BEFORE each engine call. A refused admission is reported as "never attempted" only when + /// this node had already been asked to stop, and reading the flag afterwards could not tell that + /// apart from a flag the refusal itself set. + const auto sample_cancelled = [&] { return environment.cancelled && environment.cancelled(); }; + bool cancelled = sample_cancelled(); const uint64_t wall_ms = now_ms_fn(); const uint64_t attempt_start_boot_ms = boot_clock(); @@ -1629,18 +1666,70 @@ MountRenewResult MountLeaseRenewer::renewOn( result.attempt_start_boot_ms = attempt_start_boot_ms; CasOperation op = plane.admit(environment.live); - op.observeRequests(environment.requests); + /// Counted here rather than taken from one engine call: the background renewal may span several. + uint32_t attempts_sent = 0; + uint64_t last_attempt_start_boot_ms = attempt_start_boot_ms; + op.observeRequests(RequestObserver{ + .on_write_attempt = [&](uint32_t) + { + ++attempts_sent; + last_attempt_start_boot_ms = boot_clock(); + if (environment.requests.on_write_attempt) + environment.requests.on_write_attempt(attempts_sent); + }, + .on_failure = environment.requests.on_failure, + }); + const Etag expected = precondition(); + const Retry policy = until_definitive + ? Retry::untilDefinitive() + : Retry::untilLeaseSafe(confirmed_deadline_boot_ms, static_cast(lease_safety_margin.count())); std::optional written; try { - written = op.replace(key, body, precondition(), - Retry::untilLeaseSafe(confirmed_deadline_boot_ms, static_cast(lease_safety_margin.count()))); + written = op.replace(key, body, expected, policy); + while (until_definitive) + { + uint64_t pause_ms = 0; + if (std::holds_alternative(*written)) + { + /// `writeLoop` refuses only when no attempt of the call is unclear, so no copy of ours + /// can land behind a new call with the same tuple. + pause_ms = environment.refusal_pause_ms; + } + else if (Conflict * conflict = std::get_if(&*written)) + { + if (std::holds_alternative(conflict->seen)) + conflict->seen = readUntilObserved(op, policy, environment, last_attempt_start_boot_ms); + if (std::holds_alternative(conflict->seen)) + { + /// Counted as sent: the refused precondition says the slot may have moved, so a farewell + /// naming our last token could meet a holder entitled to it. + written = GaveUp{GaveUp::Why::FenceLost, GaveUp::Source::Policy, /*sent_any=*/true, NotObserved{}, attempts_sent}; + break; + } + /// Our last confirmed body under the token we named: the slot did not move. A new call is + /// safe because a call with an unclear attempt returns `Conflict` only once the token + /// moved, so this call left no copy of ours that could land. + const Object * seen = std::get_if(&conflict->seen); + if (!seen || seen->etag != expected || seen->bytes != last_body) + break; + pause_ms = Retry::pauseAfterAttempt(*policy.attempt_spacing_ms, last_attempt_start_boot_ms, boot_clock()); + } + else + { + break; + } + op.pause(pause_ms); + cancelled = sample_cancelled(); + written = op.replace(key, body, expected, policy); + } } catch (...) { /// The engine surfaces a deterministic local failure unchanged rather than reissuing it. markMountRenewTermination(MountRenewTerminalClassification::DeterministicFailure); result.failure = std::current_exception(); + result.attempts_sent = attempts_sent; return terminalResult(std::move(result)); } @@ -1648,22 +1737,23 @@ MountRenewResult MountLeaseRenewer::renewOn( { seq = next_seq; last_etag = std::move(committed->etag); + last_body = body; last_committed_attempt_start_boot_ms = attempt_start_boot_ms; const uint64_t ttl_ms = static_cast(ttl.count()); confirmed_deadline_boot_ms = attempt_start_boot_ms > std::numeric_limits::max() - ttl_ms ? std::numeric_limits::max() : attempt_start_boot_ms + ttl_ms; result.outcome = MountRenewOutcome::Committed; - result.attempts_sent = committed->attempts_sent; + result.attempts_sent = attempts_sent; result.resolved_by_read = committed->resolved_by_read; - result.sent_any = committed->attempts_sent != 0; + result.sent_any = attempts_sent != 0; return result; } if (const Conflict * conflict = std::get_if(&*written)) { result.sent_any = true; - result.attempts_sent = conflict->attempts_sent; + result.attempts_sent = attempts_sent; try { throwRenewConflict(conflict->seen); @@ -1678,7 +1768,7 @@ MountRenewResult MountLeaseRenewer::renewOn( if (const Refused * refused = std::get_if(&*written)) { result.sent_any = true; - result.attempts_sent = refused->attempts_sent; + result.attempts_sent = attempts_sent; markMountRenewTermination(MountRenewTerminalClassification::DeterministicFailure); result.failure = std::make_exception_ptr(Exception( refused->store_error, @@ -1688,13 +1778,15 @@ MountRenewResult MountLeaseRenewer::renewOn( if (const GaveUp * gave_up = std::get_if(&*written)) { - result.sent_any = gave_up->sent_any; - result.attempts_sent = gave_up->attempts_sent; + /// Across every engine call of this renewal, not only the call that gave up. + result.sent_any = attempts_sent != 0; + result.attempts_sent = attempts_sent; if (gave_up->why == GaveUp::Why::Deadline) result.deadline_source = gave_up->deadline_source; - /// Nothing was sent and the node was already stopping: the lease is exactly as it was, so this - /// is a renewal that never ran, not one that lost its authority. + /// The call that gave up sent nothing, every earlier call of a series ended in a refusal or in a + /// read that found the slot as our last commit left it, and the node was already stopping: the + /// renewal never ran and the farewell may still hand the slot back. if (gave_up->why == GaveUp::Why::FenceLost && !gave_up->sent_any && cancelled) { markMountRenewTermination(MountRenewTerminalClassification::Cancelled); @@ -1706,7 +1798,8 @@ MountRenewResult MountLeaseRenewer::renewOn( switch (gave_up->why) { case GaveUp::Why::FenceLost: - classification = cancelled + /// Sampled again: a stop that arrived during the call still names why it ended. + classification = sample_cancelled() ? MountRenewTerminalClassification::Cancelled : MountRenewTerminalClassification::FenceOrLifecycleLost; break; @@ -1725,7 +1818,7 @@ MountRenewResult MountLeaseRenewer::renewOn( "observation: {})", key, terminalClassificationName(classification), - gave_up->sent_any ? "at least one" : "no", + result.sent_any ? "at least one" : "no", detail::renderObservation(gave_up->last_seen))); return terminalResult(std::move(result)); } @@ -1783,6 +1876,7 @@ void MountLeaseRenewer::terminate(CasOperation & op) { seq += 1; last_etag = std::move(committed->etag); + last_body = body; emitMountEvent( event_sink, CasEventType::MountRelease, srid, "farewell", nullptr, "graceful release -- lease stamped already-expired and watermark retired"); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index 9cc34bf54efbc..320e00934a9f4 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -67,11 +67,15 @@ struct MountRenewOperationEnvironment /// Facts the mount fence cannot see (park requested, pool no longer live, shutdown). FALSE ends /// the renewal exactly as a lost fence does; the engine does not need to know which refused. std::function live; - /// Sampled ONCE before the write. A renewal refused before its first attempt is reported as + /// Sampled before each engine call. A call refused before its first attempt ends the renewal as /// `NotAttempted` rather than terminal only when this node had already been asked to stop -- /// sampling it afterwards would read a flag that the refusal itself may have set. std::function cancelled; RequestObserver requests; + /// The background renewal's wait after the store refused a request: the store said it cannot serve + /// it, and each engine call may refresh the credentials once, so a pace of one spacing would hammer + /// the store and refresh every second. + uint64_t refusal_pause_ms = 0; }; /// Validate a `server_root_id` — the explicit, configured identity of the content-addressed layout @@ -529,12 +533,13 @@ bool isCreatorFenceTerminal(CasOperation & op, const Layout & layout, const Stri /// - foreign uuid → fail closed; /// - absent → `create`; expired-our-uuid (any epoch) → `replace` reclaim. /// -/// PLANES. Only the RENEWAL is admitted under the mount fence, because only a renewal writes under -/// authority the fence is tracking. The claim and the farewell are admitted off it: a self-remount -/// claims with the fence already latched lost, so a claim gated on the fence could never reclaim, and -/// a farewell refused because the fence has run down would leave the slot looking live until GC -/// fences it out. Neither is unguarded: a claim's safety is its own conditional write, and a caller -/// that has shutdown facts hands them over as a `Liveness`. +/// PLANES. The bounded renewal (`renew`: startup and direct calls) is admitted under the mount fence, +/// because it writes under the authority the fence tracks. The background renewal, the claim, the +/// remount re-anchor and the farewell run on the lease plane, whose fence is open: the background +/// renewal must outlive its lease, a self-remount claims with the fence already latched lost, and a +/// farewell refused because the fence ran down would leave the slot looking live until GC fences it +/// out. None is unguarded: every write is conditional, and a caller with shutdown facts hands them +/// over as a `Liveness`. class MountLeaseRenewer { public: @@ -552,7 +557,7 @@ class MountLeaseRenewer /// Adopt the already-claimed mount. Returns the exact pre-I/O BOOTTIME anchor. `liveness` carries /// the caller's shutdown terms; the mount fence is deliberately not consulted here. uint64_t start(Liveness liveness = {}); - /// The steady-state renewal, admitted under the mount fence. + /// The bounded renewal, admitted under the mount fence: the startup re-anchor and direct calls. MountRenewResult renew(const MountRenewOperationEnvironment & environment); /// The remount's re-anchor, which is bootstrap control rather than steady state: a remount renews /// BEFORE it arms the fence for the new incarnation, so the fence is still latched lost and an @@ -560,6 +565,10 @@ class MountLeaseRenewer /// renewer's own open plane, the one the claim and the farewell use, so there is no plane for a /// caller to get wrong. Same policy and same verdicts as `renew`. MountRenewResult renewForRemount(const MountRenewOperationEnvironment & environment = {}); + /// The background renewal, on the lease plane, retried until a definitive answer: a commit, a + /// conflict the read observed, a local failure, or `environment.live` refusing. A refusal sends the + /// same body again, and a refused precondition whose read failed repeats only the read. + MountRenewResult renewUntilDefinitive(const MountRenewOperationEnvironment & environment); void release(); MountLeaseRenewerState state() const { return renewer_state; } @@ -571,8 +580,14 @@ class MountLeaseRenewer /// The incarnation every guarded write of this slot names. Engaged for exactly the states that /// admit such a write: `start` establishes it and each committed renewal replaces it. const Etag & precondition() const; - /// One renewal admitted on `plane`; `renew` and `renewForRemount` differ only in which they pass. - MountRenewResult renewOn(CasRequests & plane, const MountRenewOperationEnvironment & environment); + /// One renewal admitted on `plane`. `renew` and `renewForRemount` pass `until_definitive = false` (bounded by + /// the lease); `renewUntilDefinitive` passes `true` (the background renewal's unbounded retry). + MountRenewResult renewOn(CasRequests & plane, const MountRenewOperationEnvironment & environment, bool until_definitive); + /// Repeats the read of a slot whose precondition was refused until it observes the slot: one spacing + /// after the write attempt that started at `write_started_boot_ms`, then `refusal_pause_ms` after each + /// refused read. `NotObserved` only when `op` is no longer admitted. + Observation readUntilObserved(CasOperation & op, const Retry & policy, + const MountRenewOperationEnvironment & environment, uint64_t write_started_boot_ms) const; Etag claim(CasOperation & op, const String & body); [[noreturn]] void throwRenewConflict(const Observation & seen) const; MountRenewResult terminalResult(MountRenewResult result); @@ -598,6 +613,8 @@ class MountLeaseRenewer /// The incarnation our last landed write created; every renewal and the farewell name it as the /// precondition. Unset only before `start` has landed one. std::optional last_etag; + /// The bytes `last_etag` names. A read that returns them under that token says the slot did not move. + String last_body; uint64_t confirmed_deadline_boot_ms = 0; uint64_t last_committed_attempt_start_boot_ms = 0; }; From bb5fa5979827761b462ee09366a8cbc46f9fb0c3 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 16 Oct 2026 02:36:10 +0200 Subject: [PATCH 4/7] Keep the CAS decommission renewal bounded by the lease The decommission tool opens its pool with the background renewal. An admin claim that cannot renew should end at its lease rather than pause, so `Pool::openForDecommission` sets `bounded_background_renewal`. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- src/Common/ProfileEvents.cpp | 4 ++-- .../MetadataStorages/ContentAddressed/Pool/CasPool.cpp | 3 +++ .../MetadataStorages/ContentAddressed/Pool/CasPool.h | 3 +++ 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 7e6cb47c6df2b..1bdbbf52ce8a9 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -956,8 +956,8 @@ The server successfully detected this situation and will download merged part fr M(CASMountRenewalRetries, "Number of physical conditional renewal PUTs sent after the first attempt of one logical CAS mount-lease renewal.", ValueType::Number) \ M(CASMountRenewalResolved, "Number of CAS mount-lease renewals whose committed outcome was proved by an exact resolving GET.", ValueType::Number) \ M(CASMountRenewalRecovered, "Number of logical CAS mount-lease renewals that committed after a physical retry or exact resolving GET.", ValueType::Number) \ - M(CASMountRenewalDeadlineExceeded, "Number of logical CAS mount-lease renewals stopped by the external lease-safety deadline. Growth means the last confirmed lease no longer had enough safe time for another physical attempt.", ValueType::Number) \ - M(CASMountLeaseLost, "Counts exactly once per operational CAS mount-lease Live-to-TransientNotLive loss/recovery generation. The initiating external loss or the first ordinary terminal renewal consumer owns the increment, including external lease-safety deadline exhaustion; parked/classification/shutdown paths do not duplicate it.", ValueType::Number) \ + M(CASMountRenewalDeadlineExceeded, "Number of bounded CAS mount-lease renewals (the startup and remount re-anchors, and the background renewal of a pool opened for decommission) stopped by the external lease-safety deadline. Growth means the last confirmed lease no longer had enough safe time for another physical attempt.", ValueType::Number) \ + M(CASMountLeaseLost, "Counts exactly once per operational CAS mount-lease Live-to-TransientNotLive loss/recovery generation. The initiating external loss or the first ordinary terminal renewal consumer owns the increment; parked/classification/shutdown paths do not duplicate it.", ValueType::Number) \ M(CASRemountAttempts, "Number of invocations of the CAS whole-chain remount attempt.", ValueType::Number) \ M(CASRemountSucceeded, "Number of CAS whole-chain remount attempts that restored Live under a fresh writer epoch.", ValueType::Number) \ M(CASRemountFailed, "Number of CAS whole-chain remount attempts that returned without restoring Live, including caught step exceptions.", ValueType::Number) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index 698681efb3cf6..0bae5f3b2e13f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -923,6 +923,9 @@ PoolPtr Pool::openForDecommission(BackendPtr backend, PoolConfig config, const S /// The admin claim must be RENEWED like any writable mount: the host disk may be observe-only /// (background_watermark=false), but an unrenewed claim (TTL ~30s) aborts any long drain midway. config.background_watermark = true; + /// Bounded by the lease, like the startup re-anchor: an admin claim that cannot renew ends rather than + /// pauses. + config.bounded_background_renewal = true; validateWritableMountTiming(config); Layout layout(config.pool_prefix); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index 34ee0316c3435..cf5a324fe902c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -184,6 +184,8 @@ struct PoolConfig uint64_t gc_io_concurrency = 16; /// Tests drive `renewWatermarkOnce` explicitly; gates both persistent runtime workers. bool background_watermark = false; + /// See `MountConfig::bounded_background_renewal`. + bool bounded_background_renewal = false; /// Installed on the pool before a writable mount can start its runtime-owned workers. CasEventSink event_sink = {}; @@ -355,6 +357,7 @@ struct PoolConfig .boot_ms_fn = boot_ms_fn, .wait_sleep_fn = wait_sleep_fn, .worker_factory = worker_factory, + .bounded_background_renewal = bounded_background_renewal, }; } }; From 57ae5a395f476ef5f1e5515c5e46b41a17545d2f Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 16 Oct 2026 02:36:23 +0200 Subject: [PATCH 5/7] Pause CAS writes while the lease cannot admit one, and report it Writes stop being admitted once the lease has less than two attempt envelopes plus the safety margin left. `system.cas_mounts` now shows `not_live` / `lease_expired` from that same moment, computed by the predicate admission uses, and one WARNING per pause names the writer epoch and the last renewal error. The refusal texts a write meets during a pause say that writes resume once the renewal is confirmed. A clean shutdown during a pause skips the farewell, which has no lease room left, and leaves the slot for GC to observe. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../ContentAddressedMetadataStorage.cpp | 4 +- .../ContentAddressedMetadataStorage.h | 10 ++- .../ContentAddressed/Pool/CasMountRuntime.cpp | 68 ++++++++++++++++--- .../ContentAddressed/Pool/CasMountRuntime.h | 15 +++- .../ContentAddressed/Pool/CasPool.cpp | 10 +++ .../ContentAddressed/Pool/CasPool.h | 2 + .../ContentAddressed/Pool/CasRefLedger.cpp | 3 +- .../StorageSystemContentAddressedMounts.cpp | 6 +- 8 files changed, 95 insertions(+), 23 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index c409cd318f59f..addb9f4992d51 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp @@ -465,8 +465,8 @@ CasLifecycleSnapshot ContentAddressedMetadataStorage::lifecycleSnapshot() const } const Cas::Pool::LifecycleSnapshot ps = pool->lifecycleSnapshot(); - snap.lifecycle = casLifecycleToString(ps.lifecycle); - snap.reason = casLifecycleReasonWord(ps.lifecycle); + snap.lifecycle = ps.writes_paused ? "not_live" : casLifecycleToString(ps.lifecycle); + snap.reason = ps.writes_paused ? "lease_expired" : casLifecycleReasonWord(ps.lifecycle); snap.detail = ps.detail; snap.since = ps.since; return snap; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h index 5f0aadb2d7d40..28bf1020a083f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h @@ -95,13 +95,17 @@ enum class CasOpAdmission : uint8_t /// - `lifecycle` — one of `live` / `not_live` / `identity_lost` / `vanished` (a live pool), or /// `constructing` / `shutdown` (no pool published). /// - `reason` — the ENUM-CLEAN sub-state word: `replaced` / `forgotten` for a -/// `vanished` pool, empty otherwise. Kept a small closed vocabulary so a downstream +/// `vanished` pool, `lease_expired` for a `not_live` row whose pool is `Live` but +/// whose lease has too little time left to admit a write, empty otherwise. Kept a +/// small closed vocabulary so a downstream /// `lifecycle || '(' || reason || ')'` yields e.g. exactly `vanished(forgotten)` — /// the [D5] free text lives in `detail`, never here. /// - `detail` — the full [D5] reason text naming the actual failure (the replaced /// diagnosis, the timestamped `FORGET` message, or the identity-loss message); spec §1 -/// requires it appear verbatim in the snapshot. Empty while `live` and for a null pool. -/// - `since` — wall-clock second the current non-`live` state was entered; 0 while `live`/no pool. +/// requires it appear verbatim in the snapshot. Empty while `live`, for +/// `lease_expired`, and for a null pool. +/// - `since` — wall-clock second the current non-`live` state was entered (for +/// `lease_expired`, when writes paused); 0 while `live`/no pool. /// - `pool_id` — last-known pool UUID (empty before the first `startup`); the disk stays /// introspectable under its identity even once the pool is gone. /// - `server_root_id` — this server's node-local root id owning the mount slot. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp index 3e0d2ad5d4c40..7028e4f657215 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp @@ -131,9 +131,10 @@ void CasMountRuntime::checkFenceOrThrow(uint64_t admitted_generation) const if (!mayMutate() || fenceGeneration() != admitted_generation) throwCasTransientUnavailable( fmt::format("content-addressed pool '{}'", server_root_id), - "mount fence tripped: the durable write is refused because this node no longer holds the mount " - "incarnation it was admitted under -- either a lease loss the disk auto-recovers from, or a " - "FORGET decommission / lost identity that does NOT recover; consult " + "mount fence tripped: the durable write is refused because the mount lease has too little time " + "left (writes are paused and resume once the lease renewal is confirmed), or because this node " + "no longer holds the mount incarnation it was admitted under -- either a lease loss the disk " + "auto-recovers from, or a FORGET decommission / lost identity that does NOT recover; consult " "system.cas_mounts for the disk's lifecycle before retrying"); } @@ -153,15 +154,49 @@ Fence::Admit CasMountRuntime::admit(uint64_t admitted_generation, uint64_t neede return Fence::Admit::Ok; } -bool CasMountRuntime::refAppendFenceOk() const +uint64_t CasMountRuntime::writeReservationMs() const { - /// Two envelopes' worth of room under the live generation -- a write and its settlement read, which - /// is what `writeLoop` reserves -- so a ref-log attempt is not started when it cannot plausibly - /// finish, safety margin included, before the lease expires. const uint64_t envelope_ms = cas_request_budget.attemptEnvelopeMs(); - const uint64_t needed_ms = envelope_ms > std::numeric_limits::max() / 2 - ? std::numeric_limits::max() : 2 * envelope_ms; - return admit(fenceGeneration(), needed_ms) == Fence::Admit::Ok; + return envelope_ms > std::numeric_limits::max() / 2 ? std::numeric_limits::max() : 2 * envelope_ms; +} + +bool CasMountRuntime::refAppendFenceOk() const +{ + /// A write and its settlement read, which is what `writeLoop` reserves, so a ref-log attempt is not + /// started when it cannot plausibly finish, safety margin included, before the lease expires. + return admit(fenceGeneration(), writeReservationMs()) == Fence::Admit::Ok; +} + +std::optional CasMountRuntime::writesPausedSinceBootMs() const +{ + if (lifecycle() != PoolLifecycle::Live || mount_fence.lost.load(std::memory_order_acquire) || refAppendFenceOk()) + return std::nullopt; + const uint64_t deadline = mount_fence.deadline_boot_ms.load(std::memory_order_acquire); + const uint64_t reservation_ms = writeReservationMs(); + const uint64_t closing_ms = reservation_ms > std::numeric_limits::max() - cas_request_budget.lease_safety_margin_ms + ? std::numeric_limits::max() + : reservation_ms + cas_request_budget.lease_safety_margin_ms; + return deadline > closing_ms ? deadline - closing_ms : 0; +} + +void CasMountRuntime::warnOnceIfWritesPaused() +{ + const std::optional paused_at = writesPausedSinceBootMs(); + if (!paused_at || writes_paused_warned.exchange(true, std::memory_order_relaxed)) + return; + std::optional last_error; + { + std::lock_guard lock(renewal_progress_mutex); + last_error = renewal_progress.last_error; + } + const uint64_t now = bootMsNow(); + /// Written at the first attempt after the pause began, so a renewal stuck in a request reports late; + /// the age says by how much. + LOG_WARNING(getLogger("CasPool"), + "CAS mount lease of '{}' under writer_epoch {} has too little time left for a write: writes are paused " + "since {} ms ago and resume once the renewal, which keeps retrying, is confirmed with enough lease left. " + "Last renewal error: {}", + server_root_id, liveWriterEpoch(), now > *paused_at ? now - *paused_at : 0, last_error.value_or("none")); } void CasMountRuntime::setMountDeadline(uint64_t deadline_boot_ms) @@ -174,6 +209,7 @@ void CasMountRuntime::armMountFence(UInt128 server_uuid, uint64_t writer_epoch, mount_fence.server_uuid = server_uuid; mount_fence.writer_epoch = writer_epoch; mount_fence.deadline_boot_ms.store(deadline_boot_ms, std::memory_order_release); + writes_paused_warned.store(false, std::memory_order_relaxed); /// A fresh lease incarnation is a fresh generation too: a durable-effect caller admitted under the /// PRIOR incarnation must re-check and abort rather than ride this re-arm through (rev.7 [C2]). fence_generation.fetch_add(1, std::memory_order_acq_rel); @@ -450,6 +486,7 @@ MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_c .requests = RequestObserver{ .on_write_attempt = [this](uint32_t attempts_sent) { + warnOnceIfWritesPaused(); ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalAttempts); if (attempts_sent > 1) ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalRetries); @@ -555,6 +592,8 @@ void CasMountRuntime::publishRenewedDeadline(uint64_t attempt_start_boot_ms) ? std::numeric_limits::max() : attempt_start_boot_ms + ttl_ms; setMountDeadline(deadline); + if (!writesPausedSinceBootMs()) + writes_paused_warned.store(false, std::memory_order_relaxed); } uint64_t CasMountRuntime::renewRenewerOnce( @@ -1180,7 +1219,14 @@ void CasMountRuntime::finishTeardown(bool drained) if (!mount_renewer) return; - if (drained && mount_renewer->state() == MountLeaseRenewerState::Active) + if (drained && mount_renewer->state() == MountLeaseRenewerState::Active && writesPausedSinceBootMs()) + { + /// The farewell needs the lease room a pause no longer has; GC fences the slot after observing it. + LOG_INFO(getLogger("CasPool"), + "CAS mount lease of '{}' has too little time left for the farewell; the slot is left for observation", + server_root_id); + } + else if (drained && mount_renewer->state() == MountLeaseRenewerState::Active) { try { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h index d6c3ee0f11309..1b69c885d092f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h @@ -322,11 +322,15 @@ class CasMountRuntime return static_cast(lifecycle_since_wall_s.load(std::memory_order_acquire)); } - /// Extends `mayMutate` with a remaining-budget check. A ref-log attempt is refused unless the - /// current lease has room for its configured timeout and safety margin, so work is not started when - /// it cannot plausibly finish before the fence expires. + /// Extends `mayMutate` with a remaining-budget check: room for one write (two attempt envelopes) plus + /// the safety margin before the deadline. bool refAppendFenceOk() const; + /// When `refAppendFenceOk` stopped admitting writes under the current deadline, while the pool is + /// `Live` and the fence is not lost: the renewal keeps retrying and writes are paused. Empty while + /// writes are admitted. I/O-free. + std::optional writesPausedSinceBootMs() const; + /// TRUE once the pool has reached — or is being driven toward — a state on which the self-remount /// worker must stop: a published terminal `Vanished` intent (`vanished_intent` — set early by /// FORGET, or by a natural `enterVanished`, and already subsuming every settled `Vanished*` state since @@ -463,6 +467,10 @@ class CasMountRuntime bool propagate_failure); /// Publish a committed renewal's deadline, its start plus the lease. Requires `driver_mutex`. void publishRenewedDeadline(uint64_t attempt_start_boot_ms); + /// One `WARNING` per pause, at the first renewal attempt after writes paused. + void warnOnceIfWritesPaused(); + /// Two attempt envelopes, saturating: what one write reserves. + uint64_t writeReservationMs() const; void renewalLoop(); void remountLoop(); ThreadFromGlobalPool makeWorker(std::function body); @@ -543,6 +551,7 @@ class CasMountRuntime /// Fence-generation token (rev.7 [C2]): bumped by `tripMountLost` and `armMountFence`. See /// `fenceGeneration`/`checkFenceOrThrow`. std::atomic fence_generation{0}; + std::atomic writes_paused_warned{false}; std::function arm_mount_fence_interposition_hook_for_test; /// The pool lifecycle condition (rev.7 §1). Starts `Live`. Non-terminal transitions diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index 0bae5f3b2e13f..25bf63f52a361 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -382,6 +382,16 @@ Pool::LifecycleSnapshot Pool::lifecycleSnapshot() const snap.lifecycle = mount_runtime.lifecycle(); snap.detail = lifecycleReasonDetail(snap.lifecycle); snap.since = mount_runtime.lifecycleSinceWallS(); + if (snap.lifecycle == PoolLifecycle::Live) + { + if (const std::optional paused_at = mount_runtime.writesPausedSinceBootMs()) + { + snap.writes_paused = true; + const uint64_t now_ms = mount_runtime.bootMsNow(); + const uint64_t ago_s = (now_ms > *paused_at ? now_ms - *paused_at : 0) / 1000; + snap.since = ::time(nullptr) - static_cast(ago_s); + } + } return snap; } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index cf5a324fe902c..ca9415369eeae 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -546,6 +546,8 @@ class Pool : public std::enable_shared_from_this PoolLifecycle lifecycle = PoolLifecycle::Live; String detail; time_t since = 0; + /// `Live`, but the lease has too little time left to admit a write; `since` is then when writes paused. + bool writes_paused = false; }; LifecycleSnapshot lifecycleSnapshot() const; RenewalProgress renewalProgress() const { return mount_runtime.renewalProgress(); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasRefLedger.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasRefLedger.cpp index 196984a9aa68c..6f9f1f242aa71 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasRefLedger.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasRefLedger.cpp @@ -262,7 +262,8 @@ void CasRefLedger::refuseUnlessAdmitted(const CasOperation & op, std::string_vie throwCasTransientUnavailable( fmt::format("content-addressed pool '{}'", server_root_id), fmt::format("{}: the operation is no longer admitted -- the mount lease has too little time left " - "for another request, or this table's runtime was detached", what)); + "for another request (writes are paused and resume once the lease renewal is confirmed), " + "or this table's runtime was detached", what)); } void CasRefLedger::setCasRetrySleepForTest(std::function sleep_fn) diff --git a/src/Storages/System/StorageSystemContentAddressedMounts.cpp b/src/Storages/System/StorageSystemContentAddressedMounts.cpp index 79285618518c0..6c07aa43b9c53 100644 --- a/src/Storages/System/StorageSystemContentAddressedMounts.cpp +++ b/src/Storages/System/StorageSystemContentAddressedMounts.cpp @@ -54,9 +54,9 @@ StorageSystemContentAddressedMounts::StorageSystemContentAddressedMounts(const S {"last_success_age_seconds", std::make_shared(std::make_shared()), "Seconds since this disk's GC last led a round (0 if it never led). NULL on rows describing other servers' mounts."}, {"wedged_namespace_count", std::make_shared(std::make_shared()), "Ref-append lanes currently wedged on this disk. NULL on rows describing other servers' mounts."}, {"lifecycle", std::make_shared(), "This server's content-addressed pool lifecycle for the disk (non-gated snapshot, always populated so a not-live disk stays visible): live, not_live, identity_lost, vanished, constructing (never started) or shutdown (torn down)."}, - {"lifecycle_reason", std::make_shared(), "The enum-clean sub-state word for a vanished disk: replaced or forgotten. Empty for every other lifecycle (so lifecycle || '(' || lifecycle_reason || ')' reads e.g. vanished(forgotten))."}, - {"lifecycle_detail", std::make_shared(), "The full typed reason text naming the actual cause when not live: the vanish diagnosis (data root replaced by a foreign pool / decommissioned by SYSTEM CAS FORGET at