diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index 24c90c6dd670..8b2e1a35eb1b 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -52,7 +52,7 @@ follower or a deferred round execution returns before that commit. | 8 | `fold_ref_intake` | fold | `GET` every new ref-log record and every referenced manifest, extracting blob source edges | | 9 | `fold_reduce` | fold | Merge prior edges, new deltas and the parent's condemned rows: spare, condemn, graduate or redelete each candidate; compute `suppress_destructive` | | 10 | `fold_seal_write` | fold | Write the new fold seal once, write-once deterministic, adopting a byte-identical replay instead of rewriting it | -| 11 | `pending_deletes` | fold | The single blob-body delete site: exact-token delete of entries a *previous* round marked `delete_pending`, up to the redelete budget, plus the outcome-log writes | +| 11 | `pending_deletes` | fold | The single blob-body delete site: `HEAD` each entry a *previous* round marked `delete_pending`, then exact-token conditional delete chunks where the mounted backend proved that capability, plus the outcome-log writes | | 12 | `meta_pool_wait` | fold | Drain the bounded pool of async `.meta` condemn-marker writes queued during the fold | | 13 | `round_commit` | fold | Retention-prune old generations, then publish the single `gc/state` `CAS` that adopts the whole round | | 14 | `handoff_reclaim` | post-`CAS` | Reclaim a generation a ref moved off during this round, which the ordinary retention prune already skipped and will not revisit | @@ -400,8 +400,11 @@ forensic outcome logs. - **Reads:** one `HEAD` per `redelete` entry — the persisted condemned token cannot itself be a precondition, so the round observes the blob first, which also settles the absent case without spending a conditional delete -- **Writes / deletes:** a `DELETE` conditional on the observed etag for each `redelete` entry whose - live etag matches the condemned token; one write-once outcome log per shard with settled entries +- **Writes / deletes:** conditional delete chunks for entries whose live etag matches the condemned + token. A chunk is bounded by `cas_gc_bulk_delete_chunk_keys`, the backend conditional-batch limit, + and 1000 keys. An unprobed, unsupported, or `skip_access_check` mount uses one-key chunks. `.meta` + cleanup remains one per settled blob; it is not part of the conditional delete request. One + write-once outcome log is written per shard with settled entries - **Safety:** an entry is deletable only because a previously committed fold seal published it `delete_pending` — durable state from an earlier commit, safe at any leader staleness. Exact token means a stale leader cannot delete a fresh incarnation; `NotFound` and `TokenMismatch` are @@ -409,7 +412,15 @@ forensic outcome logs. - **Fails the round if:** a backend delete marker appears in a response — object versioning on a mis-provisioned pool (`LOGICAL_ERROR`) - **Observability:** phase row `pending_deletes`; metrics `deleted`, `absent`, `redeleted`, - `graduated`, `replaced`, `spared`, `outcome_logs_written` + `graduated`, `replaced`, `spared`, `outcome_logs_written`, `delete_batches`. `delete_batches` + counts logical conditional-delete chunks submitted by GC. `DiskS3DeleteObjects` counts physical S3 + delete attempts, including request retries + +The project RustFS integration backend supports exact-token singleton deletes but does not implement +AWS per-object `ETag` conditions in `DeleteObjects`. Its mount probe therefore keeps one-key chunks; +integration coverage verifies this fallback and physical blob reclamation. Positive AWS batching is +covered by deterministic GC and scripted S3 transport tests. A live AWS run is still required to +validate the production service boundary end to end. Under `suppress_destructive` the `redelete` set is empty by construction, so nothing is deleted, and there is no graduation either (would-be graduates are carried unchanged); sparing and superseding @@ -802,7 +813,7 @@ the chunk size reject `0`: | `cas_manifest_sweep_delete_budget_keys` | 100 | orphan-manifest sweep `DELETE` budget per round (phases 9, 18) | | `cas_gc_round_sweep_namespace_budget` | 20 | namespaces whose protection view the sweep may build per page (phase 9) | | `cas_gc_round_sweep_recovery_op_budget` | 5000 | committed-tail ref-log reads the sweep's recovery walk may spend (phase 9) | -| `cas_gc_bulk_delete_chunk_keys` | 1000 | keys per batch `DELETE` request for write-once families (phases 15, 17); `1` to `1000` | +| `cas_gc_bulk_delete_chunk_keys` | 1000 | keys per batch `DELETE` request for blob redelete and write-once families (phases 11, 15, 17); `1` to `1000` | | `cas_gc_meta_pool_size` | 16 | bounded pool for condemn-marker writes (phase 12) | | `cas_gc_io_concurrency` | 16 | bounded pool for the fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s (phases 8, 9), the graduation `.meta` reads (phase 9), the orphan-sweep planning reads (phase 9), the rebuild read-ahead and the `pending_deletes` `HEAD` + conditional `DELETE` fan-out (phase 11); other GC requests run on the round thread; `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead | diff --git a/docs/en/antalya/cas/bucket-requirements.md b/docs/en/antalya/cas/bucket-requirements.md index ea4bfe0dbafe..23a1338bfbf1 100644 --- a/docs/en/antalya/cas/bucket-requirements.md +++ b/docs/en/antalya/cas/bucket-requirements.md @@ -131,7 +131,7 @@ establish acceptance of Google's multipart, native-copy, and exact-delete wire b | AWS S3 | ✓ | Native `ETag`-based conditional dialect for mutable objects and exact deletion; blob publication is unconditional | | Google Cloud Storage | implementation complete; release gate pending | Generation-token dialect for mutable objects/native-token `HEAD`/exact deletion, opted into via `http_client = gcs_hmac` or `gcp_oauth`; blob publication uses ordinary copy/multipart. Real credentialed GCS groups have not run yet | | Azure Blob Storage | probably | Azure's REST API documents the equivalent conditional headers, but ClickHouse's Azure object-storage backend does not yet wire up a `CAS` conditional dialect the way the S3 and GCS paths do — untested, not validated by the capability probe | -| Other S3-compatible stores | only with enforced conditional operations | The capability probe is the actual gate: a store that silently ignores `If-None-Match`/`If-Match` (accepting and applying the write regardless) fails the probe and is refused. `RustFS` passes the full battery and is used as the project's test backend; `Garage` was evaluated and rejected because it silently ignores conditional operations | +| Other S3-compatible stores | only with enforced conditional operations | The capability probe is the actual gate: a store that silently ignores `If-None-Match`/`If-Match` (accepting and applying the write regardless) fails the probe and is refused. `RustFS` passes the required singleton conditional-operation battery and is used as the project's test backend, but it does not implement AWS per-object `ETag` conditions in `DeleteObjects`; conditional GC deletion therefore stays on one-key chunks. `Garage` was evaluated and rejected because it silently ignores conditional operations | The full mechanics of the two dialects — how the backend detects which one a given endpoint speaks, what the capability probe actually checks, and how exact-token deletes map onto each provider's diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index bd77ad820345..fb6f46480599 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -106,7 +106,7 @@ entirely before release. Treat this table as a snapshot of the current build, no | `cas_part_folder_cache_max_entry_bytes` | 16 MiB | Oversized part-folder views bypass retention above this size | | `cas_manifest_decode_cache_bytes` | 128 MiB | Manifest decode cache byte budget (`0` disables) | | `cas_gc_meta_pool_size` | `16` | Bounded pool size for GC per-hash freshness-meta writes | -| `cas_gc_io_concurrency` | `16` | Bounded pool size for GC object-storage requests that run in parallel: the fold's read-ahead (checkpoints, ref logs, manifests, zero-candidate HEADs), the orphan-manifest sweep planning reads, the `SYSTEM CAS GC REBUILD` read-ahead, the `pending_deletes` blob `HEAD` + conditional `DELETE` fan-out, and the namespace janitor's batch deletes of dead ref logs and snapshots. Not covered: meta writes (`cas_gc_meta_pool_size`) and all other GC requests, which run on the round thread. `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead | +| `cas_gc_io_concurrency` | `16` | Bounded pool size for GC object-storage requests that run in parallel: the fold's read-ahead (checkpoints, ref logs, manifests, zero-candidate HEADs), the orphan-manifest sweep planning reads, the `SYSTEM CAS GC REBUILD` read-ahead, the `pending_deletes` blob `HEAD` + conditional `DELETE` chunks, and the namespace janitor's batch deletes of dead ref logs and snapshots. S3 multi-key conditional chunks are enabled only after the writable mount proves their per-object ETag semantics; unsupported storage and `skip_access_check` use one-key conditional deletes. Not covered: meta writes (`cas_gc_meta_pool_size`) and all other GC requests, which run on the round thread. `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead | | `cas_attempt_timeout_ms` | `5000` | Budget for one HTTP attempt of a writable Native mount's control-plane requests (read, head, list, remove, conditional write), at least 1. Together with the connect cap it forms the attempt envelope (`cas_attempt_timeout_ms + 2 × cap`; the cap is `cas_attempt_timeout_ms` itself when the disk's `connect_timeout_ms` is `0`, else `min(connect_timeout_ms, cas_attempt_timeout_ms)`) that the lease arithmetic reserves: one TCP connect and one TLS handshake under the cap each, send/receive bounded per socket operation by `cas_attempt_timeout_ms`. With background renewal the cadence check requires `cas_mount_renew_period_ms + 2 × envelope + cas_lease_safety_margin_ms < cas_mount_lease_ttl_ms`, which puts an effective ceiling on the frozen connect cap: under the defaults (TTL 30000, period 10000, margin 2000) the envelope must stay under 9000, so a disk `connect_timeout_ms` of 2000 ms or more refuses to open writable — lower the connect timeout or raise the TTL if you hit this | | `cas_lease_safety_margin_ms` | `2000` | Startup-only margin validated against the mount lease TTL: the attempt envelope + `cas_lease_safety_margin_ms` must be strictly less than the mount lease TTL, and `cas_mount_renew_period_ms` + 2 × envelope + `cas_lease_safety_margin_ms` too, or the disk refuses to open writable | | `cas_unsafe_remount_no_delay` | `0` | Reclaim a mount slot that carries this server's own uuid at once after a hard restart, without observing the slot's token for the lease TTL. Unsafe whenever two processes can hold the same `server_uuid` (a copied uuid file, a stalled predecessor). After such a reclaim the predecessor can still start conditional writes until its own cutoff (`confirmed deadline − cas_lease_safety_margin_ms − 2 × envelope`) or until its next renewal meets the token guard, and a request it already sent may still materialize later. That is not a data hazard: ref-log keys carry `(writer_epoch, sequence)` and creates are conditional, so two writers can never commit different bodies to one key, and recovery's epoch seal settles any straggler (recovery fails closed after 64 successive seal-create attempts displaced by newly materializing old-epoch transactions). The exposure is availability, not data. Intended for test stands and deployments that guarantee one process per uuid | diff --git a/docs/en/operations/system-tables/cas_gc_log.md b/docs/en/operations/system-tables/cas_gc_log.md index 53309724de68..6f75a3b043b6 100644 --- a/docs/en/operations/system-tables/cas_gc_log.md +++ b/docs/en/operations/system-tables/cas_gc_log.md @@ -81,7 +81,7 @@ The phases, in execution order: | `fold_ref_intake` | Read and fold every new ref log and the manifest bodies its edges name. | one `GET` per new log, one `GET` per manifest edge | | `fold_reduce` | The per-shard in-degree merge: condemn, spare, graduate. | prior-run streaming `GET`s, one `HEAD` per zero-transition candidate, run `PUT`s | | `fold_seal_write` | Publish the new fold seal. | one `PUT` | -| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. `phase_metrics` carries `jobs_scheduled` (entries sent to the GC I/O pool; `0` for no candidates, sequential execution, singleton batches, or refusal before the first submission) and `jobs_failed`. | one `HEAD` and at most one `DELETE` per entry, one outcome-log `PUT` per shard with rows | +| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. `phase_metrics` carries `jobs_scheduled` (entries submitted to the parallel observation wave; `0` for no candidates or sequential execution), `jobs_failed`, and `delete_batches` (logical conditional-delete chunks). Unsupported or unprobed storage uses one-key chunks. | one `HEAD` per entry, conditional delete chunks, one outcome-log `PUT` per shard with rows | | `meta_pool_wait` | Drain the round's per-hash freshness-meta writes. | none on this thread — see the caveat below | | `round_commit` | The generation-retention prune and the round's single `gc/state` compare-and-swap. | prune `LIST`s and deletes, one compare-and-swap | | `handoff_reclaim` | Wholesale-reclaim generations a moved run ref stranded below the retention cursor. | prefix `LIST`s and deletes | diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasBackend.h index eb502f9724c2..2735bb55eb14 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasBackend.h @@ -20,6 +20,11 @@ #include #include +namespace DB::ErrorCodes +{ + extern const int NOT_IMPLEMENTED; +} + namespace DB::Cas { @@ -169,6 +174,11 @@ class Backend /// `DeleteMarker` is a removal that did NOT reclaim: a versioned bucket archived a noncurrent /// version instead. Distinct from `Removed` because reclaiming the storage is the point. enum class RawRemoval : uint8_t { Removed, Gone, Mismatch, DeleteMarker }; + struct RawConditionalRemove + { + String key; + String expected_token; + }; /// Reads the whole object, or nullopt when the key is absent. virtual std::optional read (const String & key, TransportAccess &) = 0; @@ -201,10 +211,23 @@ class Backend /// absence is success. virtual void removeManyWriteOnce(const std::vector & keys, TransportAccess &) = 0; + virtual std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) + { + if (removals.size() > 1) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Conditional batch removal is not implemented by this backend"); + if (removals.empty()) + return {}; + return {remove(removals.front().key, removals.front().expected_token, access)}; + } + /// The most keys one `removeManyWriteOnce` sends to the storage as one request; at least 1. A decorator /// must forward it: the default of 1 would turn every batch below it into one-key requests. virtual size_t bulkDeleteKeyLimit() const { return 1; } + virtual size_t conditionalBatchDeleteKeyLimit() const { return 1; } + virtual void pinConditionalBatchDeleteSupport(bool) {} + /// Authoritative, cache-bypassing probe of one key -- see `ProbeOutcome`. DEFAULT (used by every /// backend without sharper raw-error evidence, e.g. `InMemoryBackend`): derived from `head`/`read` /// alone, so it can only distinguish `Present` from `KeyAbsent`, and ANY exception from either diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.cpp index b6bf981a40f1..d99607716673 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.cpp @@ -326,6 +326,20 @@ void InMemoryBackend::removeManyWriteOnce(const std::vector & keys } } +std::vector InMemoryBackend::removeManyByConditional( + const std::vector & removals, TransportAccess &) +{ + for (const RawConditionalRemove & removal : removals) + checkExpectedValue(removal.key, removal.expected_token); + + std::lock_guard lock(mutex_); + std::vector results; + results.reserve(removals.size()); + for (const RawConditionalRemove & removal : removals) + results.push_back(applyDelete(removal.key, removal.expected_token)); + return results; +} + void InMemoryBackend::failNextBulkRemoveWith(std::exception_ptr error) { std::lock_guard lock(mutex_); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.h index 9e3100efdf3e..b7a279724390 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.h @@ -51,6 +51,8 @@ class InMemoryBackend : public Backend /// `hold_deletes_` exactly as `remove` does: a held delete is queued and lands only on /// `landPendingDelete`. void removeManyWriteOnce(const std::vector & keys, TransportAccess & access) override; + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override; /// The next `removeManyWriteOnce` throws `error` instead of deleting anything; one-shot, like /// `failNextWriteWith`. void failNextBulkRemoveWith(std::exception_ptr error); @@ -62,6 +64,8 @@ class InMemoryBackend : public Backend /// The emulation deletes keys one by one under its lock, so the CAS maximum is the only bound. size_t bulkDeleteKeyLimit() const override { return kBulkDeleteMaxKeys; } + size_t conditionalBatchDeleteKeyLimit() const override { return conditional_batch_delete_supported ? kBulkDeleteMaxKeys : 1; } + void pinConditionalBatchDeleteSupport(bool supported) override { conditional_batch_delete_supported = supported; } /// Creates the key when `expected_value` is empty, or replaces the incarnation it names. A /// refused precondition leaves the store unchanged. Value enforcement can be disabled with @@ -227,6 +231,7 @@ class InMemoryBackend : public Backend std::vector armed_bulk_remove_failures_; std::function before_bulk_remove_hook_; size_t bulk_remove_calls_ = 0; + std::atomic_bool conditional_batch_delete_supported = false; }; } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.cpp index 4782bd3f1011..90d4b410cee3 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.cpp @@ -154,4 +154,13 @@ void InstrumentedBackend::removeManyWriteOnce(const std::vector & incrementCasEvent(classifyCasNs(key.str()), CasOp::Delete); } +std::vector InstrumentedBackend::removeManyByConditional( + const std::vector & removals, TransportAccess & access) +{ + std::vector results = inner->removeManyByConditional(removals, access); + for (const RawConditionalRemove & removal : removals) + incrementCasEvent(classifyCasNs(removal.key), CasOp::Delete); + return results; +} + } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.h index 5d1f3f36e953..041ad1d7b519 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.h @@ -130,6 +130,8 @@ class InstrumentedBackend final : public Backend /// carried, the request counter says how many requests it took. Out of line, like `publish`, so /// the header need not declare the `CASBulkDeleteRequests` extern. void removeManyWriteOnce(const std::vector & keys, TransportAccess & access) override; + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override; /// Count a create and a replacement separately, and each of them separately from its refusal: /// they cost the same one request, but a pool whose creates are mostly refused and one whose @@ -162,6 +164,8 @@ class InstrumentedBackend final : public Backend uint64_t attemptTimeoutMs() const override { return inner->attemptTimeoutMs(); } uint64_t attemptEnvelopeMs() const override { return inner->attemptEnvelopeMs(); } size_t bulkDeleteKeyLimit() const override { return inner->bulkDeleteKeyLimit(); } + size_t conditionalBatchDeleteKeyLimit() const override { return inner->conditionalBatchDeleteKeyLimit(); } + void pinConditionalBatchDeleteSupport(bool supported) override { inner->pinConditionalBatchDeleteSupport(supported); } bool refreshCredentials() override { return inner->refreshCredentials(); } private: diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp index 2b63dc0ef9d7..4a52aa05ee74 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp @@ -1009,6 +1009,18 @@ size_t ObjectStorageBackend::bulkDeleteKeyLimit() const return mode == Mode::Native ? object_storage->batchDeleteKeyLimit() : kBulkDeleteMaxKeys; } +size_t ObjectStorageBackend::conditionalBatchDeleteKeyLimit() const +{ + if (mode != Mode::Native || !conditional_batch_delete_supported.load()) + return 1; + return object_storage->batchDeleteKeyLimit(); +} + +void ObjectStorageBackend::pinConditionalBatchDeleteSupport(bool supported) +{ + conditional_batch_delete_supported.store(supported); +} + void ObjectStorageBackend::removeManyWriteOnce(const std::vector & keys, TransportAccess & access) { if (keys.empty()) @@ -1036,6 +1048,48 @@ void ObjectStorageBackend::removeManyWriteOnce(const std::vector & } } +std::vector ObjectStorageBackend::removeManyByConditional( + const std::vector & removals, TransportAccess & access) +{ + if (removals.empty()) + return {}; + if (mode != Mode::Native) + { + std::vector results; + results.reserve(removals.size()); + for (const RawConditionalRemove & removal : removals) + results.push_back(removeUnder(removal.key, removal.expected_token, controlRequest(access.attemptNo()))); + return results; + } + + std::vector objects; + objects.reserve(removals.size()); + for (const RawConditionalRemove & removal : removals) + { + if (!isValidTokenValue(dialect(), removal.expected_token)) + throw Exception(ErrorCodes::LOGICAL_ERROR, "CAS backend: malformed conditional removal token for {}", removal.key); + objects.push_back({StoredObject(removal.key), removal.expected_token}); + } + const auto storage_results = object_storage->removeObjectsIfTokensMatch(objects, controlRequest(access.attemptNo())); + if (storage_results.size() != removals.size()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "CAS backend: conditional batch removal returned {} results for {} keys", storage_results.size(), removals.size()); + + std::vector results; + results.reserve(storage_results.size()); + for (const ConditionalRemoveResult & result : storage_results) + { + switch (result.outcome) + { + case ConditionalRemoveOutcome::Removed: + results.push_back(result.created_delete_marker ? RawRemoval::DeleteMarker : RawRemoval::Removed); + break; + case ConditionalRemoveOutcome::TokenMismatch: results.push_back(RawRemoval::Mismatch); break; + case ConditionalRemoveOutcome::NotFound: results.push_back(RawRemoval::Gone); break; + } + } + return results; +} + Backend::RawListPage ObjectStorageBackend::list(const String & prefix, const String & cursor, size_t limit, TransportAccess & access) { return listUnder(prefix, cursor, limit, controlRequest(access.attemptNo())); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h index cc3c83f4f143..52dddfefefe8 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h @@ -101,10 +101,14 @@ class ObjectStorageBackend final : public Backend /// under the control-plane profile; `EmulatedSingleProcess` deletes each present key under the /// emulation lock with the same token bookkeeping as the single-key delete. void removeManyWriteOnce(const std::vector & keys, TransportAccess & access) override; + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override; /// Native: the object storage's own limit. Emulated: `kBulkDeleteMaxKeys`, since this backend deletes /// the keys one by one under its emulation lock. size_t bulkDeleteKeyLimit() const override; + size_t conditionalBatchDeleteKeyLimit() const override; + void pinConditionalBatchDeleteSupport(bool supported) override; /// Native mints its store's own dialect (ETag or GCS generation); the emulated adapter mints its /// own values. Dialect dialect() const override { return mode == Mode::Native ? native_token_type : Dialect::Emulated; } @@ -214,6 +218,7 @@ class ObjectStorageBackend final : public Backend private: const ObjectStoragePtr object_storage; const Mode mode; + std::atomic_bool conditional_batch_delete_supported = false; Dialect native_token_type = Dialect::ETag; /// See the constructor: what the READ-class requests (read, head, list, remove) carry. const bool single_attempt_control_plane; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.cpp index 625c04af8b41..04260010e2a9 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.cpp @@ -44,11 +44,13 @@ Removal removeOrReportDeleteMarker(CasOperation & op, const String & key, const } -void runCapabilityProbe(CasOperation & op, const String & probe_prefix) +CapabilityProbeResult runCapabilityProbe(CasOperation & op, const String & probe_prefix) { // Sub-directory style ("probe_prefix/token") ensures that list(probe_prefix, …) works for both the // in-memory backend (prefix match) and the LocalObjectStorage backend (directory listing). const String key = probe_prefix + "/token"; + const String first_key = probe_prefix + "/batch-first"; + const String second_key = probe_prefix + "/batch-second"; // Best-effort cleanup — runs at function exit regardless of outcome. auto cleanup = [&]() noexcept @@ -59,13 +61,17 @@ void runCapabilityProbe(CasOperation & op, const String & probe_prefix) // constructible here — the gate below is the only way to reach `remove` at all. try { - const auto h = op.head(key, Retry::standard()); - if (h) - op.remove(key, h->etag, Retry::standard()); + for (const String * cleanup_key : {&key, &first_key, &second_key}) + { + const auto h = op.head(*cleanup_key, Retry::standard()); + if (h) + op.remove(*cleanup_key, h->etag, Retry::standard()); + } } catch (...) {} /// NOLINT(bugprone-empty-catch) }; + CapabilityProbeResult result; try { // ---- Step 1: create fresh -> Committed; read-after-write returns the bytes. ---- @@ -195,6 +201,25 @@ void runCapabilityProbe(CasOperation & op, const String & probe_prefix) "CasProbe: list-after-delete failed — the deleted probe key '{}' is still visible in the listing under prefix '{}'", key, probe_prefix); } + + const Etag first_token = std::get(op.create(first_key, "probe-batch-first", Retry::standard())).etag; + const Etag second_token = std::get(op.create(second_key, "probe-batch-second-v1", Retry::standard())).etag; + const Etag second_replacement = std::get(op.replace(second_key, "probe-batch-second-v2", second_token, Retry::standard())).etag; + try + { + const std::vector removals = op.removeManyByConditionalForProbe( + {{first_key, first_token}, {second_key, second_token}}, Retry::standard()); + if (removals.size() == 2 && removals[0] == Removal::Removed && removals[1] == Removal::Mismatch) + { + const auto second = op.read(second_key, Retry::standard()); + result.conditional_batch_delete_supported = second && second->bytes == "probe-batch-second-v2"; + } + } + catch (const DB::Exception & e) + { + if (e.code() != DB::ErrorCodes::NOT_IMPLEMENTED) + throw; + } } catch (...) { @@ -206,6 +231,7 @@ void runCapabilityProbe(CasOperation & op, const String & probe_prefix) // Normal-exit cleanup (the key was already deleted in step 7). Call cleanup anyway to handle any // partial state edge cases — it is a no-op if the key is gone. cleanup(); + return result; } } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.h index 99ec36581af2..783261ddbcd2 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasProbe.h @@ -5,6 +5,11 @@ namespace DB::Cas { +struct CapabilityProbeResult +{ + bool conditional_batch_delete_supported = false; +}; + /// Run the capability battery against `op`, using throwaway keys under `probe_prefix`. /// /// `op` must already be admitted. The two store-level precondition hooks — @@ -31,6 +36,6 @@ namespace DB::Cas /// Cleanup of the probe key is best-effort and runs unconditionally: after the battery completes, or on /// the failure path immediately before the check-failing exception is rethrown. Cleanup itself suppresses /// exceptions so that it cannot hide the capability-check failure. -void runCapabilityProbe(CasOperation & op, const String & probe_prefix); +CapabilityProbeResult runCapabilityProbe(CasOperation & op, const String & probe_prefix); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp index 898c0c8c1504..4ac1abbcc209 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.cpp @@ -18,9 +18,11 @@ #include +#include #include #include #include +#include #include namespace ProfileEvents @@ -613,6 +615,69 @@ void CasOperation::removeManyWriteOnce(const std::vector & keys, c }); } +std::vector CasOperation::removeManyByConditional( + const std::vector & removals, const Retry & policy) +{ + const size_t limit = std::min(kBulkDeleteMaxKeys, owner.backend->conditionalBatchDeleteKeyLimit()); + return removeManyByConditionalImpl(removals, policy, limit); +} + +std::vector CasOperation::removeManyByConditionalForProbe( + const std::vector & removals, const Retry & policy) +{ + return removeManyByConditionalImpl(removals, policy, kBulkDeleteMaxKeys); +} + +std::vector CasOperation::removeManyByConditionalImpl( + const std::vector & removals, const Retry & policy, size_t limit) +{ + if (removals.size() > limit) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "CAS removeManyByConditional: {} keys in one chunk, the limit is {}; the consumer chunks its input", + removals.size(), limit); + if (removals.empty()) + return {}; + + std::set keys; + std::vector raw_removals; + raw_removals.reserve(removals.size()); + for (const ConditionalRemove & removal : removals) + { + if (!keys.emplace(removal.key).second) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "CAS removeManyByConditional: duplicate key '{}' in one chunk", removal.key); + raw_removals.push_back({removal.key, owner.valueFor(removal.key, removal.expected_token)}); + } + + const Retry::Bound bound = policy.bind(owner.now_ms()); + const String subject = fmt::format("{} (+{} keys)", removals.front().key, removals.size() - 1); + const std::vector raw_results = readLoop( + "removeManyByConditional", subject, policy, bound, [&](auto & access) + { + return owner.backend->removeManyByConditional(raw_removals, access); + }); + if (raw_results.size() != removals.size()) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "CAS removeManyByConditional: backend returned {} results for {} keys", raw_results.size(), removals.size()); + + std::vector results; + results.reserve(raw_results.size()); + for (const Backend::RawRemoval raw : raw_results) + { + switch (raw) + { + case Backend::RawRemoval::Removed: results.push_back(Removal::Removed); break; + case Backend::RawRemoval::Gone: results.push_back(Removal::Gone); break; + case Backend::RawRemoval::Mismatch: results.push_back(Removal::Mismatch); break; + case Backend::RawRemoval::DeleteMarker: + throw Exception(ErrorCodes::CAS_DELETE_MARKER, + "CAS conditional batch removal archived a noncurrent version instead of reclaiming an object: " + "the bucket has object versioning enabled"); + } + } + return results; +} + Removal CasOperation::remove(const String & key, const Etag & seen, const Retry & policy) { return removeUnder(key, owner.valueFor(key, seen), policy, policy.bind(owner.now_ms())); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h index dd79f1a9db4b..54f88d9b0b79 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h @@ -21,6 +21,10 @@ namespace DB::Cas { +class CasOperation; +struct CapabilityProbeResult; +CapabilityProbeResult runCapabilityProbe(CasOperation & op, const String & probe_prefix); + /// TRUE when the store's own answer proves this write never applied: a malformed request, an entity /// too large, an access denial, or a credential failure. Whether a STALE CREDENTIAL explains it is not /// asked here -- the engine asks the backend for fresh credentials first, and refuses only when no @@ -234,6 +238,12 @@ class CasRequests /// The cap on one `removeManyWriteOnce` chunk -- also the ceiling a batch-delete request can carry. inline constexpr size_t kBulkDeleteMaxKeys = 1000; +struct ConditionalRemove +{ + String key; + Etag expected_token; +}; + /// One admitted operation: the unit a policy, a fence generation and a liveness predicate apply to. /// Move-only, and every request it makes re-checks its admission -- before each attempt, before each /// sleep, and once more after a proven commit, so a write whose fence was lost while it was in flight @@ -288,6 +298,7 @@ class CasOperation /// success. Throws when the policy is exhausted. More keys than the cap is a caller bug: the /// consumer chunks, so that every chunk that succeeded is recorded before a later one can fail. void removeManyWriteOnce(const std::vector & keys, const Retry & policy); + std::vector removeManyByConditional(const std::vector & removals, const Retry & policy); /// The one primitive that reports failure as a value, so the policy reissues on the OUTCOME: /// `Indeterminate` is retried, the four authoritative outcomes return at once, and an /// `Indeterminate` that outlives the bound is returned rather than thrown. Admission refused before @@ -309,6 +320,13 @@ class CasOperation WriteResult readModifyWriteOnPresence(const String & key, const DecideOnMeta & decide, const Retry & policy); private: + friend CapabilityProbeResult runCapabilityProbe(CasOperation & op, const String & probe_prefix); + + std::vector removeManyByConditionalForProbe( + const std::vector & removals, const Retry & policy); + std::vector removeManyByConditionalImpl( + const std::vector & removals, const Retry & policy, size_t limit); + friend class CasRequests; /// The lane waits on this operation's own admission and reads and writes through its verbs; it /// needs the gate, the reservation, the resolve read and the give-up helpers, never the transport. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasThrottlingBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasThrottlingBackend.h index 8f69d4464347..9789cb55c9f7 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasThrottlingBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasThrottlingBackend.h @@ -95,6 +95,14 @@ class ThrottlingBackend final : public Backend inner->removeManyWriteOnce(keys, access); } + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override + { + for (const RawConditionalRemove & removal : removals) + refuseOrPass(removal.key); + return inner->removeManyByConditional(removals, access); + } + std::expected write(const String & key, const String & bytes, const std::optional & expected_value, TransportAccess & access) override { @@ -126,6 +134,8 @@ class ThrottlingBackend final : public Backend uint64_t attemptTimeoutMs() const override { return inner->attemptTimeoutMs(); } uint64_t attemptEnvelopeMs() const override { return inner->attemptEnvelopeMs(); } size_t bulkDeleteKeyLimit() const override { return inner->bulkDeleteKeyLimit(); } + size_t conditionalBatchDeleteKeyLimit() const override { return inner->conditionalBatchDeleteKeyLimit(); } + void pinConditionalBatchDeleteSupport(bool supported) override { inner->pinConditionalBatchDeleteSupport(supported); } bool refreshCredentials() override { return inner->refreshCredentials(); } void checkPoolPreconditions() override { inner->checkPoolPreconditions(); } void checkSkipAccessCheckSupport() override { inner->checkSkipAccessCheckSupport(); } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 58dfd377cca8..cf9a9e134d92 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -574,29 +574,57 @@ size_t Gc::bulkDeleteChunkKeys() const return std::clamp(std::min(configured, storage_limit), 1, kBulkDeleteMaxKeys); } -Gc::RedeleteIo Gc::performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op) +size_t Gc::conditionalDeleteChunkKeys() const { - RedeleteIo io; - io.blob_key = layout.blobKey(entry.ref); + const uint64_t configured = store->poolConfig().gc_bulk_delete_chunk_keys; + const uint64_t storage_limit = store->poolBackendPtr()->conditionalBatchDeleteKeyLimit(); + return std::clamp(std::min(configured, storage_limit), 1, kBulkDeleteMaxKeys); +} + +void Gc::observeRetiredBlob( + const RetiredEntry & entry, const Layout & layout, CasOperation & op, RedeleteBlobState & redelete_blob_state) +{ + redelete_blob_state.blob_key = layout.blobKey(entry.ref); /// The condemned incarnation is a PERSISTED pair and cannot itself be a precondition, so /// the round observes the blob and compares the two renderings. Observing first also /// settles the absent case without spending a conditional delete against a key that is /// already gone. - const std::optional observed = op.head(io.blob_key, Retry::standard()); - if (observed) - io.del = entry.token.matches(observed->etag) ? op.remove(io.blob_key, observed->etag, Retry::standard()) : Removal::Mismatch; - return io; + const std::optional observed = op.head(redelete_blob_state.blob_key, Retry::standard()); + if (!observed) + redelete_blob_state.removal = Removal::Gone; + else if (entry.token.matches(observed->etag)) + redelete_blob_state.live_token = observed->etag; + else + redelete_blob_state.removal = Removal::Mismatch; +} + +std::vector Gc::prepareConditionalDeleteChunks( + const std::vector & redelete_blob_states, size_t chunk_keys) +{ + std::vector chunks; + for (size_t i = 0; i < redelete_blob_states.size(); ++i) + { + const RedeleteBlobState & redelete_blob_state = redelete_blob_states[i]; + if (redelete_blob_state.error || !redelete_blob_state.live_token) + continue; + if (chunks.empty() || chunks.back().indices.size() == chunk_keys) + chunks.emplace_back(); + chunks.back().indices.push_back(i); + chunks.back().removals.push_back({redelete_blob_state.blob_key, *redelete_blob_state.live_token}); + } + return chunks; } -void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteIo & io) +void Gc::recordRedeleteSuccess(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteBlobState & redelete_blob_state) { - if (const auto & hook = store->poolConfig().gc_redelete_apply_hook_for_test) + if (const auto & hook = store->poolConfig().gc_redelete_record_hook_for_test) hook(entry.ref); - const OutcomeKind outcome_kind = io.del == Removal::Removed ? OutcomeKind::Deleted - : io.del == Removal::Gone ? OutcomeKind::Absent - : OutcomeKind::Replaced; + const Removal removal = *redelete_blob_state.removal; + const OutcomeKind outcome_kind = removal == Removal::Removed ? OutcomeKind::Deleted + : removal == Removal::Gone ? OutcomeKind::Absent + : OutcomeKind::Replaced; OutcomeEntry outcome{.kind = entry.kind, .ref = entry.ref, .token = entry.token, .outcome = outcome_kind}; - const String del_outcome{removalName(io.del)}; + const String removal_outcome{removalName(removal)}; /// The single content-delete site is attributable per row. A mismatch (a writer recreated /// the incarnation) is terminal-OK: the fresh incarnation is a live object. EventEmitter{*store}.emit( @@ -608,11 +636,11 @@ void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & e e.token = renderIncarnation(entry.token); e.round = ctx.new_round; e.gen = ctx.snap_generation; - e.outcome = del_outcome; + e.outcome = removal_outcome; e.reason = "delete_pending published by a prior pass; exact-incarnation delete (pre-CAS)"; - e.detail = {{"condemn_round", std::to_string(entry.condemn_round)}, {"key", io.blob_key}}; + e.detail = {{"condemn_round", std::to_string(entry.condemn_round)}, {"key", redelete_blob_state.blob_key}}; }); - /// The audit row is observability only -- the delete in `performRedeleteIo` already executed + /// The audit row is observability only -- the delete in `redeleteBlobs` already executed /// regardless of this cap. Skipping it here bounds the per-shard `GcOutcomes` body without /// skipping or deferring any destructive work. if (ctx.round_work_budget.outcomeEntryAvailable()) @@ -622,18 +650,27 @@ void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & e } ++ctx.report.redeleted; ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleted); +} + +void Gc::scheduleBlobMetaDeletes( + const std::vector & entries, const std::vector & redelete_blob_states) +{ /// Drop the per-hash meta only on a removal or a proven absence — a mismatch means a /// writer already published a fresh incarnation at this hash, and that writer's own /// republication path (`PartWriteTxn::ensureBlobPresent`) reconciles the meta back to Clean /// right after the publication; deleting it here would race that legitimate Clean write for /// no reason (the meta is advisory, but there is no reason to touch it on that path at all). - if (io.del == Removal::Removed || io.del == Removal::Gone) + for (size_t i = 0; i < redelete_blob_states.size(); ++i) { - meta_writer->scheduleConfirmedMetaDelete(entry.ref, entry.condemn_round); + const RedeleteBlobState & redelete_blob_state = redelete_blob_states[i]; + if (redelete_blob_state.error) + continue; + if (redelete_blob_state.removal == Removal::Removed || redelete_blob_state.removal == Removal::Gone) + meta_writer->scheduleConfirmedMetaDelete(entries[i].ref, entries[i].condemn_round); } } -void Gc::reportRedeleteFailure( +void Gc::recordRedeleteFailure( RedeleteRoundContext & ctx, const RetiredEntry & entry, const Layout & layout, @@ -649,110 +686,179 @@ void Gc::reportRedeleteFailure( unsent_count ? fmt::format("; {} re-delete entries were not submitted", *unsent_count) : String{}); } -void Gc::redeleteBlobs( - RedeleteRoundContext & ctx, - const std::vector & entries, +Gc::IoJobsResult Gc::runRedeleteIoJobs( ThreadPool & pool, size_t concurrency, - const Layout & layout, - CasOperation & op) + size_t count, + size_t refuse_offset, + CasOperation & op, + const std::function & job) { - std::vector io_results(entries.size()); - std::vector io_errors(entries.size()); - std::optional schedule_error; - size_t scheduled = 0; - if (concurrency <= 1 || entries.size() <= 1) + IoJobsResult result; + result.errors.resize(count); + if (concurrency <= 1 || count <= 1) { - for (size_t i = 0; i < entries.size(); ++i) + for (; result.scheduled < count; ++result.scheduled) { try { - io_results[i] = performRedeleteIo(entries[i], layout, op); + job(result.scheduled, op); } catch (...) { - io_errors[i] = std::current_exception(); + result.errors[result.scheduled] = std::current_exception(); } } - scheduled = entries.size(); + return result; } - else - { - ThreadPoolCallbackRunnerLocal runner(pool, ThreadName::CAS_GC_REDELETE); - std::vector::Task>> handles; - handles.reserve(entries.size()); - SCOPE_EXIT_SAFE({ ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); }); - const uint64_t admitted_generation = op.generation(); - try + result.on_pool = true; + ThreadPoolCallbackRunnerLocal runner(pool, ThreadName::CAS_GC_REDELETE); + std::vector::Task>> handles; + handles.reserve(count); + SCOPE_EXIT_SAFE({ ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); }); + const uint64_t admitted_generation = op.generation(); + try + { + for (; result.scheduled < count; ++result.scheduled) { - for (; scheduled < entries.size(); ++scheduled) + if (io_pool_refuse_at_for_test == refuse_offset + result.scheduled) { - if (io_pool_refuse_at_for_test == scheduled) - { - io_pool_refuse_at_for_test.reset(); - throw Exception(ErrorCodes::CANNOT_SCHEDULE_TASK, "Injected CAS GC re-delete enqueue refusal at index {}", scheduled); - } - handles.emplace_back(runner.enqueueAndGiveOwnership( - [slot = &io_results[scheduled], - entry = entries[scheduled], - layout_ptr = &layout, - pool_store = store, - admitted_generation] - { - CasOperation job_op = pool_store->openRequests().resume(admitted_generation); - *slot = performRedeleteIo(entry, *layout_ptr, job_op); - })); + io_pool_refuse_at_for_test.reset(); + throw Exception( + ErrorCodes::CANNOT_SCHEDULE_TASK, + "Injected CAS GC re-delete enqueue refusal at index {}", + refuse_offset + result.scheduled); } + handles.emplace_back(runner.enqueueAndGiveOwnership( + [&job, index = result.scheduled, pool_store = store, admitted_generation] + { + CasOperation job_op = pool_store->openRequests().resume(admitted_generation); + job(index, job_op); + })); + } + } + catch (...) + { + result.schedule_error = IndexedException{result.scheduled, std::current_exception()}; + } + + ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); + for (size_t i = 0; i < handles.size(); ++i) + { + try + { + handles[i]->future.get(); } catch (...) { - schedule_error = IndexedException{scheduled, std::current_exception()}; + result.errors[i] = std::current_exception(); } - ctx.jobs_scheduled += handles.size(); + } + return result; +} - ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); - for (size_t i = 0; i < handles.size(); ++i) +void Gc::redeleteBlobs( + RedeleteRoundContext & ctx, + const std::vector & entries, + ThreadPool & pool, + size_t concurrency, + const Layout & layout, + CasOperation & op) +{ + std::vector redelete_blob_states(entries.size()); + const IoJobsResult observe_jobs = runRedeleteIoJobs( + pool, + concurrency, + entries.size(), + 0, + op, + [&](size_t index, CasOperation & job_op) { observeRetiredBlob(entries[index], layout, job_op, redelete_blob_states[index]); }); + if (observe_jobs.on_pool) + ctx.jobs_scheduled += observe_jobs.scheduled; + for (size_t i = 0; i < observe_jobs.scheduled; ++i) + redelete_blob_states[i].error = observe_jobs.errors[i]; + + const std::vector delete_chunks + = prepareConditionalDeleteChunks(redelete_blob_states, conditionalDeleteChunkKeys()); + const IoJobsResult delete_jobs = runRedeleteIoJobs( + pool, + concurrency, + delete_chunks.size(), + entries.size(), + op, + [&](size_t chunk, CasOperation & job_op) { - try - { - handles[i]->future.get(); - } - catch (...) - { - io_errors[i] = std::current_exception(); - } - } + const ConditionalDeleteChunk & delete_chunk = delete_chunks[chunk]; + const auto outcomes = job_op.removeManyByConditional(delete_chunk.removals, Retry::standard()); + for (size_t pos = 0; pos < delete_chunk.indices.size(); ++pos) + redelete_blob_states[delete_chunk.indices[pos]].removal = outcomes[pos]; + }); + ctx.delete_batches += delete_jobs.scheduled; + for (size_t chunk = 0; chunk < delete_jobs.scheduled; ++chunk) + { + if (!delete_jobs.errors[chunk]) + continue; + for (const size_t index : delete_chunks[chunk].indices) + redelete_blob_states[index].error = delete_jobs.errors[chunk]; } + std::vector schedule_failures; + if (observe_jobs.schedule_error) + schedule_failures.push_back({*observe_jobs.schedule_error, entries.size() - observe_jobs.schedule_error->index}); + if (delete_jobs.schedule_error) + { + size_t delete_unsent = 0; + for (size_t chunk = delete_jobs.schedule_error->index; chunk < delete_chunks.size(); ++chunk) + delete_unsent += delete_chunks[chunk].indices.size(); + schedule_failures.push_back( + {{delete_chunks[delete_jobs.schedule_error->index].indices.front(), delete_jobs.schedule_error->exception}, delete_unsent}); + } + const std::optional first_error + = recordRedeleteResults(ctx, entries, layout, redelete_blob_states, observe_jobs.scheduled, schedule_failures); + scheduleBlobMetaDeletes(entries, redelete_blob_states); + if (first_error) + std::rethrow_exception(first_error->exception); +} + +std::optional Gc::recordRedeleteResults( + RedeleteRoundContext & ctx, + const std::vector & entries, + const Layout & layout, + const std::vector & redelete_blob_states, + size_t observed, + const std::vector & schedule_failures) +{ std::optional first_error; - for (size_t i = 0; i < scheduled; ++i) + auto keep_first_error = [&](const IndexedException & error) + { + if (!first_error || error.index < first_error->index) + first_error = error; + }; + + for (size_t i = 0; i < observed; ++i) { - if (!io_errors[i]) + const std::exception_ptr & error = redelete_blob_states[i].error; + if (!error) continue; - reportRedeleteFailure(ctx, entries[i], layout, io_errors[i]); - if (!first_error) - first_error = IndexedException{i, io_errors[i]}; + recordRedeleteFailure(ctx, entries[i], layout, error); + keep_first_error({i, error}); } - if (schedule_error) + for (const RedeleteScheduleFailure & failure : schedule_failures) { - reportRedeleteFailure( - ctx, entries[schedule_error->index], layout, schedule_error->exception, entries.size() - schedule_error->index); - if (!first_error || schedule_error->index < first_error->index) - first_error = schedule_error; + recordRedeleteFailure(ctx, entries[failure.error.index], layout, failure.error.exception, failure.unsent); + keep_first_error(failure.error); } - /// Every submitted delete of the batch has already run. If `applyRedeleteOutcome` throws, the batch stays deleted + /// Every submitted delete of the batch has already run. If `recordRedeleteSuccess` throws, the batch stays deleted /// but unrecorded: the round fails before its outcome logs and `gc/state` commit, every entry stays /// `delete_pending`, and the next round records the deleted ones as `Absent`. - for (size_t i = 0; i < scheduled; ++i) + for (size_t i = 0; i < observed; ++i) { - if (!io_errors[i]) - applyRedeleteOutcome(ctx, entries[i], io_results[i]); + if (!redelete_blob_states[i].error && redelete_blob_states[i].removal) + recordRedeleteSuccess(ctx, entries[i], redelete_blob_states[i]); } - - if (first_error) - std::rethrow_exception(first_error->exception); + return first_error; } RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool allow_steal, UniversePolicy policy, @@ -1055,10 +1161,12 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al const uint64_t graduated_before = report.graduated; const uint64_t redelete_failed_before = report.redelete_failed; uint64_t redelete_jobs_scheduled = 0; + uint64_t redelete_delete_batches = 0; const auto record_redelete_job_metrics = [&] { pending_deletes_timer->metric("jobs_scheduled", redelete_jobs_scheduled); pending_deletes_timer->metric("jobs_failed", report.redelete_failed - redelete_failed_before); + pending_deletes_timer->metric("delete_batches", redelete_delete_batches); }; SCOPE_EXIT_SAFE({ if (pending_deletes_timer) @@ -1092,7 +1200,8 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al .round_work_budget = round_work_budget, .report = report, .outcomes = outcomes, - .jobs_scheduled = redelete_jobs_scheduled}; + .jobs_scheduled = redelete_jobs_scheduled, + .delete_batches = redelete_delete_batches}; redeleteBlobs(redelete_ctx, redelete_now, *io_pool, store->poolConfig().gc_io_concurrency, layout, op); for (const RetiredEntry & entry : merge.spared) { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index 316f413fbb66..227a03991f71 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -510,6 +510,8 @@ class Gc /// limit, clamped to [1, `kBulkDeleteMaxKeys`] because an empty request would never advance. size_t bulkDeleteChunkKeys() const; + size_t conditionalDeleteChunkKeys() const; + void setRebuildEdgeBudgetForTest(uint64_t n) { rebuild_edge_budget_override = n; } /// TEST SEAM: disable the round's journal trim so a folded event stays in the journal @@ -541,6 +543,23 @@ class Gc /// (throwaway keys) remove non-content objects they own. Adding a second content-delete site is a /// protocol defect. + struct RedeleteBlobState + { + String blob_key; + std::optional live_token; + std::optional removal; + std::exception_ptr error; + }; + + struct ConditionalDeleteChunk + { + std::vector indices; + std::vector removals; + }; + + static std::vector prepareConditionalDeleteChunks( + const std::vector & redelete_blob_states, size_t chunk_keys); + private: /// Lease acquire/renew/steal per the documented observation protocol. On success `state` holds the /// committed gc/state (with our lease) and `state_etag` the etag that write created. @@ -741,18 +760,18 @@ class Gc /// every destructive-work family the round touches — see `GcRoundWorkBudget`. GcRoundWorkBudget & work_budget); - struct RedeleteIo - { - String blob_key; - Removal del = Removal::Gone; - }; - struct IndexedException { size_t index; std::exception_ptr exception; }; + struct RedeleteScheduleFailure + { + IndexedException error; + size_t unsent; + }; + struct RedeleteRoundContext { uint64_t new_round; @@ -762,19 +781,48 @@ class Gc RoundReport & report; std::map & outcomes; uint64_t & jobs_scheduled; + uint64_t & delete_batches; }; - static RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); + static void observeRetiredBlob( + const RetiredEntry & entry, const Layout & layout, CasOperation & op, RedeleteBlobState & redelete_blob_state); - void applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteIo & io); + void recordRedeleteSuccess(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteBlobState & redelete_blob_state); - void reportRedeleteFailure( + void recordRedeleteFailure( RedeleteRoundContext & ctx, const RetiredEntry & entry, const Layout & layout, std::exception_ptr exception, std::optional unsent_count = std::nullopt); + std::optional recordRedeleteResults( + RedeleteRoundContext & ctx, + const std::vector & entries, + const Layout & layout, + const std::vector & redelete_blob_states, + size_t observed, + const std::vector & schedule_failures); + + void scheduleBlobMetaDeletes( + const std::vector & entries, const std::vector & redelete_blob_states); + + struct IoJobsResult + { + bool on_pool = false; + size_t scheduled = 0; + std::vector errors; + std::optional schedule_error; + }; + + IoJobsResult runRedeleteIoJobs( + ThreadPool & pool, + size_t concurrency, + size_t count, + size_t refuse_offset, + CasOperation & op, + const std::function & job); + void redeleteBlobs( RedeleteRoundContext & ctx, const std::vector & entries, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp index 8ea206618492..d3ad439b8244 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.cpp @@ -498,7 +498,9 @@ PoolPtr Pool::open(BackendPtr backend, PoolConfig config) /// namespace only (never the content planes) — acceptable. const UInt128 probe_uid = (static_cast(thread_local_rng()) << 64) | thread_local_rng(); CasOperation probe_op = bootstrap_requests.admit(); - runCapabilityProbe(probe_op, config.pool_prefix + "/_probe/" + u128ToHex(probe_uid)); + const CapabilityProbeResult probe_result = runCapabilityProbe( + probe_op, config.pool_prefix + "/_probe/" + u128ToHex(probe_uid)); + backend->pinConditionalBatchDeleteSupport(probe_result.conditional_batch_delete_supported); } else { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index c566a9164f1a..0d81d2586043 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -213,7 +213,7 @@ struct PoolConfig std::function teardown_phase2_throw_for_test = {}; std::function teardown_phase3_throw_for_test = {}; - std::function gc_redelete_apply_hook_for_test = {}; + std::function gc_redelete_record_hook_for_test = {}; std::optional gc_io_pool_refuse_at_for_test = std::nullopt; /// Mount-lease TTL: how long a freshly-renewed mount lease is valid. The local diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.cpp b/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.cpp index f51f274c0ba4..f1a32cafc812 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.cpp +++ b/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.cpp @@ -63,6 +63,17 @@ ObjectStorageIteratorPtr IObjectStorage::iterate( return std::make_shared(std::move(files)); } +std::vector IObjectStorage::removeObjectsIfTokensMatch( + const std::vector & objects, const ObjectStorageControlRequest & request) +{ + if (objects.size() > 1) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, + "Conditional batch object removal is not implemented for {} object storage", getName()); + if (objects.empty()) + return {}; + return {removeObjectIfTokenMatches(objects.front().object, objects.front().expected_token, request)}; +} + ObjectStorageIteratorPtr IObjectStorage::iterate( const std::string & path_prefix, size_t max_keys, diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h b/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h index 55f7293bad67..09ddbb2084fb 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h +++ b/src/Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h @@ -184,6 +184,12 @@ struct ConditionalRemoveResult bool created_delete_marker = false; /// backend reported a versioning delete marker }; +struct ConditionalRemoveObject +{ + StoredObject object; + std::string expected_token; +}; + /// Base class for all object storages which implement some subset of ordinary filesystem operations. /// /// Examples of object storages are S3, Azure Blob Storage, HDFS. @@ -326,6 +332,9 @@ class IObjectStorage virtual ConditionalRemoveResult removeObjectIfTokenMatches( const StoredObject & object, const std::string & etag, const ObjectStorageControlRequest & request); + virtual std::vector removeObjectsIfTokensMatch( + const std::vector & objects, const ObjectStorageControlRequest & request); + /// Removes every object in ONE request with no per-key precondition; an absent object is success. /// Content-addressed callers use it for write-once keys only, at most 1000 per call. Throws on a /// request-level failure and on any per-key error other than "not found", naming the failed keys. diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp index c42d10b8b056..3ec485ca072a 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp +++ b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp @@ -44,6 +44,8 @@ #include #include +#include + namespace ProfileEvents { extern const Event S3ListObjects; @@ -622,6 +624,147 @@ ConditionalRemoveResult S3ObjectStorage::removeObjectIfTokenMatches( object, etag, clientForRetryProfile(request), request.attempt_number); }); } +std::vector S3ObjectStorage::removeObjectsIfTokensMatch( + const std::vector & objects, const ObjectStorageControlRequest & request) +{ + if (objects.empty()) + return {}; + if (objects.size() == 1) + return {removeObjectIfTokenMatches(objects.front().object, objects.front().expected_token, request)}; + + std::vector identifiers; + identifiers.reserve(objects.size()); + for (const ConditionalRemoveObject & object : objects) + { + Aws::S3::Model::ObjectIdentifier identifier; + identifier.SetKey(object.object.remote_path); + identifier.SetETag(object.expected_token); + identifiers.push_back(std::move(identifier)); + } + Aws::S3::Model::Delete delete_request; + delete_request.SetObjects(std::move(identifiers)); + delete_request.SetQuiet(false); + S3::DeleteObjectsRequest request_s3; + request_s3.SetBucket(uri.bucket); + request_s3.SetDelete(std::move(delete_request)); + if (request.attempt_number != 0) + S3::setClickhouseAttemptNumber(request_s3, request.attempt_number); + auto blob_storage_log = BlobStorageLogWriter::create(disk_name); + const auto log_object = [&](const StoredObject & object, Int32 error_code, const String & error_message) + { + if (blob_storage_log) + blob_storage_log->addEvent( + BlobStorageLogElement::EventType::Delete, + uri.bucket, + object.remote_path, + object.local_path, + object.bytes_size, + 0, + error_code, + error_message); + }; + const auto log_all = [&](Int32 error_code, const String & error_message) + { + for (const auto & object : objects) + log_object(object.object, error_code, error_message); + }; + const auto outcome = refreshAndRetryOnExpiredCredentials([&] + { + ProfileEvents::increment(ProfileEvents::DiskS3DeleteObjects); + auto current_outcome = clientForRetryProfile(request)->DeleteObjects(request_s3); + if (!current_outcome.IsSuccess()) + { + const auto & error = current_outcome.GetError(); + log_all(static_cast(error.GetErrorType()), error.GetMessage()); + if (error.GetExceptionName() == "InvalidRequest" || error.GetExceptionName() == "InvalidArgument" + || error.GetExceptionName() == "NotImplemented") + { + s3_capabilities.setIsBatchDeleteSupported(false); + throw Exception(ErrorCodes::NOT_IMPLEMENTED, "{} does not support DeleteObjects", getName()); + } + throw S3Exception( + PreformattedMessage::create("{} while conditionally removing objects from S3", error.GetMessage()), + error.GetErrorType(), error.GetExceptionName()); + } + return current_outcome; + }); + s3_capabilities.setIsBatchDeleteSupported(true); + + std::unordered_set expected_keys; + expected_keys.reserve(objects.size()); + for (const ConditionalRemoveObject & object : objects) + expected_keys.insert(object.object.remote_path); + + std::unordered_set response_keys; + response_keys.reserve(objects.size()); + for (const auto & deleted : outcome.GetResult().GetDeleted()) + { + const String key = deleted.GetKey(); + if (!expected_keys.contains(key) || !response_keys.insert(key).second) + { + log_all(static_cast(Aws::S3::S3Errors::UNKNOWN), "DeleteObjects response has unexpected or duplicate keys"); + throw Exception(ErrorCodes::LOGICAL_ERROR, "S3 conditional DeleteObjects response has unexpected or duplicate key {}", key); + } + } + for (const auto & error : outcome.GetResult().GetErrors()) + { + const String key = error.GetKey(); + if (!expected_keys.contains(key) || !response_keys.insert(key).second) + { + log_all(static_cast(Aws::S3::S3Errors::UNKNOWN), "DeleteObjects response has unexpected or duplicate keys"); + throw Exception(ErrorCodes::LOGICAL_ERROR, "S3 conditional DeleteObjects response has unexpected or duplicate key {}", key); + } + } + if (response_keys.size() != objects.size()) + { + log_all(static_cast(Aws::S3::S3Errors::UNKNOWN), "DeleteObjects response omitted keys"); + throw Exception(ErrorCodes::LOGICAL_ERROR, "S3 conditional DeleteObjects response omitted keys"); + } + + std::vector results; + results.reserve(objects.size()); + std::optional first_error_code; + String first_error_name; + String first_error_message; + String first_error_key; + for (const ConditionalRemoveObject & object : objects) + { + const auto deleted = std::find_if(outcome.GetResult().GetDeleted().begin(), outcome.GetResult().GetDeleted().end(), + [&](const auto & entry) { return entry.GetKey() == object.object.remote_path; }); + if (deleted != outcome.GetResult().GetDeleted().end()) + { + log_object(object.object, 0, ""); + results.push_back({ConditionalRemoveOutcome::Removed, deleted->GetDeleteMarker()}); + continue; + } + const auto error = std::find_if(outcome.GetResult().GetErrors().begin(), outcome.GetResult().GetErrors().end(), + [&](const auto & entry) { return entry.GetKey() == object.object.remote_path; }); + if (error == outcome.GetResult().GetErrors().end()) + { + log_all(static_cast(Aws::S3::S3Errors::UNKNOWN), "DeleteObjects response omitted keys"); + throw Exception(ErrorCodes::LOGICAL_ERROR, "S3 conditional DeleteObjects response omitted key {}", object.object.remote_path); + } + const auto code = classifyDeleteObjectsErrorCode(error->GetCode()); + log_object(object.object, static_cast(code), error->GetMessage()); + if (error->GetCode() == "PreconditionFailed") + results.push_back({ConditionalRemoveOutcome::TokenMismatch, false}); + else if (error->GetCode() == "NoSuchKey") + results.push_back({ConditionalRemoveOutcome::NotFound, false}); + else if (!first_error_code) + { + first_error_code = code; + first_error_name = error->GetCode(); + first_error_message = error->GetMessage(); + first_error_key = error->GetKey(); + } + } + if (first_error_code) + throw S3Exception( + PreformattedMessage::create("{} while conditionally removing object {} from S3", first_error_message, first_error_key), + *first_error_code, first_error_name); + return results; +} + ConditionalRemoveResult S3ObjectStorage::removeObjectIfTokenMatchesImpl( const StoredObject & object, const std::string & etag, const std::shared_ptr & used_client, size_t attempt_seed) diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h index 27660043af7c..46df73a910f5 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h +++ b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h @@ -128,6 +128,9 @@ class S3ObjectStorage : public IObjectStorage ConditionalRemoveResult removeObjectIfTokenMatches( const StoredObject & object, const std::string & etag, const ObjectStorageControlRequest & request) override; + std::vector removeObjectsIfTokensMatch( + const std::vector & objects, const ObjectStorageControlRequest & request) override; + /// One `DeleteObjects` for the given objects (the caller chunks to at most 1000); absence is success. /// Exactly one object is always a plain `DeleteObject` instead (never gated on `s3_capabilities`: a /// single physical request per call, so there is nothing here for that capability to say no to). diff --git a/src/Disks/tests/cas_gc_marker_test_support.h b/src/Disks/tests/cas_gc_marker_test_support.h index efba25b9d7ac..cf3a6c1a18d2 100644 --- a/src/Disks/tests/cas_gc_marker_test_support.h +++ b/src/Disks/tests/cas_gc_marker_test_support.h @@ -46,13 +46,13 @@ inline BlobRef gateRef(const UInt128 & hash) /// A pool that folds every round, so a round's number is the number of rounds run. template PoolPtr openGatePool(std::shared_ptr backend, uint64_t io_concurrency = 1, uint64_t graduation_budget = 0, - std::function apply_hook = {}, uint64_t gc_shards = 1) + std::function record_hook = {}, uint64_t gc_shards = 1) { PoolConfig config{.pool_prefix = "p", .server_root_id = "test", .gc_fold_max_defer_rounds = 0, .gc_io_concurrency = io_concurrency}; config.gc_shards = gc_shards; config.gc_round_graduation_budget = graduation_budget; - config.gc_redelete_apply_hook_for_test = std::move(apply_hook); + config.gc_redelete_record_hook_for_test = std::move(record_hook); return Pool::open(std::move(backend), std::move(config)); } diff --git a/src/Disks/tests/gtest_cas_bulk_delete_backend.cpp b/src/Disks/tests/gtest_cas_bulk_delete_backend.cpp index b41851c8aa30..5aaced5a704a 100644 --- a/src/Disks/tests/gtest_cas_bulk_delete_backend.cpp +++ b/src/Disks/tests/gtest_cas_bulk_delete_backend.cpp @@ -141,6 +141,27 @@ TEST(CASBulkDeleteBackend, InstrumentedCountsOneRequestAndOneDeletePerKeyClass) EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASRootDelete].load() - root_before, 1u); } +TEST(CASConditionalBatchDeleteBackend, InstrumentedBackendForwardsMultiKeyConditionalRemove) +{ + auto inner = std::make_shared(); + inner->pinConditionalBatchDeleteSupport(true); + auto backend = std::make_shared(inner); + CasRequests requests = openRequestsForTest(backend); + CasOperation op = requests.admit(); + const String first_key = "p/conditional/instrumented/first"; + const String second_key = "p/conditional/instrumented/second"; + const Etag first_token = std::get(op.create(first_key, "v1", Retry::once())).etag; + const Etag second_token = std::get(op.create(second_key, "v1", Retry::once())).etag; + + EXPECT_EQ(backend->conditionalBatchDeleteKeyLimit(), kBulkDeleteMaxKeys); + const std::vector results = op.removeManyByConditional( + {{first_key, first_token}, {second_key, second_token}}, Retry::once()); + + EXPECT_EQ(results, (std::vector{Removal::Removed, Removal::Removed})); + EXPECT_FALSE(op.head(first_key, Retry::once()).has_value()); + EXPECT_FALSE(op.head(second_key, Retry::once()).has_value()); +} + #if USE_AWS_S3 TEST(CASBulkDeleteBackend, ThrottlingBackendForwardsStorageLimit) { diff --git a/src/Disks/tests/gtest_cas_bulk_delete_engine.cpp b/src/Disks/tests/gtest_cas_bulk_delete_engine.cpp index fc84d67ebc61..0cf521c9b6c6 100644 --- a/src/Disks/tests/gtest_cas_bulk_delete_engine.cpp +++ b/src/Disks/tests/gtest_cas_bulk_delete_engine.cpp @@ -34,6 +34,12 @@ std::vector manifestKeys(uint32_t count) return keys; } +class UnlimitedConditionalBatchBackend : public InMemoryBackend +{ +public: + size_t conditionalBatchDeleteKeyLimit() const override { return kBulkDeleteMaxKeys + 1; } +}; + } TEST(CASBulkDeleteEngine, AFailedAttemptReissuesTheWholeChunkAndAlreadyDeletedKeysAreSuccess) @@ -70,6 +76,35 @@ TEST(CASBulkDeleteEngine, AnEmptyChunkIsNoRequest) EXPECT_EQ(backend->bulkRemoveCalls(), 0u); } +TEST(CASConditionalBatchDeleteEngine, ReturnsMixedOutcomesInInputOrder) +{ + auto backend = std::make_shared(); + backend->pinConditionalBatchDeleteSupport(true); + CasRequests requests = openRequestsForTest(backend); + CasOperation op = requests.admit(); + const String removed_key = "p/conditional/removed"; + const String replaced_key = "p/conditional/replaced"; + const String gone_key = "p/conditional/gone"; + + const Etag removed_token = std::get(op.create(removed_key, "v1", Retry::once())).etag; + const Etag stale_token = std::get(op.create(replaced_key, "v1", Retry::once())).etag; + const Etag gone_token = std::get(op.create(gone_key, "v1", Retry::once())).etag; + ASSERT_TRUE(std::holds_alternative(op.replace(replaced_key, "v2", stale_token, Retry::once()))); + ASSERT_EQ(op.remove(gone_key, gone_token, Retry::once()), Removal::Removed); + + const std::vector results = op.removeManyByConditional( + {{removed_key, removed_token}, {replaced_key, stale_token}, {gone_key, gone_token}}, Retry::once()); + + ASSERT_EQ(results.size(), 3u); + EXPECT_EQ(results[0], Removal::Removed); + EXPECT_EQ(results[1], Removal::Mismatch); + EXPECT_EQ(results[2], Removal::Gone); + EXPECT_FALSE(op.head(removed_key, Retry::once()).has_value()); + const auto replaced = op.read(replaced_key, Retry::once()); + ASSERT_TRUE(replaced.has_value()); + EXPECT_EQ(replaced->bytes, "v2"); +} + TEST(CASBulkDeleteEngine, ExactlyOneThousandKeysIsOneRequest) { auto backend = std::make_shared(); @@ -89,3 +124,20 @@ TEST(CASBulkDeleteEngineDeathTest, MoreThanOneThousandKeysIsACallerBug) EXPECT_EQ(backend->bulkRemoveCalls(), 0u); } #endif + +TEST(CASConditionalBatchDeleteEngine, MoreThanOneThousandKeysIsRejectedEvenIfBackendReportsALargerLimit) +{ + auto backend = std::make_shared(); + backend->pinConditionalBatchDeleteSupport(true); + CasRequests requests = openRequestsForTest(backend); + CasOperation op = requests.admit(); + std::vector removals; + removals.reserve(kBulkDeleteMaxKeys + 1); + for (size_t i = 0; i <= kBulkDeleteMaxKeys; ++i) + { + const String key = "p/conditional/limit/" + std::to_string(i); + removals.push_back({key, std::get(op.create(key, "v", Retry::once())).etag}); + } + + EXPECT_THROW(op.removeManyByConditional(removals, Retry::once()), DB::Exception); +} diff --git a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp index 2c1f23ceaacd..7a8cc141ecc0 100644 --- a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp +++ b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp @@ -178,7 +178,7 @@ TEST(CASGcMetaWriter, ThrowingRoundDrainsBeforeReturning) namespace { -/// The delete job, scheduled the way `Gc::applyRedeleteOutcome` schedules it, and drained. +/// The delete job, scheduled the way `Gc::scheduleBlobMetaDeletes` schedules it, and drained. void runConfirmedMetaDelete(Gc & gc, const BlobRef & ref, uint64_t condemn_round) { gc.metaWriterForTest().scheduleConfirmedMetaDelete(ref, condemn_round); diff --git a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp index 3592f7172bc3..e6609e42d48c 100644 --- a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -82,6 +82,15 @@ class WorkerFaultBackend : public InMemoryBackend return InMemoryBackend::remove(key, expected_value, access); } + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override + { + for (const RawConditionalRemove & removal : removals) + if (takeFault(remove_keys, removal.key)) + throw std::runtime_error("injected worker remove fault for key " + removal.key); + return InMemoryBackend::removeManyByConditional(removals, access); + } + std::optional head(const String & key, TransportAccess & access) override { if (takeFault(head_keys, key)) @@ -165,6 +174,16 @@ class BlobRemoveWitnessBackend : public InMemoryBackend return result; } + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override + { + std::vector results; + results.reserve(removals.size()); + for (const RawConditionalRemove & removal : removals) + results.push_back(remove(removal.key, removal.expected_token, access)); + return results; + } + private: std::set blob_keys; size_t k_overlap = 1; @@ -184,7 +203,8 @@ PoolPtr openPoolWithIoConcurrency( uint64_t gc_shards = 1, uint64_t outcome_entry_budget = 0, CasEventSink event_sink = {}, - std::optional refuse_at_for_test = std::nullopt) + std::optional refuse_at_for_test = std::nullopt, + uint64_t gc_bulk_delete_chunk_keys = kBulkDeleteMaxKeys) { PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; config.gc_io_concurrency = concurrency; @@ -192,6 +212,8 @@ PoolPtr openPoolWithIoConcurrency( config.gc_round_outcome_entry_budget = outcome_entry_budget; config.event_sink = std::move(event_sink); config.gc_io_pool_refuse_at_for_test = refuse_at_for_test; + config.gc_bulk_delete_chunk_keys = gc_bulk_delete_chunk_keys; + config.skip_access_check = true; return Pool::open(std::move(backend), std::move(config)); } @@ -362,6 +384,46 @@ TEST(CASGCRedeleteConcurrency, ParallelRedeleteReclaimsEveryBlob) EXPECT_EQ(rows->total("jobs_failed"), 0u); } +TEST(CASGCRedeleteConcurrency, ConditionalDeleteBatchesAreRecorded) +{ + auto backend = std::make_shared(); + backend->pinConditionalBatchDeleteSupport(true); + auto store = openPoolWithIoConcurrency(backend, 4, 1, 0, {}, std::nullopt, 2); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kReplacedBlob)); + + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows)); + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + + EXPECT_EQ(rep.deleted, kBlobs); + ASSERT_EQ(rows->rows.size(), 1u); + EXPECT_EQ(rows->total("delete_batches"), 3u); +} + +TEST(CASGCRedeleteConcurrency, DeleteWaveEnqueueRefusalWaitsForSubmittedChunks) +{ + auto backend = std::make_shared(); + backend->pinConditionalBatchDeleteSupport(true); + auto store = openPoolWithIoConcurrency(backend, 4, 1, 0, {}, kBlobs + 1, 2); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kReplacedBlob)); + + RoundReport progress; + EXPECT_THROW( + gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress), + DB::Exception); + store->renewWatermarkOnce(); + + EXPECT_EQ(progress.redeleted, 2u); + EXPECT_EQ(progress.redelete_failed, 1u); + EXPECT_EQ(runRegularRoundReclaiming(gc).redeleted, kBlobs); + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); +} + TEST(CASGCRedeleteConcurrency, BlobDeletesActuallyOverlap) { auto backend = std::make_shared(); @@ -520,6 +582,21 @@ class RemoveRaceBackend : public InMemoryBackend bool replacementCommitted() const { return replacement_committed.load(); } RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + injectReplacement(key, access); + return InMemoryBackend::remove(key, expected_value, access); + } + + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override + { + for (const RawConditionalRemove & removal : removals) + injectReplacement(removal.key, access); + return InMemoryBackend::removeManyByConditional(removals, access); + } + +private: + void injectReplacement(const String & key, TransportAccess & access) { if (armed.load() && key == raced_key && armed.exchange(false)) { @@ -528,10 +605,8 @@ class RemoveRaceBackend : public InMemoryBackend if (current) replacement_committed.store(InMemoryBackend::write(key, current->bytes, current->value, access).has_value()); } - return InMemoryBackend::remove(key, expected_value, access); } -private: String raced_key; std::thread::id owner; std::atomic armed{false}; @@ -707,11 +782,11 @@ void expectBodiesPresentOnlyFor(Backend & backend, const Pool & store, const std EXPECT_EQ(blobAbsent(backend, store.layout(), blobHash(b)), !present.contains(b)) << "blob " << b; } -struct ApplyHookState +struct RecordHookState { std::atomic armed{false}; BlobRef target{}; - size_t applied = 0; + size_t recorded = 0; }; } @@ -990,22 +1065,22 @@ TEST(CASGCRedeleteConcurrency, ShardedPoolDeletesEveryShard) EXPECT_EQ(shards_with_rows, expected_shards); } -TEST(CASGCRedeleteConcurrency, ApplyPhaseFaultLeavesEveryEntryForTheNextRound) +TEST(CASGCRedeleteConcurrency, RecordPhaseFaultLeavesEveryEntryForTheNextRound) { - auto state = std::make_shared(); + auto state = std::make_shared(); auto backend = std::make_shared(); PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; config.gc_io_concurrency = 4; - config.gc_redelete_apply_hook_for_test = [state](const BlobRef & ref) + config.gc_redelete_record_hook_for_test = [state](const BlobRef & ref) { if (!state->armed.load()) return; if (ref == state->target) { state->armed.store(false); - throw std::runtime_error("injected apply fault"); + throw std::runtime_error("injected record fault"); } - ++state->applied; + ++state->recorded; }; auto store = Pool::open(backend, std::move(config)); Gc gc(store, kGc); @@ -1027,10 +1102,10 @@ TEST(CASGCRedeleteConcurrency, ApplyPhaseFaultLeavesEveryEntryForTheNextRound) } store->renewWatermarkOnce(); - EXPECT_NE(message.find("injected apply fault"), String::npos) << message; - EXPECT_EQ(progress.redeleted, state->applied); - EXPECT_EQ(failedCounter() - failed_before, 0u) << "a failed apply is not a failed delete"; - EXPECT_TRUE(allBlobsAbsent(*backend, *store)) << "every delete finished before the apply walk started"; + EXPECT_NE(message.find("injected record fault"), String::npos) << message; + EXPECT_EQ(progress.redeleted, state->recorded); + EXPECT_EQ(failedCounter() - failed_before, 0u) << "a failed record is not a failed delete"; + EXPECT_TRUE(allBlobsAbsent(*backend, *store)) << "every delete finished before the record walk started"; const RoundReport next = runRegularRoundReclaiming(gc); EXPECT_EQ(next.redeleted, kBlobs); @@ -1164,3 +1239,89 @@ TEST(CASGCRedeleteConcurrency, RoundsWithoutRedeletesWriteNoOutcomeLog) ADD_FAILURE() << "a round without redeletes or spares wrote the outcome log " << key << " with " << rows.size() << " row(s)"; EXPECT_TRUE(logs.empty()); } + +namespace +{ + +Gc::RedeleteBlobState liveRedeleteBlobState(CasOperation & op, const String & key) +{ + Gc::RedeleteBlobState redelete_blob_state; + redelete_blob_state.blob_key = key; + redelete_blob_state.live_token = std::get(op.create(key, "v", Retry::once())).etag; + return redelete_blob_state; +} + +Gc::RedeleteBlobState settledRedeleteBlobState(const String & key, Removal removal) +{ + Gc::RedeleteBlobState redelete_blob_state; + redelete_blob_state.blob_key = key; + redelete_blob_state.removal = removal; + return redelete_blob_state; +} + +std::vector> chunkIndices(const std::vector & chunks) +{ + std::vector> indices; + for (const auto & chunk : chunks) + indices.push_back(chunk.indices); + return indices; +} + +} + +TEST(CASGCConditionalDeleteChunks, OnlyLiveTokensWithoutErrorsAreChunked) +{ + auto backend = std::make_shared(); + OperationForTest op(backend); + std::vector redelete_blob_states; + redelete_blob_states.push_back(liveRedeleteBlobState(*op, "a")); + redelete_blob_states.push_back(settledRedeleteBlobState("gone", Removal::Gone)); + redelete_blob_states.push_back(liveRedeleteBlobState(*op, "b")); + redelete_blob_states.push_back(settledRedeleteBlobState("replaced", Removal::Mismatch)); + redelete_blob_states.push_back(liveRedeleteBlobState(*op, "failed")); + redelete_blob_states.back().error = std::make_exception_ptr(std::runtime_error("head failed")); + redelete_blob_states.emplace_back(); + redelete_blob_states.push_back(liveRedeleteBlobState(*op, "c")); + + const auto chunks = Gc::prepareConditionalDeleteChunks(redelete_blob_states, 10); + + ASSERT_EQ(chunkIndices(chunks), (std::vector>{{0, 2, 6}})); + const auto & removals = chunks.front().removals; + ASSERT_EQ(removals.size(), 3u); + for (size_t pos = 0; pos < removals.size(); ++pos) + { + const auto & redelete_blob_state = redelete_blob_states[chunks.front().indices[pos]]; + EXPECT_EQ(removals[pos].key, redelete_blob_state.blob_key); + EXPECT_EQ(removals[pos].expected_token, *redelete_blob_state.live_token); + } +} + +TEST(CASGCConditionalDeleteChunks, ChunksAreBoundedByChunkKeys) +{ + auto backend = std::make_shared(); + OperationForTest op(backend); + std::vector redelete_blob_states; + for (const String key : {"a", "b", "c", "d", "e"}) + redelete_blob_states.push_back(liveRedeleteBlobState(*op, key)); + + EXPECT_EQ( + chunkIndices(Gc::prepareConditionalDeleteChunks(redelete_blob_states, 2)), + (std::vector>{{0, 1}, {2, 3}, {4}})); + EXPECT_EQ( + chunkIndices(Gc::prepareConditionalDeleteChunks(redelete_blob_states, 1)), + (std::vector>{{0}, {1}, {2}, {3}, {4}})); + EXPECT_EQ( + chunkIndices(Gc::prepareConditionalDeleteChunks(redelete_blob_states, 5)), + (std::vector>{{0, 1, 2, 3, 4}})); +} + +TEST(CASGCConditionalDeleteChunks, NothingEligibleGivesNoChunks) +{ + std::vector redelete_blob_states; + redelete_blob_states.push_back(settledRedeleteBlobState("gone", Removal::Gone)); + redelete_blob_states.push_back(settledRedeleteBlobState("replaced", Removal::Mismatch)); + redelete_blob_states.emplace_back(); + + EXPECT_TRUE(Gc::prepareConditionalDeleteChunks(redelete_blob_states, 2).empty()); + EXPECT_TRUE(Gc::prepareConditionalDeleteChunks({}, 2).empty()); +} diff --git a/src/Disks/tests/gtest_cas_probe.cpp b/src/Disks/tests/gtest_cas_probe.cpp index 6b9da190dcad..43c20b980791 100644 --- a/src/Disks/tests/gtest_cas_probe.cpp +++ b/src/Disks/tests/gtest_cas_probe.cpp @@ -33,6 +33,41 @@ TEST(CASProbe, PassesOnEnforcingBackend) EXPECT_TRUE(op.list("p/.cas_probe", "", 10, Retry::once()).keys.empty()); // probe cleans up after itself } +namespace +{ + +class IgnoresConditionalBatchTokensBackend : public InMemoryBackend +{ +public: + std::vector removeManyByConditional( + const std::vector & removals, TransportAccess & access) override + { + std::vector results; + results.reserve(removals.size()); + for (const RawConditionalRemove & removal : removals) + { + const auto current = InMemoryBackend::head(removal.key, access); + results.push_back(current ? InMemoryBackend::remove(removal.key, current->value, access) : RawRemoval::Gone); + } + return results; + } +}; + +} + +TEST(CASProbe, ReportsConditionalBatchCapability) +{ + InMemoryBackend enforcing; + auto enforcing_requests = makeRequests(enforcing); + auto enforcing_op = enforcing_requests.admit(); + EXPECT_TRUE(runCapabilityProbe(enforcing_op, "p/.cas_probe").conditional_batch_delete_supported); + + IgnoresConditionalBatchTokensBackend ignores; + auto ignores_requests = makeRequests(ignores); + auto ignores_op = ignores_requests.admit(); + EXPECT_FALSE(runCapabilityProbe(ignores_op, "p/.cas_probe").conditional_batch_delete_supported); +} + TEST(CASProbe, FailsClosedOnNonEnforcingDelete) { auto b = std::make_shared(); @@ -90,6 +125,20 @@ TEST(CASProbe, ConcurrentMountsDoNotCollide) EXPECT_TRUE(probe_op.read("p/_probe/token", Retry::once()).has_value()); } +TEST(CASProbe, PoolPinsConditionalBatchCapabilityOnlyAfterProbe) +{ + auto backend = std::make_shared(); + EXPECT_EQ(backend->conditionalBatchDeleteKeyLimit(), 1u); + + auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + EXPECT_EQ(store->poolBackendPtr()->conditionalBatchDeleteKeyLimit(), kBulkDeleteMaxKeys); + + auto skipped_backend = std::make_shared(); + auto skipped = Pool::open(skipped_backend, PoolConfig{ + .pool_prefix = "skip", .server_root_id = "test", .skip_access_check = true}); + EXPECT_EQ(skipped->poolBackendPtr()->conditionalBatchDeleteKeyLimit(), 1u); +} + /// RFC cas-s3-timeout-retry-control: a Native-mode mount over an object storage that does not support /// the SingleAttempt retry profile must never silently proceed under the disk's default (~500-attempt) /// transparent retry policy — see Backend::checkConditionalWriteSingleAttemptSupport. This calls the @@ -225,7 +274,7 @@ TEST(CASProbe, ReorderedProbePassesOnAllThreeDialects) auto op = requests.admit(); EXPECT_NO_THROW(runCapabilityProbe(op, "p/.cas_probe")) << "dialect " << static_cast(dialect); EXPECT_TRUE(op.list("p/.cas_probe", "", 10, Retry::once()).keys.empty()) << "dialect " << static_cast(dialect); - EXPECT_EQ(b.write_reached, 4) << "dialect " << static_cast(dialect); - EXPECT_EQ(b.remove_reached, 2) << "dialect " << static_cast(dialect); + EXPECT_EQ(b.write_reached, 7) << "dialect " << static_cast(dialect); + EXPECT_EQ(b.remove_reached, 3) << "dialect " << static_cast(dialect); } } diff --git a/src/Disks/tests/gtest_cas_s3_bulk_delete_fallback.cpp b/src/Disks/tests/gtest_cas_s3_bulk_delete_fallback.cpp index d5bbf95a8dba..1750b00b5bb1 100644 --- a/src/Disks/tests/gtest_cas_s3_bulk_delete_fallback.cpp +++ b/src/Disks/tests/gtest_cas_s3_bulk_delete_fallback.cpp @@ -80,10 +80,13 @@ class ScriptedS3Server std::lock_guard lock(owner.mutex); owner.methods_seen.push_back(request.getMethod()); } - /// `DeleteObjects` carries a request body (the XML `` payload); leaving it unread on a - /// keep-alive connection makes Poco parse those leftover bytes as the start of the NEXT - /// request once this handler returns, corrupting the very next `DeleteObject` this test expects. - request.stream().ignore(std::numeric_limits::max()); + const std::string body( + std::istreambuf_iterator(request.stream()), + std::istreambuf_iterator()); + { + std::lock_guard lock(owner.mutex); + owner.bodies_seen.push_back(body); + } owner.responder(request, response); } }; @@ -108,6 +111,7 @@ class ScriptedS3Server Responder responder; mutable std::mutex mutex; std::vector methods_seen; + std::vector bodies_seen; public: explicit ScriptedS3Server(Responder responder_) @@ -127,13 +131,19 @@ class ScriptedS3Server std::lock_guard lock(mutex); return static_cast(std::count(methods_seen.begin(), methods_seen.end(), method)); } + + std::string lastBody() const + { + std::lock_guard lock(mutex); + return bodies_seen.empty() ? std::string{} : bodies_seen.back(); + } }; void sendXml(Poco::Net::HTTPServerResponse & response, Poco::Net::HTTPResponse::HTTPStatus status, const std::string & body) { response.setContentType("application/xml"); response.setContentLength(body.size()); - response.setStatus(status); + response.setStatusAndReason(status); auto & out = response.send(); out << body; out.flush(); @@ -199,8 +209,22 @@ void sendDeleteObjectsResult(Poco::Net::HTTPServerResponse & response, const std sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, body); } +void sendConditionalDeleteObjectsResult(Poco::Net::HTTPServerResponse & response) +{ + const std::string body = + "" + "" + "first" + "secondPreconditionFailedstale" + ""; + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, body); +} + std::shared_ptr makeStorageForTest( - const std::string & endpoint, const DB::S3Capabilities & capabilities, std::optional objects_chunk_size_to_delete = {}) + const std::string & endpoint, + const DB::S3Capabilities & capabilities, + std::optional objects_chunk_size_to_delete = {}, + const DB::S3ObjectStorage::S3CredentialsRefreshCallback & credentials_refresh_callback = {}) { DB::RemoteHostFilter remote_host_filter; DB::S3::PocoHTTPClientConfiguration cfg = DB::S3::ClientFactory::instance().createClientConfiguration( @@ -238,7 +262,7 @@ std::shared_ptr makeStorageForTest( return std::make_shared( std::move(client), std::move(settings), DB::S3::URI(endpoint + "/test-bucket/"), capabilities, - DB::ObjectStorageKeyGeneratorPtr{}, "disk"); + DB::ObjectStorageKeyGeneratorPtr{}, "disk", true, credentials_refresh_callback); } DB::ContextPtr contextForTest() @@ -294,6 +318,145 @@ TEST(S3BulkDeleteFallback, PerKeyErrorsWithinASuccessfulBatchAreUnchanged) EXPECT_EQ(server.countMethod("DELETE"), 0u); } +TEST(CASS3ConditionalBatchDelete, ReturnsPerKeyOutcomesInInputOrder) +{ + (void)contextForTest(); + + ScriptedS3Server server([](const Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response) + { + ASSERT_EQ(request.getMethod(), "POST"); + sendConditionalDeleteObjectsResult(response); + }); + auto storage = makeStorageForTest(server.getUrl(), DB::S3Capabilities{}); + + const auto results = storage->removeObjectsIfTokensMatch( + {{DB::StoredObject("first"), "etag-first"}, {DB::StoredObject("second"), "etag-second"}}, + DB::ObjectStorageControlRequest{}); + + ASSERT_EQ(results.size(), 2u); + EXPECT_EQ(results[0].outcome, DB::ConditionalRemoveOutcome::Removed); + EXPECT_EQ(results[1].outcome, DB::ConditionalRemoveOutcome::TokenMismatch); + EXPECT_EQ(server.countMethod("POST"), 1u); + EXPECT_NE(server.lastBody().find("first"), std::string::npos); + EXPECT_NE(server.lastBody().find("etag-first"), std::string::npos); + EXPECT_NE(server.lastBody().find("second"), std::string::npos); + EXPECT_NE(server.lastBody().find("etag-second"), std::string::npos); +} + +TEST(CASS3ConditionalBatchDelete, ConvertsExplicitUnsupportedToNotImplemented) +{ + (void)contextForTest(); + + ScriptedS3Server server([](const Poco::Net::HTTPServerRequest &, Poco::Net::HTTPServerResponse & response) + { + sendBatchNotImplemented(response); + }); + auto storage = makeStorageForTest(server.getUrl(), DB::S3Capabilities{}); + + expectNotImplemented([&] + { + storage->removeObjectsIfTokensMatch( + {{DB::StoredObject("first"), "etag-first"}, {DB::StoredObject("second"), "etag-second"}}, + DB::ObjectStorageControlRequest{}); + }); + EXPECT_EQ(storage->batchDeleteKeyLimit(), 1u); +} + +TEST(CASS3ConditionalBatchDelete, RefreshesExpiredCredentialsBeforeReturningFailure) +{ + (void)contextForTest(); + + std::atomic requests{0}; + ScriptedS3Server server([&](const Poco::Net::HTTPServerRequest &, Poco::Net::HTTPServerResponse & response) + { + if (requests.fetch_add(1) == 0) + { + sendSingleDeleteError(response, Poco::Net::HTTPResponse::HTTP_FORBIDDEN, "ExpiredToken", "expired"); + return; + } + sendConditionalDeleteObjectsResult(response); + }); + + std::atomic refreshes{0}; + auto storage = makeStorageForTest( + server.getUrl(), DB::S3Capabilities{}, {}, + [&]() -> std::unique_ptr + { + ++refreshes; + DB::RemoteHostFilter remote_host_filter; + DB::S3::PocoHTTPClientConfiguration cfg = DB::S3::ClientFactory::instance().createClientConfiguration( + "us-east-1", remote_host_filter, 100, + DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = 0}, + false, false, false, true, {}, {}); + cfg.endpointOverride = server.getUrl(); + cfg.s3_use_adaptive_timeouts = false; + return DB::S3::ClientFactory::instance().create( + cfg, + DB::S3::ClientSettings{ + .use_virtual_addressing = false, + .disable_checksum = false, + .gcs_issue_compose_request = false, + .is_s3express_bucket = false, + }, + "ACCESS_KEY_ID", "SECRET_ACCESS_KEY", "", {}, {}, DB::S3::CredentialsConfiguration{}); + }); + + const auto results = storage->removeObjectsIfTokensMatch( + {{DB::StoredObject("first"), "etag-first"}, {DB::StoredObject("second"), "etag-second"}}, + DB::ObjectStorageControlRequest{}); + + ASSERT_EQ(results.size(), 2u); + EXPECT_EQ(refreshes.load(), 1u); + EXPECT_EQ(requests.load(), 2u); +} + +TEST(CASS3ConditionalBatchDelete, DoesNotTreatMissingBucketAsMissingObject) +{ + (void)contextForTest(); + + ScriptedS3Server server([](const Poco::Net::HTTPServerRequest &, Poco::Net::HTTPServerResponse & response) + { + sendDeleteObjectsResult(response, {{"first", "NoSuchBucket"}, {"second", "PreconditionFailed"}}); + }); + auto storage = makeStorageForTest(server.getUrl(), DB::S3Capabilities{}); + + try + { + storage->removeObjectsIfTokensMatch( + {{DB::StoredObject("first"), "etag-first"}, {DB::StoredObject("second"), "etag-second"}}, + DB::ObjectStorageControlRequest{}); + FAIL() << "expected NoSuchBucket to remain a hard error"; + } + catch (const DB::S3Exception & e) + { + EXPECT_EQ(e.getExceptionName(), "NoSuchBucket"); + } +} + +TEST(CASS3ConditionalBatchDelete, RejectsDuplicateResponseKey) +{ + (void)contextForTest(); + + ScriptedS3Server server([](const Poco::Net::HTTPServerRequest &, Poco::Net::HTTPServerResponse & response) + { + const std::string body = + "" + "" + "first" + "first" + "secondPreconditionFailedstale" + ""; + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, body); + }); + auto storage = makeStorageForTest(server.getUrl(), DB::S3Capabilities{}); + + EXPECT_THROW( + storage->removeObjectsIfTokensMatch( + {{DB::StoredObject("first"), "etag-first"}, {DB::StoredObject("second"), "etag-second"}}, + DB::ObjectStorageControlRequest{}), + DB::Exception); +} + TEST(S3BulkDeleteFallback, UnsupportedBatchReplyRecordsCapabilityFalseAndThrowsNotImplemented) { (void)contextForTest(); diff --git a/tests/integration/test_cas_gc_conditional_batch_delete/__init__.py b/tests/integration/test_cas_gc_conditional_batch_delete/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/integration/test_cas_gc_conditional_batch_delete/configs/storage_conf.xml b/tests/integration/test_cas_gc_conditional_batch_delete/configs/storage_conf.xml new file mode 100644 index 000000000000..efb13348faab --- /dev/null +++ b/tests/integration/test_cas_gc_conditional_batch_delete/configs/storage_conf.xml @@ -0,0 +1,29 @@ + + + + + object_storage + s3 + cas + 30 + 10000 + itest-cas-gc-conditional-batch + http://rustfs1:11121/test/cas_gc_conditional_batch/ + clickhouse + clickhouse + 1 + 3600 + 2 + + + + + +
+ disk_cas_gc_conditional_batch +
+ + + + + diff --git a/tests/integration/test_cas_gc_conditional_batch_delete/test.py b/tests/integration/test_cas_gc_conditional_batch_delete/test.py new file mode 100644 index 000000000000..32d9a72dd677 --- /dev/null +++ b/tests/integration/test_cas_gc_conditional_batch_delete/test.py @@ -0,0 +1,115 @@ +import pytest + +from helpers.cluster import ClickHouseCluster + +cluster = ClickHouseCluster(__file__) + +STORAGE_POLICY = "cas_gc_conditional_batch" +DISK_NAME = "disk_cas_gc_conditional_batch" +BLOBS_PREFIX = "cas_gc_conditional_batch/blobs/" +CHUNK_KEYS = 2 +NUM_INSERTS = 6 +MAX_GC_ROUNDS = 12 + + +@pytest.fixture(scope="module", autouse=True) +def start_cluster(): + cluster.add_instance( + "node", + main_configs=["configs/storage_conf.xml"], + with_rustfs=True, + stay_alive=True, + ) + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def list_blob_bodies(): + return { + o.object_name + for o in cluster.rustfs_client.list_objects( + cluster.rustfs_bucket, BLOBS_PREFIX, recursive=True + ) + if not o.object_name.endswith(".meta") + } + + +def latest_gc_round(node): + node.query("SYSTEM FLUSH LOGS") + return int( + node.query( + f"SELECT max(round) FROM system.cas_gc_log WHERE disk_name = '{DISK_NAME}'" + ).strip() + ) + + +def gc_pending_delete_rows(node, after_round): + node.query("SYSTEM FLUSH LOGS") + rows = ( + node.query( + "SELECT p.phase_metrics['redeleted'], p.phase_metrics['delete_batches'], " + "p.phase_metrics['jobs_failed'] " + "FROM system.cas_gc_log AS p INNER JOIN " + "(SELECT round_id FROM system.cas_gc_log " + f"WHERE disk_name = '{DISK_NAME}' AND event_type = 'Finish' " + f"AND round > {after_round}) AS f USING round_id " + f"WHERE p.disk_name = '{DISK_NAME}' AND p.event_type = 'Phase' " + "AND p.phase = 'pending_deletes' AND p.phase_metrics['redeleted'] > 0 " + "ORDER BY p.event_time_microseconds" + ) + .strip() + .splitlines() + ) + return [tuple(int(x) for x in row.split("\t")) for row in rows] + + +def gc_redeleted_total(node, after_round): + node.query("SYSTEM FLUSH LOGS") + return int( + node.query( + "SELECT sum(entries_redeleted) FROM system.cas_gc_log " + f"WHERE disk_name = '{DISK_NAME}' AND event_type = 'Finish' " + f"AND round > {after_round}" + ).strip() + ) + + +def test_rustfs_pending_blob_deletes_use_singleton_fallback(): + node = cluster.instances["node"] + table = "conditional_batch_redelete" + blobs_before = list_blob_bodies() + node.query( + f"CREATE TABLE {table} (k UInt64, v String) ENGINE = MergeTree ORDER BY k " + f"SETTINGS storage_policy = '{STORAGE_POLICY}'" + ) + node.query(f"SYSTEM STOP MERGES {table}") + for i in range(NUM_INSERTS): + node.query( + f"INSERT INTO {table} SELECT number + {i * 1000}, " + f"concat(toString(number), '-{i}') FROM numbers(1000) " + "SETTINGS max_insert_block_size = 1000" + ) + + created_blobs = list_blob_bodies() - blobs_before + assert len(created_blobs) > CHUNK_KEYS + first_round = latest_gc_round(node) + node.query(f"DROP TABLE {table} SYNC") + + rows = [] + remaining = created_blobs + for _ in range(MAX_GC_ROUNDS): + node.query("SYSTEM CAS GC RUN") + rows = gc_pending_delete_rows(node, first_round) + remaining = created_blobs & list_blob_bodies() + if not remaining and rows: + break + + assert not remaining, f"GC left blob bodies in RustFS: {sorted(remaining)}" + assert rows, "pending_deletes never reported completed blob redeletes" + for redeleted, batches, jobs_failed in rows: + assert jobs_failed == 0 + assert batches == redeleted + assert gc_redeleted_total(node, first_round) >= len(created_blobs)