Retry asynchronous table loading after transient CAS mount failures

src/Common/AsyncLoader.h

@@ -404,6 +404,13 @@ class AsyncLoader : private boost::noncopyable
     // prevent scheduling of new jobs.
     void shutdown();
 
+    /// Running jobs may use this to stop retrying before shutdown waits for them.
+    bool isShutdownRequested() const
+    {
+        std::lock_guard lock{mutex};
+        return shutdown_requested;
+    }
+
     // Wait for currently executing jobs to finish, but do not run any other pending jobs.
     // Not finished jobs are left in pending state:
     //  - they can be executed by calling unpause() again;

src/Databases/DatabaseOrdinary.cpp

@@ -15,6 +15,7 @@
 #include <Databases/DatabasesCommon.h>
 #include <Databases/TablesLoader.h>
 #include <Disks/DiskObjectStorage/DiskObjectStorage.h>
+#include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h>
 #include <IO/ReadBufferFromFile.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteBufferFromFile.h>
@@ -68,6 +69,7 @@ namespace ServerSetting
 
 namespace ErrorCodes
 {
+    extern const int ABORTED;
     extern const int LOGICAL_ERROR;
     extern const int UNKNOWN_DATABASE_ENGINE;
     extern const int NOT_IMPLEMENTED;
@@ -503,10 +505,33 @@ LoadTaskPtr DatabaseOrdinary::loadTableFromMetadataAsync(
         std::move(load_after),
         TablesLoaderBackgroundLoadPoolId,
         fmt::format("load table {}", name.getFullName()),
-        [this, local_context, file_path, name, ast, mode](AsyncLoader &, const LoadJobPtr &)
+        [this, local_context, file_path, name, ast, mode](AsyncLoader & loader, const LoadJobPtr &)
         {
             SCOPE_EXIT(TransactionLog::decreaseAsyncTablesLoadingJobNumber(););
-            loadTableFromMetadata(local_context, file_path, name, ast, mode);
+            bool retry_logged = false;
+            for (;;)
+            {
+                if (loading_stopped.load() || loader.isShutdownRequested())
+                    throw Exception(ErrorCodes::ABORTED, "Table loading stopped during CAS recovery");
+                try
+                {
+                    // Reconstruct after the failed attempt unwinds. Never restart
+                    // a partially initialized Storage or resume its old CAS fence.
+                    loadTableFromMetadata(local_context, file_path, name, ast, mode);
+                    return;
+                }
+                catch (const Cas::TransientStateException &)
+                {
+                    // Fresh admission still rejects terminal identity/decommission
+                    // states. Unrelated network and metadata errors are not retried.
+                    if (!retry_logged)
+                    {
+                        tryLogCurrentException(log);
+                        retry_logged = true;
+                    }
+                    sleepForSeconds(1);
+                }
+            }
         });
 
     return load_table[name.table] = makeLoadTask(async_loader, {job});
@@ -654,6 +679,7 @@ void DatabaseOrdinary::waitDatabaseStarted() const
 
 void DatabaseOrdinary::stopLoading()
 {
+    loading_stopped.store(true);
     std::unordered_map<String, LoadTaskPtr> stop_load_table;
     std::unordered_map<String, LoadTaskPtr> stop_startup_table;
     LoadTaskPtr stop_startup_database;

src/Databases/DatabaseOrdinary.h

@@ -102,6 +102,7 @@ class DatabaseOrdinary : public DatabaseOnDisk
     std::atomic<size_t> total_tables_to_startup{0};
     std::atomic<size_t> tables_started{0};
     AtomicStopwatch startup_watch;
+    std::atomic<bool> loading_stopped{false};
 
     DatabaseMetadataDiskSettings database_metadata_disk_settings;
     DiskPtr metadata_disk_ptr;

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

@@ -112,7 +112,7 @@ std::exception_ptr makeCasWriteRetryLaterExceptionPtr(const String & why)
     /// site: retry-later is right even where the condition may turn out terminal, since the next attempt
     /// re-decides against fresh state. Any promise about HOW the condition clears belongs in `condition`,
     /// where the site that can actually prove it makes it -- `checkFenceOrThrow` provably cannot.
-    throw Exception(ErrorCodes::NETWORK_ERROR,
+    throw TransientStateException(ErrorCodes::NETWORK_ERROR,
         "{} -- {}; TRANSIENT unavailability, not damage", subject, condition);
 }
 
@@ -346,7 +346,8 @@ namespace
                                    std::string_view when)
 {
     if (admit == Fence::Admit::NoBudget)
-        throwCasWriteRetryLater(fmt::format("{} of '{}': no lease budget {}", verb, subject, when));
+        throwCasTransientUnavailable(fmt::format("CAS {} of '{}'", verb, subject),
+                                     fmt::format("no lease budget {}", when));
     throwCasTransientUnavailable(fmt::format("CAS {} of '{}'", verb, subject),
                                  fmt::format("mount fence tripped {}", when));
 }

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

@@ -6,6 +6,7 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRetry.h>
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasWriteResult.h>
 #include <IO/ReadBuffer.h>
+#include <Common/Exception.h>
 #include <base/types.h>
 
 #include <cstdint>
@@ -80,6 +81,17 @@ bool isDeterministicLocalFailure(int code);
 /// SCOPE: only the ESCAPING retry-later throws route here (fence lost or a controlled write outcome
 /// remaining uncertain). Startup/decommission and generic live-lock-brake `ABORTED` values keep
 /// their meaning and are not rerouted here.
+/// Preserves NETWORK_ERROR for existing query/replication handling while allowing
+/// table construction to distinguish a transient CAS state from generic write errors.
+/// A fresh attempt must re-check the pool lifecycle; this does not override fencing.
+class TransientStateException final : public Exception
+{
+public:
+    using Exception::Exception;
+    TransientStateException * clone() const override { return new TransientStateException(*this); }
+    void rethrow() const override { throw *this; }
+};
+
 [[noreturn]] void throwCasWriteRetryLater(const String & why);
 
 /// Same classification as `throwCasWriteRetryLater`, but returns the exception as a
@@ -89,7 +101,7 @@ bool isDeterministicLocalFailure(int code);
 std::exception_ptr makeCasWriteRetryLaterExceptionPtr(const String & why);
 
 /// Throw the recoverable "this content-addressed disk cannot serve the request right now" condition.
-/// Sibling of `throwCasWriteRetryLater`, same class for the same reasons, differing only in what it
+/// Sibling of `throwCasWriteRetryLater`, same error code for the same reasons, differing in what it
 /// describes: that one names a WRITE whose commit did not land, this one names a DISK STATE that
 /// refused the request before it started -- on either plane.
 ///

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasRefLedger.cpp

@@ -769,10 +769,9 @@ void CasRefLedger::checkRecoveryStillAdmitted(const RootNamespace & ns, RefTable
     {
         cancelled = true;
         ProfileEvents::increment(ProfileEvents::CASRefRecoveryCancelled);
-        throwCasWriteRetryLater(fmt::format(
-            "CAS ref-table recovery for namespace '{}' was cancelled by a self-remount before the mount "
-            "fence was re-armed; the last attempt's fate is unresolved and nothing is installed — the next "
-            "touch recovers under the fresh incarnation", ns.string()));
+        throwCasTransientUnavailable(fmt::format("CAS ref-table recovery for namespace '{}'", ns.string()),
+            "cancelled by a self-remount before the mount fence was re-armed; the last attempt's fate is "
+            "unresolved and nothing is installed — the next touch recovers under the fresh incarnation");
     }
 
     if (rt.catalog_life_invalidated.load(std::memory_order_acquire))
@@ -785,9 +784,8 @@ void CasRefLedger::checkRecoveryStillAdmitted(const RootNamespace & ns, RefTable
     /// whatever it recovers belongs to a dead incarnation's cache. Checked separately from the
     /// cancellation because the two are independent facts, exactly as `resolveWedgeOnce` checks them.
     if (rt.superseded_by_remount.load(std::memory_order_acquire))
-        throwCasWriteRetryLater(fmt::format(
-            "CAS ref-table recovery for namespace '{}': this cached table was superseded by a self-remount "
-            "mid-recovery — retry against the fresh mount incarnation", ns.string()));
+        throwCasTransientUnavailable(fmt::format("CAS ref-table recovery for namespace '{}'", ns.string()),
+            "this cached table was superseded by a self-remount mid-recovery — retry against the fresh mount incarnation");
 
     /// The FENCE is deliberately NOT checked here: the walk's own `CasOperation` carries the admitted
     /// generation and refuses every request under a generation the fence has moved past, so a second
@@ -1569,10 +1567,8 @@ void CasRefLedger::ensureRefTableRecovered(
                         "before recovery install — nothing is installed",
                         ns.string(), renderIncarnation(rt.life.incarnation)));
                 if (rt.superseded_by_remount.load(std::memory_order_acquire))
-                    throwCasWriteRetryLater(fmt::format(
-                        "CAS ref-table recovery for namespace '{}': this cached table was superseded by a "
-                        "self-remount while the walk was in flight — nothing is installed",
-                        ns.string()));
+                    throwCasTransientUnavailable(fmt::format("CAS ref-table recovery for namespace '{}'", ns.string()),
+                        "this cached table was superseded by a self-remount while the walk was in flight — nothing is installed");
                 if (token && token->stopping())
                     throw Exception(ErrorCodes::ABORTED,
                         "CAS ref recovery: teardown began before recovery install");

tests/integration/test_cas_load_recovery/.gitignore

@@ -0,0 +1,2 @@
+.lab/
+__pycache__/

tests/integration/test_cas_load_recovery/README.md

@@ -0,0 +1,30 @@
+# Standalone CAS loading regression
+
+Run only in an isolated Linux Docker environment with synthetic data. Docker
+Compose creates private test storage, Keeper, writer, reader and fault proxies.
+No production credentials or endpoints are used. This is a standalone reproduction,
+not yet wired into the upstream pytest integration runner.
+
+```sh
+cd tests/integration/test_cas_load_recovery
+export CLICKHOUSE_IMAGE=your-locally-built-clickhouse-image
+python3 prepare.py
+docker compose up -d store keeper writer-proxy reader-proxy
+python3 lab.py init
+python3 lab.py rls --role writer
+python3 lab.py rls --role reader
+python3 lab.py loading --role writer --expect-recovery
+python3 lab.py loading --role reader --expect-recovery
+python3 lab.py loading --role writer --expect-recovery
+python3 lab.py shutdown --role writer
+docker compose down -v
+```
+
+Always run `docker compose down -v` after saving `.lab/results.jsonl`, including
+on test failure. For the unmodified control, start a fresh environment and run only
+`python3 lab.py loading --role writer` after `init`. It requires the cached
+`ASYNC_LOAD_WAIT_FAILED`/`NETWORK_ERROR` after storage recovery; destroy that
+environment before running another control. The failed writer cannot be reused.
+To check terminal refusal, run `python3 lab.py terminal --role writer` after a
+successful recovery, then destroy the environment (do not reuse it for shutdown).
+The standalone scenario was validated on ARM64 Linux; it is not automatically run by the upstream pytest runner.

tests/integration/test_cas_load_recovery/compose.yaml

@@ -0,0 +1,51 @@
+name: cas-load-recovery
+x-clickhouse: &clickhouse
+  image: ${CLICKHOUSE_IMAGE:-altinity/clickhouse-server:26.6.4.20001.altinityantalya@sha256:4403168d63f049700a979876a8075a8ae4d7ae6be7ea7e09f4fbc6a526e420fa}
+  mem_limit: 1536m
+  environment:
+    CLICKHOUSE_SKIP_USER_SETUP: '1'
+  stop_grace_period: 30s
+  depends_on: [keeper, writer-proxy, reader-proxy]
+x-proxy: &proxy
+  image: python:3.12-slim
+  command: [python3, /proxy.py]
+  environment:
+    RUSTFS_UPSTREAM: store:11121
+  volumes: [./.lab/s3_fault_proxy.py:/proxy.py:ro]
+  depends_on: [store]
+  mem_limit: 256m
+services:
+  store:
+    image: rustfs/rustfs:1.0.0-rc.3
+    command: server --address 0.0.0.0:11121
+    environment:
+      RUSTFS_ACCESS_KEY: clickhouse
+      RUSTFS_SECRET_KEY: clickhouse
+      RUSTFS_SCANNER_ENABLED: 'false'
+      RUSTFS_HEAL_ENABLED: 'false'
+    volumes: [store:/data]
+    mem_limit: 768m
+  writer-proxy:
+    <<: *proxy
+  reader-proxy:
+    <<: *proxy
+  keeper:
+    image: altinity/clickhouse-server:26.6.4.20001.altinityantalya@sha256:4403168d63f049700a979876a8075a8ae4d7ae6be7ea7e09f4fbc6a526e420fa
+    entrypoint: [/usr/bin/clickhouse, keeper]
+    command: [--config-file=/etc/keeper.xml]
+    volumes: [./.lab/keeper.xml:/etc/keeper.xml:ro, 'keeper:/var/lib/clickhouse-keeper']
+    mem_limit: 512m
+  writer:
+    <<: *clickhouse
+    volumes: [./.lab/writer.xml:/etc/clickhouse-server/config.d/lab.xml:ro, 'writer:/var/lib/clickhouse']
+  reader:
+    <<: *clickhouse
+    volumes: [./.lab/reader.xml:/etc/clickhouse-server/config.d/lab.xml:ro, 'reader:/var/lib/clickhouse']
+networks:
+  default:
+    internal: true
+volumes:
+  store:
+  keeper:
+  writer:
+  reader: