fix: correct a Kafka version downgrade and four live-only bugs found via full-stack verification

infra/terraform/local/main.tf

@@ -1109,7 +1109,16 @@ resource "kubernetes_stateful_set" "kafka" {
       spec {
         container {
           name  = "kafka"
-          image = "apache/kafka:3.9.0"
+          # Pinned to match whatever :latest had already drifted to and
+          # formatted the PVC's KRaft metadata with (verified: the cached
+          # apache/kafka:latest image is kafka_2.13-4.3.1.jar). KRaft's
+          # on-disk metadata format is forward-compatible only — pinning to
+          # an OLDER version than what already formatted the volume crashes
+          # every broker start with "No MetadataVersion with feature level
+          # N". If this version is ever bumped, wipe the PVC first
+          # (kubectl delete pvc data-kafka-0 -n sentinel-data) or pin to a
+          # version >= whatever last wrote to it.
+          image = "apache/kafka:4.3.1"
 
           env {
             name  = "KAFKA_NODE_ID"
@@ -1259,7 +1268,10 @@ resource "kubernetes_job_v1" "kafka_topic_init" {
         restart_policy = "OnFailure"
         container {
           name  = "kafka-topic-init"
-          image = "apache/kafka:3.9.0"
+          # Matches the broker's pinned version above — this only runs
+          # kafka-topics.sh as a client, but keeping both pins identical
+          # avoids a second version to track.
+          image = "apache/kafka:4.3.1"
           command = [
             "/bin/sh", "-c",
             "/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists --topic traces.raw --partitions 3 --replication-factor 1 && echo 'topic ready'"

pipelines/drift/db.py

@@ -141,25 +141,37 @@ def write_drift_stats(
 
 
 def get_active_model_version(database_url: str) -> str | None:
-    """Return the model_version model_registry considers current.
-
-    Mirrors services/classifier/db.py's get_active_model selection exactly
-    (prefer status='active', fall back to the most recent 'staging' entry) so
-    the drift job evaluates the same version a classifier pod would load on
-    startup. Reading from `classifications` instead (whichever version wrote
-    the most recent row) was a race during rolling restarts: old- and
-    new-version pods write concurrently, so "most recent row" doesn't mean
-    "the rollout's target version" — it could attribute drift_stats to a
-    version that's already being retired.
+    """Return the model_version that's actually appearing in classifications.
+
+    Live-verified gotcha this has to account for: model_registry holds two
+    different kinds of rows that don't share a model_version namespace.
+    services/classifier/db.py's get_active_model() picks a 'active'/'staging'
+    row to decide which MinIO model_path to download — but once loaded, the
+    classifier self-registers a *separate*, new 'staging' row under its own
+    freshly-derived model_version string (services/classifier/model.py's
+    `sentinel-roberta-{deployed_at}-{quant_tag}`), and THAT string — not the
+    row it downloaded from — is what ends up in classifications.model_version
+    on every write. An earlier version of this function mirrored
+    get_active_model()'s "prefer status='active'" ordering exactly, which
+    seemed principled but was wrong in practice: reproduced live against a
+    real cluster, the 'active' row was a stale promotion from a different
+    model_version string entirely (nothing in this project's current phase —
+    Airflow/Phase 7 doesn't exist yet — reliably keeps 'active' pointed at
+    what's actually running), so it returned a model_version with zero
+    matching classification rows and drift detection silently found nothing.
+    Ordering by created_at alone — "whichever pod self-registered most
+    recently" — is what's actually running and writing classifications.
+    Still scoped to non-retired rows; still subject to the same rolling-
+    restart race noted before (old- and new-version pods can both register
+    around the same moment), just no longer compounded by a status
+    preference that points at the wrong namespace entirely.
     """
     with _connect(database_url) as conn:
         row = conn.execute(
             """
             SELECT model_version FROM model_registry
             WHERE status IN ('active', 'staging')
-            ORDER BY
-                CASE status WHEN 'active' THEN 0 ELSE 1 END,
-                COALESCE(promoted_at, created_at) DESC
+            ORDER BY created_at DESC
             LIMIT 1
             """,
         ).fetchone()

scripts/dev-start.sh

@@ -137,8 +137,13 @@ kubectl exec -n sentinel-data kafka-0 -- \
 info "Syncing PostgreSQL password from secret..."
 PG_PASSWORD=$(kubectl get secret postgresql-credentials -n sentinel-data \
     -o jsonpath='{.data.password}' | base64 -d)
-if kubectl exec -n sentinel-data postgresql-0 -- \
-    psql -U sentinel -d sentinel -v pw="$PG_PASSWORD" -c "ALTER USER sentinel PASSWORD :'pw';"; then
+# -v/:'var' substitution only happens when psql reads SQL as a script
+# (stdin/-f) — NOT via -c, which bypasses that parser and sends the string
+# through unexpanded (verified live: -c left the literal text ":'pw'" in
+# the query, which Postgres then rejected as a syntax error). Piped via
+# stdin instead (kubectl exec -i to keep stdin open across the exec).
+if echo "ALTER USER sentinel PASSWORD :'pw';" | kubectl exec -i -n sentinel-data postgresql-0 -- \
+    psql -U sentinel -d sentinel -v pw="$PG_PASSWORD"; then
     info "PostgreSQL password synced"
 else
     warn "ALTER USER failed — local socket auth may not be trusted in this setup."
@@ -236,11 +241,17 @@ if [[ -z "${MODEL_PATH:-}" ]] && command -v psql >/dev/null 2>&1; then
 
         if [[ -n "$_minio_path" ]]; then
             info "Active registry entry is a host path — switching to MinIO path: $_minio_path"
-            psql "$DATABASE_URL" -v path="$_minio_path" -c "
+            # -v/:'var' substitution only happens when psql reads SQL as a
+            # script (stdin/-f) — NOT via -c, which bypasses that parser and
+            # sends the string through unexpanded (verified live: -c left
+            # the literal text ":'path'" in the query, which Postgres then
+            # rejected as a syntax error). Piped via stdin instead.
+            echo "
                 UPDATE model_registry SET status = 'retired'
                 WHERE model_path NOT LIKE 'models/%' AND status IN ('active','staging');
                 UPDATE model_registry SET status = 'active', promoted_at = NOW()
-                WHERE model_path = :'path';" >/dev/null 2>&1 || true
+                WHERE model_path = :'path';" \
+                | psql "$DATABASE_URL" -v path="$_minio_path" >/dev/null 2>&1 || true
         elif ! find "$_registry_path" -maxdepth 1 -name "*.onnx" 2>/dev/null | grep -q .; then
             warn "Registry has a stale local path (artifacts deleted): $_registry_path"
             psql "$DATABASE_URL" -c \

services/classifier/main.py

@@ -50,6 +50,7 @@ async def lifespan(app: FastAPI):
     # The registry is the source of truth for which model version to serve.
     # Open the pool first so we can query it before loading the model.
     model_dir: Path | None = None
+    active: dict | None = None
 
     if settings.database_url:
         try:
@@ -90,28 +91,39 @@ async def lifespan(app: FastAPI):
 
     # ── 3. Record this model version in the registry (idempotent) ─────────────
     # ON CONFLICT DO NOTHING: first pod registers it, subsequent pods skip.
-    # Skip registration when model_path is an absolute local path (this pod
-    # fell back to local model discovery — MODEL_PATH env var or
-    # logs/optimizer/). That path only exists on THIS pod's filesystem;
-    # download.py treats any path starting with "/" as already-materialized
-    # and never fetches it from MinIO, so writing it to the shared registry
-    # would hand a different pod a path that doesn't exist on its own
-    # filesystem, crashing it on startup. Leaving the registry untouched
-    # here means other pods fall through to their own local discovery too,
-    # which is at least self-consistent per pod.
-    if _pool and not _classifier.model_path.startswith("/"):
+    #
+    # Register the ORIGINAL MinIO-backed path (active["model_path"]) when we
+    # resolved via a successful MinIO download — NOT _classifier.model_path.
+    # download_model() always caches the downloaded model under a local
+    # directory (/tmp/sentinel-model-cache/...), so _classifier.model_path
+    # is *always* an absolute local path, even for a MinIO-backed model.
+    # Checking model_path.startswith("/") here (an earlier version of this
+    # check) therefore skipped registration for the common case too, not
+    # just genuine local-only fallback — every classification write then
+    # violated classifications_model_version_fkey, since the model_version
+    # never existed in model_registry. Reproduced live: confirmed via
+    # kubectl logs that a classifier pod loading a real MinIO-backed model
+    # still hit "Skipping registry write" and every subsequent /v1/moderations
+    # persist attempt failed with psycopg.errors.ForeignKeyViolation.
+    #
+    # `model_dir is not None` is the correct portability signal: it's only
+    # None when Classifier() fell through to _resolve_model_dir() itself
+    # (MODEL_PATH env var or logs/optimizer/ found locally) — that's the
+    # genuine non-portable case docs/local-dev.md and this comment used to
+    # describe, and it's the only one that should skip registration.
+    if _pool and model_dir is not None and active is not None:
         try:
             await _db.register_model(
                 _pool,
                 _classifier.model_version,
-                _classifier.model_path,
+                active["model_path"],
                 _classifier.threshold,
             )
         except Exception:
             logger.exception("Failed to register model version in registry")
     elif _pool:
         logger.info(
-            "Skipping registry write — model_path is a local filesystem path, "
+            "Skipping registry write — resolved via local fallback, "
             "not portable across pods | path=%s",
             _classifier.model_path,
         )

services/classifier/model.py

@@ -123,11 +123,15 @@ def predict(self, texts: list[str]) -> list[dict]:
         inputs = self._tokenizer(
             texts,
             return_tensors="np",
-            # max_length (not dynamic padding) trades a bit of extra compute
-            # on short inputs for latency that doesn't depend on what else
-            # happens to be in the same batch — a batch containing one
-            # long input no longer inflates every other input's padding.
-            padding="max_length",
+            # Reverted from padding="max_length": every batch then tokenized
+            # to the full 512 tokens regardless of actual input length,
+            # which OOM-killed the pod (1Gi limit) twice under real load —
+            # reproduced live, not theoretical. The "latency doesn't depend
+            # on batch content" benefit isn't worth crashing the service;
+            # a real fix for that (e.g. bucketed padding, or raising the
+            # memory limit + confirming it's stable under sustained load)
+            # needs its own verification pass, not a same-session swap back.
+            padding=True,
             truncation=True,
             max_length=512,
         )

services/stream-processor/main.py

@@ -94,6 +94,19 @@ def _pg_write(
     that case, so the caller never holds a reference to a leaked/half-open
     connection: the next call always reconnects from scratch instead of
     reusing something nobody closed.
+
+    Two distinct PG failure modes need different recovery:
+    - OperationalError (connection actually lost): close + reconnect + retry once.
+    - Any other psycopg.Error (e.g. a constraint violation): the connection
+      is still alive, but the current transaction is aborted — every command
+      on it then fails with InFailedSqlTransaction until rolled back.
+      Reproduced live: a stale model_version (classifier registered a model
+      version the DB didn't have yet — see main.py's #43 fix) triggered a
+      ForeignKeyViolation on classifications, and without this rollback the
+      long-lived connection stayed permanently broken — every subsequent
+      batch failed the same way forever, not just the one that hit the FK
+      violation.
+
     ON CONFLICT DO NOTHING on (span_id, text_type) makes the retry idempotent.
     """
     try:
@@ -120,6 +133,16 @@ def _pg_write(
             except Exception:
                 pass
             raise
+    except psycopg.Error:
+        # Connection is alive but its transaction is aborted (e.g. a
+        # constraint violation) — roll back so it's usable again on the
+        # NEXT call instead of failing every subsequent batch forever.
+        logger.exception("PostgreSQL write failed — rolling back to keep the connection usable")
+        try:
+            pg_conn.rollback()
+        except Exception:
+            logger.exception("Rollback itself failed — connection may still be unusable")
+        raise
 
 
 def run() -> None: