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

[VjayRam]

Summary

Follow-up to #45. After that merge, ran the full stack live against a real k3d cluster (dev-start.sh end-to-end, sustained synthetic load, real pod restarts) instead of relying on unit tests and standalone smoke tests — this surfaced five real bugs, four of them regressions from #45's own fixes. All are now fixed and independently re-verified live.

1. Kafka version downgrade broke the cluster outright (7985996)
#45 pinned apache/kafka:latest → 3.9.0 for reproducibility. In practice :latest had already drifted to Kafka 4.3.1, which had formatted the PVC's KRaft metadata with a feature level 3.9.0 doesn't recognize (No MetadataVersion with feature level 30) — a downgrade, not a pin, and it crash-looped every broker start. Inspected the cached image directly and re-pinned to 4.3.1 (the version that actually formatted the data) instead of blindly using an arbitrary "recent stable" version.

2. psql -v/:'var' substitution silently doesn't work through -c (91ed645)
#45's SQL-injection fix (#39) switched to psql -v pw=... -c "... :'pw' ..." to avoid bare string interpolation. Reproduced live: -c bypasses the parser that performs variable interpolation entirely — Postgres received the literal, unexpanded text :'pw' and rejected it as a syntax error. -f/stdin do support it. Switched both call sites to pipe the SQL through stdin instead.

3. Three bugs only reproducible under real load/restarts (e861865)

• Classifier OOM-killed twice (1Gi limit) under real traffic — #8's padding="max_length" tokenized every batch to the full 512 tokens regardless of actual input length. Reverted to padding=True; verified flat ~324Mi memory through a 300-trace burst with zero restarts.
• Every classification write was silently failing a foreign-key constraint — #43's model_path.startswith("/") check couldn't distinguish "genuine local-only fallback" from "MinIO model, locally cached" (download_model() always caches locally), so it skipped registration for the normal case too. Now keyed on whether a MinIO download actually happened.
• Once the above hit once, the stream processor's long-lived PG connection got permanently stuck in InFailedSqlTransaction — _pg_write only rolled back on OperationalError (connection lost), not on a SQL-level error that aborts the transaction but leaves the connection alive. Added a rollback for any psycopg.Error.

4. Drift job resolved a model_version with zero matching classification rows (d248302)
#33's fix mirrored the classifier's own "prefer status='active'" registry-selection logic — reasonable in principle, wrong in practice: the classifier self-registers under its own freshly-derived model_version string on every startup, completely separate from whatever model_path an 'active' row points at. The one 'active' row in the live registry was a stale promotion under an unrelated version string, so drift detection silently computed nothing meaningful. Dropped the active-status preference in favor of created_at DESC — "whichever pod self-registered most recently" is what's actually running.

Test plan

• dev-start.sh runs end-to-end with zero errors (image builds → terraform apply → data-layer ready → Kafka topic → password sync → port-forwards → model bootstrap → app rollout)
• Kafka survives a real pod deletion: identical topic data and consumer-group offsets before/after
• Classifier stable under sustained load: 300-trace burst, memory flat at ~324Mi, zero restarts (previously OOM'd twice)
• Real trace flow end-to-end: classifications/flagged_content row counts increase correctly, zero duplicates, zero consumer lag, clean logs
• Drift job runs against live Postgres: resolves the correct model_version, computes PSI/JSD on 960 real rows, writes drift_stats correctly
• Evaluation pipeline benchmarks the live active model and its gate correctly fails a low-accuracy candidate
• Prometheus scraping the classifier successfully, all 14 alert rules loaded, real metric data flowing
• terraform plan clean (only expected kubectl rollout restart annotation drift, not Terraform-managed)

🤖 Generated with Claude Code

Summary by CodeRabbit

• Bug Fixes
  • Improved classifier stability under load by reducing memory pressure during prediction.
  • Made registry and model-version handling more reliable, especially during restarts and external model downloads.
  • Strengthened PostgreSQL error handling so failed writes recover cleanly without blocking later processing.
  • Updated Kafka-related runtime components to a newer compatible version to prevent startup and metadata issues.
• Chores
  • Refined startup and database scripts for more dependable command execution.

[coderabbitai]

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Free

Run ID: 22a8268a-077c-43d3-90ba-f0c804086568

📥 Commits

Reviewing files that changed from the base of the PR and between 63753da and d248302.

📒 Files selected for processing (6)

• infra/terraform/local/main.tf
• pipelines/drift/db.py
• scripts/dev-start.sh
• services/classifier/main.py
• services/classifier/model.py
• services/stream-processor/main.py

---

📝 Walkthrough

Walkthrough

This PR contains unrelated fixes: pinning Kafka Docker images to 4.3.1 in Terraform, changing active model version selection logic in the drift DB module, fixing psql variable substitution in dev-start.sh, reworking classifier model registration to use active registry data, reverting tokenizer padding, and adding Postgres rollback handling in stream-processor.

Changes

Kafka Image Pinning

Layer / File(s)	Summary
Kafka image version pin infra/terraform/local/main.tf	Broker StatefulSet and kafka-topic-init Job images are pinned from apache/kafka:3.9.0 to 4.3.1, with comments on KRaft PVC compatibility constraints.

Model Registry Selection and Registration

Layer / File(s)	Summary
Active model version query pipelines/drift/db.py	Selection logic changed to ORDER BY created_at DESC instead of status-prioritized ordering, with docstring rationale update.
Classifier registration logic services/classifier/main.py	Introduces an active registry variable and changes registration condition/path source to use model_dir/active["model_path"] instead of _classifier.model_path.

dev-start.sh psql Variable Substitution Fix

Layer / File(s)	Summary
psql stdin piping scripts/dev-start.sh	Password sync and model path UPDATE statements are piped via stdin into psql instead of using -c, enabling variable substitution.

Classifier Padding and Stream-Processor DB Error Handling

Layer / File(s)	Summary
Tokenizer padding revert services/classifier/model.py	Tokenizer padding changed from "max_length" to True, with comment on OOM cause.
Postgres write rollback handling services/stream-processor/main.py	Adds a psycopg.Error handler that logs, rolls back, and re-raises, plus docstring updates for _pg_write.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Estimated code review effort: 3 (Moderate) | ~25 minutes

---

Note

🎁 Summarized by CodeRabbit Free

Your organization is on the Free plan. CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please upgrade your subscription to CodeRabbit Pro by visiting https://app.coderabbit.ai/login.

_{Comment @coderabbitai help to get the list of available commands.}