diff --git a/CHANGELOG.md b/CHANGELOG.md
index f41523a..eb631b6 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/README.md b/README.md
index bb0435d..84ed368 100644
--- a/README.md
+++ b/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.
diff --git a/TODO.md b/TODO.md
index 5933264..25d3458 100644
--- a/TODO.md
+++ b/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()`.
diff --git a/deploy/kubernetes/README.md b/deploy/kubernetes/README.md
index f39a420..80c8dbc 100644
--- a/deploy/kubernetes/README.md
+++ b/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
diff --git a/s3gc.py b/s3gc.py
index 9f2a917..2b63a68 100644
--- a/s3gc.py
+++ b/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):
diff --git a/tests/dev_cluster_e2e_spec.md b/tests/dev_cluster_e2e_spec.md
new file mode 100644
index 0000000..329635e
--- /dev/null
+++ b/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.
diff --git a/tests/test_s3gc.py b/tests/test_s3gc.py
index f06141d..8021a44 100644
--- a/tests/test_s3gc.py
+++ b/tests/test_s3gc.py
@@ -1877,3 +1877,323 @@ def test_image_disables_stdout_buffering():
     dockerfile = (ROOT / "docker/Dockerfile").read_text()
 
     assert "ENV PYTHONUNBUFFERED=1" in dockerfile
+
+
+# --- Replicated-database replica preflight -----------------------------------
+#
+# Every DDL in a ClickHouse Replicated database waits for all registered
+# replicas. One dead replica turned the collect phase's first CREATE TABLE into
+# a 180 s wait ending in TIMEOUT_EXCEEDED (159), which the HTTP client surfaced
+# only as IncompleteRead. These pin the up-front check that replaces it.
+#
+# Liveness comes from system.clusters.is_active (the replica's Keeper session),
+# never from clusterAllReplicas: that connects to the other replicas as the
+# database's cluster_username, `default` unless configured, so a hardened
+# `default` user failed the check on a healthy database.
+
+
+class ReplicaLivenessCH:
+    def __init__(
+        self,
+        snapshots=(),
+        current_database="default",
+        engine="Replicated",
+        session_timeout_ms=30000,
+        liveness_error=None,
+        configured_cluster_rows=(),
+    ):
+        self.engine = engine
+        self.configured_cluster_rows = list(configured_cluster_rows)
+        self.snapshots = [list(snapshot) for snapshot in snapshots]
+        self.current_database = current_database
+        self.session_timeout_ms = session_timeout_ms
+        self.liveness_error = liveness_error
+        self.queries = []
+
+    def query(self, query):
+        self.queries.append(query)
+        if "currentDatabase()" in query:
+            return QueryResult(self.current_database)
+        if "system.databases" in query:
+            result = QueryResult(self.engine)
+            result.result_rows = [(self.engine,)] if self.engine else []
+            return result
+        if "system.zookeeper_connection" in query:
+            return QueryResult(self.session_timeout_ms)
+        if "system.clusters" in query:
+            if self.liveness_error:
+                raise self.liveness_error
+            result = QueryResult(None)
+            rows = self.snapshots.pop(0) if len(self.snapshots) > 1 else self.snapshots[0]
+            # A configured cluster named like the database shares its rows in
+            # system.clusters, but has no database replica name.
+            if "database_replica_name" not in query:
+                rows = rows + self.configured_cluster_rows
+            result.result_rows = rows
+            return result
+        raise AssertionError(query)
+
+
+BOTH_ACTIVE = [("replica-0.example", 9000, 1), ("replica-1.example", 9000, 1)]
+SECOND_INACTIVE = [("replica-0.example", 9000, 1), ("replica-1.example", 9000, 0)]
+SECOND_UNKNOWN = [("replica-0.example", 9000, 1), ("replica-1.example", 9000, None)]
+
+
+@pytest.fixture
+def sleeps(s3gc_module, monkeypatch):
+    recorded = []
+    monkeypatch.setattr(s3gc_module["time"], "sleep", recorded.append)
+    return recorded
+
+
+def _replica_preflight(s3gc_module, monkeypatch, ch, dbname="`gcdb`", margin_ms=1000, recheck=True):
+    namespace = s3gc_module["preflight_database_replicas"].__globals__
+    monkeypatch.setitem(namespace, "ch_client", ch)
+    monkeypatch.setitem(namespace, "dbname", dbname)
+    monkeypatch.setattr(namespace["args"], "replica_liveness_margin_ms", margin_ms)
+    monkeypatch.setattr(namespace["args"], "replica_liveness_recheck", recheck)
+    return s3gc_module["preflight_database_replicas"]
+
+
+def test_replica_liveness_margin_defaults_to_one_second(s3gc_module):
+    assert s3gc_module["args"].replica_liveness_margin_ms == 1000
+
+
+def test_replica_liveness_recheck_is_off_by_default(s3gc_module):
+    assert s3gc_module["args"].replica_liveness_recheck is False
+
+
+def test_replica_preflight_without_recheck_does_not_wait(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE])
+    _replica_preflight(s3gc_module, monkeypatch, ch, recheck=False)()
+
+    assert sleeps == []
+
+
+def test_replica_preflight_without_recheck_still_rejects_an_inactive_replica(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[SECOND_INACTIVE])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch, recheck=False)
+
+    with pytest.raises(s3gc_module["UserVisibleError"], match="replica-1.example:9000"):
+        preflight()
+
+
+def test_replica_preflight_ignores_a_same_named_configured_cluster(s3gc_module, monkeypatch, sleeps):
+    # Seen on 26.1: a configured cluster named like the Replicated database
+    # adds its own rows to system.clusters, with is_active NULL.
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE], configured_cluster_rows=[("localhost", 9000, None)])
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+
+@pytest.mark.parametrize(
+    "cli, env",
+    [
+        (["--replica-liveness-margin-ms", "-1"], {}),
+        ([], {"S3GC_REPLICA_LIVENESS_MARGIN_MS": "-1"}),
+    ],
+)
+def test_negative_replica_liveness_margin_is_rejected(cli, env):
+    # A negative margin would re-check before Keeper can expire a crashed
+    # replica's session, or hand time.sleep() a negative duration.
+    result = subprocess.run(
+        [sys.executable, str(ROOT / "s3gc.py"), *cli],
+        env={**os.environ, **env},
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode != 0
+    assert "replica_liveness_margin_ms" in result.stderr
+
+
+def test_replica_preflight_skips_a_non_replicated_database(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[[]], engine="Atomic")
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert not any("system.clusters" in query for query in ch.queries)
+
+
+def test_replica_preflight_does_not_wait_for_a_non_replicated_database(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[[]], engine="Atomic")
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert sleeps == []
+
+
+def test_replica_preflight_warns_when_a_replicated_database_lists_no_replicas(
+    s3gc_module, monkeypatch, caplog, sleeps
+):
+    # Nothing to check, so the run continues -- but say so, rather than let a
+    # missing system.clusters entry silently disable the check.
+    ch = ReplicaLivenessCH(snapshots=[[]])
+
+    with caplog.at_level("WARNING"):
+        _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert "lists no replicas" in caplog.text
+
+
+def test_replica_preflight_warns_when_second_liveness_read_has_no_replicas(
+    s3gc_module, monkeypatch, caplog, sleeps
+):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE, []])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch, recheck=True)
+
+    with caplog.at_level("INFO"):
+        preflight()
+
+    assert sleeps == [31]
+    assert "lists no replicas" in caplog.text
+    assert "preflight passed" not in caplog.text
+
+
+def test_replica_preflight_ignores_a_same_named_cluster_for_an_atomic_database(s3gc_module, monkeypatch, sleeps):
+    # system.clusters also lists configured clusters. An Atomic database that
+    # shares a cluster's name must not be checked: its DDL is local.
+    ch = ReplicaLivenessCH(snapshots=[SECOND_INACTIVE], engine="Atomic")
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert not any("system.clusters" in query for query in ch.queries)
+
+
+def test_replica_preflight_passes_when_every_replica_stays_active(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE])
+
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+
+def test_replica_preflight_never_connects_to_other_replicas(s3gc_module, monkeypatch, sleeps):
+    # Regression: clusterAllReplicas authenticated to the other replicas as
+    # `default`, failing a healthy database whose `default` user was hardened.
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE])
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert not any("clusterAllReplicas" in query for query in ch.queries)
+
+
+def test_replica_preflight_reads_replica_liveness_from_system_clusters(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE])
+    _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert any("is_active" in query and "system.clusters" in query for query in ch.queries)
+
+
+def test_replica_preflight_looks_up_the_auxiliary_tables_database(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE])
+    _replica_preflight(s3gc_module, monkeypatch, ch, dbname="`gcdb`")()
+
+    assert any("cluster = 'gcdb'" in query for query in ch.queries)
+
+
+def test_replica_preflight_defaults_to_the_current_database(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE], current_database="userdb")
+    _replica_preflight(s3gc_module, monkeypatch, ch, dbname=None)()
+
+    assert any("cluster = 'userdb'" in query for query in ch.queries)
+
+
+def test_replica_preflight_rejects_an_inactive_replica(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[SECOND_INACTIVE])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch)
+
+    with pytest.raises(s3gc_module["UserVisibleError"], match="replica-1.example:9000"):
+        preflight()
+
+
+def test_replica_preflight_rejects_an_inactive_replica_without_waiting(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[SECOND_INACTIVE])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch)
+
+    with pytest.raises(s3gc_module["UserVisibleError"]):
+        preflight()
+    assert sleeps == []
+
+
+def test_replica_preflight_error_does_not_name_an_active_replica(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[SECOND_INACTIVE])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch)
+
+    with pytest.raises(s3gc_module["UserVisibleError"]) as error:
+        preflight()
+    assert "replica-0.example" not in str(error.value)
+
+
+def test_replica_preflight_rejects_a_replica_of_unknown_liveness(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[SECOND_UNKNOWN])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch)
+
+    with pytest.raises(s3gc_module["UserVisibleError"], match="replica-1.example:9000"):
+        preflight()
+
+
+def test_replica_preflight_waits_the_keeper_session_timeout_plus_margin(s3gc_module, monkeypatch, sleeps):
+    # A replica that crashed moments ago stays is_active until Keeper expires
+    # its session, so look again only after that.
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE], session_timeout_ms=45000)
+    _replica_preflight(s3gc_module, monkeypatch, ch, margin_ms=1500)()
+
+    assert sleeps == [46.5]
+
+
+def test_replica_preflight_rejects_a_replica_that_expires_during_the_wait(s3gc_module, monkeypatch, sleeps):
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE, SECOND_INACTIVE])
+    preflight = _replica_preflight(s3gc_module, monkeypatch, ch)
+
+    with pytest.raises(s3gc_module["UserVisibleError"], match="replica-1.example:9000"):
+        preflight()
+
+
+def test_replica_preflight_warns_when_liveness_cannot_be_read(s3gc_module, monkeypatch, caplog, sleeps):
+    # An older server without is_active must keep the pre-check behaviour,
+    # not block every run -- but say the check was skipped.
+    from clickhouse_connect.driver.exceptions import DatabaseError
+
+    ch = ReplicaLivenessCH(snapshots=[BOTH_ACTIVE], liveness_error=DatabaseError("UNKNOWN_IDENTIFIER is_active"))
+
+    with caplog.at_level("WARNING"):
+        _replica_preflight(s3gc_module, monkeypatch, ch)()
+
+    assert "UNKNOWN_IDENTIFIER is_active" in caplog.text
+
+
+def _record_main_steps(s3gc_module, args_factory, monkeypatch, preflight_error=None):
+    namespace = s3gc_module["main"].__globals__
+    steps = []
+
+    def step(name, error=None):
+        def run(*_args, **_kwargs):
+            steps.append(name)
+            if error:
+                raise error
+        return run
+
+    monkeypatch.setitem(
+        namespace,
+        "args",
+        args_factory(usecollected_flag=False, collectonly_flag=True, debug_flag=False, dryrun_flag=False),
+    )
+    monkeypatch.setitem(namespace, "connect_to_ch", step("connect_to_ch"))
+    monkeypatch.setitem(namespace, "preflight_database_replicas", step("preflight_database_replicas", preflight_error))
+    monkeypatch.setitem(namespace, "init_run_log", step("init_run_log"))
+    monkeypatch.setitem(namespace, "connect_to_s3", step("connect_to_s3"))
+    monkeypatch.setitem(namespace, "do_collect", step("do_collect"))
+    monkeypatch.setitem(namespace, "graceful_exit", step("graceful_exit"))
+    return namespace["main"], steps
+
+
+def test_main_checks_database_replicas_before_any_ddl(s3gc_module, args_factory, monkeypatch):
+    main, steps = _record_main_steps(s3gc_module, args_factory, monkeypatch)
+    main()
+
+    assert steps[:3] == ["connect_to_ch", "preflight_database_replicas", "init_run_log"]
+
+
+def test_main_stops_before_any_ddl_when_a_replica_is_down(s3gc_module, args_factory, monkeypatch):
+    error = s3gc_module["UserVisibleError"]("replica down")
+    main, steps = _record_main_steps(s3gc_module, args_factory, monkeypatch, preflight_error=error)
+
+    with pytest.raises(SystemExit) as exit_info:
+        main()
+
+    assert (exit_info.value.code, steps) == (1, ["connect_to_ch", "preflight_database_replicas"])
diff --git a/tests/test_s3gc_dev_cluster.py b/tests/test_s3gc_dev_cluster.py
new file mode 100644
index 0000000..5b6cbc1
--- /dev/null
+++ b/tests/test_s3gc_dev_cluster.py
@@ -0,0 +1,253 @@
+"""Opt-in end-to-end coverage for a dedicated development cluster.
+
+Never run this against production: it creates and drops one uniquely named
+database and writes/deletes objects beneath one generated S3 prefix.
+"""
+
+import io
+import os
+import subprocess
+import uuid
+import xml.etree.ElementTree as ET
+from pathlib import Path
+
+import pytest
+import urllib3
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _connection(name):
+    """Read only the requested client connection; never include XML in errors."""
+    candidates = [
+        Path.home() / ".clickhouse-client" / "config.xml",
+        Path.home() / ".clickhouse-client" / "clickhouse-client.conf",
+        Path.home() / ".config" / "clickhouse-client" / "config.xml",
+    ]
+    for directory in (Path.home() / ".clickhouse-client" / "config.d",):
+        candidates.extend(sorted(directory.glob("*.xml")))
+    for path in candidates:
+        if not path.is_file():
+            continue
+        try:
+            root = ET.parse(path).getroot()
+        except (ET.ParseError, OSError):
+            continue
+        for node in root.iter("connection"):
+            values = {child.tag: (child.text or "").strip() for child in node}
+            if values.get("name") == name:
+                return values
+    pytest.fail("named ClickHouse connection was not found in standard client config")
+
+
+def _sql(client, sql):
+    try:
+        return client.command(sql)
+    except Exception as exc:
+        detail = str(exc)
+        password = os.getenv("S3GC_DEV_CH_PASSWORD", "")
+        if password:
+            detail = detail.replace(password, "<redacted>")
+        pytest.fail(f"ClickHouse setup/verification query failed: {detail[:1000]}")
+
+
+def _run_s3gc(args, env, password):
+    result = subprocess.run(
+        ["python3", str(ROOT / "s3gc.py"), *args],
+        text=True, capture_output=True, check=False, timeout=300, env=env,
+    )
+    output = result.stdout + result.stderr
+    if password:
+        output = output.replace(password, "<redacted>")
+    assert result.returncode == 0, f"s3gc failed with exit {result.returncode}:\n{output}"
+    return output
+
+
+def _env_for_endpoint(env, endpoint):
+    host, port = endpoint
+    phase_env = env.copy()
+    phase_env.update({"S3GC_CHHOST": host, "S3GC_CHPORT": str(port)})
+    return phase_env
+
+
+def _assert_s3_object_missing(client, bucket, key):
+    from minio.error import S3Error
+
+    try:
+        client.stat_object(bucket, key)
+    except S3Error as exc:
+        if exc.code == "NoSuchKey":
+            return
+        raise AssertionError(f"expected S3 NoSuchKey, got {exc.code!r}") from exc
+    raise AssertionError("S3 object is still present")
+
+
+def test_s3_object_absence_requires_no_such_key():
+    from minio.error import S3Error
+
+    class FakeS3:
+        def __init__(self, *, result=None, error_code=None, exception=None):
+            self.result = result
+            self.error_code = error_code
+            self.exception = exception
+
+        def stat_object(self, _bucket, _key):
+            if self.exception:
+                raise self.exception
+            if self.error_code:
+                raise S3Error(None, self.error_code, "fixture error", "", "", "")
+            return self.result
+
+    _assert_s3_object_missing(FakeS3(error_code="NoSuchKey"), "bucket", "missing")
+    with pytest.raises(AssertionError, match="AccessDenied"):
+        _assert_s3_object_missing(FakeS3(error_code="AccessDenied"), "bucket", "present")
+    with pytest.raises(AssertionError, match="still present"):
+        _assert_s3_object_missing(FakeS3(result=object()), "bucket", "present")
+    with pytest.raises(TimeoutError):
+        _assert_s3_object_missing(FakeS3(exception=TimeoutError("network timeout")), "bucket", "unknown")
+
+
+@pytest.mark.dev_cluster
+@pytest.mark.skipif(
+    os.getenv("S3GC_DEV_E2E") != "1",
+    reason="set S3GC_DEV_E2E=1 to opt in to the isolated dev-cluster test",
+)
+def test_replicated_inventory_collect_dryrun_delete_verify():
+    required = (
+        "S3GC_DEV_CH_CLUSTER", "S3GC_DEV_CH_HTTP_ENDPOINTS",
+        "S3GC_DEV_S3_BUCKET", "S3GC_DEV_S3_HOST", "S3GC_DEV_S3_REGION",
+    )
+    missing = [key for key in required if not os.getenv(key)]
+    if missing:
+        pytest.fail("missing required dev-test settings: " + ", ".join(missing))
+    if os.getenv("S3GC_DEV_DELETE_CONFIRMATION") != "DELETE_ORPHANS":
+        pytest.fail("set S3GC_DEV_DELETE_CONFIRMATION=DELETE_ORPHANS to enable fixture deletion")
+
+    from minio import Minio
+    from minio.credentials import IamAwsProvider
+    from clickhouse_connect import get_client
+
+    connection_name = os.getenv("S3GC_DEV_CH_CONNECTION", "")
+    if os.getenv("S3GC_DEV_CH_USER") is not None and os.getenv("S3GC_DEV_CH_PASSWORD") is not None:
+        connection = {
+            "user": os.environ["S3GC_DEV_CH_USER"],
+            "password": os.environ["S3GC_DEV_CH_PASSWORD"],
+        }
+    elif connection_name:
+        connection = _connection(connection_name)
+    else:
+        pytest.fail("provide a named client connection or Secret-backed CH_USER/CH_PASSWORD")
+    cluster = os.environ["S3GC_DEV_CH_CLUSTER"]
+    bucket = os.environ["S3GC_DEV_S3_BUCKET"]
+    host = os.environ["S3GC_DEV_S3_HOST"]
+    region = os.environ["S3GC_DEV_S3_REGION"]
+    endpoints = [item.strip() for item in os.environ["S3GC_DEV_CH_HTTP_ENDPOINTS"].split(",")]
+    parsed_endpoints = []
+    for endpoint in endpoints:
+        address, separator, port_text = endpoint.rpartition(":")
+        if not separator or not address:
+            pytest.fail("S3GC_DEV_CH_HTTP_ENDPOINTS must be comma-separated host:port values")
+        parsed_endpoints.append((address, int(port_text)))
+    assert len(parsed_endpoints) == 2 and len(set(parsed_endpoints)) == 2
+    username, password = connection.get("user", "default"), connection.get("password", "")
+
+    token = uuid.uuid4().hex
+    database = "s3gc_e2e_" + token[:20]
+    table = "inventory_s3"
+    prefix = f"s3gc-e2e/{token}/"
+    keys = [f"{prefix}fixture-{index:04d}.bin" for index in range(256)]
+    iam_http = urllib3.PoolManager(
+        timeout=urllib3.Timeout(connect=5, read=10),
+        retries=False,
+    )
+    iam_provider = IamAwsProvider(http_client=iam_http, region=region)
+    # Fail promptly when pytest is not running in a workload with AWS identity;
+    # the provider's default metadata HTTP client can otherwise wait for minutes.
+    iam_provider.retrieve()
+    client = Minio(
+        f"{host}:443", credentials=iam_provider, secure=True, region=region,
+        http_client=iam_http,
+    )
+    ch_clients = [get_client(host=host_, port=port, username=username, password=password)
+                  for host_, port in parsed_endpoints]
+    replica_names = {
+        ch.query("SELECT hostName()").result_rows[0][0]
+        for ch in ch_clients
+    }
+    assert len(replica_names) == 2, "HTTP endpoints must reach two distinct ClickHouse replicas"
+    env = os.environ.copy()
+    env.update({
+        "S3GC_CHHOST": parsed_endpoints[0][0], "S3GC_CHUSER": username, "S3GC_CHPASS": password,
+        "S3GC_CHPORT": str(parsed_endpoints[0][1]), "S3GC_S3IP": host, "S3GC_S3PORT": "443",
+        "S3GC_S3BUCKET": bucket, "S3GC_S3PATH": prefix, "S3GC_S3REGION": region,
+        "S3GC_S3AUTH": "iam", "S3GC_S3SECURE_FLAG": "true",
+        "S3GC_S3DISKNAME": "s3", "S3GC_CLUSTERNAME": cluster,
+        "S3GC_EXPECTED_REPLICAS": "2", "S3GC_COLLECTTABLEPREFIX": f"{database}.inventory_",
+        "S3GC_COLLECTDATABASE": database, "S3GC_SAMPLES": "1", "S3GC_RUNLOG_FLAG": "false",
+        "S3GC_KEEPDATA_FLAG": "true", "S3GC_USE_REMOVE_OBJECTS": "true",
+    })
+
+    created = []
+    database_created = False
+    try:
+        fixture = b"s3gc-dev-e2e-fixture"
+        for key in keys:
+            client.put_object(bucket, key, io.BytesIO(fixture), len(fixture),
+                              content_type="application/octet-stream")
+            created.append(key)
+
+        _sql(ch_clients[0], f"CREATE DATABASE `{database}` ON CLUSTER `{cluster}` ENGINE = Replicated")
+        database_created = True
+        # Replicated databases replicate their own DDL; ON CLUSTER is rejected
+        # for CREATE TABLE in this database engine.
+        _sql(ch_clients[0], f"CREATE TABLE `{database}`.`{table}` "
+             "(objpath String, size Int64, last_modified DateTime, active Bool) "
+             "ENGINE = ReplicatedReplacingMergeTree ORDER BY objpath PARTITION BY CRC32(objpath) % 1")
+        for ch in ch_clients:
+            _sql(ch, f"SYSTEM SYNC DATABASE REPLICA `{database}`")
+        first_replica_env = _env_for_endpoint(env, parsed_endpoints[0])
+        second_replica_env = _env_for_endpoint(env, parsed_endpoints[1])
+        _run_s3gc(["--collect-only", "--non-interactive"], first_replica_env, password)
+        _sql(ch_clients[0], f"SYSTEM SYNC REPLICA ON CLUSTER `{cluster}` `{database}`.`{table}` LIGHTWEIGHT")
+
+        # Backdate only this generated fixture inventory; keep s3gc's production
+        # minimum age guard (24h) intact for the dry-run and delete phases.
+        _sql(ch_clients[0], f"ALTER TABLE `{database}`.`{table}` UPDATE last_modified = toDateTime('2020-01-01 00:00:00') "
+             f"WHERE startsWith(objpath, '{prefix}') SETTINGS mutations_sync=2")
+        _sql(ch_clients[0], f"SYSTEM SYNC REPLICA ON CLUSTER `{cluster}` `{database}`.`{table}` LIGHTWEIGHT")
+
+        for ch in ch_clients:
+            rows = ch.query(f"SELECT count() FROM `{database}`.`{table}` WHERE startsWith(objpath, '{prefix}') SETTINGS final=1").result_rows[0][0]
+            assert rows == len(keys)
+
+        dryrun = _run_s3gc(
+            ["--use-collected", "--dry-run", "--debug", "--non-interactive"],
+            second_replica_env,
+            password,
+        )
+        candidates = {
+            line.split("would remove if no dryrun flag", 1)[1].split(" of size", 1)[0].strip()
+            for line in dryrun.splitlines() if "would remove if no dryrun flag" in line
+        }
+        assert candidates == set(keys)
+
+        _run_s3gc(["--use-collected", "--non-interactive"], second_replica_env, password)
+        _sql(ch_clients[0], f"SYSTEM SYNC REPLICA ON CLUSTER `{cluster}` `{database}`.`{table}` LIGHTWEIGHT")
+        for ch in ch_clients:
+            remaining = ch.query(f"SELECT count() FROM `{database}`.`{table}` WHERE startsWith(objpath, '{prefix}') AND active SETTINGS final=1").result_rows[0][0]
+            assert remaining == 0
+        for key in keys:
+            _assert_s3_object_missing(client, bucket, key)
+    finally:
+        try:
+            if database_created:
+                _sql(ch_clients[0], f"DROP DATABASE IF EXISTS `{database}` ON CLUSTER `{cluster}` SYNC")
+        finally:
+            for key in created:
+                try:
+                    client.remove_object(bucket, key)
+                except Exception:
+                    pass
+            for ch in ch_clients:
+                ch.close()