diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index d6e53da9ce99b..6ee68f6ac3ebe 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -278,6 +278,7 @@ becomes `state = 'corrupt'`, never an exception). Shows every `server_root_id` i | `state` | one of `live`, `expired`, `terminated`, `fenced`, `corrupt` | | `is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count` | GC health, process-local; **`NULL` on every peer row** — a process-local fact must never be stamped onto another server's row | | `lifecycle`, `lifecycle_reason`, `lifecycle_detail`, `lifecycle_since` | the SQL surface for the in-process `PoolLifecycle` runtime above: `lifecycle` is one of `live`, `not_live`, `identity_lost`, `vanished`, `constructing`, `shutdown`; `lifecycle_reason` distinguishes `replaced` from `forgotten` for a `vanished` disk; `lifecycle_detail` carries the full diagnosis text; `lifecycle_since` is when the current non-live state began (`NULL` while live) | +| `renewal_in_progress`, `renewal_attempts`, `last_renewal_error` | the running (or last) lease renewal: conditional `PUT`s sent so far and the last failed request, updated per attempt; **`NULL` on every peer row** | The lifecycle snapshot is I/O-free and ungated, so a not-live, never-started, or vanished disk still produces a row instead of silently disappearing from the table. diff --git a/docs/en/operations/system-tables/cas_mounts.md b/docs/en/operations/system-tables/cas_mounts.md index 5a9465db04af6..4a61e48f80f04 100644 --- a/docs/en/operations/system-tables/cas_mounts.md +++ b/docs/en/operations/system-tables/cas_mounts.md @@ -41,6 +41,9 @@ rows. - `lifecycle_reason` ([String](/sql-reference/data-types/string)) — The enum-clean sub-state word for a `vanished` disk: `replaced` or `forgotten`. Empty for every other lifecycle, so `lifecycle || '(' || lifecycle_reason || ')'` reads e.g. `vanished(forgotten)`. - `lifecycle_detail` ([String](/sql-reference/data-types/string)) — The full typed reason text naming the actual cause when not live: the vanish diagnosis (a data root replaced by a foreign pool, or decommissioned by `SYSTEM CAS FORGET` at a given time) or the identity-loss message. Empty when live. - `lifecycle_since` ([Nullable(DateTime)](/sql-reference/data-types/nullable)) — When this server entered the current non-live lifecycle state. `NULL` when live, or when the state has no backing pool to date from. +- `renewal_in_progress` ([Nullable(UInt8)](/sql-reference/data-types/nullable)) — `1` while this server renews its mount lease. +- `renewal_attempts` ([Nullable(UInt64)](/sql-reference/data-types/nullable)) — Conditional `PUT`s sent by the current renewal, or by the last one once it ended. Updated per attempt, so a renewal stuck in retries is visible before it ends. +- `last_renewal_error` ([Nullable(String)](/sql-reference/data-types/nullable)) — The last failed request of that renewal, as ` : `. `class` is `refused` when the store definitely refused the request (access denied, malformed request, entity too large), `transport` otherwise; `code` is the store's error name, such as `SlowDown`. `NULL` when no request failed. `lifecycle`/`lifecycle_reason`/`lifecycle_detail`/`lifecycle_since` are the SQL surface for diagnosing an identity-lost or forgotten disk without reading server logs — see the @@ -49,9 +52,10 @@ diagnosing an identity-lost or forgotten disk without reading server logs — se what each lifecycle value means, and [`SYSTEM CAS FORGET`](/antalya/cas/operations/debugging#sql-forget) for the command that produces `vanished(forgotten)`. -:::note Local-only GC-health columns -`is_leader`, `pending_reclaim`, `last_success_age_seconds`, and `wedged_namespace_count` are process-local -facts about *this* server's own GC scheduler. They are populated **only** on the row whose `server_root_id` matches +:::note Local-only columns +`is_leader`, `pending_reclaim`, `last_success_age_seconds`, `wedged_namespace_count`, +`renewal_in_progress`, `renewal_attempts` and `last_renewal_error` are process-local facts about +*this* server. They are populated **only** on the row whose `server_root_id` matches this server's own mount, and are `NULL` on every row describing another server's mount — stamping a local health fact onto a peer's row would misread as "the peer is the GC leader" during an incident. To see the peer's own view of these columns, query `system.cas_mounts` on that server. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp index a1366b6d61cd6..6ea453eb1005e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp @@ -1,5 +1,6 @@ #include +#include #include #include #include @@ -278,6 +279,27 @@ bool isFirstAttemptFuseTimeout([[maybe_unused]] const std::exception & e, [[mayb #endif } +bool isStoreAnswer(const std::exception & e) +{ + const auto * db = dynamic_cast(&e); + return dynamic_cast(&e) && !(db && isDeterministicLocalFailure(db->code())); +} + +String describeRequestFailure(std::string_view verb, const String & key, const std::exception & e) +{ + const auto & poco = dynamic_cast(e); + String code; +#if USE_AWS_S3 + if (const auto * s3 = dynamic_cast(&e)) + code = s3->getExceptionName(); +#endif + if (const auto * db = dynamic_cast(&e); code.empty() && db) + code = String(ErrorCodes::getName(db->code())); + if (code.empty()) + code = poco.name(); + return fmt::format("{} {}: {} {} {}", verb, key, isDefinitelyRefusedWrite(e) ? "refused" : "transport", code, poco.message()); +} + CasRequests::CasRequests(BackendPtr backend_, Fence fence_, std::function now_ms_, std::function sleep_ms_, CasHotKeys * hot_keys_) @@ -924,6 +946,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co detail::recordAttempt(); ++state.attempts_sent; state.sent_any = true; + if (observer.on_write_attempt) + observer.on_write_attempt(state.attempts_sent); /// Disengaged means the attempt threw: its fate is unproven, and nothing may be read out of it. std::optional> outcome; @@ -952,6 +976,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co { if (isDeterministicLocalFailure(e.code())) throw; + if (observer.on_failure) + observer.on_failure("PUT", key, e); /// ONE refresh per call, only for the class a credential could explain, and only when a /// reissue could sign with what it installs. So an oversized entity never triggers a /// re-acquisition, a denial fresh credentials do not fix is refused on the second look @@ -1008,6 +1034,8 @@ WriteResult CasOperation::writeLoop(const String & key, const String & bytes, co /// still have landed, so `outcome` stays disengaged and the read below settles it. if (!dynamic_cast(&e)) throw; + if (observer.on_failure) + observer.on_failure("PUT", key, e); } if (outcome && outcome->has_value()) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h index 136f9d5d1e24e..2509077b47907 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h +++ b/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; +/// Live progress of one operation, called on its thread inside the engine's retry loops; must not throw. +struct RequestObserver +{ + /// A conditional write is about to send its `attempts_sent`-th physical attempt. + std::function on_write_attempt; + /// A conditional `PUT` or a body `GET` failed with a store answer (`isStoreAnswer`). + std::function on_failure; +}; + namespace detail { /// The engine's per-attempt counters, behind functions so the header need not declare the events. @@ -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 last_read_stop; + RequestObserver observer; }; template @@ -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; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp index e89102e9c5e8c..b073de81f7d2c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -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); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h index 7f3ff1d76f36d..04ec8491f84b7 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasMountRuntime.h @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -134,6 +135,14 @@ struct MountFence std::atomic lost{false}; }; +/// The current logical renewal of the mount lease so far, or the last one once it ended. +struct RenewalProgress +{ + bool in_progress = false; + uint32_t attempts_sent = 0; + std::optional last_error; +}; + /// Owns the live writer-incarnation mechanics shared by the pool's mount and recovery orchestration: /// the `MountLeaseRenewer`, local `MountFence`, build watermark and in-flight build registry, /// `live_writer_epoch`, unclean-boundary marker, and both persistent workers. `Pool` retains the higher-level @@ -172,6 +181,8 @@ class CasMountRuntime /// Renew the merged mount heartbeat once, including its build-watermark floor. A read-only runtime /// has no renewer and fails with a logical exception rather than fabricating a heartbeat. void renewWatermarkOnce(); + /// Takes only its own lock, so a renewal stuck in a request never holds up the reader. + RenewalProgress renewalProgress() const; /// ---- local write fence ---- /// Return whether a mutable operation may start under the locally observed lease state. @@ -508,6 +519,9 @@ class CasMountRuntime uint64_t remount_handled_generation = 0; ThreadFromGlobalPool renewal_worker; ThreadFromGlobalPool remount_worker; + /// A leaf lock, never held across I/O. + mutable std::mutex renewal_progress_mutex; + RenewalProgress renewal_progress TSA_GUARDED_BY(renewal_progress_mutex); /// Counted entries into `scheduleRemount`; retained as a test-only observability seam. std::atomic schedule_remount_calls_for_test{0}; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index c566a9164f1a7..0a678e067e2cc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -545,6 +545,7 @@ class Pool : public std::enable_shared_from_this time_t since = 0; }; LifecycleSnapshot lifecycleSnapshot() const; + RenewalProgress renewalProgress() const { return mount_runtime.renewalProgress(); } /// `SYSTEM CAS FORGET` — the operator force-Vanish (spec §5). Drives THIS pool to /// `Vanished(forgotten)` with the fence-first protocol, node-locally, regardless of the current diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp index b88c02dd4af1c..74693e8e4d9b1 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.cpp @@ -1629,6 +1629,7 @@ MountRenewResult MountLeaseRenewer::renewOn( result.attempt_start_boot_ms = attempt_start_boot_ms; CasOperation op = plane.admit(environment.live); + op.observeRequests(environment.requests); std::optional written; try { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h index 5e3aecc4b9194..9cc34bf54efbc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasServerRoot.h @@ -71,6 +71,7 @@ struct MountRenewOperationEnvironment /// `NotAttempted` rather than terminal only when this node had already been asked to stop -- /// sampling it afterwards would read a flag that the refusal itself may have set. std::function cancelled; + RequestObserver requests; }; /// Validate a `server_root_id` — the explicit, configured identity of the content-addressed layout diff --git a/src/Disks/tests/gtest_cas_heartbeat.cpp b/src/Disks/tests/gtest_cas_heartbeat.cpp index c39935d7ca7f7..db8ded0afd6d8 100644 --- a/src/Disks/tests/gtest_cas_heartbeat.cpp +++ b/src/Disks/tests/gtest_cas_heartbeat.cpp @@ -181,6 +181,7 @@ MountRenewOperationEnvironment renewalEnvironment( .boot_ms = [&boot_ms] { return boot_ms; }, .live = live, .cancelled = cancelled, + .requests = {}, }; } diff --git a/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp b/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp new file mode 100644 index 0000000000000..805f94a15bfa7 --- /dev/null +++ b/src/Disks/tests/gtest_cas_mount_renewal_progress.cpp @@ -0,0 +1,107 @@ +#include + +#include +#include +#include +#include + +#include + +#include "config.h" + +#if USE_AWS_S3 +#include +#endif + +namespace ProfileEvents +{ +extern const Event CASMountRenewalAttempts; +} + +using namespace DB::Cas; + +namespace +{ + +uint64_t attemptsCounter() +{ + return ProfileEvents::global_counters[ProfileEvents::CASMountRenewalAttempts].load(); +} + +/// Fixed boot clock and no-op sleep: every armed fault is finite, so retries end without time passing. +struct Fixture +{ + std::shared_ptr backend = std::make_shared(); + PoolPtr store; + String mount_key; + + explicit Fixture(const String & prefix) + { + backend->setAttemptTimeoutMs(10); + store = Pool::open(backend, PoolConfig{ + .pool_prefix = prefix, + .server_root_id = "test", + .mount_lease_ttl_ms = std::chrono::milliseconds(1000), + .cas_request_budget = CasRequestBudget{.attempt_timeout_ms = 10, .lease_safety_margin_ms = 20, .connect_timeout_cap_ms = std::nullopt}, + .boot_ms_fn = [] { return uint64_t{100}; }, + .retry_sleep_fn = [](uint64_t) {}, + }); + mount_key = store->layout().mountKey("test"); + } +}; + +} + +TEST(CASMountRenewalProgress, AttemptsAndLastErrorAreVisibleWhileTheRenewalRuns) +{ + Fixture f("renewal-progress-in-flight"); + const auto timeout = std::make_exception_ptr(Poco::TimeoutException("injected renewal timeout")); + f.backend->failNextWriteWith(f.mount_key, timeout); + f.backend->failNextWriteWith(f.mount_key, timeout); + const uint64_t attempts_before = attemptsCounter(); + /// Read at the third attempt, the first that reaches the store. Heap state and a raw pointer: the + /// hook must not keep the pool alive, and it is removed before the pool's farewell write. + auto in_flight = std::make_shared>>(); + Pool * pool = f.store.get(); + f.backend->onBeforeWrite(f.mount_key, [in_flight, pool] + { + if (!*in_flight) + *in_flight = std::make_pair(pool->renewalProgress(), attemptsCounter()); + }); + + f.store->renewWatermarkOnce(); + f.backend->onBeforeWrite(f.mount_key, {}); + + ASSERT_TRUE(in_flight->has_value()); + const auto & [seen, attempts_counted] = **in_flight; + EXPECT_TRUE(seen.in_progress); + EXPECT_EQ(seen.attempts_sent, 3u); + EXPECT_EQ(seen.last_error, "PUT " + f.mount_key + ": transport Timeout injected renewal timeout"); + EXPECT_EQ(attempts_counted - attempts_before, 3u) << "attempts are counted as they are sent"; + + const RenewalProgress ended = f.store->renewalProgress(); + EXPECT_FALSE(ended.in_progress); + EXPECT_EQ(ended.attempts_sent, 3u); + EXPECT_EQ(ended.last_error, seen.last_error) << "an ended renewal keeps its progress"; + + f.store->renewWatermarkOnce(); + const RenewalProgress next = f.store->renewalProgress(); + EXPECT_EQ(next.attempts_sent, 1u); + EXPECT_FALSE(next.last_error.has_value()) << "the next renewal starts clean"; +} + +#if USE_AWS_S3 +TEST(CASMountRenewalProgress, AStoreRefusalIsClassedRefused) +{ + Fixture f("renewal-progress-refused"); + f.backend->failNextWriteWith(f.mount_key, std::make_exception_ptr( + DB::S3Exception("the store refused the renewal", Aws::S3::S3Errors::ACCESS_DENIED, "AccessDenied"))); + + EXPECT_THROW(f.store->renewWatermarkOnce(), DB::Exception); + + const RenewalProgress ended = f.store->renewalProgress(); + EXPECT_FALSE(ended.in_progress) << "a renewal that threw has still ended"; + EXPECT_EQ(ended.attempts_sent, 1u); + EXPECT_EQ(ended.last_error, "PUT " + f.mount_key + ": refused AccessDenied the store refused the renewal"); +} +#endif diff --git a/src/Disks/tests/gtest_cas_request_observer.cpp b/src/Disks/tests/gtest_cas_request_observer.cpp new file mode 100644 index 0000000000000..3aa6565803278 --- /dev/null +++ b/src/Disks/tests/gtest_cas_request_observer.cpp @@ -0,0 +1,83 @@ +#include + +#include +#include +#include +#include + +#include + +#include + +#include +#include + +namespace DB::ErrorCodes +{ +extern const int BAD_ARGUMENTS; +extern const int NETWORK_ERROR; +} + +using namespace DB::Cas; + +namespace +{ + +/// One seeded key and an operation whose observer records every call as a line. +struct Fixture +{ + std::shared_ptr backend = std::make_shared(); + CasRequests requests{backend, Fence::open(), [] { return uint64_t{100}; }, [](uint64_t) {}}; + Etag etag = std::get(requests.admit().create("k", "v1", Retry::standard())).etag; + std::vector events; + CasOperation op = requests.admit(); + + Fixture() + { + op.observeRequests(RequestObserver{ + .on_write_attempt = [this](uint32_t attempts_sent) { events.push_back(fmt::format("attempt {}", attempts_sent)); }, + .on_failure = [this](std::string_view verb, const String & key, const std::exception & e) + { + events.push_back(describeRequestFailure(verb, key, e)); + }, + }); + } +}; + +} + +TEST(CASRequestObserver, WriteAttemptsAndFailuresAreReportedAsTheyHappen) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(Poco::TimeoutException("injected write timeout"))); + f.backend->failNextReadWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::NETWORK_ERROR, "injected read failure"))); + + ASSERT_TRUE(std::holds_alternative(f.op.replace("k", "v2", f.etag, Retry::standard()))); + EXPECT_EQ(f.events, (std::vector{ + "attempt 1", + "PUT k: transport Timeout injected write timeout", + "GET k: transport NETWORK_ERROR injected read failure", + "attempt 2", + })); +} + +/// A local bug is not an answer from the store, so it is not published as the request's error. +TEST(CASRequestObserver, ALocalFailureIsNotReportedAsAStoreAnswer) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "injected local failure"))); + + EXPECT_THROW((void)f.op.replace("k", "v2", f.etag, Retry::standard()), DB::Exception); + EXPECT_EQ(f.events, (std::vector{"attempt 1"})); +} + +/// A local failure of the resolve read is not published either, only the store answer before it. +TEST(CASRequestObserver, ALocalFailureOfTheResolveReadIsNotReported) +{ + Fixture f; + f.backend->failNextWriteWith("k", std::make_exception_ptr(Poco::TimeoutException("injected write timeout"))); + f.backend->failNextReadWith("k", std::make_exception_ptr(DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "injected local failure"))); + + EXPECT_THROW((void)f.op.replace("k", "v2", f.etag, Retry::standard()), DB::Exception); + EXPECT_EQ(f.events, (std::vector{"attempt 1", "PUT k: transport Timeout injected write timeout"})); +} diff --git a/src/Storages/System/StorageSystemContentAddressedMounts.cpp b/src/Storages/System/StorageSystemContentAddressedMounts.cpp index ba8b8670cc105..79285618518c0 100644 --- a/src/Storages/System/StorageSystemContentAddressedMounts.cpp +++ b/src/Storages/System/StorageSystemContentAddressedMounts.cpp @@ -57,6 +57,9 @@ StorageSystemContentAddressedMounts::StorageSystemContentAddressedMounts(const S {"lifecycle_reason", std::make_shared(), "The enum-clean sub-state word for a vanished disk: replaced or forgotten. Empty for every other lifecycle (so lifecycle || '(' || lifecycle_reason || ')' reads e.g. vanished(forgotten))."}, {"lifecycle_detail", std::make_shared(), "The full typed reason text naming the actual cause when not live: the vanish diagnosis (data root replaced by a foreign pool / decommissioned by SYSTEM CAS FORGET at