cas improvements

ci/jobs/functional_tests.py

@@ -506,6 +506,14 @@ def main():
 
     # for local run check if stateful tests are present to skip prepare_stateful_data and start faster if not
     has_stateful_tests = True
+    if info.is_local_run and not tests:
+        # A local run of the WHOLE suite cannot prepare the stateful datasets: `create.sql` attaches
+        # them from a web disk on `dockerhub-proxy.dockerhub-proxy-zone`, which resolves only inside
+        # CI, so the step dies with DNS_ERROR before a single test runs. Skipping it lets the
+        # stateless suite run locally; the tests that genuinely need `test.hits`/`test.visits` fail
+        # and are triaged as environment rather than taking the whole job with them.
+        print("Local full-suite run: skipping stateful data preparation (datasets are CI-hosted)")
+        has_stateful_tests = False
     if tests and info.is_local_run:
         from glob import glob
 

ci/jobs/scripts/clickhouse_proc.py

@@ -839,7 +839,9 @@ def prepare_stateful_data(self, with_s3_storage, is_db_replicated):
         command = bootstrap_vars + command
         if with_s3_storage:
             command = "USE_S3_STORAGE_FOR_MERGE_TREE=1\n" + command
-        return Shell.check(command)
+        # verbose: this step loads the stateful datasets and it is the only place in the job
+        # that can fail without printing anything at all, which is exactly what happened.
+        return Shell.check(command, verbose=True)
 
     def insert_system_zookeeper_config(self):
         for _ in range(10):

docs/en/antalya/cas/architecture/manifests-and-refs.md

@@ -101,7 +101,7 @@ swept for that root.
 flowchart TD
     A["LIST one page of cas/manifests/<br/>freeze candidates with exact GET"] --> B{"build-prefix eligible?<br/>durable watermark fact only"}
     B -->|"epoch less than lease epoch"| ELIG["eligible, old-epoch debris"]
-    B -->|"same epoch, min_active clears build_seq"| ELIG
+    B -->|"same epoch, min_active_build_sequence clears build_seq"| ELIG
     B -->|"no lease, or epoch ahead, or build may be live"| SKIP["skip"]
     ELIG --> C["protection view: committed manifests<br/>plus live precommits<br/>plus manifests with an unfolded minus-one"]
     C -->|"key protected"| SKIP2["skip"]

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

@@ -68,7 +68,7 @@ Two failure modes this closes:
 One object, `gc/server-roots/<server_root_id>/mount`, carries **both** the liveness lease and the build
 watermark — there is no separate watermark object. `MountLease` fields: `server_uuid`,
 `writer_epoch`, `write_attempt_id`, `hostname`, `pid`, `started_at_ms`, renewal `seq`,
-`expires_at_ms`, `min_active` (the build-watermark floor), and `gc_fenced`.
+`expires_at_ms`, `min_active_build_sequence` (the build-watermark floor), and `gc_fenced`.
 
 - **Logical renewal identity.** Each holder-originated body has a fresh nonzero
   `write_attempt_id`. One logical renewal fixes one immutable `(key, bytes, expected token,
@@ -133,9 +133,9 @@ into "not found".
 
 Global build ordering is the **pair** `(writer_epoch, build_seq)` compared lexicographically — the
 exact comparison GC uses for eligibility. The durable authority for both is the mount object
-itself: no mount means no deletion authority means nothing is swept. `min_active`, the oldest
+itself: no mount means no deletion authority means nothing is swept. `min_active_build_sequence`, the oldest
 in-flight `build_seq`, rides in the same mount object as the watermark floor; `UINT64_MAX` in
-`min_active` is the farewell/retired sentinel, not a real build.
+`min_active_build_sequence` is the farewell/retired sentinel, not a real build.
 
 ## Mount claim outcomes {#claim-outcomes}
 
@@ -153,7 +153,7 @@ a `MountClaimResult::Kind` together with a `MountPriorState` describing which ce
 | `MountPriorState` | Certificate that justified the reclaim |
 |---|---|
 | `None` | no reclaim needed (fresh claim or same-epoch refresh) |
-| `Clean` | the predecessor's own graceful farewell (`min_active == UINT64_MAX`) |
+| `Clean` | the predecessor's own graceful farewell (`min_active_build_sequence == UINT64_MAX`) |
 | `Fenced` | GC's own threshold-gated fence-out (`gc_fenced`) |
 | `UncleanObserved` | this claimant's own token-stability observation held for the full `TTL + drift` window |
 
@@ -168,7 +168,7 @@ stateDiagram-v2
     Absent --> Live: claimMount putIfAbsent, seq=1
     Live --> Live: keeper beat, putOverwrite seq+1
     Live --> Fenced: GC observes a stable token past threshold, gc_fenced=1, body preserved
-    Live --> Terminated: certified drain, terminal farewell (expires_at=now, min_active=MAX)
+    Live --> Terminated: certified drain, terminal farewell (expires_at=now, min_active_build_sequence=MAX)
     Fenced --> Live: same-uuid claim with a fresh writer_epoch, instant reclaim
     Terminated --> Live: same-uuid claim with a fresh writer_epoch, instant reclaim
     Live --> Live: same-uuid claim, proven-dead token via UncleanObserved
@@ -218,7 +218,7 @@ processed before renewal resumes.
 
 **Clean unmount:** request stop and join both persistent workers, drain the ref lanes, and only if
 the drain *certified* quiescence call `MountLeaseKeeper::release` on an `Active` keeper to write the
-terminal farewell (`expires_at_ms` already expired, `min_active = UINT64_MAX`). That sentinel is what
+terminal farewell (`expires_at_ms` already expired, `min_active_build_sequence = UINT64_MAX`). That sentinel is what
 lets a successor reclaim instantly. A `RenewalTerminal` keeper, an unresolved ref write, or a sent
 renewal ambiguity writes no farewell — an unearned farewell would let a successor start mutating
 while a stale conditional request from the predecessor is still in flight.

docs/en/antalya/cas/architecture/storage-layout.md

@@ -45,7 +45,7 @@ namespace's shape and never interprets its contents.
 | `gc/gen/<gen>/attempt/<att>/outcomes/<round>/<shard>.zst` | GC outcome log | `cas_gc_outcomes` | GC |
 | `gc/server-roots/<server_root_id>/owner` | server-root owner singleton | `cas_owner` | mount |
 | `gc/server-roots/<server_root_id>/epoch` | server-root epoch singleton | `cas_epoch` | mount |
-| `gc/server-roots/<server_root_id>/mount` | mount lease (incl. `min_active` watermark) | `cas_mount_lease` | mount |
+| `gc/server-roots/<server_root_id>/mount` | mount lease (incl. `min_active_build_sequence` watermark) | `cas_mount_lease` | mount |
 | `roots/<key>` | loose mountpoint object, verbatim | — (never interpreted) | upper layers |
 | `staging/<server_root_id>/…` | S3-native upload staging scratch | — | writer, own mount only |
 

docs/en/antalya/cas/index.md

@@ -64,6 +64,14 @@ Two consequences for planning:
   Each prefix is a fully independent pool (its own refs, leases, and `GC`), so rounds stay short
   regardless of the total fleet size.
 
+:::tip
+For replicated tables on `CAS`, enable
+[`execute_merges_on_single_replica_time_threshold`](/operations/settings/merge-tree-settings#execute_merges_on_single_replica_time_threshold).
+This lets one replica perform each merge while the others wait for and fetch the resulting part,
+avoiding redundant merge work across replicas. Set the threshold higher than the usual merge
+duration for your workload.
+:::
+
 ## Status {#status}
 
 `CAS` is **experimental**. It ships in Altinity Antalya builds. Experimental means the on-disk

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

@@ -38,7 +38,7 @@ specified (it is enabled by default in the shipped `config.xml`).
 - `gc_id` ([String](/sql-reference/data-types/string)) — The GC scheduler instance id (which mounter ran the round).
 - `trigger` ([Enum8](/sql-reference/data-types/enum)) — `Scheduled` (background tick) or `Manual` (`SYSTEM` command).
 - `round` ([UInt64](/sql-reference/data-types/int-uint)) — The GC round number (`0` on a `Start` row).
-- `outcome` ([Enum8](/sql-reference/data-types/enum)) — `Unknown` (on a `Start` row), `Success` (led, folded, and completed), `NotALeader` (another replica holds the GC lease), `Deferred` (led but took the skip-unchanged fast path — no fold ran, because no changed shard reached the fold threshold and no graduation was due), or `Error` (the round threw).
+- `outcome` ([Enum8](/sql-reference/data-types/enum)) — `Unknown` (on a `Start` row), `Success` (led, folded, and completed), `NotALeader` (another replica holds the GC lease), `Deferred` (led but took the skip-unchanged fast path — no fold ran, because no changed shard reached the fold threshold and no graduation was due), `Aborted` (the round threw a transient error — backend unavailability, a lost lease, a concurrent leader; the next scheduled round retries it), or `Error` (the round threw a non-transient error — investigate).
 - `candidates_marked` ([UInt64](/sql-reference/data-types/int-uint)) — Objects retired (marked) this round.
 - `objects_deleted` ([UInt64](/sql-reference/data-types/int-uint)) — Objects physically deleted this round.
 - `objects_absent` ([UInt64](/sql-reference/data-types/int-uint)) — Retire candidates found already absent.
@@ -51,7 +51,8 @@ specified (it is enabled by default in the shipped `config.xml`).
 - `fence_outs` ([UInt64](/sql-reference/data-types/int-uint)) — Expired mounts fenced out by this round's heartbeat floor.
 - `anomalies` ([UInt64](/sql-reference/data-types/int-uint)) — Fold clamps surfaced (and survived) this round. A steady non-zero value warrants a look at the round log details.
 - `duration_ms` ([UInt64](/sql-reference/data-types/int-uint)) — The round wall-clock duration (on a `Finish` row).
-- `error` ([String](/sql-reference/data-types/string)) — The exception text when `outcome = 'Error'`.
+- `error` ([String](/sql-reference/data-types/string)) — The exception text when `outcome = 'Aborted'` or `'Error'`.
+- `error_code` ([Int32](/sql-reference/data-types/int-uint)) — The exception code when `outcome = 'Aborted'` or `'Error'`; `0` otherwise. Key monitoring on this column rather than on the `error` text. On an `Aborted` or `Error` row the counters still report everything the round completed before it threw, and `round != 0` on such a row means the round's closing compare-and-swap committed and the failure hit only post-commit cleanup.
 - `ProfileEvents` ([Map(LowCardinality(String), UInt64)](/sql-reference/data-types/map)) — On a `Start`/`Finish` row, the per-round `ProfileEvents` delta (the `CAS*` counters and S3/disk events for this round). On a `Phase` row, **that phase's** delta, so `GROUP BY phase` over `ProfileEvents['S3ListObjects']` attributes the round's `LIST` budget to the phase that spent it.
 - `round_id` ([String](/sql-reference/data-types/string)) — The correlator for every row of one round attempt: its `Start`, each of its `Phase` rows, and its `Finish`. Minted per attempt, so unlike `round` it exists even for a round that never committed and for a round that never led. Group by this column to reconstruct one round.
 - `phase` ([LowCardinality(String)](/sql-reference/data-types/lowcardinality)) — The GC phase this row describes; empty on `Start`/`Finish`. See [Per-phase rows](#per-phase-rows) for the phase list.

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

@@ -252,13 +252,6 @@ PutResult ObjectStorageBackend::nativeConditionalPut(const String & key, const S
 namespace
 {
 
-/// Keep the emulated backend's publication memory bound to one materialized body at a time.
-std::mutex & emulatedBlobPublicationMutex()
-{
-    static std::mutex mutex;
-    return mutex;
-}
-
 }
 
 /// True when an exception from `IObjectStorage::readObject` means "the object is simply not there".
@@ -485,7 +478,7 @@ Token ObjectStorageBackend::emuWrite(const String & key, const String & bytes, c
     return emuMintToken(key, metadata ? metadata->etag : String{}, /*just_wrote=*/true);
 }
 
-void ObjectStorageBackend::emuPublishBlobAtomically(const String & key, const String & bytes)
+void ObjectStorageBackend::emuPublishBlobAtomically(const String & key, const String & envelope, ReadBuffer & payload, uint64_t payload_size)
 {
     if (object_storage->getType() != ObjectStorageType::Local)
         throw Exception(
@@ -497,14 +490,30 @@ void ObjectStorageBackend::emuPublishBlobAtomically(const String & key, const St
     const String root = object_storage->getCommonKeyPrefix();
     const String destination_path = resolvePathRelativelyToBase(destination_object, root);
     const String temporary_path = resolvePathRelativelyToBase(temporary_object, root);
-    const auto existing_token_state = emu_token_state.find(key);
 
+    /// The body is STREAMED into the temporary file -- envelope, then a bounded copy of the payload --
+    /// never materialized in memory. (An earlier revision accumulated envelope+payload in one String,
+    /// whose growth doubling made the peak allocation up to 2x the payload, and serialized every
+    /// publication behind a dedicated mutex just to bound that peak to one body at a time; streaming
+    /// removes both.) The destination stays untouched until the byte count has been validated: a short
+    /// or long source aborts on the temporary file, which is then removed.
     try
     {
         auto out = object_storage->writeObject(StoredObject(temporary_object), WriteMode::Rewrite);
-        out->write(bytes.data(), bytes.size());
+        out->write(envelope.data(), envelope.size());
+        const auto copy_result = blob_publication_detail::copyBlobPayloadBounded(payload, *out, payload_size);
+        if (!copy_result.exact(payload_size))
+        {
+            out->cancel();
+            throw Exception(
+                ErrorCodes::CORRUPTED_DATA,
+                "ObjectStorageBackend::publishBlob: source yielded {}{} payload bytes for {}, declared {} -- nothing was published",
+                copy_result.has_excess ? "more than " : "",
+                copy_result.copied,
+                key,
+                payload_size);
+        }
         out->finalize();
-        std::filesystem::rename(temporary_path, destination_path);
     }
     catch (...)
     {
@@ -517,7 +526,21 @@ void ObjectStorageBackend::emuPublishBlobAtomically(const String & key, const St
     /// existing disambiguator is sufficient: if the next observation sees the same ETag, it returns
     /// a token distinct from the old incarnation; if the ETag changed, emuMintToken resets the state
     /// to that new ETag. With no existing state, this backend has issued no same-process stale token
-    /// that needs fencing. The post-rename increment cannot allocate or throw.
+    /// that needs fencing. The post-rename increment cannot allocate or throw. `emu_mutex` spans the
+    /// rename and the bump so a concurrent emulated observation never sees the new incarnation with
+    /// the old disambiguator.
+    std::lock_guard lock(emu_mutex);
+    const auto existing_token_state = emu_token_state.find(key);
+    try
+    {
+        std::filesystem::rename(temporary_path, destination_path);
+    }
+    catch (...)
+    {
+        std::error_code cleanup_error;
+        std::filesystem::remove(temporary_path, cleanup_error);
+        throw;
+    }
     if (existing_token_state != emu_token_state.end())
         ++existing_token_state->second.second;
 }
@@ -872,32 +895,9 @@ void ObjectStorageBackend::publishBlob(const BlobPublishRequest & request)
 
         if (mode != Mode::Native)
         {
-            /// The emulated adapter's writes are whole-body operations. Serialize materialization so
-            /// concurrent publications retain the existing one-body peak-memory bound.
-            std::lock_guard publish_lock(emulatedBlobPublicationMutex());
-
-            String body = streaming->fresh_envelope;
-            blob_publication_detail::BlobPayloadCopyResult copy_result;
-            {
-                WriteBufferFromString out(body, AppendModeTag{});
-                copy_result = blob_publication_detail::copyBlobPayloadBounded(*payload, out, streaming->payload_size);
-                if (copy_result.exact(streaming->payload_size))
-                    out.finalize();
-                else
-                    out.cancel();
-            }
-
-            if (!copy_result.exact(streaming->payload_size))
-                throw Exception(
-                    ErrorCodes::CORRUPTED_DATA,
-                    "ObjectStorageBackend::publishBlob: source yielded {}{} payload bytes for {}, declared {} -- nothing was published",
-                    copy_result.has_excess ? "more than " : "",
-                    copy_result.copied,
-                    request.destination_key,
-                    streaming->payload_size);
-
-            std::lock_guard lock(emu_mutex);
-            emuPublishBlobAtomically(request.destination_key, body);
+            /// Streams straight into the temporary file and renames -- see emuPublishBlobAtomically.
+            emuPublishBlobAtomically(
+                request.destination_key, streaming->fresh_envelope, *payload, streaming->payload_size);
             return;
         }
 

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

@@ -264,7 +264,11 @@ class ObjectStorageBackend final : public Backend
     /// Write a complete blob body to a sibling temporary local object, then atomically replace `key`
     /// and advance any existing same-ETag disambiguator. A failure before the rename leaves the old
     /// destination and its token state untouched and cleans the temporary.
-    void emuPublishBlobAtomically(const String & key, const String & bytes);
+    /// Streams `envelope` + exactly `payload_size` bytes of `payload` into a temporary sibling of
+    /// `key`, then renames it into place -- nothing is visible at the destination until the byte count
+    /// has been validated, and the rename keeps publication atomic. Takes `emu_mutex` itself (for the
+    /// rename + token-state bump only); the caller must NOT hold it.
+    void emuPublishBlobAtomically(const String & key, const String & envelope, ReadBuffer & payload, uint64_t payload_size);
     /// Return the current emulated token for a key we just read/HEAD'd, reflecting its on-disk etag —
     /// does NOT advance the same-etag disambiguator (that only applies to a just-completed write).
     Token emuObserveToken(const String & key);

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp

@@ -547,6 +547,9 @@ Cas::GcRoundLogger ContentAddressedMetadataStorage::makeGcRoundLogger() const
             case Cas::GcRoundLogRecord::Outcome::Deferred:
                 e.outcome = ContentAddressedGarbageCollectionLogElement::DEFERRED;
                 break;
+            case Cas::GcRoundLogRecord::Outcome::Aborted:
+                e.outcome = ContentAddressedGarbageCollectionLogElement::ABORTED;
+                break;
         }
         e.round = r.round;
         e.candidates_marked = r.candidates_marked;
@@ -562,6 +565,7 @@ Cas::GcRoundLogger ContentAddressedMetadataStorage::makeGcRoundLogger() const
         e.anomalies = r.anomalies;
         e.duration_ms = r.duration_ms;
         e.error = r.error;
+        e.error_code = r.error_code;
         e.profile_events = r.profile_events;
         e.round_id = r.round_id;
         e.phase = r.phase;

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedTransaction.cpp

@@ -113,7 +113,7 @@ ContentAddressedTransaction::~ContentAddressedTransaction()
     /// backstop for aborted/exception-unwound transactions whose publishStaging never ran.
     cleanupPendingTempFiles();
 
-    /// An uncommitted transaction's uploads become min_active-spared debris: abandon every
+    /// An uncommitted transaction's uploads become min_active_build_sequence-spared debris: abandon every
     /// still-open PartWriteTxn so its build_seq is retired. This replaces the former pin machinery.
     if (committed)
         return;
@@ -759,8 +759,8 @@ std::string ContentAddressedTransaction::buildS3StagingBlobHeader(
     header.kind = Cas::ObjectKind::Blob;
     header.incarnation_tag = (static_cast<UInt128>(thread_local_rng()) << 64) | thread_local_rng();
     header.build_id = 0;   /// not known at stream time; diagnostic-only (not read by GC/read paths)
-    /// ch = the real ClickHouse VERSION_INTEGER (diagnostic-only; consistent with `PartWriteTxn::buildHeader`).
-    /// The v3 envelope drops hash_algo/domain_id/writer_version, so forensics ride on ch + bld.
+    /// `chver` = the real ClickHouse VERSION_INTEGER (diagnostic-only; consistent with `PartWriteTxn::buildHeader`).
+    /// The envelope drops hash_algo/domain_id/writer_version, so forensics ride on `chver` + `build`.
     header.provenance = Cas::Provenance{
         /*created_at_ms*/ 0, cfg.server_id, VERSION_INTEGER, Cas::ProvenanceOp::Other};
     header.intended_ref = route.ns.string() + "/" + route.ref;