fix: re-check topology per sample, quote --useafter, unbuffer stdout

CHANGELOG.md

@@ -15,6 +15,27 @@ expose.
 
 ## [Unreleased]
 
+### Fixed
+
+- **Cluster topology is re-checked before every sample, not once per run.** The
+  preflight was a point-in-time check while the anti-join loop can run for
+  hours. A replica that dropped out mid-run took its references with it, so
+  blobs it alone held started looking orphaned — and nothing noticed. Now
+  re-verified before each sample and failed closed. `ch_client` is free at that
+  point: the previous sample's stream has closed.
+
+- **`--useafter` is quoted as a SQL string literal.** It was interpolated bare,
+  so an operator-supplied value landed as an identifier — the only unquoted
+  value in the anti-join `WHERE` clause. The strict `xfail` added when this was
+  recorded flipped to XPASS and became a real test, which is what it was for.
+
+- **The image sets `PYTHONUNBUFFERED=1`.** stdout is a pipe under Kubernetes so
+  `print()` was block-buffered, and Python's default SIGTERM handling exits
+  without flushing: a Job killed at `activeDeadlineSeconds` lost its buffered
+  tail, including the closing `s3gc: OK`, and the `dev-automation` shell echoes
+  interleaved wrongly against it. Logger records were never affected —
+  `StreamHandler.emit()` flushes per record.
+
 ### Changed
 
 - **`--useage` now has a hard floor of 24 hours, and defaults to 24** instead

TODO.md

@@ -13,15 +13,12 @@ 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.
-- [ ] Quote `--useafter` as a SQL string literal (strict `xfail` in the suite).
-- [ ] Consider re-checking cluster topology per sample, not once per run, so a
-  replica lost mid-run cannot widen the deletion scope.
-- [ ] `print()` output is block-buffered because the image sets no
-  `PYTHONUNBUFFERED` and stdout is a pipe under Kubernetes. Logger records are
-  flushed per line, but the bare prints — including the closing `s3gc: OK` —
-  are lost when `activeDeadlineSeconds` fires and the kubelet sends SIGTERM.
-  The durable run log now covers the evidence case; this remains a stdout
-  fidelity gap, and it also makes the `dev-automation` phase markers interleave
-  wrongly against the shell's unbuffered `echo`.
 - [ ] 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()`.
+  Since the per-sample re-check landed there are two call sites, and the
+  top-level one only buys failing about one query earlier. It also makes the
+  M7 mutation ("remove the preflight call") survive the suite, because removing
+  either site alone is covered by the other. Removing site A and retargeting M7
+  at the per-sample call restores a clean 11/11. Not a defect — removing both
+  sites still fails two tests.

docker/Dockerfile

@@ -2,6 +2,13 @@ FROM python:3.11-slim
 
 WORKDIR /app
 
+# stdout is a pipe under Kubernetes, so print() is block-buffered and Python's
+# default SIGTERM handling exits without flushing. A Job killed at
+# activeDeadlineSeconds therefore lost its buffered tail, including the closing
+# "s3gc: OK", and the shell echoes in the dev-automation phase interleaved
+# wrongly against it.
+ENV PYTHONUNBUFFERED=1
+
 COPY requirements.txt ./
 RUN pip install --no-cache-dir --disable-pip-version-check -r requirements.txt
 

s3gc.py

@@ -846,6 +846,17 @@ def current_phase():
     return "collect+use"
 
 
+def quote_sql_string(value):
+    """Render a Python string as a ClickHouse string literal.
+
+    --useafter was interpolated bare, so an operator-supplied value landed as an
+    identifier rather than a literal. Fail-closed in practice, but it was the
+    only unquoted value in the anti-join WHERE clause.
+    """
+    escaped = str(value).replace("\\", "\\\\").replace("'", "\\'")
+    return f"'{escaped}'"
+
+
 def _query_single_value(query):
     result = ch_client.query(query)
     if not result.result_rows or not result.result_rows[0]:
@@ -1272,7 +1283,9 @@ def do_use():
     check_samples_match_partitioning()
 
     def make_antijoin(calc_only=False, sample=None):
-        after_condition = f"AND s3o.objpath > {args.useafter} " if args.useafter else ""
+        after_condition = (
+            f"AND s3o.objpath > {quote_sql_string(args.useafter)} " if args.useafter else ""
+        )
         age_condition = f"AND s3o.last_modified < now() - interval {args.useage} hour " if args.useage else ""
         limit = f" LIMIT {args.usetotal} " if args.usetotal else ""
 
@@ -1325,6 +1338,14 @@ def make_antijoin(calc_only=False, sample=None):
     if not args.dryrun_flag and args.deletebatchsize < 1:
         raise ValueError("--deletebatchsize must be a positive integer")
     for sample in range(0, args.samples):
+        # Re-checked per sample, not once per run. The preflight above is a
+        # point-in-time check while this loop can run for hours; a replica that
+        # drops out mid-run would otherwise take its references with it and make
+        # blobs it alone holds look orphaned. ch_client is free here -- the
+        # previous sample's stream has closed.
+        if not args.dryrun_flag:
+            preflight_cluster()
+
         antijoin = make_antijoin(sample=sample)
         logger.info(f"antijoin {antijoin}")
         run_log(

tests/test_s3gc.py

@@ -1206,13 +1206,6 @@ def test_preflight_rejects_unexpected_replica_count(
         s3gc_module["preflight_cluster"]()
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="known defect: --useafter is interpolated unquoted, so the value "
-    "lands as a bare SQL identifier instead of a string literal "
-    "(s3gc.py, make_antijoin/after_condition). Fail-closed in practice, but "
-    "it is the only unquoted value in the WHERE clause.",
-)
 def test_useafter_is_quoted_as_a_string_literal(
     s3gc_module, args_factory, monkeypatch
 ):
@@ -1639,3 +1632,83 @@ def run(phase):
     dev_calls = run("dev-automation")
     assert dev_calls.count("--dev-allow-short-useage=true") == 2
     assert "--dev-allow-short-useage\n" not in dev_calls
+
+
+def test_useafter_escapes_quotes(s3gc_module, args_factory, monkeypatch):
+    """An object name containing a quote must not terminate the literal."""
+    sql = _antijoin_sql(
+        s3gc_module, args_factory, monkeypatch, useafter="odd'name"
+    )
+
+    assert r"s3o.objpath > 'odd\'name'" in sql
+
+
+def test_topology_is_rechecked_for_every_sample(
+    s3gc_module, args_factory, monkeypatch
+):
+    """The preflight is point-in-time; this loop can run for hours.
+
+    A replica that drops out mid-run takes its references with it, so blobs it
+    alone holds start looking orphaned. Re-check before each sample and fail
+    closed rather than delete against a shrunken reference scope.
+    """
+    namespace = s3gc_module["do_use"].__globals__
+
+    class ShrinkingCluster(FakeCH):
+        """Loses a replica after the first sample's preflight."""
+
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.preflights = 0
+
+        def query(self, query):
+            if "clusterAllReplicas" in query and "system.one" in query:
+                self.preflights += 1
+                return QueryResult(2 if self.preflights == 1 else 1)
+            return super().query(query)
+
+    class SuccessfulMinio:
+        def remove_objects(self, bucket, objects):
+            return iter(())
+
+    client = ShrinkingCluster(blocks=[])
+    monkeypatch.setitem(namespace, "args", args_factory(samples=2))
+    monkeypatch.setitem(namespace, "ch_client", client)
+    monkeypatch.setitem(namespace, "ch_writer", RecordingCH())
+    monkeypatch.setitem(namespace, "minio_client", SuccessfulMinio())
+
+    with pytest.raises(RuntimeError, match="replica preflight failed"):
+        s3gc_module["do_use"]()
+
+    # Once up front, then again before sample 0 -- the earliest re-check, which
+    # is where the shrunken topology is caught. The run never reaches sample 1.
+    assert client.preflights == 2
+
+
+def test_dry_run_does_not_require_a_cluster_preflight(
+    s3gc_module, args_factory, monkeypatch
+):
+    """Reading is safe; only the destructive path needs the topology check."""
+    namespace = s3gc_module["do_use"].__globals__
+
+    def refuse():
+        raise AssertionError("dry run must not require preflight")
+
+    monkeypatch.setitem(namespace, "args", args_factory(dryrun_flag=True, samples=2))
+    monkeypatch.setitem(namespace, "ch_client", FakeCH())
+    monkeypatch.setitem(namespace, "preflight_cluster", refuse)
+
+    s3gc_module["do_use"]()
+
+
+def test_image_disables_stdout_buffering():
+    """A Job killed at activeDeadlineSeconds used to lose its buffered tail.
+
+    stdout is a pipe under Kubernetes so print() is block-buffered, and Python's
+    default SIGTERM handling exits without flushing -- taking the closing
+    "s3gc: OK" with it, and interleaving the dev-automation shell echoes wrongly
+    against the Python output.
+    """
+    dockerfile = (ROOT / "docker/Dockerfile").read_text()
+
+    assert "ENV PYTHONUNBUFFERED=1" in dockerfile