fix: stored logs keep bad rows, erased accounts and unconsented queries out

Deep-review round 1 of E22.

- A NUL character (`GET /assets/%00` logs the decoded path) made Postgres
  refuse the row and with it the sink's whole batch, every user's rows
  included. `to_row` replaces NUL in every text column and in attrs.
- An erased account's lines kept arriving after purge_user's DELETE: the
  erasing request's own access line, a queued row, a background task.
  `logs.forget_user` runs first; later rows are stored without the account,
  and the check and the insert share a lock with the erasure.
- An admin setting the whole process to TRACE logged every user's query
  string. The query is now logged only under the user's own Teljes consent,
  and decoded before redaction, so `%40` no longer hides an address.
- The browser log ingest has an hourly storage budget per account (5 MiB,
  one notice when spent) and reads only a JSON Content-Type, so a
  same-site text/plain form post cannot write into a user's log.
  RateLimiter takes a per-attempt cost for the byte budget.
- The root level is applied under the elevation lock: two concurrent
  changes could leave it at the base while an elevation was active.
- Tests for the running sink (level gate, drain on stop) and for the
  admin log level applied at runtime and at boot.

backend/app/auth.py

@@ -248,7 +248,10 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
     # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table.
     db.execute(delete(User).where(User.id == user_id))
     # Stored log events name the account by id with no foreign key to cascade from (models.LogEvent),
-    # so they go explicitly — they are request history, i.e. personal data.
+    # so they go explicitly — they are request history, i.e. personal data. Forgotten FIRST: this very
+    # request's access line, a queued row, a later background task would otherwise put rows back
+    # under the id after this DELETE (logs.forget_user).
+    logs.forget_user(user_id)
     db.execute(delete(LogEvent).where(LogEvent.user_id == user_id))
     # Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the
     # canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose

backend/app/logs.py

@@ -40,6 +40,7 @@ import time
 import uuid
 from datetime import datetime, timezone
 from typing import Any
+from urllib.parse import parse_qsl
 
 from app.config import settings
 from app.ratelimit import RateLimiter, client_key
@@ -204,6 +205,14 @@ def redact_text(text: str) -> str:
     return _EMAIL_RE.sub(REDACTED_EMAIL, scrubbed)
 
 
+def redact_query(query: str) -> str:
+    """Scrub a raw query string. DECODED first: the text rules match what was typed, and a browser
+    sends it percent-encoded — `someone%40example.com` is an address `_EMAIL_RE` cannot see. Each
+    parameter also goes through the key rules, so `?new_password=` goes as `redact` would drop it."""
+    pairs = parse_qsl(query, keep_blank_values=True)
+    return "&".join(redact_text(f"{key}={redact(value, key=key)}") for key, value in pairs)
+
+
 def redact(value: Any, *, key: str = "") -> Any:
     """Recursively scrub a structure destined for the log.
 
@@ -457,13 +466,57 @@ def _apply_root_level(force: bool = False) -> None:
     if _base_level == logging.NOTSET:
         return  # not configured: the root keeps whatever the process set up
     now = datetime.now(timezone.utc)
+    # The level is SET under the lock too, not just computed: two threads that each computed a
+    # different answer would otherwise apply them in either order, and the older one could win —
+    # the root back at the base while an elevation is still active.
     with _elevations_lock:
         for user_id in [u for u, (_, until) in _elevations.items() if until <= now]:
             del _elevations[user_id]
         lowest = min([_base_level, *(lvl for lvl, _ in _elevations.values())])
-    if force or lowest != _root_applied:
-        logging.getLogger().setLevel(lowest)
-        _root_applied = lowest
+        if force or lowest != _root_applied:
+            logging.getLogger().setLevel(lowest)
+            _root_applied = lowest
+
+
+# --- erased accounts ------------------------------------------------------------------------
+# `auth.purge_user` deletes an account's `log_events` rows, but a line about that account can still
+# be on its way: the erasing request's own access line is logged AFTER the handler committed, a queued
+# row waits for the sink's writer, a background task logs later. Those rows are stored without the
+# account (`logsink` asks here). Remembered for an hour in this process — a line later than that has
+# no request left to come from, and the retention job is the backstop for any other process.
+ERASED_MEMORY_SECONDS = 3600.0
+_erased: dict[int, float] = {}
+_erased_lock = threading.Lock()
+# Held by whoever checks rows against `is_erased` AND writes them (`erasure_guard`). `forget_user`
+# takes it too, so an erasure waits for a write that already passed the check to commit — and the
+# purge's DELETE, which runs after `forget_user`, then removes those rows as well. Without it a row
+# checked a moment before the erasure could be inserted a moment after the DELETE.
+_erasure_guard = threading.Lock()
+
+
+def erasure_guard() -> threading.Lock:
+    """Hold this around "drop erased accounts from these rows, insert them, commit"."""
+    return _erasure_guard
+
+
+def forget_user(user_id: int) -> None:
+    """Mark `user_id` erased: no line logged about it from now on is stored under its id. Call BEFORE
+    deleting its rows (see `_erasure_guard`)."""
+    now = time.monotonic()
+    with _erasure_guard, _erased_lock:
+        for stale in [u for u, at in _erased.items() if now - at > ERASED_MEMORY_SECONDS]:
+            del _erased[stale]
+        _erased[user_id] = now
+    with _elevations_lock:
+        _elevations.pop(user_id, None)
+
+
+def is_erased(user_id: int | None) -> bool:
+    if user_id is None:
+        return False
+    with _erased_lock:
+        at = _erased.get(user_id)
+    return at is not None and time.monotonic() - at <= ERASED_MEMORY_SECONDS
 
 
 def current_source() -> str:
@@ -537,10 +590,10 @@ async def log_requests(request, call_next):
         "status": status,
         "duration_ms": duration_ms,
     }
-    # Teljes (TRACE, the user's or the whole process's): the query string too — redacted, and in
-    # attrs only, never in the message.
+    # Teljes (TRACE) the user consented to: the query string too — redacted, and in attrs only, never
+    # in the message.
     if request.url.query and _traced(scope):
-        attrs["query"] = redact_text(request.url.query)
+        attrs["query"] = redact_query(request.url.query)
     _request_log.log(
         level,
         "%s %s %s",
@@ -563,10 +616,10 @@ async def log_requests(request, call_next):
 
 
 def _traced(scope: _RequestScope) -> bool:
-    """Whether this request is logged at TRACE — its user's consent, or the process's base level."""
-    if scope.level is not None and scope.level <= TRACE:
-        return True
-    return _base_level != logging.NOTSET and _base_level <= TRACE
+    """Whether this request's query string is logged: only under its OWN user's Teljes consent. An
+    admin setting the whole process to TRACE gets more detail, not every user's search terms — the
+    privacy policy promises those are kept only for a user who agreed to it."""
+    return scope.level is not None and scope.level <= TRACE
 
 
 # Anonymous request lines kept per client per minute. Signed-in traffic is not capped: it is a known

backend/app/logsink.py

@@ -78,21 +78,47 @@ def should_store(record: logging.LogRecord) -> bool:
     return getattr(record, "event", None) == "http.request" and record.levelno >= logging.INFO
 
 
+def _without_nul(value: Any) -> Any:
+    """`value` with every NUL character replaced, in keys and nested values alike. Postgres refuses
+    NUL in text and `\\u0000` in JSONB, and a refused row fails the WHOLE batch it rides in — every
+    user's rows with it. Anything logged can carry one (`GET /assets/%00` logs the decoded path)."""
+    if isinstance(value, str):
+        return value.replace("\x00", "�")
+    if isinstance(value, dict):
+        return {_without_nul(k): _without_nul(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_without_nul(v) for v in value]
+    return value
+
+
 def to_row(fields: dict[str, Any]) -> dict[str, Any]:
     """`logs.record_fields` output → a `log_events` row: known columns only, text clipped to its
-    column, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout)."""
+    column and free of NUL, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout)."""
     row = {
         column: fields.get(column) for column in LogEvent.__table__.columns.keys() if column != "id"
     }
     for column, limit in _LIMITS.items():
         value = row.get(column)
-        if isinstance(value, str) and len(value) > limit:
-            row[column] = value[: limit - 1] + "…"
+        if isinstance(value, str):
+            value = _without_nul(value)
+            row[column] = value[: limit - 1] + "…" if len(value) > limit else value
     if row["attrs"] is not None:
-        row["attrs"] = json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str))
+        row["attrs"] = _without_nul(
+            json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str))
+        )
     return row
 
 
+def without_erased(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """The rows, with the account taken off any whose account was erased meanwhile
+    (`logs.forget_user`). Checked right before the INSERT, under `logs.erasure_guard()` held until
+    the commit: a row built before the erasure can still be waiting for its batch."""
+    for row in rows:
+        if logs.is_erased(row.get("user_id")):
+            row["user_id"] = None
+    return rows
+
+
 def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]:
     """A row the sink writes about ITSELF. Built directly, not logged: logging it would route it back
     through this sink (and the writer thread's own records are deliberately ignored)."""
@@ -220,8 +246,8 @@ class LogSink(logging.Handler):
         if not rows:
             return 0
         try:
-            with self._bind.begin() as conn:
-                conn.execute(insert(LogEvent), rows)
+            with logs.erasure_guard(), self._bind.begin() as conn:
+                conn.execute(insert(LogEvent), without_erased(rows))
         except Exception as exc:  # noqa: BLE001 — the database is down or refused the batch
             print(
                 f"log sink: could not store {len(rows)} log events: {logs.redact_text(str(exc))[:300]}",

backend/app/models.py

@@ -557,8 +557,9 @@ class LogEvent(Base):
 
     `user_id` has NO foreign key on purpose: the writer runs behind the event, so a line about an
     account can arrive after that account was purged, and a constraint would make it fail the whole
-    batch it travels in. `auth.purge_user` deletes an account's rows explicitly instead, and the
-    retention job (`log_retention_days`) bounds whatever a late write leaves behind."""
+    batch it travels in. `auth.purge_user` deletes an account's rows explicitly instead, after
+    `logs.forget_user` makes every later write in this process store such a line without the account;
+    the retention job (`log_retention_days`) bounds whatever another process writes late."""
 
     __tablename__ = "log_events"
     __table_args__ = (Index("ix_log_events_user_ts", "user_id", "ts"),)

backend/app/ratelimit.py

@@ -17,10 +17,14 @@ from app.config import settings
 
 
 class RateLimiter:
-    def __init__(self, max_events: int, window_seconds: float):
+    """At most `max_events` attempts per `window_seconds` per key — or, when each attempt passes a
+    `cost`, at most that much total cost (a byte budget counts bytes, not calls)."""
+
+    def __init__(self, max_events: float, window_seconds: float):
         self.max_events = max_events
         self.window = window_seconds
-        self._hits: dict[str, list[float]] = {}
+        # key → (monotonic time, cost) of every counted attempt still inside the window
+        self._hits: dict[str, list[tuple[float, float]]] = {}
         self._lock = threading.Lock()
 
     def reset(self) -> None:
@@ -36,27 +40,27 @@ class RateLimiter:
         count."""
         cutoff = time.monotonic() - self.window
         with self._lock:
-            return sum(1 for t in self._hits.get(key, []) if t > cutoff) >= self.max_events
+            return sum(c for t, c in self._hits.get(key, []) if t > cutoff) >= self.max_events
 
-    def allow(self, key: str) -> bool:
-        """Record an attempt for ``key`` and return whether it is within the limit.
+    def allow(self, key: str, cost: float = 1) -> bool:
+        """Record an attempt costing ``cost`` for ``key`` and return whether it is within the limit.
 
-        True  -> under the cap (the attempt is counted).
-        False -> the cap for the current window is already reached (attempt NOT counted, so a
+        True  -> within the cap (the attempt is counted).
+        False -> the attempt would pass the cap for the current window (attempt NOT counted, so a
                  blocked caller can't keep pushing the window forward)."""
         now = time.monotonic()
         cutoff = now - self.window
         with self._lock:
-            hits = [t for t in self._hits.get(key, []) if t > cutoff]
-            if len(hits) >= self.max_events:
+            hits = [(t, c) for t, c in self._hits.get(key, []) if t > cutoff]
+            if sum(c for _, c in hits) + cost > self.max_events:
                 self._hits[key] = hits
                 return False
-            hits.append(now)
+            hits.append((now, cost))
             self._hits[key] = hits
             # Opportunistic cleanup so abandoned keys don't accumulate unboundedly.
             if len(self._hits) > 4096:
                 for k in list(self._hits):
-                    fresh = [t for t in self._hits[k] if t > cutoff]
+                    fresh = [(t, c) for t, c in self._hits[k] if t > cutoff]
                     if fresh:
                         self._hits[k] = fresh
                     else:

backend/app/routes/diagnostics.py

@@ -7,8 +7,9 @@ the body), and the request id each event carries, which is what joins a browser
 line of the same request.
 
 Nothing from the body is trusted as-is:
-  * the whole body is capped before it is parsed, and so are the number of events and each event's
-    attributes;
+  * only a JSON Content-Type is read, and the whole body is capped before it is parsed, and so are
+    the number of events and each event's attributes;
+  * the pace is capped per account, and so is what an account may store per hour;
   * everything is redacted AGAIN with the server's own rules — the client redacts at capture, but a
     modified client would not;
   * an event below the consented level is dropped (a Részletes consent does not store what only
@@ -40,6 +41,13 @@ ATTRS_MAX_BYTES = 16 * 1024
 # The client flushes every few seconds and when its buffer fills; this is room for a burst, not a
 # pace anything legitimate keeps up for a minute.
 _batches = RateLimiter(max_events=30, window_seconds=60)
+# What one account may STORE per hour. The pace cap alone still lets a self-consented account write
+# ~15 MiB a minute into the one shared database, kept for the whole retention window; a real session
+# at Teljes stores a few hundred kB an hour. Past the budget a batch is accepted and not stored (a
+# refusal would only make the client resend it), and one notice per hour says so.
+STORED_BYTES_PER_HOUR = 5 * 1024 * 1024
+_stored_bytes = RateLimiter(max_events=STORED_BYTES_PER_HOUR, window_seconds=3600)
+_budget_notices = RateLimiter(max_events=1, window_seconds=3600)
 
 ClientLevel = Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
 
@@ -73,7 +81,14 @@ async def _capped_body(request: Request) -> bytes:
 
 async def _parsed_events(request: Request) -> ClientEventsIn:
     """The body as the model — parsed only after the size cap, which is why this is a dependency and
-    not a plain body parameter (FastAPI reads a declared body in full before anything else runs)."""
+    not a plain body parameter (FastAPI reads a declared body in full before anything else runs).
+
+    JSON only, by its Content-Type, as a declared body would be. Parsing any body as JSON would let a
+    CORS-simple `text/plain` form post from a same-site page write events into a signed-in user's
+    log; the app's own client always sends `application/json`, the unload beacon included."""
+    media_type = request.headers.get("content-type", "").split(";")[0].strip().lower()
+    if media_type != "application/json" and not media_type.endswith("+json"):
+        raise HTTPException(status_code=415, detail={"code": "json_required"})
     raw = await _capped_body(request)
     try:
         return ClientEventsIn.model_validate_json(raw)
@@ -129,7 +144,33 @@ def ingest_events(
         for item in body.events
         if logs.SEVERITY[item.level] >= floor
     ]
-    if rows:
-        db.execute(insert(LogEvent), rows)
+    if not rows:
+        return {"ok": True, "stored": 0}
+    size = sum(len(json.dumps(row, ensure_ascii=False, default=str).encode()) for row in rows)
+    stored = len(rows)
+    if not _stored_bytes.allow(str(user.id), size):
+        stored = 0
+        if not _budget_notices.allow(str(user.id)):
+            return {"ok": True, "stored": 0}
+        rows = [_budget_notice(user, now)]
+    # Under the erasure guard until the commit: an account erased meanwhile gets no rows back.
+    with logs.erasure_guard():
+        db.execute(insert(LogEvent), logsink.without_erased(rows))
         db.commit()
-    return {"ok": True, "stored": len(rows)}
+    return {"ok": True, "stored": stored}
+
+
+def _budget_notice(user: User, now: datetime) -> dict[str, Any]:
+    return logsink.to_row(
+        {
+            "ts": now,
+            "level": "WARNING",
+            "sev": logs.SEVERITY["WARNING"],
+            "source": "web",
+            "logger": "web",
+            "event": "diagnostics.budget_spent",
+            "msg": "The browser log passed its hourly storage budget; the rest of the hour is not kept",
+            "attrs": {"budget_bytes": STORED_BYTES_PER_HOUR},
+            "user_id": user.id,
+        }
+    )

backend/tests/test_diagnostics.py

@@ -13,6 +13,7 @@ Three parts:
 
 import json
 import logging
+import threading
 from datetime import datetime, timedelta, timezone
 
 import pytest
@@ -20,7 +21,8 @@ from fastapi import Depends, FastAPI
 from fastapi.testclient import TestClient
 from sqlalchemy import select
 
-from app import diagnostics, logs, logsink
+from app import diagnostics, logs, logsink, sysconfig
+from app.config import settings
 from app.models import LogEvent, User
 from app.routes import diagnostics as diagnostics_routes
 
@@ -235,6 +237,38 @@ def test_the_pace_is_capped_per_user(elevated):
     assert codes[:30] == [200] * 30 and codes[30] == 429
 
 
+def test_what_an_account_stores_per_hour_is_capped_and_said_once(elevated, monkeypatch):
+    user, client, rows = elevated("DEBUG")
+    monkeypatch.setattr(
+        diagnostics_routes, "_stored_bytes", diagnostics_routes.RateLimiter(3_000, 3600)
+    )
+    batch = {"events": [_event(msg="m" * 1_000)]}
+    stored = [client.post("/api/diagnostics/events", json=batch).json()["stored"] for _ in range(4)]
+    assert stored == [1, 1, 0, 0]
+    events = [row.event for row in rows()]
+    # Two batches, then ONE notice for the hour, not one per refused batch.
+    assert events == ["nav.view", "nav.view", "diagnostics.budget_spent"]
+    assert rows()[-1].user_id == user.id
+
+
+def test_only_a_json_body_is_read(elevated):
+    _user, client, rows = elevated("DEBUG")
+    # What a same-site `<form enctype="text/plain">` can send without a preflight.
+    r = client.post(
+        "/api/diagnostics/events",
+        content=json.dumps({"events": [_event()]}),
+        headers={"content-type": "text/plain"},
+    )
+    assert r.status_code == 415 and r.json()["detail"]["code"] == "json_required"
+    assert rows() == []
+    ok = client.post(
+        "/api/diagnostics/events",
+        content=json.dumps({"events": [_event()]}),
+        headers={"content-type": "application/json; charset=utf-8"},
+    )
+    assert ok.json()["stored"] == 1
+
+
 # --- the server side of an elevation ---------------------------------------------------------
 
 
@@ -330,6 +364,82 @@ def test_the_query_string_is_logged_only_for_a_traced_request_and_redacted(gated
     assert "q=lofi" in lines[1].attrs["query"] and "SECRET" not in lines[1].attrs["query"]
 
 
+def test_an_address_typed_into_a_search_is_redacted_though_the_browser_encodes_it(gated):
+    _app().get("/api/thing/7?level=TRACE&q=someone%40example.com&new_password=hunter2")
+    (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
+    assert line.attrs["query"] == "level=TRACE&q=[email]&new_password=[redacted]"
+
+
+def test_an_admin_level_of_trace_does_not_log_other_users_query_strings(gated):
+    logs.set_level("TRACE")
+    _app().get("/api/thing/7", params={"q": "something private"})
+    (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
+    assert "query" not in line.attrs
+
+
+def test_the_admin_level_applies_to_the_running_process_when_set_and_when_reset(
+    gated, db, make_user, as_user
+):
+    admin = as_user(make_user(role="admin"))
+    r = admin.patch("/api/admin/config/log_level", json={"value": "DEBUG"})
+    assert r.status_code == 200, r.text
+    assert logs.current_level() == "DEBUG"
+    assert admin.delete("/api/admin/config/log_level").status_code == 200
+    assert logs.current_level() == settings.log_level
+
+
+def test_the_stored_admin_level_is_applied_at_boot(gated, db, monkeypatch):
+    import asyncio
+
+    import app.main as app_main
+
+    sysconfig.set_value(db, "log_level", "DEBUG")
+    db.commit()
+    monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
+    monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
+    monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
+    monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
+
+    async def _no_sweep():
+        return None
+
+    monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
+    levels: list[str] = []
+
+    async def _cycle():
+        async with app_main.lifespan(app_main.app):
+            levels.append(logs.current_level())
+
+    asyncio.run(_cycle())
+    assert levels == ["DEBUG"]
+
+
+def test_the_root_level_is_applied_in_the_order_the_elevations_changed(gated, monkeypatch):
+    """Two requests change the elevations at once: A's user leaves (the root may go back to INFO),
+    B's user arrives (it must go to DEBUG). B's change is the later one, so DEBUG must be what stays
+    — even when A is paused between working out its level and applying it."""
+    root = logging.getLogger()
+    logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1))
+    real_set = root.setLevel
+    other: list[threading.Thread] = []
+
+    def paused_set(level):
+        if level == logging.INFO and not other:
+            b = threading.Thread(
+                target=logs._track_elevation,
+                args=(2, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)),
+            )
+            other.append(b)
+            b.start()
+            b.join(0.3)  # B runs to completion here unless A's level change holds it off
+        real_set(level)
+
+    monkeypatch.setattr(root, "setLevel", paused_set)
+    logs._track_elevation(1, None, None)  # A
+    other[0].join(5)
+    assert root.level == logging.DEBUG and 2 in logs._elevations
+
+
 def test_a_real_sign_in_carries_the_consented_level_into_the_process(gated, client, db, make_user):
     user = _signed_in(client, make_user)
     assert client.put("/api/me/diagnostics", json={"level": "DEBUG"}).status_code == 200

backend/tests/test_logs.py

@@ -102,7 +102,7 @@ def test_an_email_in_an_attribute_value_is_scrubbed_too():
     assert logs.redact({"to": "a.b+tag@mail.example.org"})["to"] == logs.REDACTED_EMAIL
 
 
-# The SAME cases frontend/src/lib/log.redaction.test.ts runs against the browser's copy of redact():
+# The SAME cases frontend/src/lib/log.contract.test.ts runs against the browser's copy of redact():
 # the one thing that keeps the two logs' idea of "safe to keep" from drifting apart.
 _REDACTION_CASES = json.loads(
     (Path(__file__).parent / "fixtures" / "log_redaction_cases.json").read_text(encoding="utf-8")

backend/tests/test_logsink.py

@@ -12,7 +12,7 @@ from datetime import datetime, timedelta, timezone
 
 import pytest
 from fastapi import BackgroundTasks
-from sqlalchemy import create_engine, select
+from sqlalchemy import create_engine, func, select
 
 from app import logs, logsink, scheduler
 from app.auth import purge_user
@@ -173,3 +173,75 @@ def test_deleting_an_account_deletes_its_stored_log_events(db, make_user):
     db.commit()
     purge_user(db, gone, BackgroundTasks())
     assert [r.user_id for r in db.execute(select(LogEvent)).scalars()] == [kept.id]
+
+
+def test_an_erased_accounts_late_lines_are_stored_without_it(stored, db, make_user):
+    """The DELETE in purge_user cannot reach a row still on its way: one queued before the erasure,
+    or the erasing request's own access line, logged after the handler committed."""
+    _sink, log, rows = stored
+    user = make_user()
+    try:
+        log.warning("queued before the erasure", extra={"user_id": user.id})
+        purge_user(db, user, BackgroundTasks())
+        log.info(
+            "DELETE /api/me/account 200", extra={**logs.event("http.request"), "user_id": user.id}
+        )
+        assert [(r.msg, r.user_id) for r in rows()] == [
+            ("queued before the erasure", None),
+            ("DELETE /api/me/account 200", None),
+        ]
+    finally:
+        logs._erased.clear()
+
+
+def test_a_nul_character_costs_nothing_but_itself(stored):
+    """Postgres refuses NUL in text and JSONB, and a refused row fails its whole batch — every other
+    user's rows with it. `GET /assets/%00` logs exactly that."""
+    _sink, log, rows = stored
+    log.warning("before")
+    log.warning("GET /assets/\x00 404", extra=logs.event("http.request", path="/assets/\x00"))
+    log.warning("after")
+    before, nul, after = rows()
+    assert (before.msg, after.msg) == ("before", "after")
+    assert nul.msg == "GET /assets/� 404" and nul.attrs["path"] == "/assets/�"
+
+
+def test_the_writer_drains_what_was_queued_when_it_is_stopped(db):
+    sink = logsink.LogSink(engine)
+    log = logging.getLogger("siftlode.sinktest.drain")
+    log.addHandler(sink)
+    try:
+        for i in range(logsink.BATCH_MAX + 5):
+            log.warning(f"line {i}")  # distinct templates: no flood limiter involved
+    finally:
+        log.removeHandler(sink)
+    sink._stopping.set()
+    sink._run()  # stopped before its first wait: everything goes through the drain
+    assert (
+        db.execute(select(func.count()).select_from(LogEvent)).scalar_one() == logsink.BATCH_MAX + 5
+    )
+
+
+def test_the_running_sink_keeps_the_base_level_for_everyone_the_elevation_is_not_for(
+    db, monkeypatch
+):
+    """`logsink.start()` as production runs it. One user's elevation lowers the ROOT logger, so an
+    ordinary request line below the base level reaches every handler — the sink's own level gate is
+    what keeps it out of the table, as stdout's keeps it off the terminal."""
+    root = logging.getLogger()
+    saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level)
+    monkeypatch.setattr(logsink.settings, "log_db_sink", True)
+    logs.set_level("WARNING")
+    logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1))
+    log = logging.getLogger("siftlode.sinktest.live")
+    try:
+        logsink.start()
+        log.info("GET /api/feed 200", extra={**logs.event("http.request"), "user_id": 2})
+        log.warning("a warning")
+    finally:
+        logsink.stop()
+        logs._base_level, logs._root_applied = saved[0], saved[2]
+        logs._elevations.clear()
+        logs._elevations.update(saved[1])
+        root.setLevel(saved[3])
+    assert [r.msg for r in db.execute(select(LogEvent)).scalars()] == ["a warning"]