From 75d7d56452226021506049663b355bb50cbb603e Mon Sep 17 00:00:00 2001 From: UsatovPavel Date: Tue, 29 Sep 2026 01:10:56 +0300 Subject: [PATCH] feat(inference): one-shot backfill BACKFILL_HOURS rewinds the cursor once per runtime volume so the stream re-predicts events of the last N hours; windows warm up before that point. A fresh incident service otherwise covers only channels with events after its start. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- inference/runner.py | 30 +++++++++++++++++++++++++++++- tests/test_runner.py | 20 ++++++++++++++++++++ 3 files changed, 50 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 3010e38..9dca93e 100644 --- a/README.md +++ b/README.md @@ -761,7 +761,7 @@ inference/ Healthcheck `python -m inference.health`: флаг `ready` после прогрева и свежий `heartbeat`. -Env (default): `INFERENCE_RUNTIME_DIR` (`runtime`, volume), `POLL_INTERVAL_SECONDS` (2), `EVENTS_BATCH_SIZE` (5000), `EVENTS_REORDER_LAG_SECONDS` (300), `TTL_INTERVAL_SECONDS` (60), `METADATA_REFRESH_SECONDS` (3600), `METADATA_RETRY_SECONDS` (600), `HEALTH_MAX_AGE_SECONDS` (120). +Env (default): `INFERENCE_RUNTIME_DIR` (`runtime`, volume), `POLL_INTERVAL_SECONDS` (2), `EVENTS_BATCH_SIZE` (5000), `EVENTS_REORDER_LAG_SECONDS` (300), `TTL_INTERVAL_SECONDS` (60), `METADATA_REFRESH_SECONDS` (3600), `METADATA_RETRY_SECONDS` (600), `HEALTH_MAX_AGE_SECONDS` (120), `BACKFILL_HOURS` (0: выкл.; >0 — один раз на том runtime, маркер `backfill.done`, курсор отматывается на события за N часов и поток прогнозирует их заново). ### Docker и деплой diff --git a/inference/runner.py b/inference/runner.py index 3a641c8..dee5a7d 100644 --- a/inference/runner.py +++ b/inference/runner.py @@ -42,8 +42,11 @@ CURSOR_FILE = "events_cursor.json" TARGETS_FILE = "incident_targets.json" +# Бэкфилл выполняется один раз на том runtime: маркер переживает рестарты. +BACKFILL_FILE = "backfill.done" SELECT_MAX_ID = "SELECT coalesce(max(id), 0) FROM events" +SELECT_BACKFILL_LOW = "SELECT min(id) - 1 FROM events WHERE ts >= %s" # Прогрев — только то, что уже учтено курсором: остальное придёт # в цикле, и второй update задвоил бы окна. @@ -80,6 +83,7 @@ class RunnerSettings: backoff_max_seconds: float = 60.0 shap_max_per_tick: int = 8 shap_budget_ms: float = 500.0 + backfill_hours: float = 0.0 @classmethod def from_env( @@ -102,6 +106,7 @@ def from_env( ), shap_max_per_tick=int(env.get("SHAP_MAX_PER_TICK") or 8), shap_budget_ms=_env_float(env, "SHAP_BUDGET_MS", 500.0), + backfill_hours=_env_float(env, "BACKFILL_HOURS", 0.0), ) @@ -216,6 +221,9 @@ def start(self, conn: psycopg.Connection) -> None: cursor = EventCursor( low, lag_seconds=self.options.reorder_lag_seconds ) + backfill = self._backfill_cursor(conn) + if backfill is not None: + cursor, fresh = backfill, True rows = self._warmup(conn, service, cursor) self.classed = classed_channels(conn) if self.settings.prediction_kind == "INCIDENT": @@ -239,6 +247,8 @@ def start(self, conn: psycopg.Connection) -> None: cursor.save(self.path(CURSOR_FILE)) if self.targets is not None: self.targets.save(self.path(TARGETS_FILE), cursor.low, cursor.seen) + if self.options.backfill_hours > 0: + self.path(BACKFILL_FILE).touch() self.path(READY_FILE).touch() log.info( "started fresh=%s low=%d seen=%d warmup_rows=%d channels=%d " @@ -248,6 +258,23 @@ def start(self, conn: psycopg.Connection) -> None: self.window.total_seconds() / 3600, ) + def _backfill_span(self) -> timedelta: + if self.options.backfill_hours <= 0 or self.path(BACKFILL_FILE).exists(): + return timedelta(0) + return timedelta(hours=self.options.backfill_hours) + + def _backfill_cursor(self, conn: psycopg.Connection) -> EventCursor | None: + """Курсор на начало окна бэкфилла: поток заново прогнозирует события + за backfill_hours, чтобы у тихих с момента старта каналов был прогноз.""" + span = self._backfill_span() + if not span: + return None + low = conn.execute(SELECT_BACKFILL_LOW, (self.clock() - span,)).fetchone()[0] + if low is None: + return None + log.info("backfill hours=%.0f low=%d", span.total_seconds() / 3600, low) + return EventCursor(low, lag_seconds=self.options.reorder_lag_seconds) + def _warmup( self, conn: psycopg.Connection, @@ -255,7 +282,8 @@ def _warmup( cursor: EventCursor, ) -> int: params = { - "since": self.clock() - self.window, + # При бэкфилле окна прогреваются и до его начала. + "since": self.clock() - self.window - self._backfill_span(), "low": cursor.low, "seen": list(cursor.seen), } diff --git a/tests/test_runner.py b/tests/test_runner.py index b093b1f..7f4223f 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -109,6 +109,26 @@ def started(make_runner, db, **kwargs) -> tuple[Runner, psycopg.Connection]: return runner, conn +def test_backfill_predicts_recent_events_once(db, make_runner) -> None: + ts = datetime.now(timezone.utc) - timedelta(hours=2) + with db.admin() as admin: + add_event(admin, CHANNEL, ts, "Неисправен") + + # Без бэкфилла первый старт не прогнозирует уже лежащие события. + runner, conn = started(make_runner, db, threshold=0.0) + runner.tick(conn) + assert predictions(db) == [] + + runner, conn = started(make_runner, db, threshold=0.0, backfill_hours=3) + runner.tick(conn) + assert [row[1] for row in predictions(db)] == [str(CHANNEL)] + + # Маркер в runtime: повторный старт окно заново не прогоняет. + runner, conn = started(make_runner, db, threshold=0.0, backfill_hours=3) + runner.tick(conn) + assert len(predictions(db)) == 1 + + def test_tick_writes_prediction_rows(db, make_runner) -> None: runner, conn = started(make_runner, db, threshold=0.0) ts = datetime.now(timezone.utc) - timedelta(minutes=5)