From 991e4c28ed59fb54d22655c5692b1b3aa03cc33e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 20 Aug 2026 16:51:02 +0500 Subject: [PATCH] =?UTF-8?q?fix(ptica):=20=D0=BD=D0=B0=20worker=5Fready=20?= =?UTF-8?q?=D0=B7=D0=BE=D0=BC=D0=B1=D0=B8=20=D0=BF=D0=BE=D0=BC=D0=B5=D1=87?= =?UTF-8?q?=D0=B0=D0=B5=D1=82=D1=81=D1=8F=20=D0=9B=D0=AE=D0=91=D0=9E=D0=99?= =?UTF-8?q?=20'running',=20=D0=B0=20=D0=BD=D0=B5=20=D1=82=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=D0=BA=D0=BE=20=D1=81=D0=BE=20=D1=81=D0=BD=D0=B0=D0=BF?= =?UTF-8?q?=D1=88=D0=BE=D1=82=D0=BE=D0=BC=20(#2464)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Докстринг _resume_zombie_runs формулирует инвариант прямо: No time threshold: by definition, on worker_ready ANY 'running' row is a zombie because there is no active worker. Previously we required heartbeat … stayed in 'running' status forever and required manual cancel/resume. А запрос добавлял `AND objects_snapshot IS NOT NULL`. Строка без снапшота в выборку не попадала и оставалась 'running' НАВСЕГДА — ровно то состояние, ради устранения которого функция и заводилась. Снапшот нужен, но не для пометки, а для ВОЗОБНОВЛЕНИЯ: resume_kn_run восстанавливает обход «using objects_snapshot» и без него упал бы. Поэтому зомби помечаются все, а resume ставится только тем, кого есть чем возобновить; остальные получают честную причину в error вместо тишины. Про тест — отдельно, потому что первая версия была негодной. Двойник сессии отдавал строки независимо от WHERE, и на origin/main главный тест («строка не помечена») ПРОХОДИЛ, а краснели два других — по ложной причине. Научил двойник соблюдать ровно тот фильтр, о котором спор, и сузил совпадение до `AND objects_snapshot IS NOT NULL`: правка выносит то же выражение в список полей SELECT, и совпадение по голой подстроке отсекало бы строки у исправленной версии тоже. Против origin/main теперь: строка без снапшота не помечена zombie → падает (UPDATE вообще не выполняется) в смешанной выборке помечены не все → падает: {1,3} вместо {1,2,3} невозобновляемому resume не ставится — контроль, зелёный с обеих сторон возобновляемый получает resume как раньше — контроль, зелёный с обеих сторон Первый контроль ловит «починку», ставящую resume всем подряд. Замер прода 20.08: строк в 'running' сейчас нет, то есть правка предотвращает, а не чинит. Из 20 исторических 'zombie' восемь — без objects_snapshot, так что случай не гипотетический. Прогоны: tests/workers rc=0. Co-Authored-By: Claude Opus 5 --- backend/app/workers/lifecycle.py | 69 ++++++-- .../workers/test_2464_zombie_any_running.py | 157 ++++++++++++++++++ 2 files changed, 208 insertions(+), 18 deletions(-) create mode 100644 backend/tests/workers/test_2464_zombie_any_running.py diff --git a/backend/app/workers/lifecycle.py b/backend/app/workers/lifecycle.py index c440c56f..8cade6fe 100644 --- a/backend/app/workers/lifecycle.py +++ b/backend/app/workers/lifecycle.py @@ -89,14 +89,24 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: db = SessionLocal() ids: list[int] = [] try: + # #2464: берём ЛЮБУЮ строку в 'running', без фильтра по objects_snapshot. + # Докстринг этой функции формулирует инвариант прямо: «by definition, on + # worker_ready ANY 'running' row is a zombie because there is no active + # worker». Фильтр ему противоречил: строка без снапшота не попадала в + # выборку и оставалась 'running' НАВСЕГДА — ровно то состояние, ради + # устранения которого функция и заводилась. + # + # Снапшот всё равно нужен — но не для пометки, а для ВОЗОБНОВЛЕНИЯ: + # resume_kn_run восстанавливает обход «using objects_snapshot». Поэтому + # помечаем зомби всех, а resume ставим только тем, кого есть чем + # возобновить. Остальные получают честную причину вместо тишины. rows = ( db.execute( text( """ - SELECT run_id + SELECT run_id, (objects_snapshot IS NOT NULL) AS resumable FROM kn_scrape_runs WHERE status = 'running' - AND objects_snapshot IS NOT NULL ORDER BY started_at ASC LIMIT 20 """ @@ -106,22 +116,45 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: .all() ) if rows: - ids = [int(r["run_id"]) for r in rows] - # Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые - # run_id со ссылкой resumed_from_run_id. - db.execute( - text( - """ - UPDATE kn_scrape_runs - SET status = 'zombie', - finished_at = NOW(), - error = COALESCE(error, - 'auto-zombie at worker_ready, resume scheduled') - WHERE run_id = ANY(:ids) - """ - ), - {"ids": ids}, - ) + ids = [int(r["run_id"]) for r in rows if r["resumable"]] + orphan_ids = [int(r["run_id"]) for r in rows if not r["resumable"]] + if ids: + # Помечаем как 'zombie' одним апдейтом — resume создаст новые + # run_id со ссылкой resumed_from_run_id. + db.execute( + text( + """ + UPDATE kn_scrape_runs + SET status = 'zombie', + finished_at = NOW(), + error = COALESCE(error, + 'auto-zombie at worker_ready, resume scheduled') + WHERE run_id = ANY(:ids) + """ + ), + {"ids": ids}, + ) + if orphan_ids: + db.execute( + text( + """ + UPDATE kn_scrape_runs + SET status = 'zombie', + finished_at = NOW(), + error = COALESCE(error, + 'auto-zombie at worker_ready, resume невозможен: ' + 'нет objects_snapshot') + WHERE run_id = ANY(:orphans) + """ + ), + {"orphans": orphan_ids}, + ) + logger.warning( + "worker_ready: %d kn-прогонов без objects_snapshot помечены zombie" + " без resume — возобновлять нечем: %s", + len(orphan_ids), + orphan_ids, + ) db.commit() else: logger.info("worker_ready: нет stale kn runs для resume") diff --git a/backend/tests/workers/test_2464_zombie_any_running.py b/backend/tests/workers/test_2464_zombie_any_running.py new file mode 100644 index 00000000..b1ba844d --- /dev/null +++ b/backend/tests/workers/test_2464_zombie_any_running.py @@ -0,0 +1,157 @@ +"""На worker_ready зомби помечается ЛЮБАЯ строка в 'running' (#2464). + +Докстринг `_resume_zombie_runs` формулирует инвариант прямо: + + No time threshold: by definition, on worker_ready ANY 'running' row is a zombie + because there is no active worker. Previously we required heartbeat … stayed in + 'running' status forever and required manual cancel/resume. + +А запрос добавлял `AND objects_snapshot IS NOT NULL`. Строка без снапшота в выборку не +попадала и оставалась `'running'` НАВСЕГДА — ровно то состояние, ради устранения которого +функция и заводилась. + +Снапшот нужен, но не для пометки, а для ВОЗОБНОВЛЕНИЯ: `resume_kn_run` восстанавливает +обход «using objects_snapshot». Поэтому зомби помечаются все, а resume ставится только тем, +кого есть чем возобновить; остальные получают честную причину. + +Тесты смотрят на выполненный SQL и на поставленные resume-задачи — то есть на поведение. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any +from unittest.mock import MagicMock, patch + + +class _Result: + def __init__(self, rows: list[dict]) -> None: + self._rows = rows + + def mappings(self) -> _Result: + return self + + def all(self) -> list[dict]: + return self._rows + + def scalar(self) -> Any: + return None + + def first(self) -> Any: + return None + + +class _Session: + """Отдаёт заданные строки на SELECT и запоминает весь выполненный SQL.""" + + def __init__(self, rows: list[dict]) -> None: + self._rows = rows + self.sql: list[tuple[str, dict]] = [] + self.commits = 0 + + def execute(self, statement: Any = None, params: Any = None, *a: Any, **kw: Any) -> _Result: + text_ = str(statement) + self.sql.append((text_, params or {})) + if "SELECT run_id" in text_ and "kn_scrape_runs" in text_: + # Двойник ОБЯЗАН соблюдать ровно тот фильтр, о котором идёт спор. + # Без этого тест краснел бы на origin/main по ложной причине: там + # строка без снапшота отсекается запросом и до пометки не доходит, + # а наивный двойник отдавал бы её всё равно — и главный тест + # («строка не помечена») проходил бы на обеих сторонах. + # Ищем именно условие в WHERE (с ведущим AND), а не подвыражение в + # SELECT: правка выносит `(objects_snapshot IS NOT NULL) AS resumable` + # в список полей, и совпадение по голой подстроке отсекало бы строки + # и у исправленной версии тоже. + if "AND objects_snapshot IS NOT NULL" in text_: + return _Result([r for r in self._rows if r["resumable"]]) + return _Result(self._rows) + return _Result([]) + + def commit(self) -> None: + self.commits += 1 + + def rollback(self) -> None: + pass + + def close(self) -> None: + pass + + +def _run(rows: list[dict]) -> tuple[_Session, list[int]]: + from app.workers import lifecycle as mod + + db = _Session(rows) + enqueued: list[int] = [] + + resume_task = MagicMock() + resume_task.apply_async.side_effect = lambda args=None, **_kw: enqueued.append(args[0]) + + # SessionLocal импортируется ВНУТРИ функции (late import), поэтому подменяем + # его в модуле-источнике, а не на lifecycle. + import app.core.db as core_db + + with ( + patch.object(core_db, "SessionLocal", lambda: db), + patch.dict( + "sys.modules", + {"app.workers.tasks.scrape_kn": MagicMock(resume_kn_run=resume_task)}, + ), + ): + mod._resume_zombie_runs() + return db, enqueued + + +def _updated_ids(db: _Session) -> set[int]: + out: set[int] = set() + for sql, params in db.sql: + if "UPDATE kn_scrape_runs" in sql and "zombie" in sql: + for key in ("ids", "orphans"): + out.update(params.get(key) or []) + return out + + +def test_running_row_without_snapshot_is_marked_zombie() -> None: + """Строка без objects_snapshot обязана быть помечена, а не остаться 'running'. + + На origin/main она вообще не попадает в выборку: фильтр отсекает её, + и никакого UPDATE по ней не выполняется. + """ + db, _ = _run([{"run_id": 7, "resumable": False}]) + + assert 7 in _updated_ids(db), ( + "прогон без снапшота не помечен zombie — останется в 'running' навсегда; " + f"выполненный SQL: {[s for s, _ in db.sql]}" + ) + + +def test_unresumable_row_gets_no_resume_task() -> None: + """Контроль: возобновлять нечем — resume не ставим. + + Ловит «починку», которая ставила бы resume всем подряд: resume_kn_run + восстанавливает обход ИЗ objects_snapshot и без него упадёт. + """ + _, enqueued = _run([{"run_id": 7, "resumable": False}]) + assert enqueued == [], f"поставлен resume для невозобновляемого прогона: {enqueued}" + + +def test_resumable_row_still_gets_resume() -> None: + """Контроль: прежнее поведение для строк со снапшотом не тронуто.""" + db, enqueued = _run([{"run_id": 3, "resumable": True}]) + assert 3 in _updated_ids(db) + assert enqueued == [3] + + +def test_mixed_batch_splits_correctly() -> None: + """Контроль: смешанная выборка — помечены все, resume только у пригодных.""" + db, enqueued = _run( + [ + {"run_id": 1, "resumable": True}, + {"run_id": 2, "resumable": False}, + {"run_id": 3, "resumable": True}, + ] + ) + assert _updated_ids(db) == {1, 2, 3}, f"помечены не все: {_updated_ids(db)}" + assert sorted(enqueued) == [1, 3], f"resume поставлен не тем: {enqueued}" -- 2.45.3