diff --git a/docs/en/antalya/cas/architecture/mounts-and-leases.md b/docs/en/antalya/cas/architecture/mounts-and-leases.md index 6ee68f6ac3ebe..c707abd30626c 100644 --- a/docs/en/antalya/cas/architecture/mounts-and-leases.md +++ b/docs/en/antalya/cas/architecture/mounts-and-leases.md @@ -81,14 +81,17 @@ watermark — there is no separate watermark object. `MountLease` fields: `serve after a flat pause and settled by the reissue's own answer (a 2xx) or by the exact `GET` that follows its `412`. The renewer adopts the result only when the complete body, including `write_attempt_id`, equals its immutable request. If the predecessor token is still current, another identical `PUT` - may follow bounded backoff. A same-pair twin, GC-fenced body, successor, foreign holder, or absent + follows after a pause. A same-pair twin, GC-fenced body, successor, foreign holder, or absent body is never treated as this renewal. -- **Absolute deadline.** Renewal uses `CLOCK_BOOTTIME`, not `CLOCK_MONOTONIC`, so a VM resumed from - suspend correctly observes itself expired. Its absolute deadline is the minimum of the existing - request-operation budget and the last confirmed lease deadline minus the safety margin. The - controller checks that one attempt envelope still fits before each backend `PUT` or resolving - `GET`, after each interruptible backoff, and before accepting success. A retry, `GET`, response - timestamp, or wall-clock step never extends authority. +- **Deadline from the start.** Renewal uses `CLOCK_BOOTTIME`, not `CLOCK_MONOTONIC`, so a VM resumed + from suspend correctly observes itself expired. A confirmed renewal sets the deadline to its own + start plus the lease; a retry, `GET`, response timestamp, or wall-clock step never moves it. The + background renewal has no deadline of its own: it retries one body every 0.8–1.2 s (measured from + the start of the previous attempt) until the store answers about the slot, or until a stop, a + remount park request, a lost fence or a pool leaving `Live` ends it. A + refusal (`AccessDenied`, `MalformedXML`, …) sends the same body again one renewal period later. + The bounded renewals — the startup re-anchor, the remount re-anchor, and the background renewal of a + pool opened for decommission — stay bounded by the last confirmed lease minus the safety margin. - **Cadence.** The runtime normally starts a logical renewal every `cas_mount_renew_period_ms` (default 10 s), with TTL `cas_mount_lease_ttl_ms` (default 30 s, TTL/3 renewal ratio). The next beat is anchored at the committed body's pre-I/O BOOTTIME start. A slow recovery therefore causes an immediate @@ -103,19 +106,32 @@ watermark — there is no separate watermark object. `MountLease` fields: `serve **Losing the lease is neither read-only mode nor a process abort.** `MountLeaseRenewer` is a synchronous durable-slot state machine. A committed result advances its token, sequence, confirmed -BOOTTIME deadline, and cadence anchor. Any admitted deterministic failure, confirmed conflict, or -ambiguity left at the deadline/attempt limit moves it to `RenewalTerminal`; it cannot mint another -body or publish a clean farewell. Owner cancellation before any request is the only -`NotAttempted` result and leaves clean release possible. Cancellation after a request was sent is -terminal because that request may still land. +BOOTTIME deadline, and cadence anchor. A deterministic failure, a confirmed conflict, or, for a +bounded renewal, ambiguity left at its deadline moves it to `RenewalTerminal`; it cannot mint another +body or publish a clean farewell. Owner cancellation before any request, or between two calls of the +background renewal after a refusal or after a read that found the slot unchanged, is `NotAttempted` and +leaves clean release possible. Cancellation after a request of the current call was sent, or while the +reads after a refused precondition repeat, is terminal: that request may still land, or the slot may +have moved. + +**An expired lease is a pause.** While the store does not answer, writes stop being admitted once the +remaining lease cannot cover one write: `2 × envelope + safety_margin` before the deadline. From that +moment `system.cas_mounts` shows `lifecycle = 'not_live'` and `lifecycle_reason = 'lease_expired'`, +and the server logs one `WARNING`, at the next renewal attempt, that ends with the last renewal error: +`CAS mount lease of '' under writer_epoch has too little time left for a write`. +Nothing is remounted. A confirmed renewal resumes writes under the same `writer_epoch` only if its start +plus the lease leaves more than `2 × envelope + safety_margin`; a renewal confirmed later than that +leaves writes paused, and the next one starts at once. A clean shutdown during a pause skips the +farewell, which has no lease room left, and leaves the slot for GC to observe. Such a late renewal stores an `expires_at_ms` already in the past, and GC still does +not fence the slot for it: GC fences only a token it watched unchanged for its whole threshold. After the renewer call returns, `CasMountRuntime` consumes the result. A terminal result trips the local fence (latches `lost`, bumps the fence generation, moves the in-process runtime to `TransientNotLive`) and latches one self-remount generation. A confirmed foreign/successor or same-pair conflict remains a typed fail-closed error; it is never adopted. A real fence still costs only an epoch: recovery reclaims with a fresh one, bounded at three whole-chain attempts. This is the -general CAS posture: doubt about the source fails closed, while transport ambiguity may retry only -inside authority already proved by the last confirmed lease. +general CAS posture: doubt about the source fails closed, while transport ambiguity is retried until a +definitive answer and writes are admitted only inside the last confirmed lease. GC's own view of a dead server is symmetric and clock-skew-immune: a slot becomes fence-eligible only after the leader observes the *same* renewal token hold stable, on its own monotonic clock, diff --git a/docs/en/antalya/cas/operations/debugging.md b/docs/en/antalya/cas/operations/debugging.md index 4e9d4b335b7f9..e8c6665a166a0 100644 --- a/docs/en/antalya/cas/operations/debugging.md +++ b/docs/en/antalya/cas/operations/debugging.md @@ -129,19 +129,21 @@ row followed by a `recovered` one — and the older `unresolved_reason`, `deadli `stop_cause` keys are gone; everything they used to distinguish is now named directly by `classification`. Interpret the sequence as follows: -- `outcome = 'recovered'` means an in-budget renewal landed, in the same epoch. `classification` +- `outcome = 'recovered'` means a renewal landed after a retry or a settling read, in the same epoch. + Writes are re-admitted only if it left more than `2 × envelope + safety_margin` of lease. `classification` says how: `committed_by_read` means an exact `GET` proved a landed request; `committed_after_retry` means a later identical physical `PUT` completed and the response itself proved it. -- `outcome = 'failed'` carries the decisive `classification`: `external_lease_deadline` (the - confirmed lease's own safety margin, not the request policy, ran out first — check object-store - latency or `BOOTTIME` advancement before anything else), `request_deadline` (the ninety-second - request policy exhausted first), `unresolved` (every attempt was ambiguous and never settled by +- `outcome = 'failed'` carries the decisive `classification`: `external_lease_deadline` (a + bounded renewal only: the confirmed lease's own safety margin, not the request policy, ran out first — check object-store + latency or `BOOTTIME` advancement before anything else), `request_deadline` (a bounded + renewal only: the ninety-second request policy exhausted first), `unresolved` (every attempt was ambiguous and never settled by the time the operation gave up), `conflict` (an exact resolve read found another body — a same-pair twin, a GC-fenced body, a successor epoch, or a foreign holder), `cancelled` (a renewal in flight was cancelled by shutdown or a remount park request; expected during graceful shutdown), `fence_or_lifecycle_lost` (another local fence loss or a terminal lifecycle transition - closed admission while the operation was active), `deterministic_failure` (the store's own - answer proved the write never applied), and `vanished` (an exact resolve read proved the mount + closed admission while the operation was active), `deterministic_failure` (a local + failure, or, for a bounded renewal, a store answer proving the write never applied; + the background renewal sends a refused write again), and `vanished` (an exact resolve read proved the mount slot absent — the pool directory was removed or renamed out of band, or a decommission raced the renewal). `terminal_unclassified` means the renewal terminated through a path that assigned no classification; that is a defect to report together with the surrounding rows, not an operator diff --git a/docs/en/antalya/cas/operations/monitoring.md b/docs/en/antalya/cas/operations/monitoring.md index 4a104120697b5..0984719047ba3 100644 --- a/docs/en/antalya/cas/operations/monitoring.md +++ b/docs/en/antalya/cas/operations/monitoring.md @@ -58,7 +58,7 @@ window and correlate them with the `server_root_id` in `system.cas_log`. | `CASMountRenewalRetries` | One per physical renewal `PUT` after the first in the same logical renewal | Positive growth shows in-period retry, not a later cadence beat | | `CASMountRenewalResolved` | One per logical renewal proved committed by an exact resolving `GET` | A response was ambiguous, but exact bytes and `write_attempt_id` proved the write | | `CASMountRenewalRecovered` | One per logical renewal committed after a retry or exact resolving `GET` | Recovered object-store blips that retained the existing mount incarnation | -| `CASMountRenewalDeadlineExceeded` | One per logical renewal stopped by the external lease-safety deadline | The last confirmed lease no longer left enough safe time; this is narrower than request-budget exhaustion | +| `CASMountRenewalDeadlineExceeded` | One per bounded logical renewal (startup or remount re-anchor, or a decommission's background renewal) stopped by the external lease-safety deadline | Growth means a bounded renewal found too little confirmed lease left; a mount's own background renewal never counts here | | `CASRemountAttempts` | One per invocation of the existing whole-chain remount attempt | Includes both successful and failed attempts | | `CASRemountSucceeded` | One per whole-chain attempt that restored `Live` under a fresh writer epoch | Must be a subset of `CASRemountAttempts` | | `CASRemountFailed` | One per whole-chain attempt that returned without restoring `Live` | Includes a named step exception or a step that returned transiently | diff --git a/docs/en/antalya/cas/operations/troubleshooting.md b/docs/en/antalya/cas/operations/troubleshooting.md index 1113ec1855dd4..2ec07485084bf 100644 --- a/docs/en/antalya/cas/operations/troubleshooting.md +++ b/docs/en/antalya/cas/operations/troubleshooting.md @@ -38,16 +38,21 @@ Start with the `watermark_renew` timeline described in `CASMountRenewalRecovered` rises while `CASMountLeaseLost` and all remount counters stay flat. No intervention is needed unless the rate is sustained; investigate backend throttling/latency before the blips consume the lease budget. -2. **External lease-safety exhaustion.** The failed row has - `classification = 'external_lease_deadline'`; `CASMountRenewalDeadlineExceeded` and - `CASMountLeaseLost` rise. The runtime correctly refused to manufacture authority beyond the last - confirmed lease. Check object-store latency and BOOTTIME/suspend history, then follow the ensuing - remount. `classification = 'request_deadline'` is the sibling case: the ninety-second request - policy exhausted first rather than the lease's own safety margin. -3. **Cancellation.** `classification = 'cancelled'` after a sent request is terminal and suppresses a - clean farewell because the request may still land. Cancellation before any request remains - `Active` and emits no failed aggregate row; during graceful shutdown that is the expected - clean-release path. +2. **External lease-safety exhaustion.** Only a bounded renewal ends this way: the startup re-anchor, + the remount re-anchor, or the background renewal of a pool opened for decommission. A mount's own + background renewal keeps retrying; a long outage shows as + `lifecycle_reason = 'lease_expired'` and resumes under the same `writer_epoch`. The failed row has + `classification = 'external_lease_deadline'` and `CASMountRenewalDeadlineExceeded` rises. The + runtime refused to claim authority beyond the last confirmed lease. Check object-store latency and + BOOTTIME/suspend history. At startup the disk does not start; during a remount the remount loop + retries. `classification = 'request_deadline'` is the sibling case for the same bounded renewals: + the ninety-second request policy exhausted first rather than the lease's own safety margin. +3. **Cancellation.** `classification = 'cancelled'` after a request of the current engine call was sent + is terminal and suppresses a clean farewell because the request may still land; so is cancellation + while the reads after a refused precondition repeat. Cancellation before any request, or between two + calls of the background renewal after a refusal, remains `Active` and emits no failed aggregate row; + during graceful shutdown that is the expected clean-release path while the lease has room for the + farewell. During a pause it has none: the slot is left for observation. 4. **Confirmed conflict.** `classification = 'conflict'` means exact resolution found another body; inspect `server_root_id`, `writer_epoch`, `seq`, and `write_attempt_id`. Same-pair twins, GC-fenced bodies, successor epochs, and foreign holders all remain fail closed. Do not delete or rewrite the diff --git a/docs/en/antalya/metrics_and_events.md b/docs/en/antalya/metrics_and_events.md index c64074e66c07d..81ad009888c49 100644 --- a/docs/en/antalya/metrics_and_events.md +++ b/docs/en/antalya/metrics_and_events.md @@ -283,10 +283,10 @@ window. | Event | Type | Description | | --- | --- | --- | | `CASMountExclusivityViolation` | `Number` | Counts CAS mount releases where a runtime that still BELIEVED IT OWNED the mount (no deposition ever observed) found a FOREIGN occupant in the slot. This is the single-writer guarantee being broken, not a failover: the release refuses, the slot is left untouched, and the write fence is latched so the runtime stops trusting itself. This must always be zero. | -| `CASMountLeaseLost` | `Number` | 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. | +| `CASMountLeaseLost` | `Number` | 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. | | `CASMountReleaseSkippedForeignOccupant` | `Number` | Counts conclusive CAS mount-renewal observations that found a FOREIGN successor in the slot. This is the EXPECTED end state of a failover: renewal fences the deposed runtime, terminal teardown skips the farewell without backend I/O, and the successor's slot is left byte-for-byte untouched. Steady non-zero values on a cluster that is not failing over are worth investigating; a value that tracks failovers is normal. | | `CASMountRenewalAttempts` | `Number` | Number of physical conditional renewal PUTs sent for CAS mount leases. This counts transport attempts, not logical renewals. | -| `CASMountRenewalDeadlineExceeded` | `Number` | 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. | +| `CASMountRenewalDeadlineExceeded` | `Number` | 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. | | `CASMountRenewalRecovered` | `Number` | Number of logical CAS mount-lease renewals that committed after a physical retry or exact resolving GET. | | `CASMountRenewalResolved` | `Number` | Number of CAS mount-lease renewals whose committed outcome was proved by an exact resolving GET. | | `CASMountRenewalRetries` | `Number` | Number of physical conditional renewal PUTs sent after the first attempt of one logical CAS mount-lease renewal. | diff --git a/docs/en/operations/system-tables/cas_mounts.md b/docs/en/operations/system-tables/cas_mounts.md index 4a61e48f80f04..0e6eba43cc8e4 100644 --- a/docs/en/operations/system-tables/cas_mounts.md +++ b/docs/en/operations/system-tables/cas_mounts.md @@ -38,9 +38,9 @@ rows. - `last_success_age_seconds` ([Nullable(UInt64)](/sql-reference/data-types/nullable)) — Seconds since this disk's GC last led a round (`0` if it has never led or GC is not running here). - `wedged_namespace_count` ([Nullable(UInt64)](/sql-reference/data-types/nullable)) — Ref-append lanes currently wedged on this disk (an uncertain `PUT` exhausted its retry budget). - `lifecycle` ([String](/sql-reference/data-types/string)) — This server's content-addressed pool lifecycle for the disk (a non-gated snapshot, always populated so a not-live disk stays visible): one of `live`, `not_live`, `identity_lost`, `vanished`, `constructing` (never started), or `shutdown` (torn down). -- `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. +- `lifecycle_reason` ([String](/sql-reference/data-types/string)) — The enum-clean sub-state word: `replaced` or `forgotten` for a `vanished` disk, `lease_expired` for a `not_live` disk whose mount lease has too little time left to admit a write while its renewal keeps retrying. 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, and for `not_live` with `lifecycle_reason = 'lease_expired'`. +- `lifecycle_since` ([Nullable(DateTime)](/sql-reference/data-types/nullable)) — When this server entered the current non-live lifecycle state. For `lease_expired`, when writes paused. `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. 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/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/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 b073de81f7d2c..7028e4f657215 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_)) @@ -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); @@ -305,6 +341,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 @@ -369,7 +410,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)); }, @@ -429,6 +470,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] @@ -439,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); @@ -459,6 +507,7 @@ MountRenewOperationEnvironment CasMountRuntime::renewalEnvironment(bool worker_c } }, }, + .refusal_pause_ms = refusal_pause_ms, }; } @@ -483,7 +532,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 +555,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 +585,17 @@ 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); + if (!writesPausedSinceBootMs()) + writes_paused_warned.store(false, std::memory_order_relaxed); +} + uint64_t CasMountRuntime::renewRenewerOnce( AdmittedRenewerCall call, RenewalDriverState active, @@ -564,14 +617,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); @@ -1161,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 04ec8491f84b7..1b69c885d092f 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 @@ -155,10 +161,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_, @@ -315,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 @@ -333,10 +344,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 @@ -453,6 +465,12 @@ 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); + /// 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); @@ -468,7 +486,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; @@ -533,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 83d21a39500d7..25bf63f52a361 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) - , 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 ? 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 @@ -242,13 +247,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(); }) @@ -377,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; } @@ -918,6 +933,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); @@ -1948,19 +1966,19 @@ 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); 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()); } 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..ca9415369eeae 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 = {}; @@ -256,10 +258,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 - /// 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 + /// 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. 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. @@ -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, }; } }; @@ -543,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(); } @@ -760,11 +765,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 @@ -1173,9 +1176,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); }; @@ -1208,7 +1211,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/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/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; }; 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); diff --git a/src/Disks/tests/gtest_cas_decommission.cpp b/src/Disks/tests/gtest_cas_decommission.cpp index 936e9655e9b14..b014e9f6c8c28 100644 --- a/src/Disks/tests/gtest_cas_decommission.cpp +++ b/src/Disks/tests/gtest_cas_decommission.cpp @@ -536,6 +536,7 @@ TEST(CASDecommission, AlwaysRenewsAdminClaimEvenWhenHostDiskIsObserveOnly) backend, PoolConfig{.pool_prefix = "p", .server_root_id = "admin", .background_watermark = false}, "victim"); ASSERT_TRUE(admin != nullptr); EXPECT_TRUE(admin->poolConfig().background_watermark); + EXPECT_TRUE(admin->poolConfig().bounded_background_renewal); } TEST(CASDecommission, RefusesUnknownMember) diff --git a/src/Disks/tests/gtest_cas_heartbeat.cpp b/src/Disks/tests/gtest_cas_heartbeat.cpp index db8ded0afd6d8..3eb6d73f1785b 100644 --- a/src/Disks/tests/gtest_cas_heartbeat.cpp +++ b/src/Disks/tests/gtest_cas_heartbeat.cpp @@ -96,6 +96,9 @@ class RenewalScriptBackend : public InMemoryBackend ReturnThenCancel, ThrowBeforeThenLandAfterResolve, ThrowConnectHint, + ThrowAccessDenied, + ThrowMalformed, + RefusePrecondition, }; struct Attempt @@ -103,12 +106,22 @@ class RenewalScriptBackend : public InMemoryBackend String key; String bytes; std::optional expected; + uint64_t at_ms = 0; }; std::deque actions; std::vector attempts; std::function cancel_after_write; uint64_t read_calls = 0; + /// The injected boot clock, when a test asserts on when each attempt was sent. + const uint64_t * clock = nullptr; + /// While positive, a read of a mount slot fails with an access denial. + uint32_t refuse_reads = 0; + /// How many guarded mount writes had been sent when each read was made. + std::vector puts_at_read; + std::vector read_at_ms; + /// When set, a read of a mount slot that succeeds returns these bytes under the slot's real token. + std::optional mount_read_bytes; /// Only a GUARDED write of a mount slot is scripted; the fixture's own seeding and every other /// key reach the store untouched. @@ -119,7 +132,7 @@ class RenewalScriptBackend : public InMemoryBackend if (!expected_value || !key.ends_with("/mount")) return InMemoryBackend::write(key, bytes, expected_value, access); - attempts.push_back({key, bytes, expected_value}); + attempts.push_back({key, bytes, expected_value, clock ? *clock : 0}); const Action action = actions.empty() ? Action::Delegate : actions.front(); if (!actions.empty()) actions.pop_front(); @@ -133,6 +146,18 @@ class RenewalScriptBackend : public InMemoryBackend throw Poco::TimeoutException("connect timed out"); #endif } + if (action == Action::ThrowAccessDenied || action == Action::ThrowMalformed) + { +#if USE_AWS_S3 + if (action == Action::ThrowAccessDenied) + throw DB::S3Exception("injected renewal access denial", Aws::S3::S3Errors::ACCESS_DENIED); + throw DB::S3Exception("injected malformed renewal request", Aws::S3::S3Errors::INVALID_REQUEST); +#else + throw DB::Exception(DB::ErrorCodes::ABORTED, "store refusals need the S3 client"); +#endif + } + if (action == Action::RefusePrecondition) + return std::unexpected(RawConflict{}); if (action == Action::ThrowBefore || action == Action::ThrowBeforeThenLandAfterResolve) { @@ -156,7 +181,20 @@ class RenewalScriptBackend : public InMemoryBackend std::optional read(const String & key, TransportAccess & access) override { ++read_calls; + puts_at_read.push_back(attempts.size()); + read_at_ms.push_back(clock ? *clock : 0); + if (refuse_reads > 0 && key.ends_with("/mount")) + { + --refuse_reads; +#if USE_AWS_S3 + throw DB::S3Exception("injected read access denial", Aws::S3::S3Errors::ACCESS_DENIED); +#else + throw DB::Exception(DB::ErrorCodes::ABORTED, "store refusals need the S3 client"); +#endif + } std::optional result = InMemoryBackend::read(key, access); + if (result && mount_read_bytes && key.ends_with("/mount")) + result->bytes = *mount_read_bytes; if (pending && pending->key == key) { const Attempt delayed = *pending; @@ -949,6 +987,287 @@ TEST(CASHeartbeat, RenewalRetriesOneImmutableBodyAndAdoptsLostResponse) decodeMountLease(backend->attempts.front().bytes).write_attempt_id); } +/// The background renewal of an adopted slot, with the script and the request records empty. +class CASHeartbeatUntilDefinitive : public ::testing::Test +{ +protected: + void SetUp() override + { + backend->clock = &boot_ms; + seedOwnClaim(ops.op, layout, "test", uuid, 9, wall_ms, 1000); + renewer.start(); + clearRecords(); + } + + void clearRecords() + { + backend->attempts.clear(); + backend->read_calls = 0; + backend->puts_at_read.clear(); + backend->read_at_ms.clear(); + } + + MountRenewResult renew(const std::function & live = {}, const std::function & cancelled = {}) + { + MountRenewOperationEnvironment environment = renewalEnvironment(boot_ms, live, cancelled); + environment.refusal_pause_ms = kRefusalPauseMs; + return renewer.renewUntilDefinitive(environment); + } + + /// Rewrites the slot behind the renewer's back. + void rewriteSlot(const std::function & change) + { + const auto got = ops.op.read(layout.mountKey("test"), Retry::standard()); + MountLease lease = decodeMountLease(got->bytes); + change(lease); + ++lease.seq; + mustCommit(ops.op.replace(layout.mountKey("test"), encodeMountLease(lease), got->etag, Retry::standard()), "rewrite"); + clearRecords(); + } + + /// A refused precondition whose first two reads are refused: the third read decides, and it must + /// end the series as a loss without another PUT. + void expectLossObservedByTheReads() + { + backend->actions = {RenewalScriptBackend::Action::RefusePrecondition}; + backend->refuse_reads = 2; + + const MountRenewResult renewed = renew(); + + const DB::Exception failure = terminalException(renewed); + EXPECT_NE(failure.code(), DB::ErrorCodes::LOGICAL_ERROR); + EXPECT_EQ(renewer.state(), MountLeaseRenewerState::RenewalTerminal); + EXPECT_EQ(backend->attempts.size(), 1u); + EXPECT_EQ(backend->puts_at_read, (std::vector{1, 1, 1})); + } + + static constexpr uint64_t kRefusalPauseMs = 10'000; + static constexpr uint64_t kShortestSpacingMs = Retry::kDefinitiveAttemptSpacingMs - Retry::kDefinitiveAttemptSpacingMs / 5; + static constexpr uint64_t kLongestSpacingMs = Retry::kDefinitiveAttemptSpacingMs + Retry::kDefinitiveAttemptSpacingMs / 5; + + std::shared_ptr backend = std::make_shared(); + Layout layout{"pool"}; + const UInt128 uuid{0x1234}; + uint64_t wall_ms = 1000; + uint64_t boot_ms = 100; + Ops ops{backend, &boot_ms}; + MountLeaseRenewer renewer{ + ops.mount, ops.farewell, layout, "test", uuid, 9, std::chrono::milliseconds(1000), + [this] { return wall_ms; }, [] { return uint64_t{7}; }, {}, std::chrono::milliseconds(20), + [this] { return boot_ms; }}; +}; + +/// The background renewal outlives its lease with one tuple, and confirms its first start. +TEST_F(CASHeartbeatUntilDefinitive, RetriesOneTuplePastTheLease) +{ + for (int i = 0; i < 10; ++i) + backend->actions.push_back(RenewalScriptBackend::Action::ThrowBefore); + const uint64_t start = boot_ms; + + const MountRenewResult renewed = renew(); + + ASSERT_EQ(renewed.outcome, MountRenewOutcome::Committed); + EXPECT_EQ(renewed.attempt_start_boot_ms, start); + EXPECT_EQ(renewed.attempts_sent, 11u); + ASSERT_EQ(backend->attempts.size(), 11u); + EXPECT_GE(backend->attempts.back().at_ms, start + 8'000) << "the renewal kept retrying past the lease"; + for (size_t i = 1; i < backend->attempts.size(); ++i) + { + EXPECT_EQ(backend->attempts[i].bytes, backend->attempts[0].bytes); + EXPECT_EQ(backend->attempts[i].expected, backend->attempts[0].expected); + EXPECT_GE(backend->attempts[i].at_ms - backend->attempts[i - 1].at_ms, kShortestSpacingMs); + EXPECT_LE(backend->attempts[i].at_ms - backend->attempts[i - 1].at_ms, kLongestSpacingMs); + } +} + +/// A renewal confirmed after an outage longer than the lease stores an `expires_at_ms` already past. A GC +/// round that watched the previous token for its whole threshold does not fence the slot: the token changed. +TEST_F(CASHeartbeatUntilDefinitive, ALateCommitRestartsTheGcObservation) +{ + const uint64_t threshold_ms = mountObservationThresholdMs(/*ttl_ms=*/1000, /*cadence_ms=*/1000); + MountObservationMap observations; + const auto gc_round = [&] + { + return computeHeartbeatFloor(ops.op, layout, wall_ms, [this] { return boot_ms; }, threshold_ms, observations); + }; + ASSERT_EQ(gc_round().live, 1u); + clearRecords(); + for (int i = 0; i < 5; ++i) + backend->actions.push_back(RenewalScriptBackend::Action::ThrowBefore); + const uint64_t start = boot_ms; + + ASSERT_EQ(renew().outcome, MountRenewOutcome::Committed); + wall_ms += boot_ms - start; + + ASSERT_GE(boot_ms - start, threshold_ms) << "the GC watched the previous token for its whole threshold"; + const MountLease stored = decodeMountLease(ops.op.read(layout.mountKey("test"), Retry::standard())->bytes); + EXPECT_LT(stored.expires_at_ms, wall_ms) << "the commit stored a lease that already expired"; + const HeartbeatFloor floor = gc_round(); + EXPECT_EQ(floor.fenced_now, 0u); + EXPECT_EQ(floor.live, 1u); + EXPECT_FALSE(decodeMountLease(ops.op.read(layout.mountKey("test"), Retry::standard())->bytes).gc_fenced); +} + +#if USE_AWS_S3 +/// A refusal is not an answer about the slot: the next engine call sends the same tuple, one refusal +/// pause later, whatever the store refused for. +TEST_F(CASHeartbeatUntilDefinitive, ARefusedRenewalIsSentAgainAfterTheRefusalPause) +{ + backend->actions = {RenewalScriptBackend::Action::ThrowMalformed, RenewalScriptBackend::Action::ThrowAccessDenied, + RenewalScriptBackend::Action::Delegate}; + const uint64_t start = boot_ms; + + const MountRenewResult renewed = renew(); + + ASSERT_EQ(renewed.outcome, MountRenewOutcome::Committed); + EXPECT_EQ(renewed.attempt_start_boot_ms, start); + EXPECT_EQ(renewed.attempts_sent, 3u); + ASSERT_EQ(backend->attempts.size(), 3u); + for (size_t i = 1; i < backend->attempts.size(); ++i) + { + EXPECT_EQ(backend->attempts[i].bytes, backend->attempts[0].bytes); + EXPECT_EQ(backend->attempts[i].expected, backend->attempts[0].expected); + EXPECT_EQ(backend->attempts[i].at_ms - backend->attempts[i - 1].at_ms, kRefusalPauseMs) << "attempt " << i; + } + EXPECT_EQ(backend->read_calls, 0u) << "a refusal owes no read"; +} + +/// A refused precondition whose read failed is not a lost slot: only the read repeats, one spacing after +/// the write and one refusal pause after each refused read, and a read that finds the slot as the last +/// confirmed renewal left it sends the same tuple again. +TEST_F(CASHeartbeatUntilDefinitive, AfterACleanPreconditionRefusalOnlyTheReadRepeats) +{ + backend->actions = {RenewalScriptBackend::Action::RefusePrecondition, RenewalScriptBackend::Action::Delegate}; + backend->refuse_reads = 3; + + const MountRenewResult renewed = renew(); + + ASSERT_EQ(renewed.outcome, MountRenewOutcome::Committed); + ASSERT_EQ(backend->attempts.size(), 2u); + EXPECT_EQ(backend->attempts[1].bytes, backend->attempts[0].bytes); + EXPECT_EQ(backend->attempts[1].expected, backend->attempts[0].expected); + EXPECT_EQ(backend->puts_at_read, (std::vector{1, 1, 1, 1})) << "no PUT between the reads"; + ASSERT_EQ(backend->read_at_ms.size(), 4u); + EXPECT_GE(backend->read_at_ms[1] - backend->read_at_ms[0], kShortestSpacingMs); + EXPECT_LE(backend->read_at_ms[1] - backend->read_at_ms[0], kLongestSpacingMs); + EXPECT_EQ(backend->read_at_ms[2] - backend->read_at_ms[1], kRefusalPauseMs); + EXPECT_EQ(backend->read_at_ms[3] - backend->read_at_ms[2], kRefusalPauseMs); +} + +/// A read that finally observes a slot that moved ends the series as a loss, with no further PUT. +TEST_F(CASHeartbeatUntilDefinitive, AReadThatFindsAForeignHolderEndsTheSeries) +{ + rewriteSlot([](MountLease & lease) { lease.server_uuid = UInt128{0x9999}; }); + expectLossObservedByTheReads(); +} + +TEST_F(CASHeartbeatUntilDefinitive, AReadThatFindsOurSlotFencedEndsTheSeries) +{ + rewriteSlot([](MountLease & lease) { lease.gc_fenced = true; }); + expectLossObservedByTheReads(); +} + +TEST_F(CASHeartbeatUntilDefinitive, AReadThatFindsTheSlotDeletedEndsTheSeries) +{ + const auto got = ops.op.read(layout.mountKey("test"), Retry::standard()); + ASSERT_EQ(ops.op.remove(layout.mountKey("test"), got->etag, Retry::standard()), Removal::Removed); + clearRecords(); + expectLossObservedByTheReads(); +} + +/// Our token over bytes that are not our last confirmed body is not "the slot did not move". +TEST_F(CASHeartbeatUntilDefinitive, OurTokenOverOtherBytesEndsTheSeries) +{ + MountLease other = decodeMountLease(ops.op.read(layout.mountKey("test"), Retry::standard())->bytes); + other.write_attempt_id = UInt128{0xABCD}; + ++other.seq; + backend->mount_read_bytes = encodeMountLease(other); + clearRecords(); + expectLossObservedByTheReads(); +} + +/// A stop between engine calls ends the series as cancelled: nothing of ours is in flight, the renewer +/// stays active and its farewell still hands the slot back. +TEST_F(CASHeartbeatUntilDefinitive, AStopBetweenCallsEndsTheSeriesAsCancelled) +{ + backend->actions = {RenewalScriptBackend::Action::ThrowMalformed}; + const auto stopped = [&] { return !backend->attempts.empty(); }; + + /// No refusal pause: the farewell has to fit in the lease the last commit confirmed. + const MountRenewResult renewed + = renewer.renewUntilDefinitive(renewalEnvironment(boot_ms, [&] { return !stopped(); }, stopped)); + + EXPECT_EQ(renewed.outcome, MountRenewOutcome::NotAttempted); + EXPECT_EQ(renewer.state(), MountLeaseRenewerState::Active); + EXPECT_EQ(backend->attempts.size(), 1u); + renewer.release(); + const MountLease left = decodeMountLease(ops.op.read(layout.mountKey("test"), Retry::standard())->bytes); + EXPECT_EQ(left.min_active_build_sequence, std::numeric_limits::max()) << "the farewell landed"; +} + +/// A stop while only the read of a refused precondition repeats ends the series as cancelled with no +/// further PUT, and without a farewell: the slot may have moved to a holder entitled to it. +TEST_F(CASHeartbeatUntilDefinitive, AStopDuringTheReadsEndsTheSeriesWithoutAFarewell) +{ + backend->actions = {RenewalScriptBackend::Action::RefusePrecondition}; + backend->refuse_reads = 5; + const auto stopped = [&] { return backend->read_calls >= 2; }; + + const MountRenewResult renewed = renew([&] { return !stopped(); }, stopped); + + const DB::Exception failure = terminalException(renewed); + EXPECT_NE(failure.message().find("(cancelled,"), String::npos) << failure.message(); + EXPECT_EQ(renewer.state(), MountLeaseRenewerState::RenewalTerminal); + EXPECT_FALSE(renewer.canRelease()); + EXPECT_EQ(backend->attempts.size(), 1u); + EXPECT_EQ(backend->puts_at_read, (std::vector{1, 1})); +} + +/// A lost liveness without a stop ends the reads as a loss. +TEST_F(CASHeartbeatUntilDefinitive, ALostLivenessDuringTheReadsEndsTheSeriesAsALoss) +{ + backend->actions = {RenewalScriptBackend::Action::RefusePrecondition}; + backend->refuse_reads = 5; + + const MountRenewResult renewed = renew([&] { return backend->read_calls < 2; }); + + const DB::Exception failure = terminalException(renewed); + EXPECT_NE(failure.message().find("(fence_or_lifecycle_lost,"), String::npos) << failure.message(); + EXPECT_EQ(renewer.state(), MountLeaseRenewerState::RenewalTerminal); + EXPECT_EQ(backend->attempts.size(), 1u); + EXPECT_EQ(backend->puts_at_read, (std::vector{1, 1})); +} + +/// A renewal that ends after an earlier engine call sent an attempt reports that attempt. +TEST_F(CASHeartbeatUntilDefinitive, AGiveUpAfterARefusedCallCountsItsAttempts) +{ + backend->actions = {RenewalScriptBackend::Action::ThrowMalformed}; + + const MountRenewResult renewed = renew([&] { return backend->attempts.empty(); }); + + ASSERT_EQ(renewed.outcome, MountRenewOutcome::Terminal); + EXPECT_EQ(renewed.attempts_sent, 1u); + EXPECT_TRUE(renewed.sent_any); + const DB::Exception e = terminalException(renewed); + EXPECT_NE(e.message().find("at least one attempt sent"), String::npos) << e.message(); +} + +/// A refusal that follows an unclear attempt does not end the engine call: the read settles the +/// unclear attempt, and our attempt that landed meanwhile is adopted, not taken for another writer. +TEST_F(CASHeartbeatUntilDefinitive, ARefusalAfterAnUnclearAttemptIsSettledInsideTheCall) +{ + backend->actions = {RenewalScriptBackend::Action::ThrowBeforeThenLandAfterResolve, + RenewalScriptBackend::Action::ThrowAccessDenied}; + + const MountRenewResult renewed = renew(); + + ASSERT_EQ(renewed.outcome, MountRenewOutcome::Committed); + EXPECT_TRUE(renewed.resolved_by_read); + EXPECT_EQ(renewed.attempts_sent, 2u); + EXPECT_EQ(backend->attempts.size(), 2u); +} +#endif + #if USE_AWS_S3 TEST(CASHeartbeat, RenewalOverConnectFailuresRecoversWithoutASettleRead) { @@ -1058,6 +1377,7 @@ TEST(CASHeartbeat, CancellationAfterSendIsTerminalAndForbidsRelease) renewalEnvironment(boot_ms, /*live=*/[&] { return !cancelled; }, /*cancelled=*/[&] { return cancelled; })); const DB::Exception failure = terminalException(result); EXPECT_EQ(failure.code(), DB::ErrorCodes::NETWORK_ERROR); + EXPECT_NE(failure.message().find("(cancelled,"), String::npos) << failure.message(); EXPECT_TRUE(result.sent_any); EXPECT_EQ(backend->read_calls, 0u) << "post-write cancellation must not start a diagnostic read"; EXPECT_EQ(renewer.state(), MountLeaseRenewerState::RenewalTerminal); diff --git a/src/Disks/tests/gtest_cas_lifecycle_snapshot.cpp b/src/Disks/tests/gtest_cas_lifecycle_snapshot.cpp index ea504c8946414..c1817a76108cb 100644 --- a/src/Disks/tests/gtest_cas_lifecycle_snapshot.cpp +++ b/src/Disks/tests/gtest_cas_lifecycle_snapshot.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -235,3 +236,54 @@ TEST(CASLifecycleSnapshot, NaturalIdentityLostMatchesThrowDetail) << "snapshot detail and the typed error must not drift\n detail: " << snap.detail << "\n thrown: " << thrown; } + +/// An expired lease while the lifecycle stays `Live`: `not_live` / `lease_expired`, `since` is before the +/// deadline, and a deadline back in the future returns the row to `live`. +TEST(CASLifecycleSnapshot, ExpiredLeaseIsNotLiveUntilRestored) +{ + auto storage = openSnapshotStorage(); + commitOnePart(*storage); + auto pool = storage->store(); + const auto wall_now_s = [] + { + return static_cast(std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); + }; + + pool->setMountDeadline(pool->bootMsNow() - 5'000); + const CasLifecycleSnapshot expired = storage->lifecycleSnapshot(); + const int64_t after_s = wall_now_s(); + + EXPECT_EQ(expired.lifecycle, "not_live"); + EXPECT_EQ(expired.reason, "lease_expired"); + EXPECT_GT(static_cast(expired.since), 0); + EXPECT_LE(static_cast(expired.since), after_s - 4); + EXPECT_EQ(pool->lifecycle(), PoolLifecycle::Live) << "the lifecycle enum is not touched"; + + pool->setMountDeadline(pool->bootMsNow() + 60'000); + const CasLifecycleSnapshot restored = storage->lifecycleSnapshot(); + EXPECT_EQ(restored.lifecycle, "live"); + EXPECT_TRUE(restored.reason.empty()) << restored.reason; + EXPECT_EQ(restored.since, 0); +} + +/// The row is `not_live` from the moment the lease has too little time left to admit a write, before the +/// deadline itself, and `since` moves with the deadline. +TEST(CASLifecycleSnapshot, NotLiveOnceTheLeaseCannotAdmitAWrite) +{ + auto storage = openSnapshotStorage(); + commitOnePart(*storage); + auto pool = storage->store(); + + pool->setMountDeadline(pool->bootMsNow() + 3'000); + const CasLifecycleSnapshot closing = storage->lifecycleSnapshot(); + EXPECT_TRUE(pool->mayMutate()) << "the deadline has not passed"; + EXPECT_EQ(closing.lifecycle, "not_live"); + EXPECT_EQ(closing.reason, "lease_expired"); + + pool->setMountDeadline(pool->bootMsNow() - 5'000); + const CasLifecycleSnapshot expired = storage->lifecycleSnapshot(); + const int64_t moved_s = static_cast(closing.since) - static_cast(expired.since); + EXPECT_GE(moved_s, 7); + EXPECT_LE(moved_s, 9); +} diff --git a/src/Disks/tests/gtest_cas_pool.cpp b/src/Disks/tests/gtest_cas_pool.cpp index 64af6da168dc8..1d1b2af336b44 100644 --- a/src/Disks/tests/gtest_cas_pool.cpp +++ b/src/Disks/tests/gtest_cas_pool.cpp @@ -16,10 +16,13 @@ #include #include #include +#include #include #include +#include "config.h" #include #include +#include #include #include #include @@ -2026,6 +2029,16 @@ class RuntimeRenewBackend final : public DB::Cas::tests::CountingBackend /// rather than reissued has to move the injected clock here -- from inside the attempt, which is the /// only point between admission and the resolve read a test can reach. std::function before_throw; + /// While it answers true, every guarded write fails with a transport timeout before `fault` is + /// consulted. Set it before the workers start. + std::function outage; + /// Runs inside every write `outage` fails, before it throws. + std::function on_outage_write; + std::atomic outage_writes{0}; + /// Runs first in every guarded write. + std::function on_guarded_write; + /// While positive, every guarded write is refused with an access denial. + std::atomic deny_writes{0}; /// The fault sits on the WRITE PRIMITIVE, and only on a CONDITIONAL one: a lease renewal is a /// replace, so a create on the same key must not consume the one-shot fault. @@ -2034,6 +2047,24 @@ class RuntimeRenewBackend final : public DB::Cas::tests::CountingBackend { if (!expected_value) return DB::Cas::tests::CountingBackend::write(key, bytes, expected_value, access); + if (on_guarded_write) + on_guarded_write(); + if (deny_writes.load() > 0) + { + deny_writes.fetch_sub(1); +#if USE_AWS_S3 + throw DB::S3Exception("injected mount PUT access denial", Aws::S3::S3Errors::ACCESS_DENIED); +#else + throw DB::Exception(DB::ErrorCodes::ABORTED, "store refusals need the S3 client"); +#endif + } + if (outage && outage()) + { + outage_writes.fetch_add(1, std::memory_order_relaxed); + if (on_outage_write) + on_outage_write(); + throw Poco::TimeoutException("injected mount PUT outage"); + } const Fault current = std::exchange(fault, Fault::None); if (current == Fault::BlockThenDelegate || current == Fault::BlockThenThrow) { @@ -2075,8 +2106,8 @@ class RuntimeUnderTest [this] { return runtime.fenceGeneration(); }, [this](uint64_t g, uint64_t needed) { return runtime.admit(g, needed); }, [this](uint64_t g) { runtime.checkFenceOrThrow(g); }}) - , farewell(backend, DB::Cas::Fence::open()) - , runtime(backend, mount, farewell, std::forward(args)...) + , lease(backend, DB::Cas::Fence::open()) + , runtime(backend, mount, lease, std::forward(args)...) { /// What the request engine reserves per attempt is the BACKEND's attempt timeout, not the /// budget field alone; every construction of this holder pairs the two via `runtimeRenewBudget`, @@ -2087,7 +2118,16 @@ class RuntimeUnderTest /// that injects one MUST inject the other, or `Retry::untilLeaseSafe` compares a synthetic /// deadline against real boottime, finds it long past, and refuses every request unsent. mount.setNowFnForTest([this] { return runtime.bootMsNow(); }); - farewell.setNowFnForTest([this] { return runtime.bootMsNow(); }); + lease.setNowFnForTest([this] { return runtime.bootMsNow(); }); + /// The lease plane's production sleep, so a stop or a park request ends a renewal's pause. + lease.setSleepFnForTest([this](uint64_t ms) { runtime.sleepInterruptibly(ms); }); + } + + /// The retry wait of the mount and lease planes. Call before the workers start. + void setRetrySleepForTest(const std::function & sleep_fn) + { + mount.setSleepFnForTest(sleep_fn); + lease.setSleepFnForTest(sleep_fn); } /// The workers are joined HERE, not only by the tests that assert on teardown: `CasMountRuntime` @@ -2109,7 +2149,7 @@ class RuntimeUnderTest private: DB::Cas::CasRequests mount; - DB::Cas::CasRequests farewell; + DB::Cas::CasRequests lease; CasMountRuntime runtime; }; @@ -4150,12 +4190,8 @@ TEST(CASPoolRemount, TerminalDepositionDoesNotTouchRenewerAfterReplacement) runtime.installRenewer(uuid, 1, [&] { return wall_ms; }); const uint64_t anchor = runtime.startRenewer(); runtime.armMountFence(uuid, 1, anchor + 1000); - backend->fault = RuntimeRenewBackend::Fault::ThrowBefore; - /// Expire the lease from inside the attempt. The fault alone no longer ends a renewal: the engine - /// settles the ambiguity by reading and then reissues, and the reissue commits. With the clock past - /// the deadline the renewal was admitted under, neither the settling read nor the reissue is - /// admitted, so the renewal ends terminal -- which is what this test deposits. - backend->before_throw = [&, deadline = anchor + 1000] { boot_ms = deadline; }; + /// A definitive answer ends the worker's renewal; a transient fault would only be retried. + fenceOutMount(*backend, layout.mountKey("test")); runtime.startBackgroundWorkers(std::chrono::milliseconds(0)); terminal_deposited.waitUntilArrived(); EXPECT_TRUE(replaced.load(std::memory_order_acquire)); @@ -4240,11 +4276,8 @@ TEST(CASPoolRemount, ImmediatePostRemountRenewalFailureIsNotDropped) runtime_ptr->armMountFence(uuid, 2, fresh_anchor + 10'000); runtime_ptr->noteRemounted(); boot_ms = 2'000; - backend->fault = RuntimeRenewBackend::Fault::ThrowBefore; - /// Expire the fresh lease from inside the attempt, so the ambiguity can be neither - /// settled by a read nor reissued: otherwise the engine reissues and the renewal - /// commits, and there is no dropped failure to catch up on. - backend->before_throw = [&, deadline = fresh_anchor + 10'000] { boot_ms = deadline; }; + /// A definitive answer ends the worker's renewal; a transient fault would only be retried. + fenceOutMount(*backend, layout.mountKey("test")); first.arriveAndWait(); return true; } @@ -4269,6 +4302,418 @@ TEST(CASPoolRemount, ImmediatePostRemountRenewalFailureIsNotDropped) runtime.finishTeardown(false); } +namespace +{ +/// A mount slot claimed at epoch 1, a runtime renewing it in the background on an injected boot clock, and +/// the fence armed at the claim's anchor plus the lease. Locals the `MountConfig` hooks capture must be +/// declared before this object, which joins the workers when it is destroyed. +class BackgroundRenewal +{ +public: + BackgroundRenewal(const String & layout_name, uint64_t ttl_ms_) + : layout(layout_name) + , ttl_ms(ttl_ms_) + { + if (claimMount(*DB::Cas::tests::OperationForTest(backend), layout, "test", uuid, 1, wall_ms, ttl_ms).kind + != MountClaimResult::Claimed) + throw DB::Exception(DB::ErrorCodes::ABORTED, "test fixture claim did not land"); + } + + /// `config` gets the lease, the background flag and the boot clock here. + CasMountRuntime & open(MountConfig config, std::function remount = [] { return false; }) + { + config.mount_lease_ttl_ms = std::chrono::milliseconds(ttl_ms); + config.background_watermark = true; + config.boot_ms_fn = [this] { return boot_ms.load(); }; + holder.emplace(backend, layout, std::move(config), "test", sink, runtimeRenewBudget(), std::move(remount)); + CasMountRuntime & opened = runtime(); + opened.installRenewer(uuid, 1, [this] { return wall_ms; }); + anchor = opened.startRenewer(); + opened.armMountFence(uuid, 1, anchor + ttl_ms); + return opened; + } + + /// Every retry pause of the mount and lease planes moves the boot clock instead of sleeping. + void advanceClockOnRetrySleep() { holder->setRetrySleepForTest([this](uint64_t ms) { boot_ms += ms; }); } + + CasMountRuntime & runtime() { return **holder; } + + void stop() + { + runtime().stopBackgroundWorkers(); + runtime().finishTeardown(false); + } + + std::shared_ptr backend = std::make_shared(); + const Layout layout; + const UInt128 uuid{1}; + const uint64_t ttl_ms; + uint64_t wall_ms = 1000; + std::atomic boot_ms{100}; + CasEventSink sink; + uint64_t anchor = 0; + std::optional holder; +}; +} + +/// A renewal of the old epoch whose result is consumed after a remount armed the new epoch does not +/// move the new epoch's deadline. +TEST(CASMountRuntime, AnOldEpochRenewalDoesNotMoveTheNewEpochDeadline) +{ + DB::Cas::tests::ManualBarrier finished; + DB::Cas::tests::ManualBarrier remounted; + DB::Cas::tests::ManualBarrier back_in_loop; + std::atomic commits{0}; + std::atomic loop_passes{0}; + BackgroundRenewal renewal("runtime-old-epoch-deadline", 10'000); + CasMountRuntime & runtime = renewal.open( + MountConfig{ + .renewal_before_driver_lock_hook_for_test = [&] + { + if (++loop_passes == 2) + back_in_loop.arriveAndWait(); + }, + .renewal_finished_hook_for_test = [&](const MountRenewResult & result) + { + if (result.outcome == MountRenewOutcome::Committed && ++commits == 1) + finished.arriveAndWait(); + }}, + [&] + { + fenceOutMount(*renewal.backend, renewal.layout.mountKey("test")); + const MountClaimResult fresh = claimMount( + *DB::Cas::tests::OperationForTest(renewal.backend), renewal.layout, "test", renewal.uuid, 2, + renewal.wall_ms, renewal.ttl_ms); + EXPECT_EQ(fresh.kind, MountClaimResult::Claimed); + renewal.runtime().installRenewer(renewal.uuid, 2, [&] { return renewal.wall_ms; }); + const uint64_t fresh_anchor = renewal.runtime().startRenewer(); + renewal.runtime().armMountFence(renewal.uuid, 2, fresh_anchor + 5 * renewal.ttl_ms); + renewal.runtime().noteRemounted(); + remounted.arriveAndWait(); + return true; + }); + /// The epoch-1 renewal is due at once and inside its lease. The epoch-2 renewer is not driven before + /// the check: the worker waits in its second loop pass. + const uint64_t renewal_start = renewal.anchor + 1'000; + renewal.boot_ms = renewal_start; + runtime.startBackgroundWorkers(std::chrono::milliseconds(1'000)); + + finished.waitUntilArrived(); + runtime.tripMountLost(); + runtime.scheduleRemount(); + remounted.waitUntilArrived(); + remounted.release(); + finished.release(); + back_in_loop.waitUntilArrived(); + + /// Past the old renewal's deadline (its start plus one lease), before the new epoch's. + renewal.boot_ms = renewal_start + 2 * renewal.ttl_ms; + EXPECT_TRUE(runtime.mayMutate()) << "the old epoch's renewal moved the new epoch's deadline"; + back_in_loop.release(); + renewal.stop(); +} + +/// A mount PUT outage three leases long is a pause: the renewal keeps its tuple, its late success leaves +/// writes closed and starts the next renewal at once, that one opens them under the same fence +/// generation, and no remount is requested. +TEST(CASMountRuntime, ALongOutageIsAPauseUnderTheSameEpoch) +{ + std::atomic remount_calls{0}; + std::vector commits; + std::vector boot_at_commit; + std::vector may_mutate_after_commit; + DB::Cas::tests::ManualBarrier second_commit; + BackgroundRenewal renewal("until-definitive-long-outage", 1000); + CasMountRuntime & runtime = renewal.open( + MountConfig{ + .renewal_finished_hook_for_test = [&](const MountRenewResult & result) + { + if (result.outcome != MountRenewOutcome::Committed || commits.size() >= 2) + return; + commits.push_back(result); + boot_at_commit.push_back(renewal.boot_ms.load()); + may_mutate_after_commit.push_back(renewal.runtime().mayMutate()); + if (commits.size() == 2) + second_commit.arriveAndWait(); + }}, + [&] + { + ++remount_calls; + return false; + }); + const uint64_t generation = runtime.fenceGeneration(); + renewal.advanceClockOnRetrySleep(); + /// The first renewal is due at once and meets three lease lengths of outage. + const uint64_t anchor = renewal.anchor; + renewal.boot_ms = anchor + 500; + renewal.backend->outage = [&renewal, anchor] { return renewal.boot_ms.load() < anchor + 3'500; }; + runtime.startBackgroundWorkers(std::chrono::milliseconds(500)); + + second_commit.waitUntilArrived(); + EXPECT_EQ(commits[0].attempt_start_boot_ms, anchor + 500) << "the start of the logical renewal, not of its last attempt"; + EXPECT_GT(commits[0].attempts_sent, 2u); + EXPECT_GE(boot_at_commit[0], anchor + 3'500); + EXPECT_FALSE(may_mutate_after_commit[0]) << "a success whose start plus lease is past opens no write"; + EXPECT_EQ(commits[1].attempt_start_boot_ms, boot_at_commit[0]) << "the next renewal starts at once"; + EXPECT_TRUE(may_mutate_after_commit[1]); + EXPECT_EQ(runtime.fenceGeneration(), generation) << "a pause neither trips nor re-arms the fence"; + EXPECT_EQ(runtime.lifecycle(), PoolLifecycle::Live); + EXPECT_EQ(remount_calls.load(), 0u); + second_commit.release(); + renewal.stop(); +} + +/// A renewal that commits after its lease expired reports a reason that is true for a late commit too. +TEST(CASMountRuntime, ALateCommitReportsARecoveredRenewalWithATruthfulReason) +{ + std::atomic commits{0}; + std::atomic may_mutate_at_commit{true}; + std::optional recovered; + DB::Cas::tests::ManualBarrier recovered_reported; + BackgroundRenewal renewal("until-definitive-late-commit-reason", 1000); + renewal.sink = [&](CasEvent event) + { + if (event.type != CasEventType::WatermarkRenew || event.outcome != "recovered" || recovered) + return; + recovered = std::move(event); + recovered_reported.arriveAndWait(); + }; + CasMountRuntime & runtime = renewal.open(MountConfig{ + .renewal_finished_hook_for_test = [&](const MountRenewResult & result) + { + if (result.outcome == MountRenewOutcome::Committed && commits.fetch_add(1) == 0) + may_mutate_at_commit = renewal.runtime().mayMutate(); + }}); + renewal.advanceClockOnRetrySleep(); + const uint64_t anchor = renewal.anchor; + renewal.boot_ms = anchor + 500; + renewal.backend->outage = [&renewal, anchor] { return renewal.boot_ms.load() < anchor + 3'500; }; + runtime.startBackgroundWorkers(std::chrono::milliseconds(500)); + + recovered_reported.waitUntilArrived(); + EXPECT_FALSE(may_mutate_at_commit.load()) << "the commit came after the lease expired"; + ASSERT_TRUE(recovered.has_value()); + EXPECT_EQ(recovered->reason, "CAS mount renewal committed after a retry or a settling read"); + recovered_reported.release(); + renewal.stop(); +} + +/// A remount request ends a renewal that is retrying past its lease, at its next admission. +TEST(CASMountRuntime, ARemountRequestEndsARenewalRetryingPastTheLease) +{ + std::atomic held{false}; + std::atomic writes_at_remount{0}; + DB::Cas::tests::ManualBarrier holding; + DB::Cas::tests::ManualBarrier remount_entered; + BackgroundRenewal renewal("until-definitive-park", 1000); + CasMountRuntime & runtime = renewal.open(MountConfig{}, [&] + { + writes_at_remount = renewal.backend->outage_writes.load(); + remount_entered.arriveAndWait(); + return false; + }); + /// Five lease lengths on: a renewal bounded by its lease ended long before. + const uint64_t past_the_lease = renewal.anchor + 5'000; + renewal.advanceClockOnRetrySleep(); + renewal.backend->on_outage_write = [&renewal, &held, &holding, past_the_lease] + { + if (renewal.boot_ms.load() >= past_the_lease && !held.exchange(true)) + holding.arriveAndWait(); + }; + renewal.backend->outage = [] { return true; }; + runtime.startBackgroundWorkers(std::chrono::milliseconds(0)); + + holding.waitUntilArrived(); + const uint64_t writes_at_park = renewal.backend->outage_writes.load(); + runtime.scheduleRemount(); + EXPECT_EQ(runtime.renewalDriverStateForTest(), RenewalDriverState::ParkRequested); + holding.release(); + remount_entered.waitUntilArrived(); + EXPECT_EQ(writes_at_remount.load(), writes_at_park) << "no request starts after the remount request"; + remount_entered.release(); + renewal.stop(); +} + +/// The decommission tool's renewal stays bounded: under a permanent outage it ends inside its lease and +/// fences the mount instead of retrying past it. +TEST(CASMountRuntime, ABoundedBackgroundRenewalEndsInsideTheLease) +{ + DB::Cas::tests::ManualBarrier remount_entered; + BackgroundRenewal renewal("bounded-background-renewal", 1000); + CasMountRuntime & runtime = renewal.open(MountConfig{.bounded_background_renewal = true}, [&] + { + remount_entered.arriveAndWait(); + return false; + }); + renewal.advanceClockOnRetrySleep(); + renewal.boot_ms = renewal.anchor + 500; + renewal.backend->outage = [] { return true; }; + runtime.startBackgroundWorkers(std::chrono::milliseconds(500)); + + remount_entered.waitUntilArrived(); + EXPECT_FALSE(runtime.mayMutate()); + EXPECT_LT(renewal.boot_ms.load(), renewal.anchor + renewal.ttl_ms) << "the renewal outlived its lease"; + remount_entered.release(); + renewal.stop(); +} + +#if USE_AWS_S3 +/// A stop during the wait that follows a store refusal ends it at once, not one renewal period later, and +/// ends the series as cancelled: the farewell still hands the slot back. +TEST(CASMountRuntime, AStopDuringTheRefusalWaitEndsItAndKeepsTheFarewell) +{ + constexpr uint64_t period_ms = 30'000; + std::atomic first_write{true}; + DB::Cas::tests::ManualBarrier denied; + BackgroundRenewal renewal("until-definitive-refusal-stop", 2 * period_ms); + CasMountRuntime & runtime = renewal.open(MountConfig{}); + renewal.backend->deny_writes = 1; + renewal.backend->on_guarded_write = [&] + { + if (first_write.exchange(false)) + denied.arriveAndWait(); + }; + renewal.boot_ms = renewal.anchor + period_ms; + runtime.startBackgroundWorkers(std::chrono::milliseconds(period_ms)); + + denied.waitUntilArrived(); + denied.release(); + const auto stop_started = std::chrono::steady_clock::now(); + runtime.stopBackgroundWorkers(); + const auto stop_ms = std::chrono::duration_cast(std::chrono::steady_clock::now() - stop_started); + EXPECT_LT(stop_ms.count(), 10'000) << "the stop waited out the renewal period"; + runtime.finishTeardown(/*drained=*/true); + const auto left = (*DB::Cas::tests::OperationForTest(renewal.backend)).read(renewal.layout.mountKey("test"), Retry::standard()); + ASSERT_TRUE(left.has_value()); + EXPECT_EQ(decodeMountLease(left->bytes).min_active_build_sequence, std::numeric_limits::max()) + << "the farewell did not run"; +} + +/// A remount request during the wait that follows a store refusal ends it at once. +TEST(CASMountRuntime, AParkRequestEndsTheRefusalWaitAtOnce) +{ + constexpr uint64_t period_ms = 30'000; + std::atomic first_write{true}; + DB::Cas::tests::ManualBarrier denied; + DB::Cas::tests::ManualBarrier remount_entered; + BackgroundRenewal renewal("until-definitive-refusal-park", 2 * period_ms); + CasMountRuntime & runtime = renewal.open(MountConfig{}, [&] + { + remount_entered.arriveAndWait(); + return false; + }); + renewal.backend->deny_writes = std::numeric_limits::max(); + renewal.backend->on_guarded_write = [&] + { + if (first_write.exchange(false)) + denied.arriveAndWait(); + }; + renewal.boot_ms = renewal.anchor + period_ms; + runtime.startBackgroundWorkers(std::chrono::milliseconds(period_ms)); + + denied.waitUntilArrived(); + denied.release(); + const auto park_requested = std::chrono::steady_clock::now(); + runtime.scheduleRemount(); + remount_entered.waitUntilArrived(); + const auto park_ms = std::chrono::duration_cast(std::chrono::steady_clock::now() - park_requested); + EXPECT_LT(park_ms.count(), 10'000) << "the park request waited out the renewal period"; + remount_entered.release(); + renewal.stop(); +} + +/// After each store refusal the worker waits one renewal period before its next engine call. +TEST(CASMountRuntime, ARefusalWaitsOneRenewalPeriod) +{ + constexpr uint64_t period_ms = 700; + std::vector write_at; + std::atomic committed_once{false}; + DB::Cas::tests::ManualBarrier committed; + BackgroundRenewal renewal("until-definitive-refusal-period", 10'000); + CasMountRuntime & runtime = renewal.open(MountConfig{ + .renewal_finished_hook_for_test = [&](const MountRenewResult & result) + { + if (result.outcome == MountRenewOutcome::Committed && !committed_once.exchange(true)) + committed.arriveAndWait(); + }}); + renewal.advanceClockOnRetrySleep(); + renewal.backend->deny_writes = 2; + renewal.backend->on_guarded_write = [&] { write_at.push_back(renewal.boot_ms.load()); }; + renewal.boot_ms = renewal.anchor + period_ms; + runtime.startBackgroundWorkers(std::chrono::milliseconds(period_ms)); + + committed.waitUntilArrived(); + ASSERT_EQ(write_at.size(), 3u); + EXPECT_EQ(write_at[1] - write_at[0], period_ms); + EXPECT_EQ(write_at[2] - write_at[1], period_ms); + committed.release(); + renewal.stop(); +} +#endif + +/// One `WARNING` per pause, written at the first attempt after writes paused, and the pause is visible +/// from the moment writes are refused: the lease less one write's reservation and the safety margin. +TEST(CASMountRuntime, AWritePauseIsLoggedOnceAndReported) +{ + ScopedRemountLogCapture logs; + std::vector> paused_at_write; + std::atomic commits{0}; + String log_at_second_commit; + DB::Cas::tests::ManualBarrier second_commit; + BackgroundRenewal renewal("until-definitive-pause-log", 1000); + CasMountRuntime & runtime = renewal.open(MountConfig{ + .renewal_finished_hook_for_test = [&](const MountRenewResult & result) + { + if (result.outcome == MountRenewOutcome::Committed && ++commits == 2) + { + log_at_second_commit = logs.captured(); + second_commit.arriveAndWait(); + } + }}); + renewal.advanceClockOnRetrySleep(); + const uint64_t anchor = renewal.anchor; + renewal.boot_ms = anchor + 500; + renewal.backend->on_outage_write = [&] { paused_at_write.push_back(renewal.runtime().writesPausedSinceBootMs()); }; + renewal.backend->outage = [&renewal, anchor] { return renewal.boot_ms.load() < anchor + 3'500; }; + runtime.startBackgroundWorkers(std::chrono::milliseconds(500)); + + second_commit.waitUntilArrived(); + const String prefix = "CAS mount lease of 'test' under writer_epoch "; + size_t warnings = 0; + for (size_t at = log_at_second_commit.find(prefix); at != String::npos; at = log_at_second_commit.find(prefix, at + 1)) + ++warnings; + EXPECT_EQ(warnings, 1u) << log_at_second_commit; + EXPECT_NE(log_at_second_commit.find("Last renewal error: PUT "), String::npos) << log_at_second_commit; + ASSERT_GE(paused_at_write.size(), 3u); + EXPECT_FALSE(paused_at_write.front().has_value()) << "the first attempt runs while writes are admitted"; + const CasRequestBudget budget = runtimeRenewBudget(); + EXPECT_EQ(paused_at_write.back(), + std::optional(anchor + 1000 - 2 * budget.attemptEnvelopeMs() - budget.lease_safety_margin_ms)); + EXPECT_FALSE(runtime.writesPausedSinceBootMs().has_value()) << "a renewal with lease left ends the pause"; + second_commit.release(); + renewal.stop(); +} + +/// A clean shutdown while writes are paused leaves the slot for observation: the farewell needs lease room +/// that the pause no longer has. +TEST(CASMountRuntime, ACleanShutdownDuringAPauseLeavesTheSlotForObservation) +{ + ScopedRemountLogCapture logs; + BackgroundRenewal renewal("pause-shutdown", 1000); + CasMountRuntime & runtime = renewal.open(MountConfig{}); + renewal.boot_ms = renewal.anchor + renewal.ttl_ms; + ASSERT_TRUE(runtime.writesPausedSinceBootMs().has_value()); + + runtime.finishTeardown(/*drained=*/true); + + const String captured = logs.captured(); + EXPECT_EQ(captured.find("release during Pool teardown failed"), String::npos) << captured; + EXPECT_NE(captured.find("the slot is left for observation"), String::npos) << captured; + const auto left = (*DB::Cas::tests::OperationForTest(renewal.backend)).read(renewal.layout.mountKey("test"), Retry::standard()); + ASSERT_TRUE(left.has_value()); + EXPECT_NE(decodeMountLease(left->bytes).min_active_build_sequence, std::numeric_limits::max()); +} + TEST(CASPoolRemount, StaleRemountAnchorPerformsParkedRedo) { auto backend = std::make_shared(); @@ -4672,10 +5117,8 @@ TEST(CASPool, DeterministicWorkerFailureFencesWithoutWaitingForCadence) runtime.installRenewer(uuid, 1, [&] { return wall_ms; }); const uint64_t anchor = runtime.startRenewer(); runtime.armMountFence(uuid, 1, anchor + 1000); - backend->fault = RuntimeRenewBackend::Fault::ThrowBefore; - /// Expire the lease from inside the attempt, so the ambiguity can be neither settled by a read nor - /// reissued: without that the engine reissues and the renewal commits, and this worker never fences. - backend->before_throw = [&, deadline = anchor + 1000] { boot_ms = deadline; }; + /// A definitive answer ends the worker's renewal; a transient fault would only be retried. + fenceOutMount(*backend, layout.mountKey("test")); runtime.startBackgroundWorkers(std::chrono::milliseconds(0)); remount_entered.waitUntilArrived(); EXPECT_FALSE(runtime.mayMutate()); 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. 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