fix(ml): old schema support

deploy/backup-equipment-window.sh

@@ -21,6 +21,16 @@ container=$(docker compose ps -q inference)
   echo 'equipment inference is not running' >&2
   exit 1
 }
+old_health=$(docker inspect -f '{{.State.Health.Status}}' "$container")
+schema_recovery=0
+old_logs=$(docker logs --tail 100 "$container" 2>&1)
+if grep -Fq 'column "journal_is_alarm" does not exist' <<<"$old_logs"; then
+  schema_recovery=1
+  echo 'old equipment is blocked by missing journal_is_alarm; preserving its window before repair'
+elif [[ $old_health != healthy ]]; then
+  echo "old equipment is not healthy: $old_health" >&2
+  exit 1
+fi
 image=$(docker image inspect -f '{{.Id}}' inference:current)
 [[ -n $image ]] || { echo 'inference:current image is missing' >&2; exit 1; }
 volume=$(docker inspect -f '{{range .Mounts}}{{if eq .Destination "/app/runtime"}}{{.Name}}{{end}}{{end}}' "$container")
@@ -43,7 +53,7 @@ cleanup() {
   local result=$?
   trap - EXIT
   if (( stopped )); then
-    if ! docker compose start inference || ! wait_healthy; then
+    if ! docker compose start inference || ! verify_old_running; then
       echo 'CRITICAL: old equipment inference did not recover after backup failure' >&2
       result=1
     fi
@@ -68,6 +78,14 @@ wait_healthy() {
   done
 }
 
+verify_old_running() {
+  if (( schema_recovery )); then
+    [[ $(docker inspect -f '{{.State.Running}}' "$container") == true ]]
+  else
+    wait_healthy
+  fi
+}
+
 # Stop before copying the cursor so it matches a completed batch.
 stopped=1
 docker compose stop inference
@@ -104,16 +122,16 @@ copy_query "SELECT id, event_id, channel_id, ts, raw_value, is_alarm FROM events
 copy_query "SELECT id, event_id, channel_id, ts, raw_value, is_alarm FROM events WHERE ts > '${snapshot_at}'::timestamptz" future_events.csv.gz
 copy_query 'SELECT channel_id, eng_system_type, sensor_type, system_tag, sensor_name, object_id FROM dim_channels_current' channels.csv.gz
 copy_query 'SELECT object_id, hierarchy_level, parent_id, object_kind, dispatcher_name FROM dim_objects_current' objects.csv.gz
-wait_healthy
+verify_old_running
 
-python3 - "$tmp" "$snapshot_at" "$image" "$sha" "$volume" <<'PY'
+python3 - "$tmp" "$snapshot_at" "$image" "$sha" "$volume" "$old_health" "$schema_recovery" <<'PY'
 import csv
 import gzip
 import json
 import sys
 from pathlib import Path
 
-directory, snapshot_at, image, sha, volume = sys.argv[1:]
+directory, snapshot_at, image, sha, volume, old_health, schema_recovery = sys.argv[1:]
 root = Path(directory)
 counts = {}
 for name in ('events', 'future_events', 'channels', 'objects'):
@@ -143,6 +161,8 @@ manifest = {
     'window_hours': 72,
     'image_id': image,
     'runtime_volume': volume,
+    'old_health': old_health,
+    'schema_recovery': schema_recovery == '1',
     'deploy_sha': sha,
     'cursor_low': data['low'],
     'cursor_seen': len(data['seen']),

inference/runner.py

@@ -54,7 +54,7 @@
 
 # Фильтр по ts отсекает старые партиции и импорт истории вне окон.
 SELECT_NEW = """
-    SELECT id, channel_id, ts, raw_value, is_alarm, journal_is_alarm FROM events
+    SELECT id, channel_id, ts, raw_value, is_alarm, {journal_alarm} FROM events
     WHERE ts >= %(since)s AND id > %(low)s
       AND id <> ALL(%(seen)s::bigint[])
     ORDER BY id
@@ -326,8 +326,11 @@ def tick(self, conn: psycopg.Connection) -> bool:
 
     def _fetch(self, conn: psycopg.Connection) -> list[EventRow]:
         assert self.cursor is not None
+        journal_alarm = (
+            "journal_is_alarm" if self.targets is not None else "NULL::boolean"
+        )
         rows = conn.execute(
-            SELECT_NEW,
+            SELECT_NEW.format(journal_alarm=journal_alarm),
             {
                 "since": self.clock() - self.window,
                 "low": self.cursor.low,

tests/test_equipment_window_backup.py

@@ -29,7 +29,12 @@ def _mock_tools(tmp_path: Path) -> Path:
   'compose ps -q inference') echo old-container ;;
   'inspect -f {{.State.Running}} old-container') echo true ;;
   'inspect -f {{range .Mounts}}{{if eq .Destination "/app/runtime"}}{{.Name}}{{end}}{{end}} old-container') echo backend_inference-runtime ;;
-  'inspect -f {{.State.Health.Status}} old-container') echo healthy ;;
+  'inspect -f {{.State.Health.Status}} old-container')
+    if [ "${MOCK_SCHEMA_ERROR:-0}" = 1 ] && [ "${MOCK_HEALTHY_ERROR:-0}" != 1 ]; then echo starting;
+    else echo healthy; fi ;;
+  'logs --tail 100 old-container')
+    if [ "${MOCK_SCHEMA_ERROR:-0}" = 1 ]; then
+      echo 'column "journal_is_alarm" does not exist'; fi ;;
   'image inspect -f {{.Id}} inference:current') echo sha256:old-image ;;
   'compose stop inference') : ;;
   'compose start inference') [ "${MOCK_FAIL_START:-0}" != 1 ] ;;
@@ -54,7 +59,9 @@ def _mock_tools(tmp_path: Path) -> Path:
 
 
 def _run(tmp_path: Path, fail_copy: bool = False, fail_start: bool = False,
-         no_inference: bool = False) -> tuple[subprocess.CompletedProcess[str], Path]:
+         no_inference: bool = False, schema_error: bool = False,
+         healthy_error: bool = False
+         ) -> tuple[subprocess.CompletedProcess[str], Path]:
     bin_dir = _mock_tools(tmp_path)
     runtime = tmp_path / "runtime"
     runtime.mkdir()
@@ -77,6 +84,8 @@ def _run(tmp_path: Path, fail_copy: bool = False, fail_start: bool = False,
         "MOCK_FAIL_CP": "1" if fail_copy else "0",
         "MOCK_FAIL_START": "1" if fail_start else "0",
         "MOCK_NO_INFERENCE": "1" if no_inference else "0",
+        "MOCK_SCHEMA_ERROR": "1" if schema_error else "0",
+        "MOCK_HEALTHY_ERROR": "1" if healthy_error else "0",
     }
     result = subprocess.run(["bash", str(SCRIPT)], env=env, text=True,
                             capture_output=True, check=False)
@@ -125,3 +134,22 @@ def test_missing_equipment_service_blocks_image_switch(tmp_path: Path) -> None:
     assert result.returncode != 0
     assert "window cannot be preserved" in result.stderr
     assert not list((backup / "inference-windows").glob("equipment-*"))
+
+
+def test_known_schema_failure_still_preserves_window(tmp_path: Path) -> None:
+    result, backup = _run(tmp_path, schema_error=True)
+    assert result.returncode == 0, result.stderr
+    snapshot = next((backup / "inference-windows").glob("equipment-*"))
+    manifest = json.loads((snapshot / "manifest.json").read_text())
+    assert manifest["old_health"] == "starting"
+    assert manifest["schema_recovery"] is True
+    assert "compose start inference" in (tmp_path / "calls.log").read_text()
+
+
+def test_schema_error_overrides_stale_healthy_status(tmp_path: Path) -> None:
+    result, backup = _run(tmp_path, schema_error=True, healthy_error=True)
+    assert result.returncode == 0, result.stderr
+    snapshot = next((backup / "inference-windows").glob("equipment-*"))
+    manifest = json.loads((snapshot / "manifest.json").read_text())
+    assert manifest["old_health"] == "healthy"
+    assert manifest["schema_recovery"] is True

tests/test_runner.py

@@ -139,6 +139,16 @@ def test_tick_writes_prediction_rows(db, make_runner) -> None:
     assert "null" not in [item["feature_value"] for item in shap["top"]]
 
 
+def test_equipment_reads_events_before_incident_migration(db, make_runner) -> None:
+    with db.admin() as admin:
+        admin.execute("ALTER TABLE events DROP COLUMN journal_is_alarm")
+    runner, conn = started(make_runner, db)
+    with db.admin() as admin:
+        add_event(admin, CHANNEL, datetime.now(timezone.utc), "Норма")
+    runner.tick(conn)
+    assert len(predictions(db)) == 1
+
+
 def test_shap_null_below_threshold(db, make_runner) -> None:
     runner, conn = started(make_runner, db, threshold=1.0)