diff --git a/docs/en/antalya/cas/architecture/backend.md b/docs/en/antalya/cas/architecture/backend.md index 8f889a3bfe038..44847af56030b 100644 --- a/docs/en/antalya/cas/architecture/backend.md +++ b/docs/en/antalya/cas/architecture/backend.md @@ -134,6 +134,14 @@ battery, skipping the battery used to skip it too, which is exactly why the thir committed, so retries on the conditional path must be explicit CAS state-machine transitions, not transparent client behavior. +Blob publication is unconditional and does not use that profile, but its object-creating request +(`PutObject`, `CopyObject`, `CompleteMultipartUpload`) is also sent once. On a store with +content-derived ETags, a resent body that lands after GC deleted its incarnation brings that ETag +back, and the pending exact-ETag delete then removes a live body. The request runs on a clone of the +disk client with the disk's `request_timeout_ms`, no adaptive first-attempt timeouts, and the connect +timeout capped as for the control plane. `CreateMultipartUpload`, `UploadPart` and `UploadPartCopy` +keep the disk client's retries. + A third staging check requires `supportsCopyMode(ObjectStorageCopyMode::NativeOnly)` when `cas_staging_backend = s3`. Writable mount fails closed when native same-store copy is unavailable; it does not silently fall back to local staging. Ordinary non-CAS `copyObject` fallback behavior is diff --git a/docs/en/antalya/cas/architecture/blob-protocol.md b/docs/en/antalya/cas/architecture/blob-protocol.md index bb72311d07c8c..1bea888fdeb84 100644 --- a/docs/en/antalya/cas/architecture/blob-protocol.md +++ b/docs/en/antalya/cas/architecture/blob-protocol.md @@ -65,11 +65,20 @@ Ordered steps in `PartWriteTxn::ensureBlobPresent`: 5. Streaming publication mints a fresh `incarnation_tag` and can use ordinary multipart. The first publication of an S3-staged source may use native same-store copy only after a miss; a condemned or subsequent publication retags and streams the staged payload. + The request that creates the object (`PutObject`, `CopyObject`, `CompleteMultipartUpload`) is + never resent by the S3 client after an ambiguous outcome; resends after a definite refusal, such as + expired credentials or a redirect, stay. On a store with content-derived ETags, a resent + envelope that lands after GC deleted it would restore the condemned incarnation. Part uploads keep + the client's retries. 6. `BlobSource::beginPublication` consumes the shared, monotonic `publication_attempted` state before backend I/O. A lost response cannot re-enable verbatim copy on a later attempt. -7. Retryable or ambiguous failures restart at `HEAD`. No dependency proof is recorded until a - present body with a `Clean` marker was observed, or publication and metadata reconciliation - completed. +7. Retryable or ambiguous failures restart at `HEAD`: up to 8 attempts, each with a fresh envelope. + The engine starts no new attempt after the 90-second deadline, but an attempt already started can + run longer: each creating request waits up to the disk's `request_timeout_ms`, and multipart part + uploads keep their own retries. A write can therefore take longer than 90 seconds before it + fails with a retryable error. No dependency + proof is recorded until a present body with a `Clean` marker was observed, or publication and + metadata reconciliation completed. **Two writers uploading identical content** may both observe absence and both publish. The last physical incarnation wins, but the key proves that both payloads have the same logical identity and diff --git a/docs/en/antalya/cas/architecture/correctness.md b/docs/en/antalya/cas/architecture/correctness.md index 8260ca02f9b47..6a0ae43069483 100644 --- a/docs/en/antalya/cas/architecture/correctness.md +++ b/docs/en/antalya/cas/architecture/correctness.md @@ -24,7 +24,7 @@ The full model index (source `.tla` files and proof-run records) lives at | Model (`docs/superpowers/models/`) | Invariant it proves | Counterexample it caught | |---|---|---| -| `CaBlobPublishCore.tla` | Split `HEAD` from unconditional publication while preserving logical content, fresh incarnation identity, monotonic `publication_attempted`, fence safety, and readiness only after metadata reconciliation | Eleven sabotage configurations cover condemned adoption, stale exact delete, ambiguous copy/PUT landing, envelope reuse after a later miss, missing precommit/meta reconciliation, fence loss, and wrong-content publication; three witnesses prove the safe paths are reachable | +| `CaBlobPublishCore.tla` | Split `HEAD` from unconditional publication while preserving logical content, fresh incarnation identity, monotonic `publication_attempted`, fence safety, and readiness only after metadata reconciliation | Sabotage configurations cover condemned adoption, stale exact delete, ambiguous copy/PUT landing, envelope reuse after a later miss, missing precommit/meta reconciliation, fence loss, and wrong-content publication; three witnesses prove the safe paths are reachable. The model also lets the S3 client resend an identical envelope below the engine: on a content-ETag store that resend restores a condemned incarnation, and a stale exact delete then removes it, which is why blob creation is never resent | | `CaIncarnationCore.tla` | `INV_NO_DANGLE`, `INV_NO_LOSS`, `INV_NO_RETURN` — the safety spine for the whole GC core | `sab_unconddelete`: replacing the exact-token delete with an unconditional one lets a stale delete kill a replacement incarnation | | `CaBuildRootPrecommit.tla` | `INV_NO_DANGLE_COMMITTED` — a committed manifest never references an absent blob | Reproduces the dangling-manifest hazard exactly: `WriteBlob → AdoptBlob → BuildDie → GcDelete → Commit` with no presence re-check publishes a manifest over a deleted blob | | `CaGcLeaseCore.tla` | `NoFalseSteal` — no leader steals leadership from a live, mid-round incumbent | Without the advisory heartbeat, a frozen `seq` during a round looks identical to a dead leader, and a second leader steals from the alive one | diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index a490428b30321..1f80bcec361af 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -107,7 +107,7 @@ entirely before release. Treat this table as a snapshot of the current build, no | `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, and the `pending_deletes` blob `HEAD` + conditional `DELETE` fan-out. 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_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. Blob body publication does not use this budget: its object-creating request is sent once under the disk's `request_timeout_ms`, with the connect timeout under the same cap | | `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 | | `cas_staging_backend` | `local` | Blob staging backend (`local` \| `s3`); `s3` is opt-in and requires native same-store copy on writable mount | diff --git a/docs/en/antalya/cas/operations/troubleshooting.md b/docs/en/antalya/cas/operations/troubleshooting.md index 1113ec1855dd4..ef63acf17a337 100644 --- a/docs/en/antalya/cas/operations/troubleshooting.md +++ b/docs/en/antalya/cas/operations/troubleshooting.md @@ -17,7 +17,7 @@ tools. | Symptom | Diagnosis | Action | |---|---|---| | A server keeps losing its mount lease and self-remounting | Check `system.cas_mounts` for the server's own `state`/`expires_at`, then correlate `watermark_renew` and `mount_remount` in `system.cas_log`; losing the lease trips a local fence and latches a remount generation | Read the failed renewal's `classification` before changing anything — it alone now says why (see [the decision flow](#mount-renewal-remount-flow)). Look for object-store latency consuming the confirmed lease or BOOTTIME advancement; see [the mount lease](/antalya/cas/architecture/mounts-and-leases#mount-lease) | -| Writes slow down or stall under load, with no exception reaching the client | S3 `SlowDown`/`ServiceUnavailable`/`RequestTimeout`/`InternalError` (5xx) responses are not on the request engine's `isDefinitelyRefusedWrite` definite-failure list (only malformed-request, entity-too-large, and access-denied that no credential refresh can fix are), so they classify as ambiguous and are retried automatically. Confirm with `sum(ProfileEvents['CASConditionalWriteUnresolved'])` rising alongside `sum(ProfileEvents['CASConditionalWriteAttempts'])` over `system.query_log` for the affected window (or `ProfileEvent_CASConditionalWriteUnresolved` in `system.metric_log` for a cumulative view across queries), and check `system.blob_storage_log` for `disk_name = ''` rows with a nonzero `error_code` around the same window | Nothing to configure per-request: the request engine retries the same `(key, bytes)` with capped-exponential backoff (200ms initial, capped at 5s, full jitter) until the 90-second operation deadline — there is no separate attempts ceiling, only the deadline — and the mount-lease renewer keeps extending the fence across the disruption — this is the "blips, throttling, partial outages" case the write path is built to survive. Confirm the mount lease itself is still renewing (`system.cas_mounts.expires_at` moving forward, `last_success_age_seconds` not climbing) — if it is, this is expected and self-resolving. If `SlowDown` responses are sustained rather than transient, check the bucket's request-rate limits against the pool's actual PUT/GET rate (see [bucket requirements](/antalya/cas/bucket-requirements)) and consider lowering `cas_blob_upload_pool_size` to reduce concurrent upload traffic; a write only surfaces a client-visible `NETWORK_ERROR` if the 90-second deadline is exhausted before the store recovers, and that error is retried by the ordinary merge/insert backoff, not silently dropped | +| Writes slow down or stall under load, with no exception reaching the client | S3 `SlowDown`/`ServiceUnavailable`/`RequestTimeout`/`InternalError` (5xx) responses are not on the request engine's `isDefinitelyRefusedWrite` definite-failure list (only malformed-request, entity-too-large, and access-denied that no credential refresh can fix are), so they classify as ambiguous and are retried automatically. Confirm with `sum(ProfileEvents['CASConditionalWriteUnresolved'])` rising alongside `sum(ProfileEvents['CASConditionalWriteAttempts'])` over `system.query_log` for the affected window (or `ProfileEvent_CASConditionalWriteUnresolved` in `system.metric_log` for a cumulative view across queries), and check `system.blob_storage_log` for `disk_name = ''` rows with a nonzero `error_code` around the same window | Nothing to configure per-request: the request engine retries the same `(key, bytes)` with capped-exponential backoff (200ms initial, capped at 5s, full jitter) until the 90-second operation deadline — for these conditional writes there is no separate attempts ceiling, only the deadline. A blob body upload is different: it is never resent as is after an ambiguous outcome, and each retry publishes a fresh envelope, up to 8 attempts, and no new attempt starts after the 90-second deadline. An attempt already started can run longer: each creating request waits up to the disk's `request_timeout_ms`, and multipart part uploads keep their own retries. `CASBlobPublishAttemptFailed` counts failed attempts; `CASBlobPublishGaveUp` counts uploads that failed at least once and then ended with an exception, at the attempt cap or the deadline. Meanwhile the mount-lease renewer keeps extending the fence across the disruption — this is the "blips, throttling, partial outages" case the write path is built to survive. Confirm the mount lease itself is still renewing (`system.cas_mounts.expires_at` moving forward, `last_success_age_seconds` not climbing) — if it is, this is expected and self-resolving. If `SlowDown` responses are sustained rather than transient, check the bucket's request-rate limits against the pool's actual PUT/GET rate (see [bucket requirements](/antalya/cas/bucket-requirements)) and consider lowering `cas_blob_upload_pool_size` to reduce concurrent upload traffic; a write only surfaces a client-visible `NETWORK_ERROR` if the 90-second deadline passes before the store recovers (an attempt started before the deadline can finish after it), and that error is retried by the ordinary merge/insert backoff, not silently dropped | | `GC` never seems to reclaim space after tables are dropped | `SELECT * FROM system.cas_gc_log WHERE event_type='Finish' ORDER BY event_time DESC LIMIT 5` — check `outcome`; also `SELECT is_leader FROM system.cas_mounts` on this node | If `outcome != 'Success'`/`'Deferred'`, see [reading GC health](/antalya/cas/operations/monitoring#gc-health); if this node is not the leader (`is_leader = 0`), it never reclaims for this disk — check the peer holding leadership. Reclamation also needs at least two full rounds past condemnation by design (the grace period is rounds, not acks) — a single manual `SYSTEM CAS GC RUN` will not finish it | | A dangling-access exception or `CORRUPTED_DATA` on read | Run `clickhouse-disks cas-fsck --detail` and check `dangling` specifically — it is the one class that means data loss, distinct from `unreachable`/`awaiting-gc`, which are just waiting for graduation | A nonzero `dangling` count is a real incident: collect the `--detail` output (see [what to collect before filing a bug](/antalya/cas/operations/debugging#filing-a-bug)) before taking any destructive action | | `SYSTEM CAS FSCK` or `clickhouse-disks cas-fsck` times out on a large pool | The scan is bounded by `--timeout` (default 600s / the `SYSTEM` form has no override); a large `roots/` prefix can make the scan slow | Retry with `--partial` to see the counts accumulated so far instead of aborting empty-handed, or `--namespace ` to scope the scan to a subset of namespaces | diff --git a/docs/en/antalya/metrics_and_events.md b/docs/en/antalya/metrics_and_events.md index c64074e66c07d..fdd1ad44e70c1 100644 --- a/docs/en/antalya/metrics_and_events.md +++ b/docs/en/antalya/metrics_and_events.md @@ -83,6 +83,8 @@ validation, and other callers. | `CASBlobHeadMiss` | `Number` | Number of successful backend HEAD requests under CAS blob paths that found no object, aggregated across all callers. | | `CASBlobList` | `Number` | Number of CAS blob LIST requests. Growing values indicate more object enumeration. | | `CASBlobOverwrite` | `Number` | Number of CAS blob overwrite requests. Growing values indicate repeated replacement writes. | +| `CASBlobPublishAttemptFailed` | `Number` | Number of CAS blob publication attempts that failed with an ambiguous or retryable error. The writer observes the blob again and publishes a fresh envelope, up to the attempt limit. | +| `CASBlobPublishGaveUp` | `Number` | Number of CAS blob publications that ended with an exception after at least one failed attempt, for example at the attempt limit or the deadline. | | `CASBlobPut` | `Number` | Number of CAS blob PUT requests. Grows with blob uploads and indicates storage write traffic. | | `CASBlobPutDeduplicated` | `Number` | Number of CAS blob deduplicating PUT requests. Grows when uploads reuse existing content. | @@ -523,7 +525,7 @@ These counters cover client-side time of individual S3 and Azure list and head c | `AzureListObjectsMicroseconds` | `Microseconds` | Time of Azure blob storage API ListObjects execution. | | `S3HeadObjectMicroseconds` | `Microseconds` | Time of S3 API HeadObject execution. | | `S3ListObjectsMicroseconds` | `Microseconds` | Time of S3 API ListObjects execution. | -| `S3SingleAttemptRetryConsultations` | `Number` | Number of AWS SDK retry consultations refused by the single-attempt S3 retry profile. Non-zero means the SDK attempted a transparent retry on a write that must make exactly one HTTP attempt. | +| `S3SingleAttemptRetryConsultations` | `Number` | Number of AWS SDK retry consultations refused by a single-attempt S3 client: the clone used by the single-attempt retry profile, or the clone that sends the object-creating requests of a write that forbids a resend. Non-zero means a request on such a client failed and was not resent. | ### Other profile events {#other-profile-events} diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 7e6cb47c6df2b..364055346a717 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -795,6 +795,8 @@ The server successfully detected this situation and will download merged part fr M(CASBlobCompareSwapConflict,"Number of CAS blob compare-and-swap conflicts. A non-zero value indicates concurrent update contention.", ValueType::Number) \ M(CASBlobHead, "Number of successful backend HEAD requests under CAS blob paths that found an object, aggregated across all callers.", ValueType::Number) \ M(CASBlobHeadMiss, "Number of successful backend HEAD requests under CAS blob paths that found no object, aggregated across all callers.", ValueType::Number) \ + M(CASBlobPublishAttemptFailed, "Number of CAS blob publication attempts that failed with an ambiguous or retryable error. The writer observes the blob again and publishes a fresh envelope, up to the attempt limit.", ValueType::Number) \ + M(CASBlobPublishGaveUp, "Number of CAS blob publications that ended with an exception after at least one failed attempt, for example at the attempt limit or the deadline.", ValueType::Number) \ M(CASBlobBodyPutAvoided, "Number of physical CAS blob body publications avoided after a safe presence observation found the object present. Growing values indicate saved upload traffic.", ValueType::Number) \ M(CASBlobUploadFanoutBatches, "Number of CAS intra-part blob-upload fan-out invocations (one per part publish that had at least one unique blob to upload). Compare with CASBlobUploadFanoutTasks for the average number of unique blobs uploaded per part.", ValueType::Number) \ M(CASBlobUploadFanoutTasks, "Number of CAS blob-upload tasks dispatched to the fan-out pool (one per unique blob ref across all part publishes). Duplicate references within a part collapse to one task.", ValueType::Number) \ @@ -981,7 +983,7 @@ The server successfully detected this situation and will download merged part fr M(S3CompleteMultipartUpload, "Number of S3 API CompleteMultipartUpload calls.", ValueType::Number) \ M(S3PutObject, "Number of S3 API PutObject calls.", ValueType::Number) \ M(S3GetObject, "Number of S3 API GetObject calls.", ValueType::Number) \ - M(S3SingleAttemptRetryConsultations, "Number of AWS SDK retry consultations refused by the single-attempt S3 retry profile. Non-zero means the SDK attempted a transparent retry on a write that must make exactly one HTTP attempt.", ValueType::Number) \ + M(S3SingleAttemptRetryConsultations, "Number of AWS SDK retry consultations refused by a single-attempt S3 client: the clone used by the single-attempt retry profile, or the clone that sends the object-creating requests of a write that forbids a resend. Non-zero means a request on such a client failed and was not resent.", ValueType::Number) \ \ M(DiskS3DeleteObjects, "Number of DiskS3 API DeleteObject(s) calls.", ValueType::Number) \ M(DiskS3CopyObject, "Number of DiskS3 API CopyObject calls.", ValueType::Number) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp index 31b0d32e3662d..0a72ef6515009 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.cpp @@ -801,10 +801,9 @@ SentinelProbeResult ObjectStorageBackend::probeSentinelUnder(const String & key, /// Settings for a genuine Native conditional write. Mark the request for the typed conditional /// dialect and, on a generation-token store, force a single PUT: GCS does not enforce the condition -/// on multipart completion. Blob publication never uses these settings; it remains an ordinary -/// unconditional multipart-capable write. CAS-mutable keys (shard manifests, gc/state, the registry) -/// also skip the racy post-upload existence/size check; a publish's manifest CAS was observed racing -/// the GC fence there. +/// on multipart completion. Blob publication uses `blobPublicationWriteSettings` instead. CAS-mutable +/// keys (shard manifests, gc/state, the registry) also skip the racy post-upload existence/size check; +/// a publish's manifest CAS was observed racing the GC fence there. WriteSettings ObjectStorageBackend::conditionalWriteSettings(size_t attempt_no) const { WriteSettings ws; @@ -874,6 +873,18 @@ std::expected ObjectStorageBackend::write( return emuWrite(key, bytes); } +/// On a content-ETag store a resent envelope that lands after GC deleted it restores the condemned +/// incarnation, and the exact-token delete of that incarnation then removes a live body. Retries belong +/// to the caller, which re-observes and publishes a fresh envelope. Requests keep the storage's own +/// request timeout, which a body upload needs. +WriteSettings ObjectStorageBackend::blobPublicationWriteSettings() const +{ + WriteSettings ws; + ws.object_storage_single_attempt_object_creation = true; + ws.object_storage_connect_timeout_cap_ms = connect_timeout_cap_ms; + return ws; +} + void ObjectStorageBackend::publish(const BlobPublishRequest & request, TransportAccess &) { if (const auto * streaming = std::get_if(&request.publication)) @@ -893,14 +904,14 @@ void ObjectStorageBackend::publish(const BlobPublishRequest & request, Transport return; } - /// Ordinary unconditional rewrite: default request mode, retry profile, and multipart policy. - /// In particular, generation stores are not restricted by the conditional single-PUT cap. + /// Unconditional rewrite: default request mode and multipart policy, so generation stores are + /// not restricted by the conditional single-PUT cap. auto out = object_storage->writeObject( StoredObject(request.destination_key), WriteMode::Rewrite, /*attributes=*/std::nullopt, DBMS_DEFAULT_BUFFER_SIZE, - WriteSettings{}); + blobPublicationWriteSettings()); out->write(streaming->fresh_envelope.data(), streaming->fresh_envelope.size()); blob_publication_detail::BlobPayloadCopyResult copy_result; try @@ -933,7 +944,7 @@ void ObjectStorageBackend::publish(const BlobPublishRequest & request, Transport ErrorCodes::NOT_IMPLEMENTED, "ObjectStorageBackend::publish: verbatim staged publication requires Native mode"); - WriteSettings write_settings; + WriteSettings write_settings = blobPublicationWriteSettings(); write_settings.object_storage_copy_mode = ObjectStorageCopyMode::NativeOnly; if (!object_storage->supportsCopyMode(write_settings.object_storage_copy_mode)) throw Exception( diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h index aef51c04816f7..f9fa31e55fbb9 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasObjectStorageBackend.h @@ -205,6 +205,8 @@ class ObjectStorageBackend final : public Backend /// physical-attempt count (see `TransportAccess::attemptNo`), carried into /// `object_storage_attempt_number` so the HTTP client sees a reissue as attempt >= 2. WriteSettings conditionalWriteSettings(size_t attempt_no) const; + /// Settings for an unconditional blob publication: each object-creating request goes out at most once. + WriteSettings blobPublicationWriteSettings() const; WriteSettings conditionalWriteSettingsForTest() const { return conditionalWriteSettings(/*attempt_no=*/1); } /// Override the emulated backend's wall clock for deterministic expiry tests. void setEmuNowNsForTest(uint64_t now_ns); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp index 0d3f0a334020c..4feb497cac86a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPartWriteTxn.cpp @@ -9,18 +9,23 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include +#include namespace ProfileEvents { extern const Event CASBlobBodyPutAvoided; + extern const Event CASBlobPublishAttemptFailed; + extern const Event CASBlobPublishGaveUp; extern const Event CASBlobAdoptTrusted; extern const Event CASMetaPut; extern const Event CASMetaCompareSwap; @@ -72,6 +77,17 @@ uint64_t nowMs() std::chrono::system_clock::now().time_since_epoch()).count()); } +/// Rate-limited, because under an outage every blob of every write fails here; the counter has them all. +void recordFailedPublicationAttempt( + const String & key, BlobPublicationTransport transport, int attempt, int max_attempts, const String & error) +{ + ProfileEvents::increment(ProfileEvents::CASBlobPublishAttemptFailed); + LogSeriesLimiter log(getLogger("CasBlobPublishAttempt"), /*allowed_count=*/5, /*interval_s=*/30); + LOG_WARNING(log, "Publication of CAS blob {} ({}) failed, attempt {} of {}: {}", + key, transport == BlobPublicationTransport::Streaming ? "streaming" : "server-side copy", + attempt + 1, max_attempts, error); +} + bool isDeterministicBlobPublicationFailure(const std::exception & error) { /// A refused write, EXCEPT the class a fresh credential fixes: the engine refreshes once before it @@ -390,6 +406,14 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) }; constexpr int max_publication_attempts = 8; + /// Every exit by exception after a failed attempt gave up on the blob, the attempt cap and the + /// loop's deadline alike. + int failed_attempts = 0; + const int exceptions_at_entry = std::uncaught_exceptions(); + SCOPE_EXIT({ + if (failed_attempts > 0 && std::uncaught_exceptions() > exceptions_at_entry) + ProfileEvents::increment(ProfileEvents::CASBlobPublishGaveUp); + }); for (int attempt = 0; attempt < max_publication_attempts; ++attempt) { requireAlive(); @@ -491,6 +515,8 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) /// every further request of this operation refuses the same way. if (isDeterministicBlobPublicationFailure(error) || !op.admitted()) throw; + ++failed_attempts; + recordFailedPublicationAttempt(key, transport, attempt, max_publication_attempts, error.what()); if (attempt + 1 == max_publication_attempts) throwCasWriteRetryLater(fmt::format( "PartWriteTxn::ensureBlobPresent: publication of {} remained ambiguous after {} " @@ -502,6 +528,9 @@ BlobUploadResult PartWriteTxn::ensureBlobPresent(const BlobUploadRequest & req) } catch (...) { + ++failed_attempts; + recordFailedPublicationAttempt( + key, transport, attempt, max_publication_attempts, getCurrentExceptionMessage(/*with_stacktrace=*/false)); if (attempt + 1 == max_publication_attempts) throwCasWriteRetryLater(fmt::format( "PartWriteTxn::ensureBlobPresent: publication of {} remained ambiguous after {} " diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp index c42d10b8b056f..86da45ceb8612 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp +++ b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp @@ -105,6 +105,27 @@ namespace ErrorCodes namespace { +/// `WriteBufferFromS3` and `copyS3File` reissue the identical request on `NO_SUCH_KEY` in their own +/// loops above the client, so neither a client retry profile nor single-attempt creation bounds them. +void applyWriteErrorRetryOverride(S3::S3RequestSettings & request_settings, const WriteSettings & write_settings) +{ + if (write_settings.object_storage_single_attempt_object_creation) + request_settings[S3RequestSetting::max_unexpected_write_error_retries] = 1; + else if (write_settings.s3_max_unexpected_write_error_retries_override) + request_settings[S3RequestSetting::max_unexpected_write_error_retries] + = write_settings.s3_max_unexpected_write_error_retries_override; +} + +void requireNativeOnlyForSingleAttemptCreation(const WriteSettings & write_settings) +{ + if (write_settings.object_storage_single_attempt_object_creation + && write_settings.object_storage_copy_mode != ObjectStorageCopyMode::NativeOnly) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "A copy with single-attempt object creation requires NativeOnly copy mode: the read-write " + "fallback would write through the retrying client"); +} + template void throwIfError(const Aws::Utils::Outcome & response) { @@ -438,15 +459,7 @@ std::unique_ptr S3ObjectStorage::writeObject( /// NOLIN request_settings[S3RequestSetting::min_upload_part_size] = cap; } - if (write_settings.s3_max_unexpected_write_error_retries_override) - { - /// WriteBufferFromS3's OWN retry loop (makeSinglepartUpload/completeMultipartUpload) reissues - /// the identical request — WITH its If-None-Match/If-Match condition — on a NO_SUCH_KEY - /// response; this sits ABOVE the S3 client, so a client-level profile override does not bound - /// it. See WriteSettings. - request_settings[S3RequestSetting::max_unexpected_write_error_retries] - = write_settings.s3_max_unexpected_write_error_retries_override; - } + applyWriteErrorRetryOverride(request_settings, write_settings); ThreadPoolCallbackRunnerUnsafe scheduler; if (write_settings.s3_allow_parallel_part_upload) @@ -473,7 +486,8 @@ std::unique_ptr S3ObjectStorage::writeObject( /// NOLIN std::move(blob_storage_log), attributes, std::move(scheduler), - disk_write_settings); + disk_write_settings, + objectCreationClient(write_settings)); } @@ -997,12 +1011,17 @@ void S3ObjectStorage::copyObjectToAnotherObjectStorage( // NOLINT IObjectStorage & object_storage_to, std::optional object_to_attributes) { + /// A non-S3 destination is rejected here under the flag: `NativeOnly` throws `NOT_IMPLEMENTED` below. + requireNativeOnlyForSingleAttemptCreation(write_settings); + /// Shortcut for S3 if (auto * dest_s3 = dynamic_cast(&object_storage_to); dest_s3 != nullptr) { auto current_client = dest_s3->client.get(); auto settings_ptr = s3_settings.get(); auto size = S3::getObjectSize(*client.get(), uri.bucket, object_from.remote_path, {}); + S3::S3RequestSettings request_settings = settings_ptr->request_settings; + applyWriteErrorRetryOverride(request_settings, write_settings); auto scheduler = threadPoolCallbackRunnerUnsafe(getThreadPoolWriter(), ThreadName::S3_COPY_POOL); const auto read_settings_to_use = patchSettings(read_settings); @@ -1017,13 +1036,14 @@ void S3ObjectStorage::copyObjectToAnotherObjectStorage( // NOLINT /*dest_s3_client=*/current_client, /*dest_bucket=*/dest_s3->uri.bucket, /*dest_key=*/object_to.remote_path, - settings_ptr->request_settings, + request_settings, read_settings_to_use, BlobStorageLogWriter::create(disk_name), scheduler, [&, this]{ return readObject(object_from, read_settings_to_use);}, object_to_attributes, - write_settings.object_storage_copy_mode); + write_settings.object_storage_copy_mode, + dest_s3->objectCreationClient(write_settings)); return; } catch (S3Exception & exc) @@ -1076,9 +1096,14 @@ void S3ObjectStorage::copyObject( // NOLINT "(allow_native_copy=false) for object storage {}", getName()); + requireNativeOnlyForSingleAttemptCreation(write_settings); + auto current_client = client.get(); auto settings_ptr = s3_settings.get(); auto size = S3::getObjectSize(*current_client, uri.bucket, object_from.remote_path, {}); + + S3::S3RequestSettings request_settings = settings_ptr->request_settings; + applyWriteErrorRetryOverride(request_settings, write_settings); auto scheduler = threadPoolCallbackRunnerUnsafe(getThreadPoolWriter(), ThreadName::S3_COPY_POOL); const auto read_settings_to_use = patchSettings(read_settings); @@ -1091,13 +1116,14 @@ void S3ObjectStorage::copyObject( // NOLINT /*dest_s3_client=*/current_client, /*dest_bucket=*/uri.bucket, /*dest_key=*/object_to.remote_path, - settings_ptr->request_settings, + request_settings, read_settings_to_use, BlobStorageLogWriter::create(disk_name), scheduler, [&, this]{ return readObject(object_from, read_settings_to_use);}, object_to_attributes, - write_settings.object_storage_copy_mode); + write_settings.object_storage_copy_mode, + objectCreationClient(write_settings)); } void S3ObjectStorage::shutdown() @@ -1235,7 +1261,8 @@ std::shared_ptr S3ObjectStorage::tryGetS3StorageClient() return client.get(); } -std::shared_ptr S3ObjectStorage::getSingleAttemptClient(uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms) const +std::shared_ptr S3ObjectStorage::getSingleAttemptClient( + uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms, bool disable_adaptive_timeouts) const { auto base = client.get(); std::lock_guard lock(single_attempt_client_mutex); @@ -1245,7 +1272,7 @@ std::shared_ptr S3ObjectStorage::getSingleAttemptClient(uint64 single_attempt_client_base = base; } - const auto cache_key = std::make_pair(request_timeout_ms, connect_timeout_cap_ms); + const auto cache_key = std::make_tuple(request_timeout_ms, connect_timeout_cap_ms, disable_adaptive_timeouts); if (auto it = single_attempt_clients.find(cache_key); it != single_attempt_clients.end()) return it->second; @@ -1262,6 +1289,8 @@ std::shared_ptr S3ObjectStorage::getSingleAttemptClient(uint64 if (request_timeout_ms != 0) cfg.requestTimeoutMs = static_cast(request_timeout_ms); + if (disable_adaptive_timeouts) + cfg.s3_use_adaptive_timeouts = false; /// One TCP/TLS connect may not cost more than the cap the mount froze at open: the engine reserves /// attempt + 2 × cap per envelope, and a reloaded base client with a wider connect timeout must not @@ -1284,18 +1313,30 @@ std::shared_ptr S3ObjectStorage::getSingleAttemptClient(uint64 bool S3ObjectStorage::hasSingleAttemptClientForTest(uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms) const { std::lock_guard lock(single_attempt_client_mutex); - return single_attempt_clients.contains(std::make_pair(request_timeout_ms, connect_timeout_cap_ms)); + return single_attempt_clients.contains(std::make_tuple(request_timeout_ms, connect_timeout_cap_ms, false)); } std::shared_ptr S3ObjectStorage::clientForRetryProfile(const ObjectStorageControlRequest & request) const { /// getSingleAttemptClient is only invoked when actually selected, so an ordinary request never /// pays for building or locking the clone. + /// The clone keeps the disk's adaptive timeouts: the caller reissues these requests itself and numbers + /// its reissues, and the HTTP client gives attempt 2 onwards the configured timeouts. if (request.profile == ObjectStorageRetryProfile::SingleAttempt) return getSingleAttemptClient(request.attempt_timeout_ms, request.connect_timeout_cap_ms); return client.get(); } +std::shared_ptr S3ObjectStorage::objectCreationClient(const WriteSettings & write_settings) const +{ + if (!write_settings.object_storage_single_attempt_object_creation) + return nullptr; + /// The creating request has no second attempt, so it must not run under the first-attempt timeouts. + return getSingleAttemptClient( + write_settings.object_storage_attempt_timeout_ms, write_settings.object_storage_connect_timeout_cap_ms, + /*disable_adaptive_timeouts=*/true); +} + bool S3ObjectStorage::tryRefreshCredentialsViaCallback() { fiu_do_on(FailPoints::object_storage_force_refresh_callback_success, { return true; }); diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h index 27660043af7c7..c7bb9c459f269 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h +++ b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.h @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -212,13 +213,15 @@ class S3ObjectStorage : public IObjectStorage /// disk client rotates (applyNewSettings/credentials refresh) — the cached clone is keyed by the /// base client's identity, so a stale clone can never outlive a rotation. /// `request_timeout_ms` overrides the clone's send/receive inactivity bound; 0 keeps the disk's. - /// `connect_timeout_cap_ms` caps the clone's connect timeout (0 = no cap); the cache key is the - /// pair, so two callers asking for the same request timeout but different caps get distinct clones. - std::shared_ptr getSingleAttemptClient(uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms = 0) const; - - /// True iff a clone for exactly this (request timeout, connect cap) pair is already cached -- - /// never builds one. Lets a test prove dispatch used a SPECIFIC key (and no other) without ever - /// creating a clone itself and without measuring anything. + /// `connect_timeout_cap_ms` caps the clone's connect timeout (0 = no cap). `disable_adaptive_timeouts` + /// turns off the adaptive first-attempt timeouts, which assume a second attempt with the configured + /// ones. The cache key is all three, so callers asking for different bounds get distinct clones. + std::shared_ptr getSingleAttemptClient( + uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms = 0, bool disable_adaptive_timeouts = false) const; + + /// True iff a clone for exactly this (request timeout, connect cap) pair, with the disk's adaptive + /// timeouts setting, is already cached -- never builds one. Lets a test prove dispatch used a SPECIFIC + /// key (and no other) without ever creating a clone itself and without measuring anything. bool hasSingleAttemptClientForTest(uint64_t request_timeout_ms, uint64_t connect_timeout_cap_ms) const; private: @@ -243,6 +246,9 @@ class S3ObjectStorage : public IObjectStorage std::shared_ptr clientForRetryProfile(const ObjectStorageControlRequest & request) const; + /// The single-attempt client for the object-creating requests of a write that asks for it, else null. + std::shared_ptr objectCreationClient(const WriteSettings & write_settings) const; + /// Runs `fn` and, if it failed because the vended credentials expired, refreshes this disk's /// client and runs it once more. `fn` must re-read the client itself, so the second run signs /// with the refreshed one. @@ -279,10 +285,10 @@ class S3ObjectStorage : public IObjectStorage std::atomic pinned_generation_dialect{-1}; /// -1 unpinned, 0 pinned ETag, 1 pinned generation mutable std::mutex single_attempt_client_mutex; - /// One clone per (requested timeout, connect cap) pair: the verbs of one operation ask for - /// different bounds, and a single slot would rebuild a whole S3 client (and lose its connection - /// pool) on every alternation between them. - mutable std::map, std::shared_ptr> single_attempt_clients; + /// One clone per (requested timeout, connect cap, adaptive timeouts disabled) key: the verbs of one + /// operation ask for different bounds, and a single slot would rebuild a whole S3 client (and lose + /// its connection pool) on every alternation between them. + mutable std::map, std::shared_ptr> single_attempt_clients; /// The base client the cached clones above were built from. Deliberately held as a shared_ptr (not /// a raw pointer): a raw pointer would be compared for identity AFTER the object it once pointed /// to could have been freed and a new client reallocated at the same address by an unrelated diff --git a/src/Disks/tests/S3ObjectCreationTestStorage.h b/src/Disks/tests/S3ObjectCreationTestStorage.h new file mode 100644 index 0000000000000..ec9fd8b1ea6d1 --- /dev/null +++ b/src/Disks/tests/S3ObjectCreationTestStorage.h @@ -0,0 +1,109 @@ +#pragma once + +#include "config.h" + +#if USE_AWS_S3 + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace DB::S3RequestSetting +{ +extern const S3RequestSettingsUInt64 max_single_operation_copy_size; +extern const S3RequestSettingsUInt64 max_single_part_upload_size; +extern const S3RequestSettingsUInt64 min_upload_part_size; +} + +/// `S3ObjectStorage` instances for tests against `CountingS3::CountingS3Server`. +namespace S3ObjectCreationTestStorage +{ + +/// A genuine `S3ObjectStorage` whose client retries a retryable failure twice, as a disk's client does. +/// `adaptive_timeouts` gives each request's first attempt the short adaptive timeouts, as on a disk +/// with the default `use_adaptive_timeouts`. +inline std::shared_ptr makeRetryingStorage( + const std::string & endpoint, const std::function & adjust_settings = {}, + bool adaptive_timeouts = false, long request_timeout_ms = 10000) +{ + /// The client configuration keeps a reference to the filter. + static const DB::RemoteHostFilter remote_host_filter; + DB::S3::PocoHTTPClientConfiguration cfg = DB::S3::ClientFactory::instance().createClientConfiguration( + "us-east-1", + remote_host_filter, + /* s3_max_redirects = */ 100, + DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = 2, .initial_delay_ms = 1, .max_delay_ms = 1}, + /* s3_slow_all_threads_after_network_error = */ false, + /* s3_slow_all_threads_after_retryable_error = */ false, + /* enable_s3_requests_logging = */ false, + /* for_disk_s3 = */ true, + /* opt_disk_name = */ {}, + /* request_throttler = */ {}); + cfg.endpointOverride = endpoint; + cfg.connectTimeoutMs = 10000; + cfg.requestTimeoutMs = request_timeout_ms; + cfg.s3_use_adaptive_timeouts = adaptive_timeouts; + /// A fresh connection per request: a pooled one can outlive this test's ephemeral-port server. + cfg.http_keep_alive_timeout = 0; + auto client = 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{}); + + auto settings = std::make_unique(); + if (adjust_settings) + adjust_settings(*settings); + return std::make_shared( + std::move(client), std::move(settings), DB::S3::URI(endpoint + "/test-bucket/"), DB::S3Capabilities{}, + DB::ObjectStorageKeyGeneratorPtr{}, "disk"); +} + +/// The `DB::Exception` code `fn` throws: 0 when it returns, -1 for any other exception. +inline int thrownCode(const std::function & fn) +{ + try + { + fn(); + } + catch (const DB::Exception & e) + { + return e.code(); + } + catch (...) + { + return -1; + } + return 0; +} + +/// A body above the single-part ceiling that fits one part, so a write is exactly one `UploadPart`. +inline void forceOnePartMultipart(DB::S3Settings & settings) +{ + settings.request_settings[DB::S3RequestSetting::max_single_part_upload_size] = 16; + settings.request_settings[DB::S3RequestSetting::min_upload_part_size] = 1024 * 1024; +} + +/// A staged object above the single-operation copy ceiling that fits one part, so a copy is exactly one +/// `UploadPartCopy`. +inline void forceOnePartMultipartCopy(DB::S3Settings & settings) +{ + settings.request_settings[DB::S3RequestSetting::max_single_operation_copy_size] = 16; + settings.request_settings[DB::S3RequestSetting::min_upload_part_size] = 1024 * 1024; +} + +} + +#endif diff --git a/src/Disks/tests/gtest_cas_backend.cpp b/src/Disks/tests/gtest_cas_backend.cpp index 04052060d7ebe..17f7c0bfe8b3a 100644 --- a/src/Disks/tests/gtest_cas_backend.cpp +++ b/src/Disks/tests/gtest_cas_backend.cpp @@ -946,7 +946,7 @@ String readStorageObject(const DB::ObjectStoragePtr & storage, const String & ke } -TEST(CASObjectStorageBackend, PublishBlobStreamingUsesOrdinaryDefaultWriteTransport) +TEST(CASObjectStorageBackend, PublishBlobStreamingSettingsRequestSingleCreation) { auto storage = makePublicationRecordingStorage(); auto backend = std::make_shared(storage, ObjectStorageBackend::Mode::Native); @@ -964,7 +964,7 @@ TEST(CASObjectStorageBackend, PublishBlobStreamingUsesOrdinaryDefaultWriteTransp ASSERT_TRUE(storage->last_write_settings.has_value()); EXPECT_EQ(storage->last_write_settings->object_storage_request_mode, DB::ObjectStorageRequestMode::Default); EXPECT_EQ(storage->last_write_settings->object_storage_retry_profile, DB::ObjectStorageRetryProfile::Default); - EXPECT_EQ(storage->last_write_settings->s3_max_unexpected_write_error_retries_override, 0u); + EXPECT_TRUE(storage->last_write_settings->object_storage_single_attempt_object_creation); EXPECT_FALSE(storage->last_write_settings->s3_force_single_part_upload); EXPECT_TRUE(storage->last_write_settings->object_storage_write_if_none_match.empty()); EXPECT_TRUE(storage->last_write_settings->object_storage_write_if_match.empty()); @@ -1155,6 +1155,7 @@ TEST(CASObjectStorageBackend, PublishBlobCopiesStagedBytesWithNativeOnlyDefaultR EXPECT_EQ(storage->last_copy_settings->object_storage_copy_mode, DB::ObjectStorageCopyMode::NativeOnly); EXPECT_EQ(storage->last_copy_settings->object_storage_request_mode, DB::ObjectStorageRequestMode::Default); EXPECT_EQ(storage->last_copy_settings->object_storage_retry_profile, DB::ObjectStorageRetryProfile::Default); + EXPECT_TRUE(storage->last_copy_settings->object_storage_single_attempt_object_creation); EXPECT_TRUE(storage->last_copy_settings->object_storage_write_if_none_match.empty()); EXPECT_TRUE(storage->last_copy_settings->object_storage_write_if_match.empty()); EXPECT_EQ(readStorageObject(storage, destination), "staged-envelopepayload"); diff --git a/src/Disks/tests/gtest_cas_blob_publish_transport.cpp b/src/Disks/tests/gtest_cas_blob_publish_transport.cpp new file mode 100644 index 0000000000000..b09661ce4d6dd --- /dev/null +++ b/src/Disks/tests/gtest_cas_blob_publish_transport.cpp @@ -0,0 +1,256 @@ +#include + +#include "config.h" + +#if USE_AWS_S3 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +/// Blob publication must put every object-creating request (`PutObject`, `CopyObject`, +/// `CompleteMultipartUpload`) on the wire at most once. These tests drive the backend's `publish` +/// against a counting S3 server whose client retries, as a disk's client does. + +namespace ProfileEvents +{ +extern const Event CASBlobPublishAttemptFailed; +} + +namespace DB::ErrorCodes +{ +extern const int S3_ERROR; +} + +using CountingS3::CountingS3Server; +using CountingS3::Verb; +using S3ObjectCreationTestStorage::forceOnePartMultipart; +using S3ObjectCreationTestStorage::makeRetryingStorage; +using S3ObjectCreationTestStorage::thrownCode; + +namespace +{ + +std::shared_ptr nativeBackend(DB::ObjectStoragePtr storage) +{ + return std::make_shared( + std::move(storage), DB::Cas::ObjectStorageBackend::Mode::Native, + /*single_attempt_control_plane_=*/true, /*attempt_timeout_ms_=*/5000, /*connect_timeout_cap_ms_=*/5000); +} + +DB::Cas::BlobPublishRequest streamingPublication() +{ + return DB::Cas::BlobPublishRequest{ + .destination_key = "blob", + .publication = DB::Cas::StreamingBlobPublication{ + .payload_size = 100, + .fresh_envelope = "envelope", + .open_payload = [] + { + return std::make_unique(std::string(100, 'p')); + }}}; +} + +/// One engine attempt of the backend's blob publication, so every counted request comes from below it. +void publishOnce(const DB::ObjectStoragePtr & storage, const DB::Cas::BlobPublishRequest & request) +{ + DB::Cas::CasRequests requests(DB::Cas::BackendPtr(nativeBackend(storage)), DB::Cas::Fence::open()); + auto op = requests.admit(); + op.publish(request, DB::Cas::Retry::once()); +} + +} + +/// A lost `CompleteMultipartUpload` is not resent by the transport. +TEST(CASBlobPublishTransport, LostCompleteIsOneWireRequest) +{ + (void)getContext(); + CountingS3Server server(CountingS3::dropping(Verb::CompleteMultipartUpload)); + auto storage = makeRetryingStorage(server.getUrl(), forceOnePartMultipart); + + EXPECT_EQ(thrownCode([&] { publishOnce(storage, streamingPublication()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); +} + +namespace +{ + +/// The control plane in memory, the blob bodies on a real `S3ObjectStorage`: the writer's blob `HEAD` and +/// its publication go through the S3 transport, so the publication loop sees the transport's failures. +class S3BlobBodiesBackend final : public DB::Cas::InMemoryBackend +{ +public: + using InMemoryBackend::head; + + explicit S3BlobBodiesBackend(std::shared_ptr blob_bodies_) : blob_bodies(std::move(blob_bodies_)) {} + + std::optional head(const String & key, DB::Cas::TransportAccess & access) override + { + if (key.contains("/blobs/") && !key.ends_with(".meta")) + return blob_bodies->head(key, access); + return InMemoryBackend::head(key, access); + } + + void publish(const DB::Cas::BlobPublishRequest & request, DB::Cas::TransportAccess & access) override + { + blob_bodies->publish(request, access); + } + +private: + std::shared_ptr blob_bodies; +}; + +DB::Cas::PoolPtr openPool(const std::string & endpoint) +{ + auto backend = std::make_shared(nativeBackend(makeRetryingStorage(endpoint))); + auto pool = DB::Cas::Pool::open(backend, DB::Cas::PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); + /// The publication loop paces its attempts on the engine's clock; a virtual one skips the backoff. + auto now = std::make_shared>(0); + pool->mountRequests().setNowFnForTest([now] { return now->load(); }); + pool->mountRequests().setSleepFnForTest([now](uint64_t ms) { now->fetch_add(ms + 1); }); + return pool; +} + +/// Stages a one-blob part for `payload` and returns its build, ready for `putBlob`. +DB::Cas::PartWriteTxnPtr startBuild(const DB::Cas::PoolPtr & pool, const std::string & payload) +{ + const DB::Cas::RootNamespace ns{"srv/tbl"}; + DB::Cas::PartWriteInfo info; + info.intended_ref = ns.string() + "/part"; + auto build = pool->beginPartWrite(info); + + DB::Cas::ManifestEntry entry; + entry.path = "a.bin"; + entry.placement = DB::Cas::EntryPlacement::Blob; + entry.ref = DB::Cas::tests::idOf(payload); + entry.blob_size = payload.size(); + build->precommitAdd(ns, "part", build->stageManifest({entry})); + return build; +} + +std::string s3Path(const std::string & key) +{ + return "/test-bucket/" + key; +} + +/// The requests of `verb` to `path`, in arrival order. +std::vector requestsTo(const CountingS3::ObjectStore & store, Verb verb, const std::string & path) +{ + std::vector matching; + for (auto & request : store.journal()) + if (request.verb == verb && request.path == path) + matching.push_back(std::move(request)); + return matching; +} + +/// Whether a `HEAD` of `path` arrived between the first and the second creating request to it. +bool observedBetweenCreations(const CountingS3::ObjectStore & store, const std::string & path) +{ + int creations = 0; + bool observed = false; + for (const auto & request : store.journal()) + { + if (request.path != path) + continue; + if (request.verb == Verb::PutObject || request.verb == Verb::CopyObject) + ++creations; + else if (request.verb == Verb::HeadObject && creations == 1) + observed = true; + } + return observed; +} + +/// The first request of `verb` reaches the store but is not applied, and its response is lost. +CountingS3::ObjectStore::FaultFn dropFirst(Verb verb) +{ + return [verb](const CountingS3::Request & request, Poco::Net::HTTPServerResponse &) + { + if (request.verb == verb && request.nth == 1) + throw CountingS3::DropResponse(); + return false; + }; +} + +uint64_t failedPublicationAttempts() +{ + return ProfileEvents::global_counters[ProfileEvents::CASBlobPublishAttemptFailed].load(); +} + +} + +/// A lost `PUT` is not resent by the transport: the writer observes the key again and publishes a new +/// envelope, so a late landing of the first one can never carry the second one's `incarnation_tag`. +TEST(CASBlobPublishTransport, LostPutIsRepublishedWithAFreshEnvelope) +{ + (void)getContext(); + CountingS3::ObjectStore store(dropFirst(Verb::PutObject)); + CountingS3Server server(store.responder()); + auto pool = openPool(server.getUrl()); + const std::string payload = "composition-payload"; + auto build = startBuild(pool, payload); + const std::string path = s3Path(pool->layout().blobKey(DB::Cas::tests::idOf(payload))); + const uint64_t failed_before = failedPublicationAttempts(); + + DB::Cas::BlobSource source; + source.size = payload.size(); + source.open = [payload] { return std::make_unique(payload); }; + EXPECT_EQ(build->putBlob(DB::Cas::tests::idOf(payload), std::move(source)).size, payload.size()); + + const auto puts = requestsTo(store, Verb::PutObject, path); + ASSERT_EQ(puts.size(), 2u); + EXPECT_TRUE(observedBetweenCreations(store, path)) << "the second publication must follow a new observation"; + const uint64_t object_size = pool->poolMeta().blob_header_len + payload.size(); + const auto first = DB::Cas::decodeEnvelopeHeader(puts[0].body, object_size, DB::Cas::ObjectKind::Blob); + const auto second = DB::Cas::decodeEnvelopeHeader(puts[1].body, object_size, DB::Cas::ObjectKind::Blob); + EXPECT_TRUE(first.incarnation_tag != second.incarnation_tag) << "the second publication resent the first envelope"; + EXPECT_EQ(store.get(path), puts[1].body); + EXPECT_EQ(failedPublicationAttempts() - failed_before, 1u); +} + +/// A lost staged `CopyObject` is not resent: the staged copy is spent, so the retry streams. +TEST(CASBlobPublishTransport, LostStagedCopyIsRepublishedByStreaming) +{ + (void)getContext(); + CountingS3::ObjectStore store(dropFirst(Verb::CopyObject)); + CountingS3Server server(store.responder()); + auto pool = openPool(server.getUrl()); + const std::string payload = "staged-composition-payload"; + auto build = startBuild(pool, payload); + const std::string path = s3Path(pool->layout().blobKey(DB::Cas::tests::idOf(payload))); + const std::string staging_key = "p/staging/test/blob"; + const std::string staged_bytes = std::string(pool->poolMeta().blob_header_len, 'h') + payload; + store.put(s3Path(staging_key), staged_bytes); + + DB::Cas::BlobSource source; + source.size = payload.size(); + source.server_side_copy_from = staging_key; + source.open = [payload] { return std::make_unique(payload); }; + EXPECT_EQ(build->putBlob(DB::Cas::tests::idOf(payload), std::move(source)).size, payload.size()); + + EXPECT_EQ(requestsTo(store, Verb::CopyObject, path).size(), 1u); + EXPECT_EQ(requestsTo(store, Verb::PutObject, path).size(), 1u); + EXPECT_TRUE(observedBetweenCreations(store, path)) << "the streamed publication must follow a new observation"; + const auto published = store.get(path); + ASSERT_TRUE(published.has_value()); + EXPECT_NE(*published, staged_bytes); + EXPECT_EQ(published->substr(pool->poolMeta().blob_header_len), payload); +} + +#endif diff --git a/src/Disks/tests/gtest_cas_s3_single_attempt_client.cpp b/src/Disks/tests/gtest_cas_s3_single_attempt_client.cpp index cf43b7ee9c101..ee3aa2670f611 100644 --- a/src/Disks/tests/gtest_cas_s3_single_attempt_client.cpp +++ b/src/Disks/tests/gtest_cas_s3_single_attempt_client.cpp @@ -53,6 +53,8 @@ #include #include #include +#include +#include /// The single-attempt client clone must cap its connect timeout at the value the mount froze at open, /// never at the disk's (possibly wider, possibly reloaded, possibly unbounded) own connect timeout. @@ -60,6 +62,7 @@ namespace ProfileEvents { extern const Event S3PutObject; +extern const Event CASConditionalWriteUnresolved; } namespace @@ -867,4 +870,33 @@ TEST(CASBackend, ConditionalPutBufferStartsAtTheBodySize) EXPECT_EQ(storage->opened_buffer_sizes, (std::vector{body.size(), 1})); } +/// `NoSuchUpload` from a conditional multipart completion does not prove that this upload created the +/// object at the key, so the write stays unresolved and the engine settles it by reading the key, which +/// here holds another writer's bytes. +TEST(CASBackend, ConditionalMultipartNoSuchUploadIsUnresolved) +{ + (void)contextForTest(); + CountingS3::ObjectStore store([](const CountingS3::Request & request, Poco::Net::HTTPServerResponse & response) + { + if (request.verb != CountingS3::Verb::CompleteMultipartUpload) + return false; + CountingS3::sendError(response, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchUpload"); + return true; + }); + store.put("/test-bucket/key", "another writer's bytes"); + CountingS3::CountingS3Server server(store.responder()); + auto backend = conditionalBackendForTest( + S3ObjectCreationTestStorage::makeRetryingStorage(server.getUrl(), S3ObjectCreationTestStorage::forceOnePartMultipart)); + DB::Cas::CasRequests requests(DB::Cas::BackendPtr(backend), DB::Cas::Fence::open()); + const uint64_t unresolved_before = ProfileEvents::global_counters[ProfileEvents::CASConditionalWriteUnresolved].load(); + + auto op = requests.admit(); + const auto result = op.create("key", std::string(100, 'c'), DB::Cas::Retry::once()); + + EXPECT_TRUE(std::holds_alternative(result)) + << "variant " << result.index() << "; another writer's object must not be taken for this upload"; + EXPECT_EQ(server.count(CountingS3::Verb::CompleteMultipartUpload), 1u); + EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASConditionalWriteUnresolved].load() - unresolved_before, 1u); +} + #endif diff --git a/src/Disks/tests/gtest_s3_object_creation_requests.cpp b/src/Disks/tests/gtest_s3_object_creation_requests.cpp new file mode 100644 index 0000000000000..4aeff38eb3db9 --- /dev/null +++ b/src/Disks/tests/gtest_s3_object_creation_requests.cpp @@ -0,0 +1,455 @@ +#include + +#include "config.h" + +#if USE_AWS_S3 + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +/// Per-verb wire counts of the requests `S3ObjectStorage` sends to create an object. + +namespace DB::ErrorCodes +{ +extern const int BAD_ARGUMENTS; +extern const int S3_ERROR; +} + +using CountingS3::CountingS3Server; +using CountingS3::Verb; +using S3ObjectCreationTestStorage::forceOnePartMultipart; +using S3ObjectCreationTestStorage::forceOnePartMultipartCopy; +using S3ObjectCreationTestStorage::makeRetryingStorage; +using S3ObjectCreationTestStorage::thrownCode; + +namespace +{ + +void copyObject(DB::S3ObjectStorage & storage, const DB::WriteSettings & write_settings) +{ + storage.copyObject(DB::StoredObject("stage"), DB::StoredObject("copy"), DB::ReadSettings{}, write_settings); +} + +void writeObject(DB::S3ObjectStorage & storage, const DB::WriteSettings & write_settings) +{ + auto buffer = storage.writeObject( + DB::StoredObject("object"), DB::WriteMode::Rewrite, {}, DB::DBMS_DEFAULT_BUFFER_SIZE, write_settings); + buffer->write(std::string(100, 'd').data(), 100); + buffer->finalize(); +} + +#ifdef DEBUG_OR_SANITIZER_BUILD +/// A copy that asks for a creation client but allows the read-write fallback; sends no request. +void copyWithCreationClientAllowingFallback() +{ + auto storage = makeRetryingStorage("http://127.0.0.1:1"); + auto client = storage->getS3StorageClient(); + DB::copyS3File( + client, "test-bucket", "stage", 0, CountingS3::object_size, client, "test-bucket", "copy", + DB::S3Settings{}.request_settings, DB::ReadSettings{}, /*blob_storage_log=*/nullptr, /*schedule=*/{}, + /*fallback_file_reader=*/{}, /*object_metadata=*/std::nullopt, DB::ObjectStorageCopyMode::Default, + /*object_creation_client=*/client); +} +#endif + +/// Captures what any logger writes at `priority` and above while in scope. +class ScopedLogCapture +{ +public: + explicit ScopedLogCapture(int priority = Poco::Message::PRIO_ERROR) + : channel(new Poco::StreamChannel(stream)) + , root_channel(Poco::Logger::root().getChannel(), /*shared=*/true) + , root_level(Poco::Logger::root().getLevel()) + { + std::vector names; + Poco::Logger::names(names); + for (const auto & name : names) + if (auto * logger = Poco::Logger::has(name)) + saved.push_back({name, Poco::AutoPtr(logger->getChannel(), /*shared=*/true), logger->getLevel()}); + Poco::Logger::setChannel(Poco::Logger::ROOT, channel.get()); + Poco::Logger::setLevel(Poco::Logger::ROOT, priority); + } + + ~ScopedLogCapture() + { + /// A logger created in scope took the capture's channel and gets the root's. + Poco::Logger::setChannel(Poco::Logger::ROOT, root_channel.get()); + Poco::Logger::setLevel(Poco::Logger::ROOT, root_level); + for (auto & [name, old_channel, old_level] : saved) + { + if (auto * logger = Poco::Logger::has(name)) + { + logger->setChannel(old_channel.get()); + logger->setLevel(old_level); + } + } + } + + std::string captured() const { return stream.str(); } + +private: + struct SavedLogger + { + std::string name; + Poco::AutoPtr channel; + int level; + }; + + std::ostringstream stream; // STYLE_CHECK_ALLOW_STD_STRING_STREAM + Poco::AutoPtr channel; + Poco::AutoPtr root_channel; + int root_level; + std::vector saved; +}; + +DB::WriteSettings singleAttemptCreation() +{ + DB::WriteSettings write_settings; + write_settings.object_storage_single_attempt_object_creation = true; + return write_settings; +} + +DB::WriteSettings singleAttemptNativeCopy() +{ + DB::WriteSettings write_settings = singleAttemptCreation(); + write_settings.object_storage_copy_mode = DB::ObjectStorageCopyMode::NativeOnly; + return write_settings; +} + +/// One way `S3ObjectStorage` creates an object, and the verb that creates it. +struct CreationPath +{ + std::string name; + Verb creating_verb; + void (*adjust_settings)(DB::S3Settings &); + bool is_copy; + + std::shared_ptr storage(const std::string & endpoint) const + { + return makeRetryingStorage(endpoint, adjust_settings ? adjust_settings : std::function{}); + } + + DB::WriteSettings singleAttemptSettings() const { return is_copy ? singleAttemptNativeCopy() : singleAttemptCreation(); } + + void create(DB::S3ObjectStorage & target, const DB::WriteSettings & write_settings) const + { + if (is_copy) + copyObject(target, write_settings); + else + writeObject(target, write_settings); + } +}; + +const CreationPath creation_paths[] = { + {"Put", Verb::PutObject, nullptr, false}, + {"Copy", Verb::CopyObject, nullptr, true}, + {"Complete", Verb::CompleteMultipartUpload, forceOnePartMultipart, false}, + {"MultipartCopyComplete", Verb::CompleteMultipartUpload, forceOnePartMultipartCopy, true}, +}; + +void PrintTo(const CreationPath & path, std::ostream * os) +{ + *os << path.name; +} + +class S3SingleAttemptObjectCreationPath : public ::testing::TestWithParam +{ +}; + +/// The paths that end in `CompleteMultipartUpload`. +class S3SingleAttemptMultipartCreationPath : public ::testing::TestWithParam +{ +}; + +} + +/// `writeObject` already applies the override; a copy reissues `CopyObject` in the same loop. +TEST(S3ObjectStorageCopyObject, HonoursUnexpectedWriteErrorRetryOverride) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing(Verb::CopyObject, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchKey")); + auto storage = makeRetryingStorage(server.getUrl()); + + EXPECT_EQ(thrownCode([&] { copyObject(*storage, DB::WriteSettings{}); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(Verb::CopyObject), 4u) << "the default `max_unexpected_write_error_retries`"; + + DB::WriteSettings write_settings; + write_settings.s3_max_unexpected_write_error_retries_override = 1; + EXPECT_EQ(thrownCode([&] { copyObject(*storage, write_settings); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(Verb::CopyObject), 5u); +} + +TEST_P(S3SingleAttemptObjectCreationPath, OneWireRequestOnLostResponse) +{ + (void)getContext(); + const CreationPath & path = GetParam(); + CountingS3Server server(CountingS3::dropping(path.creating_verb)); + + EXPECT_EQ(thrownCode([&] { path.create(*path.storage(server.getUrl()), path.singleAttemptSettings()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(path.creating_verb), 1u); +} + +/// A writer that does not ask for single-attempt creation keeps the client's retries. +TEST_P(S3SingleAttemptObjectCreationPath, DefaultWriteStillResendsLostResponse) +{ + (void)getContext(); + const CreationPath & path = GetParam(); + CountingS3Server server(CountingS3::dropping(path.creating_verb, /*only_nth=*/1)); + + EXPECT_EQ(thrownCode([&] { path.create(*path.storage(server.getUrl()), DB::WriteSettings{}); }), 0); + EXPECT_EQ(server.count(path.creating_verb), 2u); +} + +/// The caller of a single-attempt creating request resolves the outcome and retries, so one failed +/// attempt is not an error for the operator. +TEST_P(S3SingleAttemptObjectCreationPath, LostResponseLogsNothingAtError) +{ + (void)getContext(); + const CreationPath & path = GetParam(); + CountingS3Server server(CountingS3::dropping(path.creating_verb)); + auto storage = path.storage(server.getUrl()); + + ScopedLogCapture capture; + EXPECT_EQ(thrownCode([&] { path.create(*storage, path.singleAttemptSettings()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(capture.captured(), ""); + + /// Positive control: an `Error` message does reach the capture, so the emptiness above means something. + Poco::Logger::get("ScopedLogCaptureControl").error("control"); + EXPECT_NE(capture.captured().find("control"), std::string::npos); +} + +/// The flag alone bounds the `NO_SUCH_KEY` reissue loops above the client. +TEST_P(S3SingleAttemptObjectCreationPath, NoSuchKeyIsOneWireRequest) +{ + (void)getContext(); + const CreationPath & path = GetParam(); + CountingS3Server server(CountingS3::failing(path.creating_verb, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchKey")); + + EXPECT_EQ(thrownCode([&] { path.create(*path.storage(server.getUrl()), path.singleAttemptSettings()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(path.creating_verb), 1u); +} + +INSTANTIATE_TEST_SUITE_P( + S3SingleAttemptObjectCreation, + S3SingleAttemptObjectCreationPath, + ::testing::ValuesIn(creation_paths), + [](const ::testing::TestParamInfo & param_info) { return param_info.param.name; }); + +/// After a lost single-attempt `CompleteMultipartUpload` the abort finds no upload: the completion may +/// have applied, and the caller resolves that. +TEST_P(S3SingleAttemptMultipartCreationPath, AbortAfterLostCompleteLogsNothingAtError) +{ + (void)getContext(); + const CreationPath & path = GetParam(); + CountingS3Server server([](const CountingS3::Request & request, Poco::Net::HTTPServerResponse & response) + { + if (request.verb == Verb::CompleteMultipartUpload) + throw CountingS3::DropResponse(); + if (request.verb == Verb::AbortMultipartUpload) + return CountingS3::sendError(response, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchUpload"); + CountingS3::answerOk(request.verb, response); + }); + auto storage = path.storage(server.getUrl()); + + ScopedLogCapture capture; + EXPECT_EQ(thrownCode([&] { path.create(*storage, path.singleAttemptSettings()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(Verb::AbortMultipartUpload), 1u); + EXPECT_EQ(capture.captured(), ""); +} + +INSTANTIATE_TEST_SUITE_P( + S3SingleAttemptObjectCreation, + S3SingleAttemptMultipartCreationPath, + ::testing::Values(creation_paths[2], creation_paths[3]), + [](const ::testing::TestParamInfo & param_info) { return param_info.param.name; }); + +/// A `NO_SUCH_KEY` loop bounded to one request must not announce a retry. +TEST(S3SingleAttemptObjectCreation, LastNoSuchKeyAttemptDoesNotAnnounceARetry) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing(Verb::PutObject, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchKey")); + auto storage = makeRetryingStorage(server.getUrl()); + DB::WriteSettings write_settings; + write_settings.s3_max_unexpected_write_error_retries_override = 1; + + ScopedLogCapture capture(Poco::Message::PRIO_INFORMATION); + EXPECT_EQ(thrownCode([&] { writeObject(*storage, write_settings); }), DB::ErrorCodes::S3_ERROR); + EXPECT_NE(capture.captured().find("NO_SUCH_KEY"), std::string::npos) << capture.captured(); + EXPECT_EQ(capture.captured().find("will retry"), std::string::npos) << capture.captured(); +} + +/// The client resends a request whose response body fails mid-read; the `PUT` was applied by then. +TEST(S3SingleAttemptObjectCreation, PutIsOneWireRequestOnBrokenResponseBody) +{ + (void)getContext(); + CountingS3Server server(CountingS3::faulting(Verb::PutObject, [](Poco::Net::HTTPServerResponse & response) + { + response.set("ETag", "\"put-etag\""); + response.setContentType("application/xml"); + response.setContentLength(4096); + response.setStatus(Poco::Net::HTTPResponse::HTTP_OK); + response.send() << "" << std::flush; + throw CountingS3::DropResponse(); + })); + + EXPECT_EQ(thrownCode([&] { writeObject(*makeRetryingStorage(server.getUrl()), singleAttemptCreation()); }), DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(server.count(Verb::PutObject), 1u); +} + +/// A part creates no object, so a throttled part keeps the client's retries. +TEST(S3SingleAttemptObjectCreation, UploadPartStillRetried) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing( + Verb::UploadPart, Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR, "InternalError", /*only_nth=*/1)); + + EXPECT_EQ(thrownCode([&] { writeObject(*makeRetryingStorage(server.getUrl(), forceOnePartMultipart), singleAttemptCreation()); }), 0); + EXPECT_EQ(server.count(Verb::UploadPart), 2u); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); +} + +TEST(S3SingleAttemptObjectCreation, CreateMultipartUploadStillRetried) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing( + Verb::CreateMultipartUpload, Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR, "InternalError", /*only_nth=*/1)); + + EXPECT_EQ(thrownCode([&] { writeObject(*makeRetryingStorage(server.getUrl(), forceOnePartMultipart), singleAttemptCreation()); }), 0); + EXPECT_EQ(server.count(Verb::CreateMultipartUpload), 2u); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); +} + +TEST(S3SingleAttemptObjectCreation, UploadPartCopyStillRetried) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing( + Verb::UploadPartCopy, Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR, "InternalError", /*only_nth=*/1)); + + EXPECT_EQ( + thrownCode([&] { copyObject(*makeRetryingStorage(server.getUrl(), forceOnePartMultipartCopy), singleAttemptNativeCopy()); }), + 0); + EXPECT_EQ(server.count(Verb::UploadPartCopy), 2u); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); +} + +/// A definite refusal means the `CopyObject` was not applied, so the multipart copy is the only creation. +TEST(S3SingleAttemptObjectCreation, CopyRefusedAsTooLargeBecomesOneMultipartCopy) +{ + (void)getContext(); + CountingS3Server server(CountingS3::failing(Verb::CopyObject, Poco::Net::HTTPResponse::HTTP_BAD_REQUEST, "EntityTooLarge")); + + EXPECT_EQ(thrownCode([&] { copyObject(*makeRetryingStorage(server.getUrl()), singleAttemptNativeCopy()); }), 0); + EXPECT_EQ(server.count(Verb::CopyObject), 1u); + EXPECT_EQ(server.count(Verb::CreateMultipartUpload), 1u); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); +} + +/// The creating request has no second attempt, so its client must not use the short first-attempt timeouts. +TEST(S3SingleAttemptObjectCreation, CreationClientDisablesAdaptiveTimeouts) +{ + auto storage = makeRetryingStorage("http://127.0.0.1:1", {}, /*adaptive_timeouts=*/true); + ASSERT_TRUE(storage->getS3StorageClient()->getClientConfiguration().s3_use_adaptive_timeouts); + + EXPECT_FALSE(storage->getSingleAttemptClient(0, 0, /*disable_adaptive_timeouts=*/true) + ->getClientConfiguration().s3_use_adaptive_timeouts); + EXPECT_TRUE(storage->getSingleAttemptClient(0, 0)->getClientConfiguration().s3_use_adaptive_timeouts) + << "the control-plane clone keeps the disk's adaptive timeouts"; +} + +/// End to end: `writeObject` sends the `PUT` through that client. The delay is above the adaptive +/// first-attempt receive timeout of a `PUT` and below the configured one. +TEST(S3SingleAttemptObjectCreation, PutSlowerThanTheFirstAttemptTimeoutSucceedsOnce) +{ + (void)getContext(); + CountingS3Server server(CountingS3::faulting(Verb::PutObject, [](Poco::Net::HTTPServerResponse & response) + { + std::this_thread::sleep_for(std::chrono::milliseconds(3500)); + CountingS3::answerOk(Verb::PutObject, response); + })); + + EXPECT_EQ( + thrownCode([&] { writeObject(*makeRetryingStorage(server.getUrl(), {}, /*adaptive_timeouts=*/true), singleAttemptCreation()); }), + 0); + EXPECT_EQ(server.count(Verb::PutObject), 1u); +} + +/// The read-write fallback would write through the retrying client, so the storage refuses the request +/// before sending anything. +TEST(S3ObjectStorageCopyObject, SingleAttemptCreationRequiresNativeOnly) +{ + (void)getContext(); + CountingS3Server server([](const CountingS3::Request & request, Poco::Net::HTTPServerResponse & response) + { + CountingS3::answerOk(request.verb, response); + }); + + EXPECT_EQ( + thrownCode([&] { copyObject(*makeRetryingStorage(server.getUrl()), singleAttemptCreation()); }), + DB::ErrorCodes::BAD_ARGUMENTS); + EXPECT_EQ(server.count(Verb::HeadObject), 0u); + EXPECT_EQ(server.count(Verb::CopyObject), 0u); +} + +/// The S3-to-S3 shortcut creates the destination object, so it honours the flag like `copyObject`. +TEST(S3ObjectStorageCopyObject, CopyToAnotherStorageIsOneWireRequestWhenSingleAttempt) +{ + (void)getContext(); + const auto copy_to_another = [](DB::S3ObjectStorage & from, DB::S3ObjectStorage & to, const DB::WriteSettings & write_settings) + { + from.copyObjectToAnotherObjectStorage( + DB::StoredObject("stage"), DB::StoredObject("copy"), DB::ReadSettings{}, write_settings, to); + }; + + CountingS3Server dropping_server(CountingS3::dropping(Verb::CopyObject)); + auto dropping_storage = makeRetryingStorage(dropping_server.getUrl()); + EXPECT_EQ( + thrownCode([&] { copy_to_another(*dropping_storage, *dropping_storage, singleAttemptNativeCopy()); }), + DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(dropping_server.count(Verb::CopyObject), 1u); + + CountingS3Server no_such_key_server(CountingS3::failing(Verb::CopyObject, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchKey")); + auto no_such_key_storage = makeRetryingStorage(no_such_key_server.getUrl()); + EXPECT_EQ( + thrownCode([&] { copy_to_another(*no_such_key_storage, *no_such_key_storage, singleAttemptNativeCopy()); }), + DB::ErrorCodes::S3_ERROR); + EXPECT_EQ(no_such_key_server.count(Verb::CopyObject), 1u); + + EXPECT_EQ( + thrownCode([&] { copy_to_another(*dropping_storage, *dropping_storage, singleAttemptCreation()); }), + DB::ErrorCodes::BAD_ARGUMENTS); +} + +/// `copyS3File` itself only asserts it; a failed assertion aborts debug and sanitizer builds. +#ifdef DEBUG_OR_SANITIZER_BUILD +TEST(S3SingleAttemptObjectCreationDeathTest, CreationClientRequiresNativeOnlyCopy) +{ + (void)getContext(); + EXPECT_DEATH( + { + copyWithCreationClientAllowingFallback(); + std::_Exit(0); + }, + "NativeOnly"); +} +#endif + +#endif diff --git a/src/IO/S3/Client.cpp b/src/IO/S3/Client.cpp index c5280421de19d..dbb11f664a0ce 100644 --- a/src/IO/S3/Client.cpp +++ b/src/IO/S3/Client.cpp @@ -580,8 +580,13 @@ Model::CompleteMultipartUploadOutcome Client::CompleteMultipartUpload(CompleteMu const auto & key = request.GetKey(); const auto & bucket = request.GetBucket(); + /// A single-attempt client sent this request once, so `NoSuchUpload` cannot mean that its own earlier + /// attempt completed; an object at the key may come from another writer. + const bool single_attempt = usesSingleAttemptRetryStrategy(); + if (!outcome.IsSuccess() - && outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_UPLOAD) + && outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_UPLOAD + && !single_attempt) { auto check_request = HeadObjectRequest() .WithBucket(bucket) @@ -594,7 +599,8 @@ Model::CompleteMultipartUploadOutcome Client::CompleteMultipartUpload(CompleteMu outcome = Aws::S3::Model::CompleteMultipartUploadOutcome(Aws::S3::Model::CompleteMultipartUploadResult()); } - if (outcome.IsSuccess() && provider_type == ProviderType::GCS && client_settings.gcs_issue_compose_request) + /// The compose rewrites the object as a new generation, a second creating request; off by default. + if (outcome.IsSuccess() && provider_type == ProviderType::GCS && client_settings.gcs_issue_compose_request && !single_attempt) { /// For GCS we will try to compose object at the end, otherwise we cannot do a native copy /// for the object (e.g. for backups) diff --git a/src/IO/S3/copyS3File.cpp b/src/IO/S3/copyS3File.cpp index 4f9e8b0a9743f..c66eaca933812 100644 --- a/src/IO/S3/copyS3File.cpp +++ b/src/IO/S3/copyS3File.cpp @@ -16,10 +16,13 @@ #include #include +#include #include #include +#include + namespace ProfileEvents { @@ -82,8 +85,10 @@ namespace const std::optional & object_metadata_, ThreadPoolCallbackRunnerUnsafe schedule_, BlobStorageLogWriterPtr blob_storage_log_, - const LoggerPtr log_) + const LoggerPtr log_, + const std::shared_ptr & object_creation_client_ = nullptr) : client_ptr(client_ptr_) + , object_creation_client(object_creation_client_ ? object_creation_client_ : client_ptr_) , dest_bucket(dest_bucket_) , dest_key(dest_key_) , request_settings(request_settings_) @@ -100,6 +105,8 @@ namespace protected: std::shared_ptr client_ptr; + /// Sends the requests that create the destination object. + std::shared_ptr object_creation_client; const String & dest_bucket; const String & dest_key; const S3::S3RequestSettings & request_settings; @@ -206,7 +213,7 @@ namespace ProfileEvents::increment(ProfileEvents::DiskS3CompleteMultipartUpload); Stopwatch watch; - auto outcome = client_ptr->CompleteMultipartUpload(request); + auto outcome = object_creation_client->CompleteMultipartUpload(request); auto elapsed = watch.elapsedMicroseconds(); if (blob_storage_log) @@ -245,6 +252,11 @@ namespace abort_request.SetKey(dest_key); abort_request.SetUploadId(multipart_upload_id); + /// A lost single-attempt completion may have consumed the upload. + std::optional no_such_upload_is_expected; + if (object_creation_client->usesSingleAttemptRetryStrategy()) + no_such_upload_is_expected.emplace(); + Stopwatch watch; auto outcome = client_ptr->AbortMultipartUpload(abort_request); auto elapsed = watch.elapsedMicroseconds(); @@ -305,7 +317,12 @@ namespace } catch (...) { - tryLogCurrentException(log, fmt::format("While performing multipart upload of object {} in bucket {}", dest_key, dest_bucket)); + /// A caller that sends the completion once resolves a failure and retries itself, failed + /// `CreateMultipartUpload` and `UploadPartCopy` included; CAS logs each failed attempt. + tryLogCurrentException( + log, + fmt::format("While performing multipart upload of object {} in bucket {}", dest_key, dest_bucket), + object_creation_client->usesSingleAttemptRetryStrategy() ? LogsLevel::debug : LogsLevel::error); task_tracker.safeWaitAll(); @@ -616,7 +633,8 @@ namespace ThreadPoolCallbackRunnerUnsafe schedule_, BlobStorageLogWriterPtr blob_storage_log_, std::function fallback_method_, - bool allow_fallback_ = true) + bool allow_fallback_, + const std::shared_ptr & object_creation_client_) : UploadHelper( client_ptr_, dest_bucket_, @@ -625,7 +643,8 @@ namespace object_metadata_, schedule_, blob_storage_log_, - getLogger("copyS3File")) + getLogger("copyS3File"), + object_creation_client_) , src_bucket(src_bucket_) , src_key(src_key_) , offset(src_offset_) @@ -725,7 +744,7 @@ namespace if (client_ptr->isClientForDisk()) ProfileEvents::increment(ProfileEvents::DiskS3CopyObject); - auto outcome = client_ptr->CopyObject(request); + auto outcome = object_creation_client->CopyObject(request); if (outcome.IsSuccess()) { LOG_TRACE( @@ -896,8 +915,12 @@ void copyS3File( ThreadPoolCallbackRunnerUnsafe schedule, const CreateReadBuffer& fallback_file_reader, const std::optional & object_metadata, - ObjectStorageCopyMode copy_mode) + ObjectStorageCopyMode copy_mode, + std::shared_ptr object_creation_client) { + /// The read-write fallback would write through the retrying client. + chassert(!object_creation_client || copy_mode == ObjectStorageCopyMode::NativeOnly); + if (!dest_s3_client) dest_s3_client = src_s3_client; @@ -942,7 +965,8 @@ void copyS3File( schedule, blob_storage_log, std::move(fallback_method), - /*allow_fallback=*/copy_mode == ObjectStorageCopyMode::Default}; + /*allow_fallback=*/copy_mode == ObjectStorageCopyMode::Default, + object_creation_client}; helper.performCopy(); } diff --git a/src/IO/S3/copyS3File.h b/src/IO/S3/copyS3File.h index d4f7283771302..392d186b2e55a 100644 --- a/src/IO/S3/copyS3File.h +++ b/src/IO/S3/copyS3File.h @@ -34,6 +34,9 @@ using CreateReadBuffer = std::function()>; /// /// `copy_mode = NativeOnly` forbids the client-side read-write fallback. If native copy is disabled /// or cannot complete, the failure is propagated instead. +/// +/// `object_creation_client`, when set, sends `CopyObject` and `CompleteMultipartUpload`; it requires +/// `NativeOnly`, because the read-write fallback would write through the retrying client. void copyS3File( std::shared_ptr src_s3_client, const String & src_bucket, @@ -49,7 +52,8 @@ void copyS3File( ThreadPoolCallbackRunnerUnsafe schedule, const CreateReadBuffer& fallback_file_reader, const std::optional & object_metadata = std::nullopt, - ObjectStorageCopyMode copy_mode = ObjectStorageCopyMode::Default); + ObjectStorageCopyMode copy_mode = ObjectStorageCopyMode::Default, + std::shared_ptr object_creation_client = nullptr); /// Copies data from any seekable source to S3. /// The same functionality can be done by using the function copyData() and the class WriteBufferFromS3 diff --git a/src/IO/S3/tests/CountingS3Server.h b/src/IO/S3/tests/CountingS3Server.h new file mode 100644 index 0000000000000..a339b764c4a20 --- /dev/null +++ b/src/IO/S3/tests/CountingS3Server.h @@ -0,0 +1,354 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/// A local HTTP server standing in for S3 that counts requests per verb, so a test can assert how many +/// times a request went on the wire. Path-style requests only. +namespace CountingS3 +{ + +enum class Verb : size_t +{ + PutObject, + CopyObject, + CreateMultipartUpload, + UploadPart, + UploadPartCopy, + CompleteMultipartUpload, + AbortMultipartUpload, + HeadObject, + GetObject, + ComposeObject, + Other, + Count, +}; + +inline Verb classify(const Poco::Net::HTTPServerRequest & request) +{ + const std::string query = Poco::URI(request.getURI()).getRawQuery(); + const bool has_upload_id = query.contains("uploadId="); + const bool has_part_number = query.contains("partNumber="); + const bool is_copy = request.has("x-amz-copy-source"); + const std::string & method = request.getMethod(); + + if (method == Poco::Net::HTTPRequest::HTTP_HEAD) + return Verb::HeadObject; + if (method == Poco::Net::HTTPRequest::HTTP_GET && !query.contains("list-type")) + return Verb::GetObject; + if (method == Poco::Net::HTTPRequest::HTTP_PUT && query.contains("compose")) + return Verb::ComposeObject; + if (method == Poco::Net::HTTPRequest::HTTP_PUT && has_part_number) + return is_copy ? Verb::UploadPartCopy : Verb::UploadPart; + if (method == Poco::Net::HTTPRequest::HTTP_PUT) + return is_copy ? Verb::CopyObject : Verb::PutObject; + if (method == Poco::Net::HTTPRequest::HTTP_POST && has_upload_id) + return Verb::CompleteMultipartUpload; + if (method == Poco::Net::HTTPRequest::HTTP_POST && query.contains("uploads")) + return Verb::CreateMultipartUpload; + if (method == Poco::Net::HTTPRequest::HTTP_DELETE && has_upload_id) + return Verb::AbortMultipartUpload; + return Verb::Other; +} + +/// Thrown by a responder to close the connection without an answer: the client sees a network error. +/// Whether the request was applied depends on the thrower; `ObjectStore` faults fire before the put. +struct DropResponse : std::runtime_error +{ + DropResponse() : std::runtime_error("test server drops the response") {} +}; + +using VerbCounts = std::array, static_cast(Verb::Count)>; + +/// One request as the server received it. +struct Request +{ + Verb verb; + /// The 1-based count of requests of this verb, this one included. + size_t nth; + /// `/bucket/key`. + std::string path; + std::string body; +}; + +using Responder = std::function; + +class CountingHandler : public Poco::Net::HTTPRequestHandler +{ + VerbCounts & counts; + const Responder & respond; + +public: + CountingHandler(VerbCounts & counts_, const Responder & respond_) : counts(counts_), respond(respond_) {} + + void handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response) override + { + const Verb verb = classify(request); + /// Read the whole body first, so a dropped response follows a fully received request. + std::string body; + if (request.hasContentLength()) + Poco::StreamCopier::copyToString64(request.stream(), body); + const size_t nth = ++counts[static_cast(verb)]; + respond( + Request{ + .verb = verb, + .nth = nth, + .path = Poco::URI(request.getURI()).getPath(), + .body = std::move(body)}, + response); + } +}; + +class CountingHandlerFactory : public Poco::Net::HTTPRequestHandlerFactory +{ + VerbCounts & counts; + const Responder & respond; + + Poco::Net::HTTPRequestHandler * createRequestHandler(const Poco::Net::HTTPServerRequest &) override + { + return new CountingHandler(counts, respond); + } + +public: + CountingHandlerFactory(VerbCounts & counts_, const Responder & respond_) : counts(counts_), respond(respond_) {} +}; + +class CountingS3Server +{ + VerbCounts counts{}; + Responder respond; + std::unique_ptr server_socket; + Poco::ThreadPool thread_pool; + std::unique_ptr server; + +public: + explicit CountingS3Server(Responder respond_) + : respond(std::move(respond_)) + , server_socket(std::make_unique(0)) + , thread_pool("CountingS3") + , server(std::make_unique( + Poco::SharedPtr(new CountingHandlerFactory(counts, respond)), + thread_pool, + *server_socket, + Poco::AutoPtr(new Poco::Net::HTTPServerParams()))) + { + server->start(); + } + + ~CountingS3Server() + { + server->stop(); + thread_pool.joinAll(); + } + + std::string getUrl() const { return "http://127.0.0.1:" + std::to_string(server_socket->address().port()); } + size_t count(Verb verb) const { return counts[static_cast(verb)].load(); } +}; + +inline void sendXml(Poco::Net::HTTPServerResponse & response, Poco::Net::HTTPResponse::HTTPStatus status, const std::string & body) +{ + response.setStatus(status); + response.setContentType("application/xml"); + response.setContentLength(static_cast(body.size())); + response.send() << body; +} + +inline void sendError(Poco::Net::HTTPServerResponse & response, Poco::Net::HTTPResponse::HTTPStatus status, + const std::string & code, const std::string & message = "injected") +{ + sendXml(response, status, "" + code + "" + + message + ""); +} + +/// The size every `HEAD` reports. +inline constexpr size_t object_size = 64; + +/// A store on which every request succeeds. +inline void answerOk(Verb verb, Poco::Net::HTTPServerResponse & response) +{ + switch (verb) + { + case Verb::PutObject: + case Verb::UploadPart: + case Verb::ComposeObject: + response.set("ETag", "\"part-or-put-etag\""); + response.setContentLength(0); + response.setStatus(Poco::Net::HTTPResponse::HTTP_OK); + response.send(); + return; + case Verb::CopyObject: + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, + "\"copy-etag\"" + "2026-01-01T00:00:00.000Z"); + return; + case Verb::UploadPartCopy: + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, + "\"copy-part-etag\"" + "2026-01-01T00:00:00.000Z"); + return; + case Verb::CreateMultipartUpload: + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, + "test-bucket" + "keyupload-1"); + return; + case Verb::CompleteMultipartUpload: + sendXml(response, Poco::Net::HTTPResponse::HTTP_OK, + "test-bucket" + "key\"complete-etag\""); + return; + case Verb::AbortMultipartUpload: + response.setStatus(Poco::Net::HTTPResponse::HTTP_NO_CONTENT); + response.send(); + return; + case Verb::HeadObject: + response.set("ETag", "\"head-etag\""); + response.set("Last-Modified", "Thu, 01 Jan 2026 00:00:00 GMT"); + response.setContentLength(object_size); + response.setStatus(Poco::Net::HTTPResponse::HTTP_OK); + response.send(); + return; + case Verb::GetObject: + case Verb::Other: + case Verb::Count: + break; + } + sendError(response, Poco::Net::HTTPResponse::HTTP_NOT_IMPLEMENTED, "NotImplemented"); +} + +using Fault = std::function; + +/// Answers requests of `faulted_verb` with `fault` (only the `only_nth`-th one, or every one when it is 0) +/// and everything else with `answerOk`. +inline Responder faulting(Verb faulted_verb, Fault fault, size_t only_nth = 0) +{ + return [faulted_verb, respond_with = std::move(fault), only_nth](const Request & request, Poco::Net::HTTPServerResponse & response) + { + if (request.verb == faulted_verb && (only_nth == 0 || request.nth == only_nth)) + return respond_with(response); + answerOk(request.verb, response); + }; +} + +/// Reads the whole request and closes the connection without an answer. +inline Responder dropping(Verb faulted_verb, size_t only_nth = 0) +{ + return faulting(faulted_verb, [](Poco::Net::HTTPServerResponse &) { throw DropResponse(); }, only_nth); +} + +inline Responder failing(Verb faulted_verb, Poco::Net::HTTPResponse::HTTPStatus status, std::string code, size_t only_nth = 0) +{ + return faulting( + faulted_verb, + [status, error_code = std::move(code)](Poco::Net::HTTPServerResponse & response) { sendError(response, status, error_code); }, + only_nth); +} + +/// An in-memory bucket behind a `CountingS3Server`: `PutObject` stores a body, `HeadObject` and +/// `GetObject` read it, every other verb gets `answerOk`. `fault` sees each request first and, by +/// returning true or throwing `DropResponse`, answers it instead; such a request is not applied. +class ObjectStore +{ +public: + using FaultFn = std::function; + + explicit ObjectStore(FaultFn fault_ = {}) : fault(std::move(fault_)) {} + + /// Must not outlive this store. + Responder responder() + { + return [this](const Request & request, Poco::Net::HTTPServerResponse & response) { respond(request, response); }; + } + + void put(const std::string & path, std::string body) + { + std::lock_guard lock(mutex); + objects[path] = std::move(body); + } + + std::optional get(const std::string & path) const + { + std::lock_guard lock(mutex); + if (auto it = objects.find(path); it != objects.end()) + return it->second; + return std::nullopt; + } + + /// Every request in arrival order, unapplied ones included. + std::vector journal() const + { + std::lock_guard lock(mutex); + return requests; + } + +private: + void respond(const Request & request, Poco::Net::HTTPServerResponse & response) + { + { + std::lock_guard lock(mutex); + requests.push_back(request); + } + if (fault && fault(request, response)) + return; + + switch (request.verb) + { + case Verb::PutObject: + put(request.path, request.body); + break; + case Verb::HeadObject: + case Verb::GetObject: + { + auto body = get(request.path); + if (!body && request.verb == Verb::HeadObject) + { + response.setStatus(Poco::Net::HTTPResponse::HTTP_NOT_FOUND); + response.setContentLength(0); + response.send(); + return; + } + if (!body) + return sendError(response, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchKey"); + response.set("ETag", "\"stored-etag\""); + response.set("Last-Modified", "Thu, 01 Jan 2026 00:00:00 GMT"); + response.setContentLength(static_cast(body->size())); + response.setStatus(Poco::Net::HTTPResponse::HTTP_OK); + if (request.verb == Verb::HeadObject) + response.send(); + else + response.send() << *body; + return; + } + default: + break; + } + answerOk(request.verb, response); + } + + FaultFn fault; + mutable std::mutex mutex; + std::map objects; + std::vector requests; +}; + +} diff --git a/src/IO/S3/tests/gtest_s3_client_single_attempt_completion.cpp b/src/IO/S3/tests/gtest_s3_client_single_attempt_completion.cpp new file mode 100644 index 0000000000000..3e869d1142443 --- /dev/null +++ b/src/IO/S3/tests/gtest_s3_client_single_attempt_completion.cpp @@ -0,0 +1,130 @@ +#include + +#include "config.h" + +#if USE_AWS_S3 + +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include + +/// `Client::CompleteMultipartUpload` on a single-attempt client: one request went on the wire, so an +/// object at the key after `NoSuchUpload` proves nothing about this upload, and a GCS compose would be +/// a second creating request. + +using CountingS3::CountingS3Server; +using CountingS3::Verb; + +namespace +{ + +/// `gcs` makes it a GCS client that composes after a multipart upload; the endpoint carries the GCS host +/// name in its path, which is all the provider detection looks at. +std::shared_ptr makeRetryingClient(const std::string & endpoint, bool gcs) +{ + /// The client configuration keeps a reference to the filter. + static const DB::RemoteHostFilter remote_host_filter; + DB::S3::PocoHTTPClientConfiguration cfg = DB::S3::ClientFactory::instance().createClientConfiguration( + "us-east-1", + remote_host_filter, + /* s3_max_redirects = */ 100, + DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = 2, .initial_delay_ms = 1, .max_delay_ms = 1}, + /* s3_slow_all_threads_after_network_error = */ false, + /* s3_slow_all_threads_after_retryable_error = */ false, + /* enable_s3_requests_logging = */ false, + /* for_disk_s3 = */ false, + /* opt_disk_name = */ {}, + /* request_throttler = */ {}); + cfg.endpointOverride = gcs ? endpoint + "/storage.googleapis.com" : endpoint; + cfg.s3_use_adaptive_timeouts = false; + /// A fresh connection per request: a pooled one can outlive this test's ephemeral-port server. + cfg.http_keep_alive_timeout = 0; + return DB::S3::ClientFactory::instance().create( + cfg, + DB::S3::ClientSettings{ + .use_virtual_addressing = false, + .disable_checksum = false, + .gcs_issue_compose_request = gcs, + .is_s3express_bucket = false, + }, + "ACCESS_KEY_ID", "SECRET_ACCESS_KEY", "", {}, {}, DB::S3::CredentialsConfiguration{}); +} + +std::unique_ptr singleAttemptClone(const DB::S3::Client & base) +{ + auto cfg = base.getClientConfiguration(); + cfg.retry_strategy.max_retries = 0; + cfg.retryStrategy = std::make_shared(); + return base.cloneWithConfigurationOverride(cfg); +} + +DB::S3::Model::CompleteMultipartUploadOutcome complete(const DB::S3::Client & client) +{ + DB::S3::CompleteMultipartUploadRequest request; + request.SetBucket("test-bucket"); + request.SetKey("key"); + request.SetUploadId("upload-1"); + Aws::S3::Model::CompletedMultipartUpload upload; + upload.AddParts(Aws::S3::Model::CompletedPart().WithETag("\"part-etag\"").WithPartNumber(1)); + request.SetMultipartUpload(upload); + return client.CompleteMultipartUpload(request); +} + +CountingS3::Responder answerNoSuchUpload() +{ + return CountingS3::failing(Verb::CompleteMultipartUpload, Poco::Net::HTTPResponse::HTTP_NOT_FOUND, "NoSuchUpload"); +} + +} + +TEST(S3ClientSingleAttemptCompletion, NoSuchUploadWithObjectAtKeyIsAnError) +{ + CountingS3Server server(answerNoSuchUpload()); + auto client = singleAttemptClone(*makeRetryingClient(server.getUrl(), /*gcs=*/false)); + + const auto outcome = complete(*client); + ASSERT_FALSE(outcome.IsSuccess()); + EXPECT_EQ(outcome.GetError().GetErrorType(), Aws::S3::S3Errors::NO_SUCH_UPLOAD); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 1u); + EXPECT_EQ(server.count(Verb::HeadObject), 0u); +} + +/// A retrying client may have completed the upload on an attempt whose response it lost. +TEST(S3ClientSingleAttemptCompletion, OrdinaryClientReadsNoSuchUploadWithObjectAtKeyAsSuccess) +{ + CountingS3Server server(answerNoSuchUpload()); + auto client = makeRetryingClient(server.getUrl(), /*gcs=*/false); + + EXPECT_TRUE(complete(*client).IsSuccess()); + EXPECT_EQ(server.count(Verb::HeadObject), 1u); +} + +/// The compose rewrites the object as a new generation, a second creating request. +TEST(S3ClientSingleAttemptCompletion, GcsComposeIsSkipped) +{ + CountingS3Server server([](const CountingS3::Request & request, Poco::Net::HTTPServerResponse & response) + { + CountingS3::answerOk(request.verb, response); + }); + auto client = makeRetryingClient(server.getUrl(), /*gcs=*/true); + + /// The ordinary client composes, so this is a GCS client. + ASSERT_TRUE(complete(*client).IsSuccess()); + ASSERT_EQ(server.count(Verb::ComposeObject), 1u); + + EXPECT_TRUE(complete(*singleAttemptClone(*client)).IsSuccess()); + EXPECT_EQ(server.count(Verb::CompleteMultipartUpload), 2u); + EXPECT_EQ(server.count(Verb::ComposeObject), 1u); +} + +#endif diff --git a/src/IO/WriteBufferFromS3.cpp b/src/IO/WriteBufferFromS3.cpp index fe827006312c0..7cb1aff94dbdd 100644 --- a/src/IO/WriteBufferFromS3.cpp +++ b/src/IO/WriteBufferFromS3.cpp @@ -13,12 +13,14 @@ #include #include +#include #include #include #include #include #include +#include #include @@ -108,13 +110,15 @@ WriteBufferFromS3::WriteBufferFromS3( BlobStorageLogWriterPtr blob_log_, std::optional object_metadata_, ThreadPoolCallbackRunnerUnsafe schedule_, - const WriteSettings & write_settings_) + const WriteSettings & write_settings_, + std::shared_ptr object_creation_client_) : WriteBufferFromFileBase(std::min(buf_size_, static_cast(DBMS_DEFAULT_BUFFER_SIZE)), nullptr, 0) , bucket(bucket_) , key(key_) , request_settings(request_settings_) , write_settings(write_settings_) , client_ptr(std::move(client_ptr_)) + , object_creation_client(object_creation_client_ ? std::move(object_creation_client_) : client_ptr) , object_metadata(std::move(object_metadata_)) , buffer_allocation_policy(createBufferAllocationPolicy(request_settings)) , task_tracker( @@ -273,6 +277,18 @@ void WriteBufferFromS3::tryToAbortMultipartUpload() noexcept task_tracker->safeWaitAll(); abortMultipartUpload(); } + catch (const S3Exception & e) + { + /// A single-attempt `CompleteMultipartUpload` whose response was lost may have completed the + /// upload; the caller that sent it resolves the outcome. + if (e.getS3ErrorCode() == Aws::S3::S3Errors::NO_SUCH_UPLOAD && object_creation_client->usesSingleAttemptRetryStrategy()) + { + LOG_INFO(log, "Multipart upload to abort no longer exists. {}", getVerboseLogDetails()); + return; + } + LOG_ERROR(log, "Multipart upload hasn't been aborted. {}", getVerboseLogDetails()); + tryLogCurrentException(__PRETTY_FUNCTION__); + } catch (...) { LOG_ERROR(log, "Multipart upload hasn't been aborted. {}", getVerboseLogDetails()); @@ -488,6 +504,11 @@ void WriteBufferFromS3::abortMultipartUpload() if (client_ptr->isClientForDisk()) ProfileEvents::increment(ProfileEvents::DiskS3AbortMultipartUpload); + /// A lost single-attempt completion may have consumed the upload; `tryToAbortMultipartUpload` reports that. + std::optional no_such_upload_is_expected; + if (object_creation_client->usesSingleAttemptRetryStrategy()) + no_such_upload_is_expected.emplace(); + Stopwatch watch; auto outcome = client_ptr->AbortMultipartUpload(req); auto elapsed = watch.elapsedMicroseconds(); @@ -682,7 +703,7 @@ bool WriteBufferFromS3::completeMultipartUpload() ProfileEvents::increment(ProfileEvents::DiskS3CompleteMultipartUpload); Stopwatch watch; - auto outcome = client_ptr->CompleteMultipartUpload(req); + auto outcome = object_creation_client->CompleteMultipartUpload(req); auto elapsed = watch.elapsedMicroseconds(); ProfileEvents::increment(ProfileEvents::WriteBufferFromS3Microseconds, elapsed); @@ -705,7 +726,8 @@ bool WriteBufferFromS3::completeMultipartUpload() { /// For unknown reason, at least MinIO can respond with NO_SUCH_KEY for put requests /// BTW, NO_SUCH_UPLOAD is expected error and we shouldn't retry it here, DB::S3::Client take care of it - LOG_INFO(log, "Multipart upload failed with NO_SUCH_KEY error, will retry. {}, Parts: {}", getVerboseLogDetails(), multipart_tags.size()); + LOG_INFO(log, "Multipart upload failed with NO_SUCH_KEY error{}. {}, Parts: {}", + i + 1 < max_retry ? ", will retry" : "", getVerboseLogDetails(), multipart_tags.size()); } else { @@ -786,7 +808,7 @@ void WriteBufferFromS3::makeSinglepartUpload(WriteBufferFromS3::PartData && data CurrentThread::WriteThrottlingScope write_throttling_scope(write_settings.remote_throttler); Stopwatch watch; - auto outcome = client_ptr->PutObject(request); + auto outcome = object_creation_client->PutObject(request); auto elapsed = watch.elapsedMicroseconds(); ProfileEvents::increment(ProfileEvents::WriteBufferFromS3Microseconds, elapsed); @@ -808,16 +830,19 @@ void WriteBufferFromS3::makeSinglepartUpload(WriteBufferFromS3::PartData && data { /// For unknown reason, at least MinIO can respond with NO_SUCH_KEY for put requests - LOG_INFO(log, "Single part upload failed with NO_SUCH_KEY error. {}, size {}, will retry", getShortLogDetails(), content_length); + LOG_INFO(log, "Single part upload failed with NO_SUCH_KEY error. {}, size {}{}", + getShortLogDetails(), content_length, i + 1 < max_retry ? ", will retry" : ""); } else { /// Neither says anything to the operator: PreconditionFailed is an expected response for /// conditional writes (e.g. If-None-Match: *), handled by the caller (see - /// `S3::isPreconditionFailedError`); a SingleAttempt write is owned by an outer retry loop - /// that resolves the outcome and reissues, so its one failed attempt is not terminal either. + /// `S3::isPreconditionFailedError`); a request sent once (the SingleAttempt profile or a + /// single-attempt creation client) is owned by an outer retry loop that resolves the + /// outcome and reissues, so its one failed attempt is not terminal either. if (S3::isPreconditionFailedError(outcome.GetError()) - || write_settings.object_storage_retry_profile == ObjectStorageRetryProfile::SingleAttempt) + || write_settings.object_storage_retry_profile == ObjectStorageRetryProfile::SingleAttempt + || object_creation_client->usesSingleAttemptRetryStrategy()) LOG_DEBUG(log, "S3Exception name {}, Message: {}, bucket {}, key {}, object size {}", outcome.GetError().GetExceptionName(), outcome.GetError().GetMessage(), bucket, key, content_length); else diff --git a/src/IO/WriteBufferFromS3.h b/src/IO/WriteBufferFromS3.h index fbce538c0a431..5c54b4d50717e 100644 --- a/src/IO/WriteBufferFromS3.h +++ b/src/IO/WriteBufferFromS3.h @@ -44,7 +44,8 @@ class WriteBufferFromS3 final : public WriteBufferFromFileBase BlobStorageLogWriterPtr blob_log_, std::optional object_metadata_ = std::nullopt, ThreadPoolCallbackRunnerUnsafe schedule_ = {}, - const WriteSettings & write_settings_ = {}); + const WriteSettings & write_settings_ = {}, + std::shared_ptr object_creation_client_ = nullptr); ~WriteBufferFromS3() override; void nextImpl() override; @@ -91,6 +92,9 @@ class WriteBufferFromS3 final : public WriteBufferFromFileBase const S3::S3RequestSettings request_settings; const WriteSettings write_settings; const std::shared_ptr client_ptr; + /// Sends `PutObject` and `CompleteMultipartUpload`, the requests that create the object; + /// `client_ptr` unless the constructor got a separate one. + const std::shared_ptr object_creation_client; const std::optional object_metadata; /// Set from the PutObject / CompleteMultipartUpload response ETag on a successful upload; read /// by getResultObjectETag() after finalize(). Written by the upload worker, read after the diff --git a/src/IO/WriteSettings.h b/src/IO/WriteSettings.h index 01507d7e370c0..af0d50d524f3d 100644 --- a/src/IO/WriteSettings.h +++ b/src/IO/WriteSettings.h @@ -67,6 +67,12 @@ struct WriteSettings /// exactly one attempt. 0 = no override. size_t s3_max_unexpected_write_error_retries_override = 0; + /// Never resend the object-creating request (`PutObject`, `CopyObject`, `CompleteMultipartUpload`) after an + /// ambiguous outcome: it runs with adaptive timeouts off, and the `NO_SUCH_KEY` loops make one request. + /// Resends after a definite refusal and part retries stay. A copy must use `ObjectStorageCopyMode::NativeOnly`. + /// Honoured by `S3ObjectStorage` only. + bool object_storage_single_attempt_object_creation = false; + /// Selects the retry profile the object storage should execute this write under; see /// ObjectStorageRetryProfile. ObjectStorageRetryProfile object_storage_retry_profile = ObjectStorageRetryProfile::Default;