diff --git a/docs/en/antalya/cas/architecture/replication.md b/docs/en/antalya/cas/architecture/replication.md index 5ba870b0f714d..cebbd8aa19c0c 100644 --- a/docs/en/antalya/cas/architecture/replication.md +++ b/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" | diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index 4828cfff79d70..195372a1919f9 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/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} diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.cpp index d5c308dc8bea5..331cd5b2c237d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.cpp +++ b/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(); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.h index 3c0944561a01e..f7991307a9266 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/Cache/MetadataStorageFromCacheObjectStorage.h +++ b/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; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h index 2bcd45d5f4e73..3195d35bb8965 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h +++ b/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 [[noreturn]] void throwCasOwnershipConflict(FormatStringHelper fmt, Args &&... args) { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Parts/PartFolderAccess.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Parts/PartFolderAccess.cpp index 0886ffd999b8c..1238f73f79689 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Parts/PartFolderAccess.cpp +++ b/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, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h index 28d40443231f8..43c161dcd2e4c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h +++ b/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 diff --git a/src/Disks/IDisk.h b/src/Disks/IDisk.h index cfb7bfd661283..4036885d1206a 100644 --- a/src/Disks/IDisk.h +++ b/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; } diff --git a/src/Disks/MetadataStorageWithPathWrapper.h b/src/Disks/MetadataStorageWithPathWrapper.h index 0a6fe08ee6b40..34aa9b28ad9f9 100644 --- a/src/Disks/MetadataStorageWithPathWrapper.h +++ b/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 { diff --git a/src/Disks/tests/gtest_cas_part_folder_access.cpp b/src/Disks/tests/gtest_cas_part_folder_access.cpp index 4ccd1c540cd1a..2ebf9c5c29afd 100644 --- a/src/Disks/tests/gtest_cas_part_folder_access.cpp +++ b/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(); + 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 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(); diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp index da86ef58fee83..61365665b5b24 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp +++ b/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; diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskBase.h b/src/Storages/MergeTree/DataPartStorageOnDiskBase.h index 1cc3abad7e011..0d4b107a4f4c4 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.h +++ b/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; diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp index 517c25c12f4d4..ed3d1f2439167 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp +++ b/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(); -} - } diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskFull.h b/src/Storages/MergeTree/DataPartStorageOnDiskFull.h index 7e994a1b0e6ac..b19c46946e9c9 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskFull.h +++ b/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 diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp index 8f20d92e4e1ea..c8f99ddfd8752 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp +++ b/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); } } diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h index 395ff00438d5a..913fd517e558b 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h +++ b/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; } diff --git a/src/Storages/MergeTree/DataPartsExchange.cpp b/src/Storages/MergeTree/DataPartsExchange.cpp index 6ce267fd194ce..6506e34b3f4d3 100644 --- a/src/Storages/MergeTree/DataPartsExchange.cpp +++ b/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 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"} }); diff --git a/tests/clickhouse-test b/tests/clickhouse-test index 975d6e5e260f9..8c1c6eee366ae 100755 --- a/tests/clickhouse-test +++ b/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 ( diff --git a/tests/integration/test_cas_replicated_relink/test.py b/tests/integration/test_cas_replicated_relink/test.py index a9b6c7952e133..1635583418c24 100644 --- a/tests/integration/test_cas_replicated_relink/test.py +++ b/tests/integration/test_cas_replicated_relink/test.py @@ -614,20 +614,22 @@ def interserver_request(node, target_host, params): return headers, int(size) -def test_version_mix_legacy_peer_gets_bytes(): - """Task 16 step 7 — version mix: a peer that does not promise to confirm is served BYTES. - - This is the sender-side half of the mixed-build gate, and it is the half that is reachable without a - second binary: the offer is gated on `client_protocol_version >= 11` (`..._WITH_CA_CONFIRM`), so a - peer advertising 10 — a build that would relink WITHOUT confirming — must get the byte stream. - Degrading to bytes, never to an unconfirmed relink, is the whole point of moving the gate to 11. - - The control request (identical, but advertising 11) is what makes the negative meaningful: it proves - the request is otherwise perfectly relinkable, so the absence of an offer in the v10 case is the - version gate and not a malformed request. - - The receiver-side row-1 branch — a genuinely OLD sender that offers a relink with NO source token — - is NOT covered here; see the report accompanying this task. +def test_relink_is_selected_by_the_capability_not_the_version(): + """A relink is offered to a peer that advertises `cas_capabilities=relink_confirm`; the replication + protocol version never selects it. + + The capability is the receiver's promise to confirm the offer against its source before it promotes. + It travels as its own request parameter so that the protocol version stays upstream's: a peer that + advertises a higher version for some other upstream feature, but never heard of relink, must get the + byte stream, and this build must answer it with upstream's version ceiling, not a number upstream may + reuse with another meaning. + + Three raw requests at the sender, identical except for the two knobs: + - version 10 with the capability: the offer (control, and the proof that the capability selects it); + - version 11 without the capability: bytes, `server_protocol_version=10`; + - version 10 without the capability: bytes. + The receiver-side branch — a genuinely OLD sender that offers a relink with NO source token — is not + covered here. """ node1 = cluster.instances["node1"] node2 = cluster.instances["node2"] @@ -655,29 +657,40 @@ def test_version_mix_legacy_peer_gets_bytes(): ("compress", "false"), ("cas_pool_uuid", pool_uuid), ] + capability = ("cas_capabilities", "relink_confirm") + + # CONTROL — a confirm-capable peer at upstream's version: an offer, with a token, and a tiny + # manifest-only body. + headers_capable, size_capable = interserver_request( + node2, "node1", base + [("client_protocol_version", "10"), capability] + ) + assert "cas_relink=part_manifest_v2" in headers_capable, headers_capable + assert "cas_source_token=" in headers_capable, headers_capable + assert "server_protocol_version=10" in headers_capable, headers_capable - # CONTROL — a confirm-capable peer: an offer, with a token, and a tiny manifest-only body. + # THE CASE UNDER TEST — a higher version without the capability: no offer, upstream's ceiling in the + # answer, and the part's bytes. headers_v11, size_v11 = interserver_request( node2, "node1", base + [("client_protocol_version", "11")] ) - assert "cas_relink=part_manifest_v2" in headers_v11, headers_v11 - assert "cas_source_token=" in headers_v11, headers_v11 + assert "cas_relink" not in headers_v11, headers_v11 + assert "cas_source_token" not in headers_v11, headers_v11 + assert "server_protocol_version=10" in headers_v11, headers_v11 + + # Positive proof that bytes ACTUALLY moved rather than the request merely succeeding: the response + # carries the whole part, orders of magnitude more than the manifest-only relink payload. + assert size_v11 > 20 * size_capable, ( + "the peer without the capability should have received the part's bytes, got {} bytes against " + "the relink offer's {}".format(size_v11, size_capable) + ) - # THE CASE UNDER TEST — a peer advertising the pre-confirm version: no offer, and the part's bytes. - headers_v10, size_v10 = interserver_request( + # A pre-capability peer at upstream's version: bytes as well. + headers_v10, _ = interserver_request( node2, "node1", base + [("client_protocol_version", "10")] ) assert "cas_relink" not in headers_v10, headers_v10 - assert "cas_source_token" not in headers_v10, headers_v10 assert "server_protocol_version=10" in headers_v10, headers_v10 - # Positive proof that bytes ACTUALLY moved rather than the request merely succeeding: the v10 - # response carries the whole part, orders of magnitude more than the manifest-only relink payload. - assert size_v10 > 20 * size_v11, ( - "the v10 peer should have received the part's bytes, got {} bytes against the relink offer's " - "{}".format(size_v10, size_v11) - ) - drop_everywhere(table) diff --git a/tests/queries/0_stateless/05217_cas_packed_parts.sh b/tests/queries/0_stateless/05217_cas_packed_parts.sh index 93255cf809610..6bcdb4dc66268 100755 --- a/tests/queries/0_stateless/05217_cas_packed_parts.sh +++ b/tests/queries/0_stateless/05217_cas_packed_parts.sh @@ -1,5 +1,6 @@ #!/usr/bin/env bash -# Tags: no-fasttest, no-random-merge-tree-settings, no-parallel +# Tags: no-fasttest, no-random-merge-tree-settings, no-parallel, cas-storage +# cas-storage: the oracles below read system.cas_log of the default disk and use the cas_packed_mixed policy of the CAS lanes. set -euo pipefail @@ -177,10 +178,13 @@ MAIN_DISK=$(${CLICKHOUSE_CLIENT} --query " FROM system.parts WHERE database = currentDatabase() AND table = 'cas_packed_local' AND active") MAIN_DISK_IS_CAS=$(${CLICKHOUSE_CLIENT} --query "SELECT count() FROM system.cas_mounts WHERE disk = '${MAIN_DISK}'") +if [ "${MAIN_DISK_IS_CAS}" != "1" ]; then + echo "the default disk '${MAIN_DISK}' is not content-addressed; this test runs on the CAS lanes only" >&2 + exit 1 +fi -if [ "${MAIN_DISK_IS_CAS}" = "1" ]; then - ${CLICKHOUSE_CLIENT} --query "SYSTEM FLUSH LOGS cas_log" - ${CLICKHOUSE_CLIENT} --multiquery <&2 + exit 1 +fi -if [ "${HAS_MIXED_POLICY}" = "1" ]; then - ${CLICKHOUSE_CLIENT} --multiquery <&2 + exit 1 fi ${CLICKHOUSE_CLIENT} --multiquery <