fix: check Replicated-database replicas before the first DDL

CHANGELOG.md

@@ -30,6 +30,46 @@ expose.
   warning whenever the auxiliary table lived in a different database — a
   second, independent bug the same incident exposed.
 
+### Fixed
+
+- The optional Replicated-database liveness re-check now warns and skips the
+  preflight when its second replica lookup returns no rows. It no longer reports
+  that a zero-replica re-check passed.
+- **Every phase now checks, before any DDL, that all replicas of a Replicated
+  auxiliary database respond.** Every DDL in a database with
+  `ENGINE = Replicated` waits for all replicas registered in it. With one
+  registered replica down — its pod gone and its hostname no longer resolving —
+  the collect phase's first `CREATE TABLE IF NOT EXISTS` waited
+  `distributed_ddl_task_timeout` (180 s) and failed with `TIMEOUT_EXCEEDED`
+  (159), even though the tables already existed. The server had already sent
+  `200 OK`, so the HTTP client reported only `http.client.IncompleteRead`, and
+  the real cause was visible only in `system.query_log`. Each run lost about
+  six minutes (the run-log `CREATE`, then the auxiliary-table `CREATE`) before
+  crashing with a misleading traceback. `preflight_database_replicas()` now
+  runs right after the ClickHouse connection: it confirms in
+  `system.databases` that the database engine is `Replicated` (a configured
+  cluster that merely shares the database's name does not count), then reads
+  each registered replica's `is_active` from `system.clusters`. An inactive
+  or unknown replica exits 1 at once, with one message naming the database
+  and those replicas. Only rows with a `database_replica_name` count: a
+  configured cluster named like the database adds its own rows with
+  `is_active` NULL, which would otherwise block every phase. With
+  `S3GC_REPLICA_LIVENESS_RECHECK=true` the check also waits the Keeper
+  session timeout from `system.zookeeper_connection` plus
+  `S3GC_REPLICA_LIVENESS_MARGIN_MS` (default 1000 ms, never negative) and
+  reads `is_active` again: a replica that crashed just before the run stays
+  active until Keeper expires its session (30 s on the lab after `SIGKILL`).
+  The re-check is opt-in because it adds that wait to every run. A database that is not Replicated is skipped without
+  waiting; a server that cannot report liveness logs a warning and continues.
+  Liveness deliberately does not come from `clusterAllReplicas(<db>, ...)`:
+  ClickHouse runs that on the other replicas as the database's
+  `cluster_username`, `default` unless configured, not as the s3gc user. A
+  first version did, and on a lab whose `default` user had a password it
+  failed a healthy database with `default: Authentication failed` while the
+  DDL itself, which goes through Keeper, succeeded. Collect itself never
+  needed the other replicas — `INSERT` is not DDL — so the check guards only
+  the DDL the tool issues; the delete-time topology preflight is unchanged.
+
 ### Documentation
 
 - Reworked the Kubernetes guidance into a phase-based production runbook and a

README.md

@@ -21,6 +21,13 @@ Deleting an object is irreversible. `s3gc` preserves the following controls:
 - A delete Job needs the `DELETE_ORPHANS` confirmation token.
 - A clustered delete checks the configured cluster and expected replica count
   before it calls S3.
+- When the auxiliary database uses `ENGINE = Replicated`, every phase checks
+  that all of its replicas are active in Keeper before it runs any DDL. With
+  `S3GC_REPLICA_LIVENESS_RECHECK=true` it checks again after the Keeper
+  session timeout (about 30 s, plus `S3GC_REPLICA_LIVENESS_MARGIN_MS`), to
+  catch a replica that crashed just before the run. A down replica would otherwise stall
+  each `CREATE TABLE` for `distributed_ddl_task_timeout` and surface only as
+  `IncompleteRead`.
 - A failed delete Job does not retry automatically. Confirmed batches remain
   checkpointed, so a replacement Job can resume with the same collection-table
   prefix.

TODO.md

@@ -4,6 +4,12 @@ Possible, interesting, and deliberately deferred engineering or operational
 improvements. Completed behaviour belongs in `CHANGELOG.md`; this file is the
 forward-looking backlog.
 
+- [ ] In a separate documentation PR, document every CLI parameter: its purpose,
+  accepted values, default, aliases, exact `S3GC_*` environment variable name,
+  and interactions or constraints. Prefer existing kebab-case aliases in
+  examples, explain CLI/environment/configuration precedence, and include
+  examples for collect, dry-run, delete and verify. Keep existing option names
+  and aliases; a naming overhaul is unnecessary.
 - [ ] Add resumable `--collectonly` collection checkpoints. A crash currently
   restarts listing from the beginning; `--collectafter` would allow a large
   bucket collection to resume safely.
@@ -13,6 +19,11 @@ forward-looking backlog.
   fixtures, and remain excluded from CI.
 - [ ] Require the `Container / test` GitHub Actions check before pull-request
   merges in the repository branch-protection settings.
+- [ ] Surface a DDL `TIMEOUT_EXCEEDED` (159) instead of `IncompleteRead`. The
+  server sends `200 OK` before a distributed DDL times out, so the HTTP client
+  hides the cause. The Replicated-database preflight covers a replica that is
+  already down; a replica that dies after the preflight still surfaces as
+  `IncompleteRead` after `distributed_ddl_task_timeout`.
 - [ ] Consider a `TTL` on the run-log table. Volume is a few hundred rows per
   run so it is not urgent, but it grows without bound across many cleanups.
 - [ ] Drop the redundant top-level `preflight_cluster()` call in `do_use()`.

deploy/kubernetes/README.md

@@ -116,7 +116,10 @@ GRANT ON CLUSTER <cluster-name>
 
 The database grant covers the auxiliary inventory and durable run-log tables.
 `REMOTE` permits the `clusterAllReplicas` reads used by clustered dry-run and
-delete preflight. Do not grant ClickHouse's `S3` source privilege: s3gc uses
+delete preflight. When `<auxiliary-database>` uses `ENGINE = Replicated`, every
+phase, including collect, first checks that all of that database's replicas are
+active in Keeper (`S3GC_REPLICA_LIVENESS_RECHECK=true` adds a re-check after one
+Keeper session timeout, about 30 s); this reads only system tables and needs no extra grant. Do not grant ClickHouse's `S3` source privilege: s3gc uses
 its own S3 client and the Secret or workload identity instead.
 
 For `S3AUTH=static`, create the Secret from a private, shell-style credential

s3gc.py

@@ -34,6 +34,7 @@
 import logging
 import datetime
 import socket
+import time
 import uuid
 
 usage = """
@@ -325,6 +326,20 @@ def coerce_bool(value):
     type=Optional[int],
     help="Fail before deleting when clusterAllReplicas() does not return this many replicas",
 )
+parser.add_argument(
+    "--replica-liveness-margin-ms",
+    dest="replica_liveness_margin_ms",
+    type=int,
+    default=1000,
+    help="Extra wait beyond the Keeper session timeout before re-checking Replicated-database replica liveness",
+)
+parser.add_argument(
+    "--replica-liveness-recheck",
+    action="store_true",
+    dest="replica_liveness_recheck",
+    help="Re-check Replicated-database replica liveness after the Keeper session timeout, "
+    "to catch a replica that crashed just before the run",
+)
 parser.add_argument(
     "--age",
     "--hours",
@@ -578,6 +593,7 @@ def coerce_bool(value):
     "debug_flag",
     "silent_flag",
     "listoptions",
+    "replica_liveness_recheck",
 )
 
 for _dest in BOOLEAN_DESTS:
@@ -593,6 +609,12 @@ def coerce_bool(value):
             "use one of true/false, yes/no, on/off, 1/0"
         )
 
+if args.replica_liveness_margin_ms < 0:
+    parser.error(
+        f"invalid value {args.replica_liveness_margin_ms} for replica_liveness_margin_ms "
+        "(environment variable S3GC_REPLICA_LIVENESS_MARGIN_MS); it must not be negative"
+    )
+
 if args.listoptions:
     with redirect_stdout(StringIO()) as f:
         try:
@@ -923,6 +945,89 @@ class UserVisibleError(RuntimeError):
     """An operator-facing failure: reported without a traceback."""
 
 
+def _inactive_database_replicas(database):
+    replicas = ch_client.query(
+        "SELECT host_name, port, is_active FROM system.clusters "
+        f"WHERE cluster = {quote_sql_string(database)} AND database_replica_name != ''"
+    ).result_rows
+    return replicas, [f"{host}:{port}" for host, port, is_active in replicas if is_active != 1]
+
+
+def _raise_for_inactive_replicas(database, inactive):
+    if inactive:
+        raise UserVisibleError(
+            f"database replica preflight failed: Replicated database {database!r} has "
+            f"inactive or unknown replicas in Keeper: {', '.join(inactive)}. Every CREATE, "
+            "DROP and TRUNCATE in this database waits for all registered replicas and "
+            "would fail after distributed_ddl_task_timeout. Bring the missing replica back, "
+            "or remove its stale registration from the database, then rerun."
+        )
+
+
+def preflight_database_replicas():
+    """Fail fast when a replica of a Replicated database cannot take DDL.
+
+    Every DDL in a Replicated database waits for all registered replicas. One
+    dead replica turned the first CREATE TABLE IF NOT EXISTS into a 180 s
+    distributed_ddl_task_timeout and TIMEOUT_EXCEEDED, which the HTTP client
+    reported only as IncompleteRead. Checked before any DDL, so the operator
+    gets the cause in seconds instead.
+
+    Liveness is system.clusters.is_active, the replica's Keeper session. Not
+    clusterAllReplicas: that logs in to the other replicas as the database's
+    cluster_username (`default` unless configured), which a hardened `default`
+    user rejects even when every replica is healthy.
+    """
+    database = dbname.strip("`") if dbname else _query_single_value("SELECT currentDatabase()")
+    # system.clusters also lists configured clusters, so a matching name alone
+    # does not make a database Replicated. Check the engine first.
+    engines = ch_client.query(
+        f"SELECT engine FROM system.databases WHERE name = {quote_sql_string(database)}"
+    ).result_rows
+    if not engines or engines[0][0] != "Replicated":
+        logger.debug(f"database {database} is not Replicated; DDL does not wait on other replicas")
+        return
+
+    try:
+        replicas, inactive = _inactive_database_replicas(database)
+        if not replicas:
+            logger.warning(
+                f"Replicated database {database} lists no replicas in system.clusters; "
+                "skipping the database replica preflight"
+            )
+            return
+        _raise_for_inactive_replicas(database, inactive)
+        if not args.replica_liveness_recheck:
+            logger.info(f"database replica preflight passed: database={database}, replicas={len(replicas)}")
+            return
+
+        # A replica that crashed moments ago stays active until Keeper expires
+        # its session, so look again once that can have happened.
+        session_timeout_ms = _query_single_value("SELECT max(session_timeout_ms) FROM system.zookeeper_connection")
+        wait_seconds = (session_timeout_ms + args.replica_liveness_margin_ms) / 1000
+        logger.info(
+            f"database replica preflight: all {len(replicas)} replicas of {database} are active; "
+            f"re-checking in {wait_seconds} s, after the Keeper session timeout"
+        )
+        time.sleep(wait_seconds)
+        replicas, inactive = _inactive_database_replicas(database)
+        if not replicas:
+            logger.warning(
+                f"Replicated database {database} lists no replicas in system.clusters "
+                "after the liveness re-check; skipping the database replica preflight"
+            )
+            return
+    except clickhouse_connect.driver.exceptions.ClickHouseError as exc:
+        logger.warning(
+            f"cannot read replica liveness of Replicated database {database}; "
+            f"skipping the database replica preflight: {exc}"
+        )
+        return
+    _raise_for_inactive_replicas(database, inactive)
+
+    logger.info(f"database replica preflight passed: database={database}, replicas={len(replicas)}")
+
+
 def connect_to_ch():
     logger.info(
         f"Connecting to ClickHouse, host={args.chhost}, port={args.chport}, username={args.chuser}, password={args.chpass}, s3path={args.s3path}, bucket={args.s3bucket}, s3path={args.s3path}"
@@ -1509,6 +1614,7 @@ def make_antijoin(calc_only=False, sample=None):
 def main():
     try:
         connect_to_ch()
+        preflight_database_replicas()
         init_run_log()
         run_log("start", f"{current_phase()} phase started")
         if not (args.usecollected_flag and args.dryrun_flag):

tests/dev_cluster_e2e_spec.md

@@ -0,0 +1,20 @@
+# Dev-cluster E2E spec
+
+Add an opt-in test that validates s3gc's Replicated-database preflight and
+cross-replica inventory checkpoints against an isolated development ClickHouse
+cluster and S3 prefix. The test must never run in the default pytest suite.
+
+Acceptance criteria:
+
+- The test requires explicit opt-in, ClickHouse credentials from a named
+  client connection or Kubernetes Secret environment, two distinct replica
+  endpoints, the S3 bucket and region, and IAM-backed S3 access.
+- It creates a unique Replicated database and
+  `ReplicatedReplacingMergeTree` inventory, seeds only uniquely prefixed test
+  objects, collects on one replica, then runs dry-run and delete on the other.
+  Synchronization and verification check the inventory on both replicas.
+- It deletes only the seeded fixture keys, confirms the dry-run candidate set
+  matches them, verifies replicated tombstones and missing S3 keys, and cleans
+  up only its own database and objects on success or failure.
+- It remains marked `dev_cluster`, excluded from CI, and does not store or log
+  credentials or target-specific configuration.