Show CAS mount renewal progress in system.cas_mounts

docs/en/antalya/cas/architecture/mounts-and-leases.md

@@ -278,6 +278,7 @@ becomes `state = 'corrupt'`, never an exception). Shows every `server_root_id` i
 | `state` | one of `live`, `expired`, `terminated`, `fenced`, `corrupt` |
 | `is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count` | GC health, process-local; **`NULL` on every peer row** — a process-local fact must never be stamped onto another server's row |
 | `lifecycle`, `lifecycle_reason`, `lifecycle_detail`, `lifecycle_since` | the SQL surface for the in-process `PoolLifecycle` runtime above: `lifecycle` is one of `live`, `not_live`, `identity_lost`, `vanished`, `constructing`, `shutdown`; `lifecycle_reason` distinguishes `replaced` from `forgotten` for a `vanished` disk; `lifecycle_detail` carries the full diagnosis text; `lifecycle_since` is when the current non-live state began (`NULL` while live) |
+| `renewal_in_progress`, `renewal_attempts`, `last_renewal_error` | the running (or last) lease renewal: conditional `PUT`s sent so far and the last failed request, updated per attempt; **`NULL` on every peer row** |
 
 The lifecycle snapshot is I/O-free and ungated, so a not-live, never-started, or vanished disk
 still produces a row instead of silently disappearing from the table.

docs/en/operations/system-tables/cas_mounts.md

@@ -41,6 +41,9 @@ rows.
 - `lifecycle_reason` ([String](/sql-reference/data-types/string)) — The enum-clean sub-state word for a `vanished` disk: `replaced` or `forgotten`. Empty for every other lifecycle, so `lifecycle || '(' || lifecycle_reason || ')'` reads e.g. `vanished(forgotten)`.
 - `lifecycle_detail` ([String](/sql-reference/data-types/string)) — The full typed reason text naming the actual cause when not live: the vanish diagnosis (a data root replaced by a foreign pool, or decommissioned by `SYSTEM CAS FORGET` at a given time) or the identity-loss message. Empty when live.
 - `lifecycle_since` ([Nullable(DateTime)](/sql-reference/data-types/nullable)) — When this server entered the current non-live lifecycle state. `NULL` when live, or when the state has no backing pool to date from.
+- `renewal_in_progress` ([Nullable(UInt8)](/sql-reference/data-types/nullable)) — `1` while this server renews its mount lease.
+- `renewal_attempts` ([Nullable(UInt64)](/sql-reference/data-types/nullable)) — Conditional `PUT`s sent by the current renewal, or by the last one once it ended. Updated per attempt, so a renewal stuck in retries is visible before it ends.
+- `last_renewal_error` ([Nullable(String)](/sql-reference/data-types/nullable)) — The last failed request of that renewal, as `<VERB> <key>: <class> <code> <message>`. `class` is `refused` when the store definitely refused the request (access denied, malformed request, entity too large), `transport` otherwise; `code` is the store's error name, such as `SlowDown`. `NULL` when no request failed.
 
 `lifecycle`/`lifecycle_reason`/`lifecycle_detail`/`lifecycle_since` are the SQL surface for
 diagnosing an identity-lost or forgotten disk without reading server logs — see the
@@ -49,9 +52,10 @@ diagnosing an identity-lost or forgotten disk without reading server logs — se
 what each lifecycle value means, and [`SYSTEM CAS FORGET`](/antalya/cas/operations/debugging#sql-forget)
 for the command that produces `vanished(forgotten)`.
 
-:::note Local-only GC-health columns
-`is_leader`, `pending_reclaim`, `last_success_age_seconds`, and `wedged_namespace_count` are process-local
-facts about *this* server's own GC scheduler. They are populated **only** on the row whose `server_root_id` matches
+:::note Local-only columns
+`is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count`,
+`renewal_in_progress`, `renewal_attempts` and `last_renewal_error` are process-local facts about
+*this* server. They are populated **only** on the row whose `server_root_id` matches
 this server's own mount, and are `NULL` on every row describing another server's mount — stamping a local
 health fact onto a peer's row would misread as "the peer is the GC leader" during an incident. To see the
 peer's own view of these columns, query `system.cas_mounts` on that server.

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp

@@ -1,5 +1,6 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h>
 
+#include <Common/ErrorCodes.h>
 #include <Common/Exception.h>
 #include <IO/ReadBuffer.h>
 #include <Common/LoggingHelpers.h>
@@ -278,6 +279,27 @@ bool isFirstAttemptFuseTimeout([[maybe_unused]] const std::exception & e, [[mayb
 #endif
 }
 
+bool isStoreAnswer(const std::exception & e)
+{
+    const auto * db = dynamic_cast<const Exception *>(&e);
+    return dynamic_cast<const Poco::Exception *>(&e) && !(db && isDeterministicLocalFailure(db->code()));
+}
+
+String describeRequestFailure(std::string_view verb, const String & key, const std::exception & e)
+{
+    const auto & poco = dynamic_cast<const Poco::Exception &>(e);
+    String code;
+#if USE_AWS_S3
+    if (const auto * s3 = dynamic_cast<const S3Exception *>(&e))
+        code = s3->getExceptionName();
+#endif
+    if (const auto * db = dynamic_cast<const Exception *>(&e); code.empty() && db)
+        code = String(ErrorCodes::getName(db->code()));
+    if (code.empty())
+        code = poco.name();
+    return fmt::format("{} {}: {} {} {}", verb, key, isDefinitelyRefusedWrite(e) ? "refused" : "transport", code, poco.message());
+}
+
 CasRequests::CasRequests(BackendPtr backend_, Fence fence_,
                          std::function<uint64_t()> now_ms_, std::function<void(uint64_t)> sleep_ms_,
                          CasHotKeys * hot_keys_)
@@ -924,6 +946,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co
         detail::recordAttempt();
         ++state.attempts_sent;
         state.sent_any = true;
+        if (observer.on_write_attempt)
+            observer.on_write_attempt(state.attempts_sent);
 
         /// Disengaged means the attempt threw: its fate is unproven, and nothing may be read out of it.
         std::optional<std::expected<String, Backend::RawConflict>> outcome;
@@ -952,6 +976,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co
         {
             if (isDeterministicLocalFailure(e.code()))
                 throw;
+            if (observer.on_failure)
+                observer.on_failure("PUT", key, e);
             /// ONE refresh per call, only for the class a credential could explain, and only when a
             /// reissue could sign with what it installs. So an oversized entity never triggers a
             /// re-acquisition, a denial fresh credentials do not fix is refused on the second look
@@ -1008,6 +1034,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co
             /// still have landed, so `outcome` stays disengaged and the read below settles it.
             if (!dynamic_cast<const Poco::Exception *>(&e))
                 throw;
+            if (observer.on_failure)
+                observer.on_failure("PUT", key, e);
         }
 
         if (outcome && outcome->has_value())

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h

@@ -48,6 +48,10 @@ bool isFirstAttemptFuseTimeout(const std::exception & e, size_t attempt_no);
 /// same failure and buries the root cause behind a retryable exception. The set is `LOGICAL_ERROR`,
 /// `NOT_IMPLEMENTED`, `BAD_ARGUMENTS` and `CORRUPTED_DATA`.
 bool isDeterministicLocalFailure(int code);
+/// An exception the store or the transport raised, as opposed to a local fault.
+bool isStoreAnswer(const std::exception & e);
+/// `VERB key: refused|transport code message`; `code` is the store's error name when it sent one.
+String describeRequestFailure(std::string_view verb, const String & key, const std::exception & e);
 
 /// Throw the recoverable "CAS write could not be committed, retry later" condition.
 ///
@@ -132,6 +136,15 @@ struct ListPage
 /// The walk's callback: FALSE stops the walk.
 using ListedKeyFn = std::function<bool(const ListedKey &)>;
 
+/// Live progress of one operation, called on its thread inside the engine's retry loops; must not throw.
+struct RequestObserver
+{
+    /// A conditional write is about to send its `attempts_sent`-th physical attempt.
+    std::function<void(uint32_t attempts_sent)> on_write_attempt;
+    /// A conditional `PUT` or a body `GET` failed with a store answer (`isStoreAnswer`).
+    std::function<void(std::string_view verb, const String & key, const std::exception & e)> on_failure;
+};
+
 namespace detail
 {
 /// The engine's per-attempt counters, behind functions so the header need not declare the events.
@@ -256,6 +269,7 @@ class CasOperation
     /// The verdict point: is this operation still admitted? For the sites that guard a decision rather
     /// than a request.
     bool admitted() const { return gate(0) == Gate::Ok; }
+    void observeRequests(RequestObserver observer_) { observer = std::move(observer_); }
 
     /// The write lane for keys several writers of this pool share.
     CasHotKeys & hotKeys() const { return *owner.hot_keys; }
@@ -445,6 +459,7 @@ class CasOperation
     /// Written immediately before a read-class give-up throws, cleared and read only by the resolve
     /// read that swallows it. Every other caller lets the exception carry the verdict.
     std::optional<ReadStop> last_read_stop;
+    RequestObserver observer;
 };
 
 template <typename Fn>
@@ -476,6 +491,8 @@ auto CasOperation::readLoop(std::string_view verb, const String & subject, const
         }
         catch (const std::exception & e)
         {
+            if (observer.on_failure && verb == "read" && isStoreAnswer(e))
+                observer.on_failure("GET", subject, e);
             bool refreshed = false;
             if (refreshAndClassifyReadFault(e, refresh_attempted, refreshed))
                 throw;

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp

@@ -5,6 +5,7 @@
 #include <Common/logger_useful.h>
 #include <Common/setThreadName.h>
 #include <Common/thread_local_rng.h>
+#include <base/scope_guard.h>
 #include <algorithm>
 #include <chrono>
 #include <ctime>
@@ -420,6 +421,12 @@ uint64_t CasMountRuntime::startRenewer()
     return anchor;
 }
 
+RenewalProgress CasMountRuntime::renewalProgress() const
+{
+    std::lock_guard lock(renewal_progress_mutex);
+    return renewal_progress;
+}
+
 MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_call)
 {
     return MountRenewOperationEnvironment{
@@ -429,6 +436,29 @@ MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_c
             return config.renewal_live_for_test ? config.renewal_live_for_test() : renewalLive(worker_call);
         },
         .cancelled = [this] { return renewalCancelled(); },
+        .requests = RequestObserver{
+            .on_write_attempt = [this](uint32_t attempts_sent)
+            {
+                ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalAttempts);
+                if (attempts_sent > 1)
+                    ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalRetries);
+                std::lock_guard lock(renewal_progress_mutex);
+                renewal_progress.attempts_sent = attempts_sent;
+            },
+            .on_failure = [this](std::string_view verb, const String & key, const std::exception & e)
+            {
+                /// Runs inside the engine's catch; a formatting failure must not change how the renewal ends.
+                try
+                {
+                    String error = describeRequestFailure(verb, key, e);
+                    std::lock_guard lock(renewal_progress_mutex);
+                    renewal_progress.last_error = std::move(error);
+                }
+                catch (...)
+                {
+                }
+            },
+        },
     };
 }
 
@@ -464,13 +494,6 @@ void CasMountRuntime::consumeRenewResult(
 {
     /// Driver ownership has already been restored by `DriverLease::finish`; this is the single logical
     /// consumption boundary and it runs without `driver_mutex` or renewer access.
-    /// The physical counters come off the result rather than off a per-attempt callback, so they count
-    /// the same on every ending: a renewal that gave up still sent what it sent.
-    if (result.attempts_sent > 0)
-    {
-        ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalAttempts, result.attempts_sent);
-        ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalRetries, result.attempts_sent - 1);
-    }
     if (result.resolved_by_read)
         ProfileEvents::incrementNoTrace(ProfileEvents::CASMountRenewalResolved);
 
@@ -534,9 +557,17 @@ uint64_t CasMountRuntime::renewRenewerOnce(
     /// so it renews on the renewer's open plane: admitted under the mount fence it could only ever give
     /// up, and every remount would fail at this step. `RemountCall` is reached from
     /// `renewRenewerForRemountOnce` alone.
-    const MountRenewResult result = active == RenewalDriverState::RemountCall
-        ? call.renewer->renewForRemount(renewalEnvironment(worker_call))
-        : call.renewer->renew(renewalEnvironment(worker_call));
+    {
+        std::lock_guard lock(renewal_progress_mutex);
+        renewal_progress = RenewalProgress{.in_progress = true, .attempts_sent = 0, .last_error = std::nullopt};
+    }
+    MountRenewResult result;
+    {
+        SCOPE_EXIT({ std::lock_guard lock(renewal_progress_mutex); renewal_progress.in_progress = false; });
+        result = active == RenewalDriverState::RemountCall
+            ? call.renewer->renewForRemount(renewalEnvironment(worker_call))
+            : call.renewer->renew(renewalEnvironment(worker_call));
+    }
     const RenewalDriverState destination = active == RenewalDriverState::WorkerCall
         ? RenewalDriverState::WorkerIdle
         : (active == RenewalDriverState::RemountCall ? RenewalDriverState::Parked : RenewalDriverState::Dormant);

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h

@@ -7,6 +7,7 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Primitives/CasEvent.h>
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h>
 #include <Common/ThreadPool.h>
+#include <base/defines.h>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -134,6 +135,14 @@ struct MountFence
     std::atomic<bool> lost{false};
 };
 
+/// The current logical renewal of the mount lease so far, or the last one once it ended.
+struct RenewalProgress
+{
+    bool in_progress = false;
+    uint32_t attempts_sent = 0;
+    std::optional<String> last_error;
+};
+
 /// Owns the live writer-incarnation mechanics shared by the pool's mount and recovery orchestration:
 /// the `MountLeaseRenewer`, local `MountFence`, build watermark and in-flight build registry,
 /// `live_writer_epoch`, unclean-boundary marker, and both persistent workers. `Pool` retains the higher-level
@@ -172,6 +181,8 @@ class CasMountRuntime
     /// Renew the merged mount heartbeat once, including its build-watermark floor. A read-only runtime
     /// has no renewer and fails with a logical exception rather than fabricating a heartbeat.
     void renewWatermarkOnce();
+    /// Takes only its own lock, so a renewal stuck in a request never holds up the reader.
+    RenewalProgress renewalProgress() const;
 
     /// ---- local write fence ----
     /// Return whether a mutable operation may start under the locally observed lease state.
@@ -508,6 +519,9 @@ class CasMountRuntime
     uint64_t remount_handled_generation = 0;
     ThreadFromGlobalPool renewal_worker;
     ThreadFromGlobalPool remount_worker;
+    /// A leaf lock, never held across I/O.
+    mutable std::mutex renewal_progress_mutex;
+    RenewalProgress renewal_progress TSA_GUARDED_BY(renewal_progress_mutex);
     /// Counted entries into `scheduleRemount`; retained as a test-only observability seam.
     std::atomic<uint64_t> schedule_remount_calls_for_test{0};
 

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h

@@ -545,6 +545,7 @@ class Pool : public std::enable_shared_from_this<Pool>
         time_t since = 0;
     };
     LifecycleSnapshot lifecycleSnapshot() const;
+    RenewalProgress renewalProgress() const { return mount_runtime.renewalProgress(); }
 
     /// `SYSTEM CAS FORGET` — the operator force-Vanish (spec §5). Drives THIS pool to
     /// `Vanished(forgotten)` with the fence-first protocol, node-locally, regardless of the current

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp

@@ -1629,6 +1629,7 @@ MountRenewResult MountLeaseRenewer::renewOn(
     result.attempt_start_boot_ms = attempt_start_boot_ms;
 
     CasOperation op = plane.admit(environment.live);
+    op.observeRequests(environment.requests);
     std::optional<WriteResult> written;
     try
     {

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h

@@ -71,6 +71,7 @@ struct MountRenewOperationEnvironment
     /// `NotAttempted` rather than terminal only when this node had already been asked to stop --
     /// sampling it afterwards would read a flag that the refusal itself may have set.
     std::function<bool()> cancelled;
+    RequestObserver requests;
 };
 
 /// Validate a `server_root_id` — the explicit, configured identity of the content-addressed layout

src/Disks/tests/gtest_cas_heartbeat.cpp

@@ -181,6 +181,7 @@ MountRenewOperationEnvironment renewalEnvironment(
         .boot_ms = [&boot_ms] { return boot_ms; },
         .live = live,
         .cancelled = cancelled,
+        .requests = {},
     };
 }