From fe7435bc1304f73a201884462b3ca9d6b707de6f Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 15 Oct 2026 07:41:52 +0200 Subject: [PATCH 1/4] Add failing tests: stale CAS mounts are waited out one after another Server start creates each table's `disk = disk(...)` custom disk inside `Context::getOrCreateDisk`, which holds `storage_policies_mutex` while the disk starts. A CAS disk whose previous process left no farewell waits out the stale mount lease during that start, so N stale mounts cost N observation windows and every disk or storage-policy lookup in the server blocks meanwhile. Both tests fail on the current code: three stale mounts take 10820 ms against a 3450 ms window, and `getDisksMap` blocks for 3603 ms. CI report: https://altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2516&sha=b7e424247ab997ae2b0d3c06853e10a2c445f2a5&name_0=PR&name_1=Stress%20test%20%28amd_debug%29 PR: https://github.com/Altinity/ClickHouse/pull/2516 Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../tests/gtest_cas_startup_stale_mounts.cpp | 305 ++++++++++++++++++ 1 file changed, 305 insertions(+) create mode 100644 src/Disks/tests/gtest_cas_startup_stale_mounts.cpp diff --git a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp new file mode 100644 index 0000000000000..aa7bbb65e94ad --- /dev/null +++ b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp @@ -0,0 +1,305 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/// Server start re-creates the `disk = disk(...)` custom disks of every table it loads. A CAS disk +/// whose previous process died without a farewell waits out the stale mount lease inside that +/// creation. These tests pin how such waits compose with the rest of startup. + +namespace +{ + +namespace fs = std::filesystem; +using namespace DB::Cas; + +constexpr uint64_t lease_ttl_ms = 3000; +constexpr uint64_t renew_period_ms = 600; +/// `mountObservationThresholdMs`: TTL + TTL/20 + renew_period/2. +constexpr uint64_t observation_ms = lease_ttl_ms + lease_ttl_ms / 20 + renew_period_ms / 2; + +uint64_t elapsedMs(std::chrono::steady_clock::time_point since) +{ + return static_cast( + std::chrono::duration_cast(std::chrono::steady_clock::now() - since).count()); +} + +uint64_t wallNowMs() +{ + return static_cast(std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); +} + +struct StaleCasDisk +{ + String name; + String server_root_id; + fs::path pool_dir; + fs::path scratch_dir; + + std::map settings() const + { + return { + {"type", "object_storage"}, + {"object_storage_type", "local"}, + {"metadata_type", "cas"}, + {"path", pool_dir.string() + "/"}, + {"cas_server_root_id", server_root_id}, + {"cas_scratch_path", scratch_dir.string() + "/"}, + {"cas_mount_lease_ttl_ms", std::to_string(lease_ttl_ms)}, + {"cas_mount_renew_period_ms", std::to_string(renew_period_ms)}, + {"cas_attempt_timeout_ms", "200"}, + {"cas_lease_safety_margin_ms", "200"}, + {"cas_gc_enabled", "0"}, + }; + } + + /// The table-metadata form: what server start parses from `SETTINGS disk = disk(...)`. + DB::ASTPtr diskFunctionAst() const + { + String text = "disk(name = '" + name + "'"; + for (const auto & [key, value] : settings()) + text += ", " + key + " = '" + value + "'"; + text += ")"; + DB::ParserFunction parser; + return DB::parseQuery(parser, text, "", 0, 0, 0); + } + + Poco::AutoPtr xmlConfig() const + { + std::ostringstream xml; + xml << ""; + for (const auto & [key, value] : settings()) + xml << "<" << key << ">" << value << ""; + xml << ""; + std::istringstream in(xml.str()); + return Poco::AutoPtr(new Poco::Util::XMLConfiguration(in)); + } +}; + +/// Leaves `disk` as a process killed after it minted a writer epoch and claimed the mount: a live- +/// looking lease with no farewell, held by this server's own uuid under an older epoch. +void seedStaleMount(const StaleCasDisk & disk) +{ + { + auto config = disk.xmlConfig(); + auto seeded = DB::DiskFactory::instance().create( + disk.name + "_seed", *config, "disk", getContext().context, {}, /*attach=*/true, /*custom_disk=*/true); + seeded->shutdown(); + } + + auto object_storage = std::make_shared( + DB::LocalObjectStorageSettings(disk.name + "_seed_backend", disk.pool_dir.string() + "/", /*read_only_=*/false)); + auto backend = std::make_shared(object_storage, ObjectStorageBackend::Mode::EmulatedSingleProcess); + DB::Cas::tests::OperationForTest op(backend); + const Layout layout{"ca"}; + + const auto mount = (*op).read(layout.mountKey(disk.server_root_id), Retry::standard()); + ASSERT_TRUE(mount.has_value()); + const MountLease farewell = decodeMountLease(mount->bytes); + ASSERT_EQ(farewell.min_active_build_sequence, std::numeric_limits::max()) + << "the seeding open must end in a clean farewell"; + + const auto epoch_object = (*op).read(layout.epochKey(disk.server_root_id), Retry::standard()); + ASSERT_TRUE(epoch_object.has_value()); + const uint64_t crashed_epoch = decodeServerEpoch(epoch_object->bytes).next_writer_epoch; + ASSERT_TRUE(std::holds_alternative((*op).replace( + layout.epochKey(disk.server_root_id), encodeServerEpoch(ServerEpoch{.next_writer_epoch = crashed_epoch + 1}), + epoch_object->etag, Retry::standard()))); + + const MountClaimResult claim = claimMount( + *op, layout, disk.server_root_id, farewell.server_uuid, crashed_epoch, wallNowMs(), lease_ttl_ms); + ASSERT_EQ(claim.kind, MountClaimResult::Claimed); +} + +/// Signals when a CAS mount starts waiting out a stale lease. +class StaleMountWaitChannel : public Poco::Channel +{ +public: + void log(const Poco::Message & msg) override + { + if (msg.getText().find("stale-looking mount lease") == std::string::npos) + return; + std::lock_guard lock(mutex); + ++waits_started; + cv.notify_all(); + } + + bool waitForWaitStarted(size_t count) + { + std::unique_lock lock(mutex); + return cv.wait_for(lock, std::chrono::seconds(30), [&] { return waits_started >= count; }); + } + +private: + std::mutex mutex; + std::condition_variable cv; + size_t waits_started = 0; +}; + +class CASStartupStaleMounts : public ::testing::Test +{ +public: + static void SetUpTestSuite() + { + DB::ServerUUID::setRandomForUnitTests(); + DB::clearDiskRegistry(); + DB::registerDisks(/*global_skip_access_check=*/true); + } + + static void TearDownTestSuite() { DB::clearDiskRegistry(); } + +protected: + void SetUp() override + { + root = fs::temp_directory_path() + / ("cas_startup_stale_mounts_" + std::to_string(::getpid()) + "_" + + ::testing::UnitTest::GetInstance()->current_test_info()->name()); + fs::remove_all(root); + fs::create_directories(root); + } + + void TearDown() override + { + for (const auto & disk : disks) + if (auto created = getContext().context->getDisksMap(); created.contains(disk.name)) + created.at(disk.name)->shutdown(); + fs::remove_all(root); + } + + StaleCasDisk & addStaleDisk(const String & suffix) + { + const String test_name = ::testing::UnitTest::GetInstance()->current_test_info()->name(); + StaleCasDisk disk{ + .name = "cas_stale_" + test_name + "_" + suffix, + .server_root_id = "stale_" + suffix, + .pool_dir = root / ("pool_" + suffix), + .scratch_dir = root / ("scratch_" + suffix), + }; + fs::create_directories(disk.scratch_dir); + seedStaleMount(disk); + return disks.emplace_back(std::move(disk)); + } + + fs::path root; + std::vector disks; +}; + +} + +/// Each stale mount must be waited out once, but the waits of different disks are independent. +/// Waiting them one after another makes server start N × the observation window, which is what +/// missed the start timeout with three stale mounts. +TEST_F(CASStartupStaleMounts, StaleMountsOfDifferentDisksAreWaitedOutConcurrently) +{ + constexpr size_t disk_count = 3; + for (size_t i = 0; i < disk_count; ++i) + addStaleDisk(std::to_string(i)); + ASSERT_FALSE(HasFatalFailure()); + + std::vector per_disk_ms(disk_count, 0); + std::vector errors(disk_count); + const auto start = std::chrono::steady_clock::now(); + { + std::vector loaders; + for (size_t i = 0; i < disk_count; ++i) + loaders.emplace_back([&, i] + { + try + { + DB::DiskFromAST::createCustomDisk(disks[i].diskFunctionAst(), getContext().context, /*attach=*/true); + } + catch (...) + { + errors[i] = std::current_exception(); + } + per_disk_ms[i] = elapsedMs(start); + }); + for (auto & loader : loaders) + loader.join(); + } + const uint64_t total_ms = elapsedMs(start); + + for (size_t i = 0; i < disk_count; ++i) + { + ASSERT_FALSE(errors[i]) << DB::getExceptionMessage(errors[i], false); + EXPECT_GE(per_disk_ms[i], observation_ms) << "disk " << i << " had no stale mount to wait out"; + } + EXPECT_LT(total_ms, 2 * observation_ms) + << disk_count << " stale mounts took " << total_ms << " ms to open; one observation window is " + << observation_ms << " ms, so the waits ran one after another"; +} + +/// A disk lookup by an unrelated table load must not queue behind a stale-mount wait: the wait is +/// up to a full lease TTL, and every table load and system log flush resolves disks. +TEST_F(CASStartupStaleMounts, DiskLookupDoesNotWaitForAStaleMountObservation) +{ + const StaleCasDisk & stale = addStaleDisk("0"); + ASSERT_FALSE(HasFatalFailure()); + + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + Poco::AutoPtr previous_channel(pool_logger.getChannel(), /*shared=*/true); + const int previous_level = pool_logger.getLevel(); + Poco::AutoPtr channel(new StaleMountWaitChannel); + pool_logger.setChannel(channel); + pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); + SCOPE_EXIT({ + pool_logger.setChannel(previous_channel); + pool_logger.setLevel(previous_level); + }); + + std::exception_ptr error; + std::thread loader([&] + { + try + { + DB::DiskFromAST::createCustomDisk(stale.diskFunctionAst(), getContext().context, /*attach=*/true); + } + catch (...) + { + error = std::current_exception(); + } + }); + + ASSERT_TRUE(channel->waitForWaitStarted(1)) << "the stale mount was never observed"; + const auto lookup_start = std::chrono::steady_clock::now(); + (void)getContext().context->getDisksMap(); + const uint64_t lookup_ms = elapsedMs(lookup_start); + loader.join(); + + ASSERT_FALSE(error) << DB::getExceptionMessage(error, false); + EXPECT_LT(lookup_ms, observation_ms / 3) + << "a disk lookup took " << lookup_ms << " ms while a stale mount was being observed for up to " + << observation_ms << " ms"; +} From 19176553849ee9bbf3f33fba945d4886ec711aba Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 15 Oct 2026 10:43:04 +0200 Subject: [PATCH 2/4] Create custom disks outside storage_policies_mutex `Context::getOrCreateDisk` ran the disk creator under the server-wide `storage_policies_mutex`. A CAS disk waiting out a stale mount lease (about 36.5 s) inside the creator blocked every other disk creation and every disk and storage policy lookup, so stale mounts were waited out one after another at server start. The mutex now guards only the map of in-flight creations and the final insert into `DiskSelector`. Concurrent requests for one name share one creation and get the same disk or a copy of the same exception. A failed creation is still not cached. CI report: https://github.com/Altinity/ClickHouse/pull/2516 Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- .../tests/gtest_cas_startup_stale_mounts.cpp | 95 +++++++++++++++++++ src/Interpreters/Context.cpp | 92 ++++++++++++++++-- 2 files changed, 180 insertions(+), 7 deletions(-) diff --git a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp index aa7bbb65e94ad..2d16319c09cb3 100644 --- a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp +++ b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -38,6 +39,11 @@ /// whose previous process died without a farewell waits out the stale mount lease inside that /// creation. These tests pin how such waits compose with the rest of startup. +namespace DB::ErrorCodes +{ +extern const int BAD_ARGUMENTS; +} + namespace { @@ -303,3 +309,92 @@ TEST_F(CASStartupStaleMounts, DiskLookupDoesNotWaitForAStaleMountObservation) << "a disk lookup took " << lookup_ms << " ms while a stale mount was being observed for up to " << observation_ms << " ms"; } + +/// Two table loads that name the same custom disk must share one creation: a second open of the +/// same CAS root while the first is still waiting out the stale mount would be a live twin. +TEST_F(CASStartupStaleMounts, ConcurrentCreationsOfOneDiskShareOneOpen) +{ + const StaleCasDisk & stale = addStaleDisk("0"); + ASSERT_FALSE(HasFatalFailure()); + + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + Poco::AutoPtr previous_channel(pool_logger.getChannel(), /*shared=*/true); + const int previous_level = pool_logger.getLevel(); + Poco::AutoPtr channel(new StaleMountWaitChannel); + pool_logger.setChannel(channel); + pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); + SCOPE_EXIT({ + pool_logger.setChannel(previous_channel); + pool_logger.setLevel(previous_level); + }); + + const auto config = stale.xmlConfig(); + std::atomic creator_calls = 0; + const auto creator = [&](const DB::DisksMap & disks_map) + { + ++creator_calls; + return DB::DiskFactory::instance().create( + stale.name, *config, "disk", getContext().context, disks_map, /*attach=*/true, /*custom_disk=*/true); + }; + + DB::DiskPtr first_disk; + DB::DiskPtr second_disk; + std::exception_ptr first_error; + std::exception_ptr second_error; + std::thread first([&] + { + try + { + first_disk = getContext().context->getOrCreateDisk(stale.name, creator); + } + catch (...) + { + first_error = std::current_exception(); + } + }); + /// Without this wait the second request could find the disk already created. + const bool first_is_waiting = channel->waitForWaitStarted(1); + std::thread second([&] + { + try + { + second_disk = getContext().context->getOrCreateDisk(stale.name, creator); + } + catch (...) + { + second_error = std::current_exception(); + } + }); + first.join(); + second.join(); + + ASSERT_TRUE(first_is_waiting) << "the stale mount was never observed"; + ASSERT_FALSE(first_error) << DB::getExceptionMessage(first_error, false); + ASSERT_FALSE(second_error) << DB::getExceptionMessage(second_error, false); + EXPECT_EQ(creator_calls.load(), 1u); + EXPECT_EQ(first_disk, second_disk); + EXPECT_EQ(first_disk, getContext().context->getDisk(stale.name)); +} + +/// A failed creation is not remembered: the next request for the name creates the disk again. +TEST_F(CASStartupStaleMounts, FailedDiskCreationIsRetriedByTheNextRequest) +{ + const String name = "cas_stale_failed_creation"; + std::vector calls; + EXPECT_THROW( + getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + calls.push_back("first"); + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "creation failed"); + }), + DB::Exception); + EXPECT_FALSE(getContext().context->getDisksMap().contains(name)); + EXPECT_THROW( + getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + calls.push_back("second"); + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "creation failed again"); + }), + DB::Exception); + EXPECT_EQ(calls, (std::vector{"first", "second"})); +} diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 1109c5ae62036..40e6d3f822ee4 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -664,6 +665,8 @@ struct ContextSharedPart : boost::noncopyable mutable std::unique_ptr compression_codec_selector TSA_GUARDED_BY(mutex); /// Storage disk chooser for MergeTree engines mutable std::shared_ptr merge_tree_disk_selector TSA_GUARDED_BY(storage_policies_mutex); + /// Custom disks being created by `getOrCreateDisk`, by name + mutable std::unordered_map> disks_being_created TSA_GUARDED_BY(storage_policies_mutex); /// Storage policy chooser for MergeTree engines mutable std::shared_ptr merge_tree_storage_policy_selector TSA_GUARDED_BY(storage_policies_mutex); @@ -6775,20 +6778,95 @@ DiskPtr Context::getDisk(const String & name) const return disk_selector->get(name); } +namespace +{ + +/// Every waiter of one creation rethrows the same stored exception. Copy it, so a caller that +/// modifies its exception (e.g. `addMessage`) does not race with the other waiters. +DiskPtr getDiskCreationResult(const std::shared_future & creation) +{ + try + { + return creation.get(); + } + catch (const Exception & e) + { + throw Exception(e); + } +} + +} + DiskPtr Context::getOrCreateDisk(const String & name, DiskCreator creator) const { - std::lock_guard lock(shared->storage_policies_mutex); + std::promise promise; + std::shared_future creation; + bool is_creator = false; + DisksMap disks_map; + { + std::lock_guard lock(shared->storage_policies_mutex); - auto disk_selector = getDiskSelector(lock); + auto disk_selector = getDiskSelector(lock); + if (auto disk = disk_selector->tryGet(name)) + return disk; + + if (auto it = shared->disks_being_created.find(name); it != shared->disks_being_created.end()) + { + creation = it->second; + } + else + { + disks_map = disk_selector->getDisksMap(); + creation = promise.get_future().share(); + shared->disks_being_created.emplace(name, creation); + is_creator = true; + } + } - auto disk = disk_selector->tryGet(name); - if (!disk) + if (!is_creator) + return getDiskCreationResult(creation); + + /// Creating a disk can take long (a CAS disk waits out a stale mount lease), so it runs without + /// `storage_policies_mutex`. Concurrent requests for the same name wait for this creation rather + /// than open the same storage a second time. + DiskPtr disk; + std::exception_ptr error; + try { - disk = creator(getDisksMap(lock)); - const_cast(disk_selector.get())->addToDiskMap(name, disk); + disk = creator(disks_map); + } + catch (...) + { + error = std::current_exception(); + } + + { + std::lock_guard lock(shared->storage_policies_mutex); + shared->disks_being_created.erase(name); + if (!error) + { + try + { + auto disk_selector = getDiskSelector(lock); + /// A config reload may have added a disk with this name meanwhile; it wins, as if it ran first. + if (auto existing = disk_selector->tryGet(name)) + disk = existing; + else + const_cast(disk_selector.get())->addToDiskMap(name, disk); + } + catch (...) + { + error = std::current_exception(); + } + } } - return disk; + if (error) + promise.set_exception(error); + else + promise.set_value(disk); + + return getDiskCreationResult(creation); } StoragePolicyPtr Context::getStoragePolicy(const String & name) const From 9cce8e96981239f657044418870210b0510185aa Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 15 Oct 2026 11:09:06 +0200 Subject: [PATCH 3/4] Keep reloads off in-flight disk names, keep exception types A configuration reload could create a configured disk whose name was being created as a custom disk, opening the same storage twice. The reload now skips such names and logs a warning. Callers of a shared disk creation rethrow a copy of the stored exception through `Poco::Exception::rethrow`, so `ErrnoException` and other derived types keep their type and fields. Other exceptions become a `DB::Exception` with the same code and message. Tests: shared failure with independent copies, config reload during a creation, unique disk names so the suite survives `--gtest_repeat`, and the CAS logger channel is restored only after the disks stop. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- src/Disks/DiskSelector.cpp | 7 +- src/Disks/DiskSelector.h | 5 +- .../tests/gtest_cas_startup_stale_mounts.cpp | 190 ++++++++++++++---- src/Interpreters/Context.cpp | 31 +-- src/Interpreters/Context.h | 4 + 5 files changed, 188 insertions(+), 49 deletions(-) diff --git a/src/Disks/DiskSelector.cpp b/src/Disks/DiskSelector.cpp index 441451ad5ab37..23c4cad14e37e 100644 --- a/src/Disks/DiskSelector.cpp +++ b/src/Disks/DiskSelector.cpp @@ -143,7 +143,7 @@ catch (...) } DiskSelectorPtr DiskSelector::updateFromConfig( - const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context) const + const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, const NameSet & disks_to_skip) const { assertInitialized(); @@ -166,6 +166,11 @@ DiskSelectorPtr DiskSelector::updateFromConfig( auto disk_config_prefix = config_prefix + "." + disk_name; if (!result->getDisksMap().contains(disk_name)) { + if (disks_to_skip.contains(disk_name)) + { + LOG_WARNING(getLogger("DiskSelector"), "Disk {} is not created from the configuration: a custom disk with this name is being created", backQuote(disk_name)); + continue; + } auto created_disk = factory.create( disk_name, config, disk_config_prefix, context, result->getDisksMap(), /*attach*/ false, /*custom_disk*/ false, skip_types); if (created_disk) diff --git a/src/Disks/DiskSelector.h b/src/Disks/DiskSelector.h index acf491daa1c3e..066c5e30b4b54 100644 --- a/src/Disks/DiskSelector.h +++ b/src/Disks/DiskSelector.h @@ -29,10 +29,13 @@ class DiskSelector using DiskValidator = std::function; void initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator = {}); + /// Disks in `disks_to_skip` that are not created yet are left out: their names are being created + /// as custom disks, and creating them here would open the same storage twice. DiskSelectorPtr updateFromConfig( const Poco::Util::AbstractConfiguration & config, const String & config_prefix, - ContextPtr context) const; + ContextPtr context, + const NameSet & disks_to_skip = {}) const; /// Get disk by name DiskPtr get(const String & name) const; diff --git a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp index 2d16319c09cb3..51e47379a1039 100644 --- a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp +++ b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp @@ -1,9 +1,11 @@ #include +#include #include #include #include #include +#include #include #include #include @@ -22,7 +24,6 @@ #include #include #include -#include #include #include @@ -184,7 +185,10 @@ class CASStartupStaleMounts : public ::testing::Test DB::registerDisks(/*global_skip_access_check=*/true); } - static void TearDownTestSuite() { DB::clearDiskRegistry(); } + static void TearDownTestSuite() + { + DB::clearDiskRegistry(); + } protected: void SetUp() override @@ -201,14 +205,27 @@ class CASStartupStaleMounts : public ::testing::Test for (const auto & disk : disks) if (auto created = getContext().context->getDisksMap(); created.contains(disk.name)) created.at(disk.name)->shutdown(); + /// Restored only after the disks stop, so no CAS worker logs while the channel is swapped. + if (stale_mount_waits) + { + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + pool_logger.setChannel(previous_pool_channel); + pool_logger.setLevel(previous_pool_level); + } fs::remove_all(root); } + /// Disks stay in the global context after a test, so every name is unique within the process. + static String uniqueDiskName(const String & prefix) + { + static std::atomic counter = 0; + return prefix + "_" + ::testing::UnitTest::GetInstance()->current_test_info()->name() + "_" + std::to_string(++counter); + } + StaleCasDisk & addStaleDisk(const String & suffix) { - const String test_name = ::testing::UnitTest::GetInstance()->current_test_info()->name(); StaleCasDisk disk{ - .name = "cas_stale_" + test_name + "_" + suffix, + .name = uniqueDiskName("cas_stale_" + suffix), .server_root_id = "stale_" + suffix, .pool_dir = root / ("pool_" + suffix), .scratch_dir = root / ("scratch_" + suffix), @@ -218,15 +235,28 @@ class CASStartupStaleMounts : public ::testing::Test return disks.emplace_back(std::move(disk)); } + StaleMountWaitChannel & watchStaleMountWaits() + { + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + previous_pool_channel = Poco::AutoPtr(pool_logger.getChannel(), /*shared=*/true); + previous_pool_level = pool_logger.getLevel(); + stale_mount_waits = new StaleMountWaitChannel; + pool_logger.setChannel(stale_mount_waits); + pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); + return *stale_mount_waits; + } + fs::path root; std::vector disks; + Poco::AutoPtr stale_mount_waits; + Poco::AutoPtr previous_pool_channel; + int previous_pool_level = 0; }; } /// Each stale mount must be waited out once, but the waits of different disks are independent. -/// Waiting them one after another makes server start N × the observation window, which is what -/// missed the start timeout with three stale mounts. +/// Waiting them one after another makes server start N × the observation window. TEST_F(CASStartupStaleMounts, StaleMountsOfDifferentDisksAreWaitedOutConcurrently) { constexpr size_t disk_count = 3; @@ -273,17 +303,7 @@ TEST_F(CASStartupStaleMounts, DiskLookupDoesNotWaitForAStaleMountObservation) { const StaleCasDisk & stale = addStaleDisk("0"); ASSERT_FALSE(HasFatalFailure()); - - Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); - Poco::AutoPtr previous_channel(pool_logger.getChannel(), /*shared=*/true); - const int previous_level = pool_logger.getLevel(); - Poco::AutoPtr channel(new StaleMountWaitChannel); - pool_logger.setChannel(channel); - pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); - SCOPE_EXIT({ - pool_logger.setChannel(previous_channel); - pool_logger.setLevel(previous_level); - }); + StaleMountWaitChannel & waits = watchStaleMountWaits(); std::exception_ptr error; std::thread loader([&] @@ -298,13 +318,18 @@ TEST_F(CASStartupStaleMounts, DiskLookupDoesNotWaitForAStaleMountObservation) } }); - ASSERT_TRUE(channel->waitForWaitStarted(1)) << "the stale mount was never observed"; - const auto lookup_start = std::chrono::steady_clock::now(); - (void)getContext().context->getDisksMap(); - const uint64_t lookup_ms = elapsedMs(lookup_start); + const bool wait_started = waits.waitForWaitStarted(1); + uint64_t lookup_ms = 0; + if (wait_started) + { + const auto lookup_start = std::chrono::steady_clock::now(); + (void)getContext().context->getDisksMap(); + lookup_ms = elapsedMs(lookup_start); + } loader.join(); ASSERT_FALSE(error) << DB::getExceptionMessage(error, false); + ASSERT_TRUE(wait_started) << "the stale mount was never observed"; EXPECT_LT(lookup_ms, observation_ms / 3) << "a disk lookup took " << lookup_ms << " ms while a stale mount was being observed for up to " << observation_ms << " ms"; @@ -316,17 +341,7 @@ TEST_F(CASStartupStaleMounts, ConcurrentCreationsOfOneDiskShareOneOpen) { const StaleCasDisk & stale = addStaleDisk("0"); ASSERT_FALSE(HasFatalFailure()); - - Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); - Poco::AutoPtr previous_channel(pool_logger.getChannel(), /*shared=*/true); - const int previous_level = pool_logger.getLevel(); - Poco::AutoPtr channel(new StaleMountWaitChannel); - pool_logger.setChannel(channel); - pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); - SCOPE_EXIT({ - pool_logger.setChannel(previous_channel); - pool_logger.setLevel(previous_level); - }); + StaleMountWaitChannel & waits = watchStaleMountWaits(); const auto config = stale.xmlConfig(); std::atomic creator_calls = 0; @@ -353,7 +368,7 @@ TEST_F(CASStartupStaleMounts, ConcurrentCreationsOfOneDiskShareOneOpen) } }); /// Without this wait the second request could find the disk already created. - const bool first_is_waiting = channel->waitForWaitStarted(1); + const bool first_is_waiting = waits.waitForWaitStarted(1); std::thread second([&] { try @@ -379,7 +394,7 @@ TEST_F(CASStartupStaleMounts, ConcurrentCreationsOfOneDiskShareOneOpen) /// A failed creation is not remembered: the next request for the name creates the disk again. TEST_F(CASStartupStaleMounts, FailedDiskCreationIsRetriedByTheNextRequest) { - const String name = "cas_stale_failed_creation"; + const String name = uniqueDiskName("failed_creation"); std::vector calls; EXPECT_THROW( getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr @@ -398,3 +413,110 @@ TEST_F(CASStartupStaleMounts, FailedDiskCreationIsRetriedByTheNextRequest) DB::Exception); EXPECT_EQ(calls, (std::vector{"first", "second"})); } + +/// A request that arrives while another creation of the same name fails gets that failure without +/// running its own creator. Each caller gets its own copy of the exception, of the original type. +TEST_F(CASStartupStaleMounts, ConcurrentRequestsShareOneFailure) +{ + const String name = uniqueDiskName("shared_failure"); + std::mutex mutex; + std::condition_variable cv; + bool first_started = false; + bool second_returned = false; + std::atomic creator_calls = 0; + + const auto failing_creator = [&](const DB::DisksMap &) -> DB::DiskPtr + { + ++creator_calls; + std::unique_lock lock(mutex); + first_started = true; + cv.notify_all(); + /// Gives the second request time to join this creation. It returns early only if it did not join. + cv.wait_for(lock, std::chrono::seconds(2), [&] { return second_returned; }); + DB::ErrnoException::throwWithErrno(DB::ErrorCodes::BAD_ARGUMENTS, ENOSPC, "creation failed"); + }; + + struct Outcome + { + bool got_errno_exception = false; + int code = 0; + int saved_errno = 0; + String message; + }; + const auto request = [&](const String & caller, Outcome & outcome) + { + try + { + getContext().context->getOrCreateDisk(name, failing_creator); + } + catch (DB::ErrnoException & e) + { + outcome.got_errno_exception = true; + outcome.code = e.code(); + outcome.saved_errno = e.getErrno(); + e.addMessage("seen by " + caller); + outcome.message = e.message(); + } + catch (...) + { + outcome.message = DB::getCurrentExceptionMessage(false); + } + }; + + Outcome first_outcome; + Outcome second_outcome; + std::thread first([&] { request("first", first_outcome); }); + { + std::unique_lock lock(mutex); + cv.wait(lock, [&] { return first_started; }); + } + std::thread second([&] + { + request("second", second_outcome); + std::lock_guard lock(mutex); + second_returned = true; + cv.notify_all(); + }); + first.join(); + second.join(); + + EXPECT_EQ(creator_calls.load(), 1u); + for (const Outcome * outcome : {&first_outcome, &second_outcome}) + { + EXPECT_TRUE(outcome->got_errno_exception) << outcome->message; + EXPECT_EQ(outcome->code, DB::ErrorCodes::BAD_ARGUMENTS); + EXPECT_EQ(outcome->saved_errno, ENOSPC); + } + EXPECT_TRUE(first_outcome.message.contains("seen by first")) << first_outcome.message; + EXPECT_FALSE(first_outcome.message.contains("seen by second")) << first_outcome.message; + EXPECT_TRUE(second_outcome.message.contains("seen by second")) << second_outcome.message; + EXPECT_FALSE(second_outcome.message.contains("seen by first")) << second_outcome.message; + EXPECT_FALSE(getContext().context->getDisksMap().contains(name)); +} + +/// A configuration reload during a custom disk creation must not open a configured disk of the +/// same name: both could describe the same storage. +TEST_F(CASStartupStaleMounts, ConfigReloadSkipsADiskBeingCreated) +{ + const String name = uniqueDiskName("reload_race"); + const fs::path config_disk_path = root / "config_disk"; + const fs::path custom_disk_path = root / "custom_disk"; + fs::create_directories(config_disk_path); + fs::create_directories(custom_disk_path); + + std::istringstream xml( + "<" + name + ">local" + config_disk_path.string() + + "/"); + Poco::AutoPtr reloaded_config(new Poco::Util::XMLConfiguration(xml)); + + DB::DiskPtr created; + const auto disk = getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + getContext().context->updateStorageConfiguration(*reloaded_config); + created = std::make_shared(name, custom_disk_path.string() + "/"); + return created; + }); + + EXPECT_EQ(disk, created); + EXPECT_EQ(getContext().context->getDisk(name), created); +} diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 40e6d3f822ee4..8bc1d02070418 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -6781,17 +6781,22 @@ DiskPtr Context::getDisk(const String & name) const namespace { -/// Every waiter of one creation rethrows the same stored exception. Copy it, so a caller that -/// modifies its exception (e.g. `addMessage`) does not race with the other waiters. +/// All callers of one creation get the same stored exception. Each rethrows its own copy, so a +/// caller that modifies its exception (e.g. `addMessage`) does not race with the others. DiskPtr getDiskCreationResult(const std::shared_future & creation) { try { return creation.get(); } - catch (const Exception & e) + catch (const Poco::Exception & e) { - throw Exception(e); + /// Virtual, so the copy keeps the dynamic type (e.g. `ErrnoException`). + e.rethrow(); + } + catch (...) + { + throw Exception::createRuntime(getCurrentExceptionCode(), getCurrentExceptionMessage(/* with_stacktrace */ false)); } } @@ -6826,7 +6831,7 @@ DiskPtr Context::getOrCreateDisk(const String & name, DiskCreator creator) const if (!is_creator) return getDiskCreationResult(creation); - /// Creating a disk can take long (a CAS disk waits out a stale mount lease), so it runs without + /// A remote disk's startup can block on the network for a long time, so the creator runs without /// `storage_policies_mutex`. Concurrent requests for the same name wait for this creation rather /// than open the same storage a second time. DiskPtr disk; @@ -6847,12 +6852,7 @@ DiskPtr Context::getOrCreateDisk(const String & name, DiskCreator creator) const { try { - auto disk_selector = getDiskSelector(lock); - /// A config reload may have added a disk with this name meanwhile; it wins, as if it ran first. - if (auto existing = disk_selector->tryGet(name)) - disk = existing; - else - const_cast(disk_selector.get())->addToDiskMap(name, disk); + const_cast(getDiskSelector(lock).get())->addToDiskMap(name, disk); } catch (...) { @@ -6969,8 +6969,13 @@ void Context::updateStorageConfiguration(const Poco::Util::AbstractConfiguration std::lock_guard lock(shared->storage_policies_mutex); Strings disks_to_reinit; if (shared->merge_tree_disk_selector) - shared->merge_tree_disk_selector - = shared->merge_tree_disk_selector->updateFromConfig(config, "storage_configuration.disks", shared_from_this()); + { + NameSet disks_being_created; + for (const auto & [name, _] : shared->disks_being_created) + disks_being_created.insert(name); + shared->merge_tree_disk_selector = shared->merge_tree_disk_selector->updateFromConfig( + config, "storage_configuration.disks", shared_from_this(), disks_being_created); + } if (shared->merge_tree_storage_policy_selector) { diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h index 5a9943af651d8..9427c4b39198e 100644 --- a/src/Interpreters/Context.h +++ b/src/Interpreters/Context.h @@ -1685,6 +1685,10 @@ class Context: public ContextData, public std::enable_shared_from_this /// Provides storage disks DiskPtr getDisk(const String & name) const; using DiskCreator = std::function; + /// Returns the disk `name`, creating it with `creator` if it does not exist yet. The creator runs + /// without the storage lock and gets a snapshot of the disks. Concurrent calls for the same name + /// share one creation: they all get its disk or a copy of its exception. A failed creation is not + /// remembered. The creator must not request a disk of its own name. DiskPtr getOrCreateDisk(const String & name, DiskCreator creator) const; StoragePoliciesMap getPoliciesMap() const; From c905ab39e7a6c09d80a94afa422db2ade13a64c4 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Thu, 15 Oct 2026 11:46:50 +0200 Subject: [PATCH 4/4] Copy shared disk-creation exceptions with copyMutableException Use the existing `copyMutableException` helper: `Poco` exceptions are copied through their virtual `rethrow`, other exceptions are rethrown unchanged. State the `disks_to_skip` contract in `DiskSelector.h`, wrap a long log line, and bound the waits in the shared-failure test. Signed-off-by: Mikhail Filimonov Co-Authored-By: Claude Opus 5.5 --- src/Disks/DiskSelector.cpp | 5 ++++- src/Disks/DiskSelector.h | 4 ++-- src/Disks/tests/gtest_cas_startup_stale_mounts.cpp | 9 ++++++--- src/Interpreters/Context.cpp | 7 +------ 4 files changed, 13 insertions(+), 12 deletions(-) diff --git a/src/Disks/DiskSelector.cpp b/src/Disks/DiskSelector.cpp index 23c4cad14e37e..130286306ac0e 100644 --- a/src/Disks/DiskSelector.cpp +++ b/src/Disks/DiskSelector.cpp @@ -168,7 +168,10 @@ DiskSelectorPtr DiskSelector::updateFromConfig( { if (disks_to_skip.contains(disk_name)) { - LOG_WARNING(getLogger("DiskSelector"), "Disk {} is not created from the configuration: a custom disk with this name is being created", backQuote(disk_name)); + LOG_WARNING( + getLogger("DiskSelector"), + "Disk {} is not created from the configuration: a custom disk with this name is being created", + backQuote(disk_name)); continue; } auto created_disk = factory.create( diff --git a/src/Disks/DiskSelector.h b/src/Disks/DiskSelector.h index 066c5e30b4b54..f85820db1bfe9 100644 --- a/src/Disks/DiskSelector.h +++ b/src/Disks/DiskSelector.h @@ -29,8 +29,8 @@ class DiskSelector using DiskValidator = std::function; void initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator = {}); - /// Disks in `disks_to_skip` that are not created yet are left out: their names are being created - /// as custom disks, and creating them here would open the same storage twice. + /// Configured disks named in `disks_to_skip` are not created by this reload: their names are being + /// created as custom disks, and creating them here could open the same storage twice. DiskSelectorPtr updateFromConfig( const Poco::Util::AbstractConfiguration & config, const String & config_prefix, diff --git a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp index 51e47379a1039..34e6e9390987a 100644 --- a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp +++ b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp @@ -431,8 +431,9 @@ TEST_F(CASStartupStaleMounts, ConcurrentRequestsShareOneFailure) std::unique_lock lock(mutex); first_started = true; cv.notify_all(); - /// Gives the second request time to join this creation. It returns early only if it did not join. - cv.wait_for(lock, std::chrono::seconds(2), [&] { return second_returned; }); + /// Gives the second request time to join this creation; it returns early only if it did not join. + /// A late scheduler can only cause a false failure, never a false pass. + cv.wait_for(lock, std::chrono::seconds(5), [&] { return second_returned; }); DB::ErrnoException::throwWithErrno(DB::ErrorCodes::BAD_ARGUMENTS, ENOSPC, "creation failed"); }; @@ -466,9 +467,10 @@ TEST_F(CASStartupStaleMounts, ConcurrentRequestsShareOneFailure) Outcome first_outcome; Outcome second_outcome; std::thread first([&] { request("first", first_outcome); }); + bool first_is_creating = false; { std::unique_lock lock(mutex); - cv.wait(lock, [&] { return first_started; }); + first_is_creating = cv.wait_for(lock, std::chrono::seconds(30), [&] { return first_started; }); } std::thread second([&] { @@ -480,6 +482,7 @@ TEST_F(CASStartupStaleMounts, ConcurrentRequestsShareOneFailure) first.join(); second.join(); + ASSERT_TRUE(first_is_creating) << "the first creator never started"; EXPECT_EQ(creator_calls.load(), 1u); for (const Outcome * outcome : {&first_outcome, &second_outcome}) { diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 8bc1d02070418..2c411a8aa7bb6 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -6789,14 +6789,9 @@ DiskPtr getDiskCreationResult(const std::shared_future & creation) { return creation.get(); } - catch (const Poco::Exception & e) - { - /// Virtual, so the copy keeps the dynamic type (e.g. `ErrnoException`). - e.rethrow(); - } catch (...) { - throw Exception::createRuntime(getCurrentExceptionCode(), getCurrentExceptionMessage(/* with_stacktrace */ false)); + std::rethrow_exception(copyMutableException(std::current_exception())); } }