CAS after #2647: relink capability, Packed index, test gate, oracles

docs/en/antalya/cas/architecture/replication.md

@@ -31,8 +31,8 @@ sequenceDiagram
     participant Snd as Sender
     participant S3 as Shared pool
 
-    R->>Snd: GET part, cas_pool_uuid = every pool of R's policy, client_protocol_version = 11
-    Note over R: advertising 11 is a promise to confirm before promoting
+    R->>Snd: GET part, cas_pool_uuid = every pool of R's policy, cas_capabilities = relink_confirm
+    Note over R: the capability is a promise to confirm before promoting
     Snd->>Snd: same disk pool uuid? identity, never endpoint plus prefix
     Snd->>S3: resolve the offer once -- manifest bytes and confirm token from the SAME view
     Snd-->>R: cookies cas_relink = part_manifest_v2, cas_source_token = ..., cas_pool_uuid = the matched pool -- body = manifest bytes
@@ -64,7 +64,7 @@ sequenceDiagram
 | # | Gate | What it enforces |
 |---|---|---|
 | 1 | Pool identity | The receiver advertises `cas_pool_uuid` — the pool uuids of every content-addressed disk of its storage policy that is not read-only, as one list — and the sender offers relink only if its own disk's pool uuid is **in** it, naming that uuid in a `cas_pool_uuid` response cookie. Matching by endpoint and prefix was tried and rejected — a minted pool uuid is the identity |
-| 2 | Protocol version 11 | On the receiver side, advertising it is a promise to run the confirm round trip before promoting |
+| 2 | The `relink_confirm` capability | The receiver sends `cas_capabilities=relink_confirm` with the fetch; the sender offers a relink only to a receiver that promised to run the confirm round trip before promoting. The replication protocol version stays upstream's and never selects a relink |
 | 3 | One resolution for two outputs | The manifest bytes and the confirm token come from the **same** view. Two separate calls would allow a repoint in between and hand the receiver a token naming a manifest whose entries it never adopted |
 | 4 | The receiver trusts nothing from the wire but the entry list | The sender's manifest id, namespace and payload digest are ignored; the target namespace and ref come from the receiver's own router, and manifest path hygiene is validated at decode |
 | 5 | The confirm is I/O-free and fail-closed | A cold, evicted, unfenced or terminal mount answers `Unknown`. `No` and `Unknown` both go on the wire as `unproven`, because the fence check is evaluated last, so a `No` cannot be distinguished from "cannot prove it right now" |

docs/en/antalya/cas/configuration.md

@@ -259,6 +259,8 @@ A 26.8 server renames a committed part with one ref-log `Move` record (`cas_log`
 
 Upgrade order: stop every 26.6.x server that mounts the pool, upgrade all of them to 26.8, then start them. There is no per-server switch: a 26.8 server always raises the floor on its first writable mount. With GC enabled on the 26.6.x servers, follow [Upgrading a pool with GC enabled](operations/debugging.md#upgrade-gc) first. A 26.6.x server that stays up after the raise cannot re-mount the pool on its own: after a lost mount lease or fence it stays unmounted until it is upgraded.
 
+Replication while the versions are mixed: a 26.6.x replica that fetches from a 26.8 server can fail the fetch. Both builds advertise replication protocol version 10 with different meanings — 26.8 sends `invalidated_system_columns.txt`, which a 26.6.x receiver counts into the checksums. Fetches resume once both sides run 26.8; a fetch-by-relink needs the `relink_confirm` capability, which only 26.8 advertises.
+
 What a 26.6.x server does if it meets a generation-2 pool: it refuses to mount after a restart (`UNKNOWN_FORMAT_VERSION` at startup: `CAS pool meta: pool requires reader generation 2 but this build supports at most 1`). An already mounted 26.6.x server keeps serving namespaces that have no `ref_move` record; direct recovery of a namespace that has one cannot cross the record until a later generation-1 snapshot gives it a new base; its GC round holds on the undecodable log and performs no deletions. Fetch-by-relink is protected by namespace ownership and the mount fence, not by the log version. No data is deleted or lost. Downgrading a pool whose floor is 2 is not supported.
 
 ## Server-level settings {#server-settings}

src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.cpp

@@ -54,6 +54,11 @@ bool MetadataStorageFromCacheObjectStorage::appliesOperationsEagerly() const
     return underlying->appliesOperationsEagerly();
 }
 
+bool MetadataStorageFromCacheObjectStorage::allowsPackedParts() const
+{
+    return underlying->allowsPackedParts();
+}
+
 bool MetadataStorageFromCacheObjectStorage::areBlobPathsRandom() const
 {
     return underlying->areBlobPathsRandom();

src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.h

@@ -28,6 +28,7 @@ class MetadataStorageFromCacheObjectStorage : public IMetadataStorage
     bool supportsEmptyFilesWithoutBlobs() const override;
     bool supportsInlineData() const override;
     bool appliesOperationsEagerly() const override;
+    bool allowsPackedParts() const override;
     bool areBlobPathsRandom() const override;
     ObjectStorageKeyGeneratorPtr getKeyGenerator() const override;
 

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

@@ -102,8 +102,10 @@ std::exception_ptr makeCasWriteRetryLaterExceptionPtr(const String & why);
 
 /// Throw `CAS_OWNERSHIP_CONFLICT`: the committed ref a caller read moved, dropped or changed owner before the
 /// caller's write landed; the caller must re-resolve the part before trying again. Counted in
-/// `ProfileEvents::CASOwnershipConflict`, so every conflict site goes through here (the counter counts
-/// throws; a ref-log builder that is re-run counts again).
+/// `ProfileEvents::CASOwnershipConflict`, so every conflict site goes through here. The counter counts
+/// throws on the throwing thread: inside an `appendRefOps` builder that is the append leader, which may be
+/// another query's thread, so per-query attribution in `system.query_log` is best-effort; the global
+/// counter is exact (a builder runs once per append).
 template <typename... Args>
 [[noreturn]] void throwCasOwnershipConflict(FormatStringHelper<Args...> fmt, Args &&... args)
 {

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Parts/PartFolderAccess.cpp

@@ -615,6 +615,8 @@ bool CachedPartFolderAccess::moveRefExact(const PartRefKey & src, const PartRefK
             if (d != committed.end())
             {
                 /// One manifest has one committed owner (`owned_manifests`); two rows naming `expected` is a broken table.
+                /// `CORRUPTED_DATA`, not a conflict: the ledger is replayed from the ref log, so a damaged log is the
+                /// only way here, and no MergeTree caller keys a rename on this code. Not counted as a conflict.
                 if (s != committed.end() && s->second.manifest_ref == expected && d->second.manifest_ref == expected)
                     throw Exception(ErrorCodes::CORRUPTED_DATA,
                         "moveRefExact: '{}' and '{}' both name manifest {}", src.ns.string() + "/" + src.ref,

src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h

@@ -369,6 +369,7 @@ class IMetadataStorage : private boost::noncopyable
     /// disk transaction delegates writes to the metadata transaction's content-addressed buffer.
     virtual bool isContentAddressed() const { return false; }
 
+    /// Whether a new part may be written in the `Packed` storage format (see `IDisk::allowsPackedParts`).
     virtual bool allowsPackedParts() const { return true; }
 
     /// True when a file write through this metadata storage publishes atomically, i.e. no partial

src/Disks/IDisk.h

@@ -479,6 +479,8 @@ class IDisk : public Space
     /// True when a file write on this disk publishes atomically (see `IDataPartStorage::supportsAtomicFileWrites`).
     virtual bool supportsAtomicFileWrites() const { return false; }
 
+    /// Whether a new part may be written in the `Packed` storage format on this disk. The format chooser
+    /// downgrades `Packed` to `Full` on a disk that answers false; parts already in `Packed` form stay readable.
     virtual bool allowsPackedParts() const { return true; }
 
     virtual bool isWriteOnce() const { return false; }

src/Disks/MetadataStorageWithPathWrapper.h

@@ -197,6 +197,7 @@ class MetadataStorageWithPathWrapper final : public IMetadataStorage
     bool supportsInlineData() const override { return delegate->supportsInlineData(); }
 
     bool appliesOperationsEagerly() const override { return delegate->appliesOperationsEagerly(); }
+    bool allowsPackedParts() const override { return delegate->allowsPackedParts(); }
 
     bool areBlobPathsRandom() const override
     {

src/Disks/tests/gtest_cas_part_folder_access.cpp

@@ -1494,6 +1494,28 @@ TEST(CASPartFolderAccess, MoveRefExactAbortsOnDestinationPrecommit)
     EXPECT_TRUE(access.existsRef({ns, "src"}, Cas::Freshness::ForceFresh));
 }
 
+/// The audit event is emitted after the Move or drop landed; a sink that throws loses the row, never the mutation.
+TEST(CASPartFolderAccess, AuditSinkFailureDoesNotFailTheDurableMoveOrDrop)
+{
+    auto backend = std::make_shared<CountingBackend>();
+    auto store = openPoolForTest(backend);
+    const Cas::RootNamespace ns{"srv/t1"};
+    Cas::CachedPartFolderAccess access(store);
+    publishPart(store, ns, "src", {inlineEntry("f", "v1")});
+    const auto m = access.resolve({ns, "src"}, Cas::Freshness::ForceFresh)->manifest_id.ref;
+    std::vector<Cas::CasEventType> seen;
+    store->setEventSink([&](Cas::CasEvent e) { seen.push_back(e.type); throw std::runtime_error("audit sink is down"); });
+
+    EXPECT_TRUE(access.moveRefExact({ns, "src"}, {ns, "dst"}, m));
+    EXPECT_FALSE(access.existsRef({ns, "src"}, Cas::Freshness::ForceFresh));
+    EXPECT_TRUE(access.existsRef({ns, "dst"}, Cas::Freshness::ForceFresh));
+    EXPECT_EQ(access.dropRefExact({ns, "dst"}, m), Cas::Removal::Removed);
+    EXPECT_FALSE(access.existsRef({ns, "dst"}, Cas::Freshness::ForceFresh));
+    EXPECT_EQ(std::ranges::count(seen, Cas::CasEventType::RefMove), 1) << "the move's audit event reached the sink";
+    EXPECT_EQ(std::ranges::count(seen, Cas::CasEventType::RefDrop), 1) << "the drop's audit event reached the sink";
+    store->setEventSink(nullptr);
+}
+
 TEST(CASPartFolderAccess, SecondMoveFailureLeavesFirstMoved)
 {
     auto backend = std::make_shared<CountingBackend>();

src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp

@@ -1352,6 +1352,14 @@ void DataPartStorageOnDiskBase::createDirectories()
     executeWriteOperation([&](auto & disk) { disk.createDirectories(fs::path(root_path) / part_dir); });
 }
 
+void DataPartStorageOnDiskBase::checkpointTransaction()
+{
+    if (isContentAddressed())
+        return;
+
+    IDataPartStorage::checkpointTransaction();
+}
+
 bool DataPartStorageOnDiskBase::hasActiveTransaction() const
 {
     return transaction != nullptr;

src/Storages/MergeTree/DataPartStorageOnDiskBase.h

@@ -175,6 +175,9 @@ class DataPartStorageOnDiskBase : public IDataPartStorage
 
     SyncGuardPtr getDirectorySyncGuard() const override;
     bool hasActiveTransaction() const override;
+    /// A content-addressed part is published once, as a whole, by its final commit: a mid-build commit
+    /// would publish the files written so far as a separate partial part, so the checkpoint is a no-op there.
+    void checkpointTransaction() override;
 
     bool isCaseInsensitive() const override;
 

src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp

@@ -295,14 +295,4 @@ void DataPartStorageOnDiskFull::commitTransaction()
     FailPointInjection::pauseFailPoint(FailPoints::part_storage_pause_after_commit_transaction);
 }
 
-void DataPartStorageOnDiskFull::checkpointTransaction()
-{
-    /// A content-addressed part is published once, as a whole, by its final commit: a mid-build
-    /// commit would publish the files written so far as a separate partial part.
-    if (isContentAddressed())
-        return;
-
-    IDataPartStorage::checkpointTransaction();
-}
-
 }

src/Storages/MergeTree/DataPartStorageOnDiskFull.h

@@ -45,7 +45,6 @@ class DataPartStorageOnDiskFull final : public DataPartStorageOnDiskBase
 
     void beginTransaction() override;
     void commitTransaction() override;
-    void checkpointTransaction() override;
     void precommitTransaction() override {}
     bool hasActiveTransaction() const override { return transaction != nullptr; }
 #if CLICKHOUSE_CLOUD

src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp

@@ -581,14 +581,6 @@ void DataPartStorageOnDiskPacked::commitTransaction()
     FailPointInjection::pauseFailPoint(FailPoints::part_storage_pause_after_commit_transaction);
 }
 
-void DataPartStorageOnDiskPacked::checkpointTransaction()
-{
-    if (isContentAddressed())
-        return;
-
-    IDataPartStorage::checkpointTransaction();
-}
-
 #if CLICKHOUSE_CLOUD
 TransactionCommitOutcomeVariant DataPartStorageOnDiskPacked::tryCommitTransaction(const TransactionCommitOptionsVariant & options)
 {
@@ -714,7 +706,8 @@ void DataPartStorageOnDiskPacked::resetReader(const ReadSettings & read_settings
     auto data_path = getRelativeDataPath();
     if (existsFileInTransactionOrDisk(data_path))
     {
-        auto buffer = readArchive(read_settings, std::nullopt);
+        /// Only the archive index is read here; a full-size (prefetching) buffer per part load is waste.
+        auto buffer = readArchive(read_settings.adjustBufferSize(4096), std::nullopt);
         reader.emplace(*buffer);
     }
 }

src/Storages/MergeTree/DataPartStorageOnDiskPacked.h

@@ -81,7 +81,6 @@ class DataPartStorageOnDiskPacked final : public DataPartStorageOnDiskBase
 
     void beginTransaction() override;
     void commitTransaction() override;
-    void checkpointTransaction() override;
 
     void setPreferredFileOrder(const Strings & file_names) override { preferred_file_order = file_names; }
 

src/Storages/MergeTree/DataPartsExchange.cpp

@@ -98,11 +98,25 @@ constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_PROJECTION = 7;
 constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_METADATA_VERSION = 8;
 constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_COLUMNS_SUBSTREAMS = 9;
 constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_INVALIDATED_SYSTEM_COLUMNS = 10;
-/// CAS replication, publish-then-confirm. A relink offer is now accompanied by a
-/// source token, and the endpoint answers a second, part-less request that asks whether that token is
-/// still exactly what the sender's ref names. A server advertising this version serves the confirm
-/// action; a receiver advertising it must confirm before it promotes.
-constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_CA_CONFIRM = 11;
+/// The newest protocol version this build speaks. Fork features ride on their own request parameters
+/// (`cas_capabilities`, `cas_pool_uuid`), never on a version number upstream may assign another meaning.
+constexpr auto REPLICATION_PROTOCOL_VERSION_LATEST = REPLICATION_PROTOCOL_VERSION_WITH_INVALIDATED_SYSTEM_COLUMNS;
+
+/// CAS capabilities a receiver advertises on the fetch request, as a comma-separated list. Each token is
+/// a promise about how the receiver treats what the sender offers.
+constexpr auto CA_CAPABILITIES_PARAM = "cas_capabilities";
+/// CAS replication, publish-then-confirm: the receiver confirms a relink offer against its source token
+/// (a second, part-less request) before it promotes. A relink is offered only to a receiver that promises
+/// this; a receiver without it gets the bytes, never an unconfirmed relink.
+constexpr auto CA_CAPABILITY_RELINK_CONFIRM = "relink_confirm";
+
+bool hasCasCapability(const String & advertised, std::string_view capability)
+{
+    for (const auto token : std::views::split(advertised, ','))
+        if (std::string_view(token) == capability)
+            return true;
+    return false;
+}
 
 std::string getEndpointId(const std::string & node_id)
 {
@@ -331,7 +345,7 @@ void Service::processQuery(const HTMLForm & params, ReadBufferPtr body, WriteBuf
     MergeTreePartInfo::fromPartName(part_name, data.format_version);
 
     /// We pretend to work as older server version, to be sure that client will correctly process our version
-    response.addCookie({"server_protocol_version", toString(std::min(client_protocol_version, REPLICATION_PROTOCOL_VERSION_WITH_CA_CONFIRM))});
+    response.addCookie({"server_protocol_version", toString(std::min(client_protocol_version, REPLICATION_PROTOCOL_VERSION_LATEST))});
 
     LOG_TRACE(log, "Sending part {}", part_name);
 
@@ -402,13 +416,12 @@ void Service::processQuery(const HTMLForm & params, ReadBufferPtr body, WriteBuf
         /// Strictly gated on a matching pool_uuid: a non-CA part, a CA part on a different pool, or a
         /// receiver without the capability all fall through to the unchanged byte path below.
         ///
-        /// The gate is `..._WITH_CA_CONFIRM`: a receiver is offered a relink
-        /// only once it advertises that it will confirm the offer before promoting it. A receiver that
-        /// still advertises 10 (a build whose relink had no confirm step) gets the bytes — mixed versions
-        /// degrade to bytes, never to an unconfirmed relink. This gate and the version the client advertises
-        /// (`fetchSelectedPart`) are one change in two places; separated in either order they either
-        /// disable relink outright or hand an unconfirmed relink to a receiver that claimed it confirms.
-        if (client_protocol_version >= REPLICATION_PROTOCOL_VERSION_WITH_CA_CONFIRM
+        /// The gate is the `relink_confirm` capability: a receiver is offered a relink only once it
+        /// advertises that it will confirm the offer before promoting it. A receiver without it (an older
+        /// build, or an upstream build whatever protocol version it speaks) gets the bytes — mixed versions
+        /// degrade to bytes, never to an unconfirmed relink. This gate and the capability the client
+        /// advertises (`fetchSelectedPart`) are one change in two places.
+        if (hasCasCapability(params.get(CA_CAPABILITIES_PARAM, ""), CA_CAPABILITY_RELINK_CONFIRM)
             && part->getDataPartStorage().isContentAddressed())
         {
             /// The receiver advertises every pool its storage policy has a writable content-addressed
@@ -703,11 +716,10 @@ std::pair<MergeTreeData::MutableDataPartPtr, scope_guard> Fetcher::fetchSelected
     {
         {"endpoint",                endpoint_id},
         {"part",                    part_name},
-        /// Advertising `..._WITH_CA_CONFIRM` is a PROMISE, not a capability list: this receiver will
-        /// confirm a relink offer against its source before it promotes (`relinkPartToDisk`). It is the
-        /// pair of the sender-side offer gate on the same constant, and the two cannot be separated —
-        /// see the comment there.
-        {"client_protocol_version", toString(REPLICATION_PROTOCOL_VERSION_WITH_CA_CONFIRM)},
+        {"client_protocol_version", toString(REPLICATION_PROTOCOL_VERSION_LATEST)},
+        /// A PROMISE: this receiver confirms a relink offer against its source before it promotes
+        /// (`relinkPartToDisk`). It is the pair of the sender-side offer gate on the same token.
+        {CA_CAPABILITIES_PARAM,      CA_CAPABILITY_RELINK_CONFIRM},
         {"compress",                "false"}
     });
 

tests/clickhouse-test

@@ -1607,6 +1607,7 @@ class FailureReason(enum.Enum):
     COVERAGE = "coverage"
     S3_STORAGE = "s3-storage"
     CAS_STORAGE = "cas-storage"
+    CAS_STORAGE_ONLY = "requires a CAS disk as the default MergeTree storage (cas-storage tag)"
     AZURE_BLOB_STORAGE = "azure-blob-storage"
     BUILD = "not running for current build"
     NO_PARALLEL_REPLICAS = "smth is not supported with parallel replicas"
@@ -3236,6 +3237,8 @@ class TestCase:
             and args.cas_storage
         ):
             return FailureReason.CAS_STORAGE
+        if tags and ("cas-storage" in tags) and not args.cas_storage:
+            return FailureReason.CAS_STORAGE_ONLY
         if tags and ("no-azure-blob-storage" in tags) and args.azure_blob_storage:
             return FailureReason.AZURE_BLOB_STORAGE
         if (