fix: e2e and post leave nothing behind

Makefile

@@ -324,8 +324,9 @@ PROXY_NETWORK_REUSE := $(filter $(NETWORK_REUSE_GOALS),$(or $(MAKECMDGOALS),help
 # Compose names the default network after the project, which is COMPOSE_PROJECT_NAME or this directory.
 PROXY_PROJECT := $(or $(strip $(COMPOSE_PROJECT_NAME)),$(notdir $(CURDIR)))
 # The certificate pair, the subnet and the network as Docker holds it now; the helper prints the first problem and nothing when there is none.
+# A helper that exits non-zero printed no problem because it checked nothing, so that exit becomes the problem and the start is refused.
 ifneq ($(PROXY_TLS_START)$(PROXY_NETWORK_REUSE),)
-    PROXY_TLS_PROBLEM := $(shell python3 scripts/proxy_tls.py precheck $(if $(PROXY_TLS_START),--certs --overlap) $(if $(PROXY_NETWORK_REUSE),--toggle) --cert-dir '$(DFE_PROXY_CERT_DIR)' --project '$(PROXY_PROJECT)' --subnet '$(strip $(DFE_NETWORK_SUBNET))' --tls $(if $(filter true,$(strip $(DFE_PROXY_TLS))),true,false))
+    PROXY_TLS_PROBLEM := $(shell python3 scripts/proxy_tls.py precheck $(if $(PROXY_TLS_START),--certs --overlap) $(if $(PROXY_NETWORK_REUSE),--toggle) --cert-dir '$(DFE_PROXY_CERT_DIR)' --project '$(PROXY_PROJECT)' --subnet '$(strip $(DFE_NETWORK_SUBNET))' --tls $(if $(filter true,$(strip $(DFE_PROXY_TLS))),true,false) || echo "scripts/proxy_tls.py precheck failed before it could check the console TLS setup, so the start is refused rather than assumed safe. Its error is printed above")
     ifneq ($(strip $(PROXY_TLS_PROBLEM)),)
         $(error $(PROXY_TLS_PROBLEM))
     endif

docs/developing.md

@@ -368,7 +368,7 @@ The stored rule has to scan `dfe.main` with no SQL errors and a WHERE that carri
 
 The side panel only offers values rows already hold, so both event sets are sent once before the search. The hunt's first window can reach those rows, so their near misses are judged too, and their matches are not.
 
-The console holds every page behind its setup wizard, so the test first makes the `e2e` organisation and one user of the deployment's own, as the wizard would. uv installs Playwright from the script's inline metadata, and the test drives Google Chrome (`uvx playwright install chrome` where it is absent). `DFE_E2E_HEADED=1` shows the browser, and `DFE_E2E_SHOTS_DIR` keeps the step screenshots, which otherwise go to a new temporary directory named in the log.
+The console holds every page behind its setup wizard, so the test first makes the `e2e` organisation and one user of the deployment's own (`e2e-analyst`), as the wizard would, and deletes whichever of the two it made once the hunt and the rule are gone. uv installs Playwright from the script's inline metadata, and the test drives Google Chrome (`uvx playwright install chrome` where it is absent). `DFE_E2E_HEADED=1` shows the browser, and `DFE_E2E_SHOTS_DIR` keeps the step screenshots, which otherwise go to a new temporary directory named in the log.
 
 Under console TLS the suite first verifies the console certificate itself, then pins Chrome to that certificate's public key, because Chrome takes no CA bundle.
 

docs/observability.md

@@ -207,7 +207,8 @@ Node metrics are still not collected.
   nothing.
 - **Routing**, where the engine renders the tier's app config. A source created
   through the API reaches the running receiver, and a record matching its rule
-  lands in a table of its own.
+  lands in a table of its own. The source is deleted afterwards and then that
+  table is dropped, because the engine's source delete leaves the table.
 
 Once every claim has run, the three ingest rows are deleted: a lightweight `DELETE`, or a mutation (`ALTER TABLE ... DELETE`) on a table whose projections refuse one. A delete that does not finish within 60s is reported, never failed.
 

scripts/post.py

@@ -64,6 +64,7 @@
 import functools
 import json
 import os
+import re
 import stat
 import subprocess
 import sys
@@ -300,8 +301,13 @@
 # built from those parts in `_routing_timeout_seconds`.
 ROUTING_SERVICES = ("dfe-engine", "dfe-receiver", "dfe-loader")
 ROUTING_SOURCE_PREFIX = "postroute"
+# The prefix and the last 8 hex digits of a `_run_id`.
+ROUTED_TABLE_NAME = re.compile(rf"{ROUTING_SOURCE_PREFIX}[0-9a-f]{{8}}")
 RECEIVER_CONFIG_POLL_SECONDS = 5.0
 ROUTING_INTERVAL_SECONDS = 3.0
+# The engine deletes a source's two topics before it answers, each delete allowed
+# 30s after a 10s topic listing, which a broker that does not answer uses in full.
+ROUTING_DELETE_TIMEOUT_SECONDS = 90
 # `.profile.mk`'s answer to "does the engine render this tier's app config": the
 # claim is only makeable where it does.
 APP_CONFIG_DIR_KEY = "DFE_ENGINE_APP_CONFIG_DIR"
@@ -1283,15 +1289,44 @@ def _verify_routing_applied(*, database: str) -> Claim:
     fault = _create_routed_source(base=base, name=name, token=token)
     if fault:
         _print(msg=f"FAIL  {fault}")
-        _remove(f"{base}/sources/{name}", kind="source", name=name, token=token)
+        _remove_routed_source(base=base, database=database, name=name, token=token)
         return Claim(asserted=True, failed=1)
 
     try:
         return Claim(
             asserted=True, failed=_await_routed_row(database=database, name=name)
         )
     finally:
-        _remove(f"{base}/sources/{name}", kind="source", name=name, token=token)
+        _remove_routed_source(base=base, database=database, name=name, token=token)
+
+
+def _remove_routed_source(*, base: str, database: str, name: str, token: str) -> None:
+    """Delete the routing claim's source, then the table its deploy created.
+
+    The engine's source delete takes the topics, the fetcher instance and the
+    receiver rule, and leaves the table. It is dropped only once the source is
+    gone, because a table pulled from under a live source fails its loads.
+    """
+    removed = _remove(
+        f"{base}/sources/{name}",
+        kind="source",
+        name=name,
+        token=token,
+        timeout=ROUTING_DELETE_TIMEOUT_SECONDS,
+    )
+    if not (removed):
+        _print(msg=f"      note: {database}.{name} is kept while its source remains")
+        return
+    # Only a name this claim makes, so the drop can never reach another table.
+    if not (ROUTED_TABLE_NAME.fullmatch(name)):
+        return
+    dropped, answer = ch_execute(
+        f"DROP TABLE IF EXISTS {database}.{name} SYNC",
+        timeout=CLEANUP_STATEMENT_TIMEOUT_SECONDS,
+    )
+    if not (dropped):
+        refusal = answer.splitlines()[0] if answer else "no answer"
+        _print(msg=f"      note: {database}.{name} was not dropped: {refusal}")
 
 
 def _routing_timeout_seconds(*, bus: bool) -> float:
@@ -1831,11 +1866,17 @@ def _create_hunt(
     return "" if status == 201 else f"POST /hunts returned HTTP {status}: {body}"
 
 
-def _remove(url: str, *, kind: str, name: str, token: str) -> None:
-    """Delete one thing this run created, and say so when it does not go."""
-    status = _api_delete(url, token=token)
-    if status not in (200, 204):
+def _remove(url: str, *, kind: str, name: str, token: str, timeout: int = 10) -> bool:
+    """Delete one thing this run created, say so when it does not go, and return whether it went."""
+    try:
+        status = _api_delete(url, token=token, timeout=timeout)
+    except (ApiUnreachable, OSError, subprocess.TimeoutExpired) as error:
+        _print(msg=f"      note: {kind} {name!r} was not removed: {error}")
+        return False
+    if status not in (200, 204, 404):
         _print(msg=f"      note: {kind} {name!r} was not removed (HTTP {status})")
+        return False
+    return True
 
 
 def _hunt_runner_alive(*, base: str, token: str) -> bool:

scripts/proxy_tls.py

@@ -103,15 +103,20 @@ def cert_problem(*, cert_dir: str) -> str:
         return f"{setting} must be a path starting with /, ./ or ../ because compose reads a bare name as a named volume"
     if not (Path(cert_dir).is_dir()):
         return f"{setting} is not a directory. It must hold {_CERT} and {_KEY}"
+    if not (os.access(cert_dir, os.X_OK)):
+        return f"{setting} exists but this user cannot open it. Under TLS both proxies run as the user who runs make, so that user must be able to read {_CERT} and {_KEY} there. Run make as the user who owns it rather than widening its mode"
     paths = {}
     for name, kind in ((_CERT, "certificate"), (_KEY, "private key")):
         shown = f"{cert_dir.rstrip('/')}/{name}"
         path = Path(cert_dir) / name
         paths[name] = path
-        if path.is_symlink():
-            return f"{shown} is a symlink, which dangles inside the container (certbot's live/ directory is all symlinks). Copy the file itself with `cp -L`"
-        if not (path.is_file()):
-            return f"DFE_PROXY_TLS=true needs the {kind} {shown} and it is missing. Generate the pair as docs/configuration.md (Console TLS) shows. Set DFE_PROXY_TLS=false to start without TLS"
+        try:
+            if path.is_symlink():
+                return f"{shown} is a symlink, which dangles inside the container (certbot's live/ directory is all symlinks). Copy the file itself with `cp -L`"
+            if not (path.is_file()):
+                return f"DFE_PROXY_TLS=true needs the {kind} {shown} and it is missing. Generate the pair as docs/configuration.md (Console TLS) shows. Set DFE_PROXY_TLS=false to start without TLS"
+        except OSError as error:
+            return f"{shown} cannot be checked ({error.strerror}), so the start is refused rather than assumed safe"
         if not (os.access(path, os.R_OK)):
             return f"{shown} exists but this user cannot read it. Under TLS both proxies run as the user who runs make, so the {kind} must be readable by that user. Keep the key 0600 and chown it rather than widening the mode"
     if shutil.which("openssl") is None:

scripts/test_e2e.py

@@ -2832,13 +2832,22 @@ def verify_detections(
 
 # ------------------------------------------------------------------------------
 # Remove Detection Objects
-# - Deletes what the test created, the hunt first because the engine refuses to
-#   delete a rule a hunt still names, and proves both are gone
-# ------------------------------------------------------------------------------
+# - Deletes what the test created in the order given, the hunt before the rule
+#   because the engine refuses to delete a rule a hunt still names, and proves
+#   each is gone
+# ------------------------------------------------------------------------------
+DETECTION_OBJECT_PATHS = {
+    "account": "/auth/accounts",
+    "hunt": "/hunts",
+    "organisation": "/orgs",
+    "rule": "/rules",
+}
+
+
 def remove_detection_objects(ctx, test, token, created):
     left = []
     for kind, name in created:
-        path = f"/{kind}s/{quote(name)}"
+        path = f"{DETECTION_OBJECT_PATHS[kind]}/{quote(name)}"
         deleted, _ = engine_call("DELETE", path, token)
         status, _ = engine_call("GET", path, token)
         if status != 404:
@@ -2945,16 +2954,20 @@ def run_detection(ctx, test, ingest_url):
 # Complete Setup
 # - The organisation the hunt runs for and a user of the deployment's own, which
 #   the console's setup wizard holds every other page back for
+# - Each one this run had to make goes to the front of `created`, so the test
+#   removes it after the hunt and the rule; one that already existed stays
 # ------------------------------------------------------------------------------
-def complete_setup(ctx, test, token):
+def complete_setup(ctx, test, token, created):
     org_path = f"/orgs/{quote(DETECTION_CUSTOMER)}"
     status, org = engine_call("GET", org_path, token)
     if status == 404:
         request = {
             "name": DETECTION_CUSTOMER,
             "display_name": DETECTION_CUSTOMER.upper(),
         }
-        engine_call("POST", "/orgs", token, request)
+        made, _ = engine_call("POST", "/orgs", token, request)
+        if made == 201:
+            created.insert(0, ("organisation", DETECTION_CUSTOMER))
         status, org = engine_call("GET", org_path, token)
     if status != 200 or not (isinstance(org, dict)):
         mark_fail(
@@ -2973,7 +2986,9 @@ def complete_setup(ctx, test, token):
             "password": secrets.token_urlsafe(24),
             "name": "E2E Analyst",
         }
-        engine_call("POST", "/auth/accounts", token, account)
+        made, _ = engine_call("POST", "/auth/accounts", token, account)
+        if made in (200, 201):
+            created.insert(0, ("account", SETUP_FIRST_USER))
         status, setup = engine_call("GET", "/auth/setup-status", token)
     initial = (setup or {}).get("initial_setup") or {}
     if status != 200 or initial.get("complete") is not True:
@@ -3136,7 +3151,7 @@ def run_search_detection(ctx, test, ingest_url, token):
     landed = None
     console = None
     try:
-        organisation = complete_setup(ctx, test, token)
+        organisation = complete_setup(ctx, test, token, created)
         if organisation is None or not (hyperdx_ready(ctx, test, token)):
             return
         print()

scripts/tests/test_engine_only_schema_control.py

@@ -91,6 +91,17 @@
     "scripts/tests/test_engine_only_schema_control.py",
 }
 
+# The self test's routing claim drops the throwaway table its own source
+# deployed, once that source is gone, because the engine's source delete leaves
+# the table. This one statement, and nothing else in post.py, is let through.
+ALLOWED_LINES = {
+    ("scripts/post.py", 'f"DROP TABLE IF EXISTS {database}.{name} SYNC",'),
+    (
+        "scripts/tests/test_post.py",
+        'assert statements == ["DROP TABLE IF EXISTS dfe.postroute1a2b3c4d SYNC"]',
+    ),
+}
+
 # Services whose start order the engine's schema pass governs: each reads a table
 # or a topic dfe-schemas declares and dfe-engine makes.
 GATED_SERVICES = (
@@ -198,7 +209,7 @@ def _offenders(pattern: re.Pattern[str]) -> list[str]:
         except (UnicodeDecodeError, OSError):
             continue
         for number, line in enumerate(body.splitlines(), start=1):
-            if pattern.search(line):
+            if pattern.search(line) and (rel, line.strip()) not in ALLOWED_LINES:
                 hits.append(f"{rel}:{number}: {line.strip()[:120]}")
     return hits
 

scripts/tests/test_ingest_send.py

@@ -25,11 +25,23 @@
 
 
 class _HoldingReceiver(BaseHTTPRequestHandler):
-    """Answers each POST after HOLD_SECONDS, 503 for a body asking to be refused."""
+    """Answers each POST after HOLD_SECONDS, 503 for a body asking to be refused.
+
+    It counts the requests it is holding at once, and keeps the most it held.
+    """
+
+    lock = threading.Lock()
+    holding = 0
+    most_held = 0
 
     def do_POST(self) -> None:
         body = self.rfile.read(int(self.headers["Content-Length"]))
+        with self.lock:
+            type(self).holding += 1
+            type(self).most_held = max(type(self).most_held, type(self).holding)
         time.sleep(HOLD_SECONDS)
+        with self.lock:
+            type(self).holding -= 1
         self.send_response(503 if b"refuse" in body else 200)
         self.end_headers()
 
@@ -39,6 +51,8 @@ def log_message(self, *args) -> None:
 
 @pytest.fixture
 def receiver_url():
+    _HoldingReceiver.holding = 0
+    _HoldingReceiver.most_held = 0
     server = ThreadingHTTPServer(("127.0.0.1", 0), _HoldingReceiver)
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
@@ -61,13 +75,14 @@ def test_one_sender_at_a_time_pays_the_hold_per_event(receiver_url: str) -> None
 
     assert statuses == [200] * EVENTS
     assert elapsed >= EVENTS * HOLD_SECONDS
+    assert _HoldingReceiver.most_held == 1
 
 
 def test_concurrent_senders_share_the_hold(receiver_url: str) -> None:
-    statuses, elapsed = _timed(receiver_url, workers=EVENTS)
+    statuses, _ = _timed(receiver_url, workers=EVENTS)
 
     assert statuses == [200] * EVENTS
-    assert elapsed < EVENTS * HOLD_SECONDS / 2
+    assert _HoldingReceiver.most_held > 1
 
 
 def test_each_answer_stays_with_its_own_request(receiver_url: str) -> None:

scripts/tests/test_make_apply.py

@@ -48,12 +48,16 @@
 
 @pytest.fixture
 def checkout(tmp_path: Path) -> Path:
-    """The Makefile, a stand-in resolver, and no .env."""
+    """The Makefile, a stand-in resolver, the network precheck, and no .env."""
     shutil.copy(REPO_ROOT / "Makefile", tmp_path / "Makefile")
     (tmp_path / "scripts").mkdir()
     (tmp_path / "scripts" / "resolve_profile.py").write_text(
         _RESOLVER, encoding="utf-8", newline="\n"
     )
+    # The goals that reuse a live network run it while make parses, and a precheck that cannot run refuses the start.
+    shutil.copy(
+        REPO_ROOT / "scripts" / "proxy_tls.py", tmp_path / "scripts" / "proxy_tls.py"
+    )
     return tmp_path
 
 

scripts/tests/test_post.py

@@ -1058,3 +1058,85 @@ def test_the_rows_are_deleted_after_the_console_reads_them_back(
     assert post.main() == 0
     assert calls[-1] == "_cleanup"
     assert calls.index("_verify_ui_query") < calls.index("_cleanup")
+
+
+def _stub_routed_removal(
+    monkeypatch: pytest.MonkeyPatch, delete
+) -> tuple[list[tuple[str, int]], list[str]]:
+    """Record each source DELETE (url, timeout) and each ClickHouse statement."""
+    deletes: list[tuple[str, int]] = []
+    statements: list[str] = []
+
+    def _delete(url: str, token: str = "", timeout: int = 10) -> int:
+        deletes.append((url, timeout))
+        return delete()
+
+    def _execute(sql: str, *, timeout: int) -> tuple[bool, str]:
+        statements.append(sql)
+        return True, ""
+
+    monkeypatch.setattr(post, "_api_delete", _delete)
+    monkeypatch.setattr(post, "ch_execute", _execute)
+    return deletes, statements
+
+
+@pytest.mark.parametrize("status", [204, 404])
+def test_the_routing_table_is_dropped_once_its_source_is_gone(
+    monkeypatch: pytest.MonkeyPatch, status: int
+) -> None:
+    deletes, statements = _stub_routed_removal(monkeypatch, lambda: status)
+
+    post._remove_routed_source(
+        base="http://engine/api/v1", database="dfe", name="postroute1a2b3c4d", token="t"
+    )
+
+    assert deletes == [
+        (
+            "http://engine/api/v1/sources/postroute1a2b3c4d",
+            post.ROUTING_DELETE_TIMEOUT_SECONDS,
+        )
+    ]
+    assert statements == ["DROP TABLE IF EXISTS dfe.postroute1a2b3c4d SYNC"]
+
+
+def test_the_routing_table_stays_while_its_source_remains(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
+) -> None:
+    _, statements = _stub_routed_removal(monkeypatch, lambda: 500)
+
+    post._remove_routed_source(
+        base="http://engine/api/v1", database="dfe", name="postroute1a2b3c4d", token="t"
+    )
+
+    assert statements == []
+    assert "dfe.postroute1a2b3c4d is kept while its source remains" in (
+        capsys.readouterr().err
+    )
+
+
+@pytest.mark.parametrize("name", ["main", "postroute", "postroute1a2b3c4d; x"])
+def test_a_table_this_claim_did_not_name_is_never_dropped(
+    monkeypatch: pytest.MonkeyPatch, name: str
+) -> None:
+    _, statements = _stub_routed_removal(monkeypatch, lambda: 204)
+
+    post._remove_routed_source(
+        base="http://engine/api/v1", database="dfe", name=name, token="t"
+    )
+
+    assert statements == []
+
+
+def test_a_source_delete_with_no_answer_keeps_the_table_and_does_not_raise(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def _unanswered() -> int:
+        raise post.ApiUnreachable("DELETE timed out")
+
+    _, statements = _stub_routed_removal(monkeypatch, _unanswered)
+
+    post._remove_routed_source(
+        base="http://engine/api/v1", database="dfe", name="postroute1a2b3c4d", token="t"
+    )
+
+    assert statements == []