From c646a71001ac67b7c31b7aedfb09dc9ce0c19e61 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 27 Jun 2026 12:36:50 +0500 Subject: [PATCH] fix(freshness): output-floor on cycle-SUM over fresh_days window (#1945, #1947) Adversarial review found the single-newest last_success_work basis fragile: a load cycle = MULTIPLE done-runs (objective per group_name, kn per region + resume), so the newest run alone misrepresents the cycle. Replace with SUM(work_col) FILTER(done) over the source's own fresh_days window. - recent_output = SUM(COALESCE(work_col,0)) FILTER done in fresh_days window (make_interval secs => fresh_days*86400; no :: cast; NULL-counter -> 0) - downgrade when status==ok and recent_output < min_output_rows (strict <) - registry floors kept: kn_flats=50000 (healthy ~376k, broken sum <=3670), objective=1000 (SUM 7d ~946k); comments rewritten to cycle-sum basis + kn_flats zombie-resume KNOWN LIMITATION - window=fresh_days keeps the #1947 aging-FP fix (7-8d run stays in window) - tests: low-output-cycle->failed, above-floor->ok, boundary==floor->ok, weekly-aging->ok, age-stale precedence, registry floors Verified on prod: domrf_kn_flats latest snapshot 2026-06-22=9 flats vs healthy 2026-05-17=376604 (broken 5+ weeks); objective SUM(rows_lots,7d)=946264. --- backend/app/api/v1/admin_scrape.py | 138 +++++++++--- .../api/v1/test_admin_scrape_freshness.py | 209 ++++++++++++++---- 2 files changed, 273 insertions(+), 74 deletions(-) diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 6e56b194..53554c40 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1280,12 +1280,21 @@ class FreshnessSource(BaseModel): stale_days: float = 30.0 # < stale_days → stale, иначе critical critical: bool = False # критичный источник (для Sentry-алёртов в beat-таске) # True → источник, который прогнался успешно И недавно (был бы fresh по возрасту), - # НО произвёл 0 work-строк в окне (SUM(work_col) FILTER(status='done') == 0), - # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-ничего-не-извлёк" - # (напр. WAF-break / тихий пустой extract: status=done, но flats_count=0). По - # умолчанию OFF — включаем только там, где 0 строк = поломка (не legitimate-0, - # как nspd/nspd_geo/cadastre). Применимо только к run-ledger (не data-table) режиму. + # НО SUM(work_col) по done-прогонам цикла (окно fresh_days) < min_output_rows, + # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-почти-ничего-не-извлёк" + # (напр. WAF-break / тихий частичный extract: status=done, но flats_count суммарно + # 9 при здоровых ~376k). По умолчанию OFF — включаем только там, где малый выход = + # поломка (не legitimate-low, как nspd/nspd_geo/cadastre). Применимо только к + # run-ledger (не data-table) режиму. alert_on_zero_output: bool = False + # Пол (floor) для recent_output = SUM(work_col) по done-прогонам ЦИКЛА в окне + # fresh_days (НЕ один прогон — цикл объединяет несколько прогонов: objective по + # group_name, kn по region_codes+resume). Проверяется ТОЛЬКО при + # alert_on_zero_output=True: recent_output < min_output_rows → даунгрейд (строгое <). + # Окно = fresh_days, поэтому здоровый недельный прогон возрастом 7–8d остаётся + # внутри окна и не ложно-срабатывает (#1947 fix). Default 1 → флагует только если + # суммарный выход цикла = 0 (безопасный минимальный catch). + min_output_rows: int = 1 # Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на @@ -1295,9 +1304,12 @@ class FreshnessSource(BaseModel): # забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs"). # Пороги: # kn — beat ~еженедельно (scrape_kn cron, Mon) → fresh<8d, stale<14d -# kn_flats — та же run-таблица, work_col=flats_count + alert_on_zero_output (issue -# #1945: kn-run status=done но flats_count=0 ~5 недель — монитор молчал, -# т.к. kn смотрит objects_count ~1548, не flats_count) → fresh<8d, stale<14d +# kn_flats — та же run-таблица, work_col=flats_count + alert_on_zero_output + +# min_output_rows=50000 (issue #1945: kn-run status=done но flats_count +# мал/нестабилен — монитор смотрел objects_count ~1548, не flats_count; +# #1947 fix: floor сравнивается с SUM(flats_count) по done-прогонам цикла +# в окне fresh_days=8d — ловит low-output 9/2296/3670 и не ложит на 7–8d +# aging) → fresh<8d, stale<14d # objective — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d # nspd — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d # nspd_geo — ручной bulk geo-job → fresh<7d, stale<30d @@ -1318,11 +1330,23 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ ), FreshnessSource( # kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count - # (квартиры). Источник истины для "лоадер прогнался, но ничего не извлёк" - # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count=0 ~5 недель, + # (квартиры). Источник истины для "лоадер прогнался, но почти ничего не извлёк" + # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count мал/нестабилен, # монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги, - # что у kn, но work_col=flats_count + alert_on_zero_output → 0 квартир за неделю - # done-прогонов даунгрейдит otherwise-fresh источник до failed. + # что у kn, но work_col=flats_count + alert_on_zero_output + min_output_rows. + # + # Floor сравнивается с recent_output = SUM(flats_count) по ВСЕМ done-прогонам в + # окне fresh_days=8d (kn пишет строку на region_codes + resume — это НЕ один + # прогон). Магнитуды (прод gendesign-postgres-1, #1947): healthy цикл ~376k + # квартир (snapshot 05-17=376604); сломанное состояние — суммарно ≤3670, сейчас + # SUM(flats_count,8d)=9 (latest done = resume, считает только хвост; ground truth + # snapshot 2026-06-22=9, сломан 5+ недель). Порог 50000 чисто разделяет: + # healthy ≫ 50k ≫ broken ≤3670. + # + # KNOWN LIMITATION: zombie-resume, загрузивший лишь малый хвост БЕЗ полного + # прогона в окне, может транзиентно недосчитать → максимум одно-цикловый ложный + # алёрт, самоисцеляется следующим полным прогоном. Приемлемо (а в текущем реальном + # resume-кейсе snapshot реально сломан=9 → флаг корректен). source="kn_flats", label="DOM.РФ kn-API (квартиры)", table="kn_scrape_runs", @@ -1331,6 +1355,7 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ stale_days=14.0, critical=True, alert_on_zero_output=True, + min_output_rows=50000, ), FreshnessSource( source="objective", @@ -1340,9 +1365,16 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ fresh_days=7.0, stale_days=14.0, critical=True, - # Объектив-sync, прогнавшийся с 0 лотов за неделю — тоже поломка (тихий - # пустой extract), а не легитимный 0. Даунгрейд otherwise-fresh до failed. + # Объектив-sync, прогнавшийся с почти нулём лотов — тоже поломка (тихий + # частичный extract), а не легитимный low. objective пишет строку на group_name + # (4 группы: Свердловская/Тюмень/Пермь/Челябинск), поэтому floor сравнивается с + # recent_output = SUM(rows_lots) по группам в окне fresh_days=7d (прод: SUM 7d= + # 946264; даже наименьшая отдельная группа Челябинск=68387). Floor 1000 ловит + # вырожденный цикл с огромным запасом, не задевая нормальный разброс — и + # робастен к тому, КАКАЯ группа дописалась последней (single-newest был бы + # хрупок: 324570 или 68387 в зависимости от порядка финиша). alert_on_zero_output=True, + min_output_rows=1000, ), FreshnessSource( source="nspd", @@ -1403,15 +1435,21 @@ def compute_freshness(db: Session) -> dict[str, Any]: - last_attempt_at — max(started_at) (или created_at fallback) по всем прогонам - last_status — status последнего прогона - objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне + - recent_output — SUM(work_col) по done-прогонам цикла в окне fresh_days (NULL + если успехов в окне нет) - age_days — возраст last_success в днях (NULL если успехов не было) - freshness — fresh / stale / critical / unknown (по порогам источника) - status — ok / stale / failed (для UI-светофора) - - reason — пояснение к zero-output-даунгрейду (иначе None) + - reason — пояснение к output-floor-даунгрейду (иначе None) - Zero-output (issue #1945): для run-ledger источников с alert_on_zero_output=True - источник, который иначе был бы fresh→ok (недавний успешный прогон), но произвёл - 0 work-строк за 7d окно (objects_updated_7d == 0), ДАУНГРЕЙДИТСЯ до failed — ловит - "лоадер-прогнался-но-ничего-не-извлёк" (kn flats_count=0 ~5 недель при status=done). + Output-floor (issue #1945, redesigned #1947): для run-ledger источников с + alert_on_zero_output=True источник, который иначе был бы fresh→ok (недавний + успешный прогон), но SUM(work_col) по ВСЕМ done-прогонам цикла в окне fresh_days + (recent_output) < min_output_rows, ДАУНГРЕЙДИТСЯ до failed — ловит + "лоадер-прогнался-но-почти-ничего-не-извлёк" (kn flats_count суммарно 9 при + здоровых ~376k). Cycle-SUM (не один прогон) робастен к multi-run циклам: objective + пишет строку на group_name, kn — на region_codes + resume. Окно fresh_days (не + хардкод 7d) → здоровый недельный прогон 7–8d остаётся в окне (нет aging-FP). Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить. Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps) @@ -1496,7 +1534,25 @@ def compute_freshness(db: Session) -> dict[str, Any]: SELECT status FROM {src.table} ORDER BY {order_col} DESC NULLS LAST LIMIT 1 - ) AS last_status + ) AS last_status, + -- Cycle-sum: SUM(work_col) по ВСЕМ done-прогонам в окне + -- fresh_days. Логический цикл загрузки = НЕСКОЛЬКО прогонов + -- (objective — по строке на group_name, kn — по region_codes + -- + resume), поэтому суммируем, а не берём один прогон. + -- Окно = fresh_days (per-source), а НЕ хардкод 7d: здоровый + -- недельный прогон возрастом 7–8d остаётся ВНУТРИ окна → + -- нет aging-FP (#1947). Inner COALESCE(work_col,0) — NULL- + -- счётчик на done-прогоне вносит 0, не NULL. fresh_days — + -- float из доверенного реестра → f-string безопасен. Окно + -- через make_interval(secs => fresh_days*86400): secs — + -- double precision, поэтому дробный fresh_days валиден (у + -- days int-only → 7.5 упал бы). NB без двоеточийного cast. + SUM(COALESCE({src.work_col}, 0)) FILTER ( + WHERE status = 'done' + AND COALESCE(finished_at, {started}) + > now() + - make_interval(secs => {src.fresh_days} * 86400) + ) AS recent_output FROM {src.table} """ ), @@ -1509,6 +1565,13 @@ def compute_freshness(db: Session) -> dict[str, Any]: last_success = row["last_success_at"] if row else None last_attempt = row["last_attempt_at"] if row else None last_status = row["last_status"] if row else None + # SUM(work_col) по done-прогонам в окне fresh_days (run-ledger режим = весь + # цикл загрузки). В data-table режиме колонки нет → None; .get() терпит + # отсутствие ключа в mapping-строке. None также когда done-прогонов в окне нет. + recent_output_raw = row.get("recent_output") if row else None + recent_output: int | None = ( + int(recent_output_raw) if recent_output_raw is not None else None + ) age_days: float | None = None if last_success is not None: @@ -1530,25 +1593,32 @@ def compute_freshness(db: Session) -> dict[str, Any]: else: # critical status = "failed" - # Zero-output downgrade (issue #1945): источник прогнался успешно И недавно - # (был бы fresh→ok по возрасту), НО за окно done-прогонов произвёл 0 work-строк - # (objects_updated_7d == SUM(work_col) FILTER(status='done') в 7d == 0) → - # ДАУНГРЕЙД до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила. - # Только для run-ledger (timestamp_col is None) с alert_on_zero_output=True и - # ТОЛЬКО когда источник иначе был бы fresh/ok — по возрасту stale/failed уже - # покрыт, не двойним. last_success is None уже отсеян (status=='ok' гарантирует - # недавний успех), поэтому это ДОПОЛНИТЕЛЬНЫЙ способ завалить otherwise-fresh. + # Output-floor downgrade (issue #1945, redesigned #1947): источник прогнался + # успешно И недавно (был бы fresh→ok по возрасту), НО SUM(work_col) по всем + # done-прогонам ЦИКЛА (окно fresh_days) ниже порога min_output_rows → ДАУНГРЕЙД + # до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила. + # - cycle-SUM (а не один прогон): цикл = несколько done-прогонов (objective — + # по group_name, kn — по region_codes + resume), сумма даёт магнитуду цикла + # (objective ~946k, kn_flats ~376k healthy) → робастно к multi-run. + # - окно fresh_days (а не хардкод 7d): здоровый недельный прогон 7–8d остаётся + # внутри окна → нет aging-FP первой версии #1947. + # Только run-ledger (timestamp_col is None) c alert_on_zero_output=True и ТОЛЬКО + # когда источник иначе был бы fresh/ok (по возрасту stale/failed уже покрыт — не + # двойним). status=='ok' гарантирует недавний успех (last_success is not None). + # Строгое '<': SUM ровно на пороге → ok (порог = минимально-приемлемый объём). reason: str | None = None if ( src.alert_on_zero_output and src.timestamp_col is None and status == "ok" - and upd_7d == 0 + and recent_output is not None + and recent_output < src.min_output_rows ): status = "failed" reason = ( - f"zero-output: прогон успешен (age={age_days}d), но {src.work_col}=0 " - f"за 7d done-прогонов — лоадер прогнался, но ничего не извлёк (#1945)" + f"low-output: done-прогоны за {src.fresh_days:g}d дали " + f"SUM({src.work_col})={recent_output} < порога {src.min_output_rows} " + f"(#1945, #1947)" ) if src.critical and severity[status] > severity[worst_critical]: @@ -1564,11 +1634,15 @@ def compute_freshness(db: Session) -> dict[str, Any]: "last_status": last_status, "objects_updated_24h": upd_24h, "objects_updated_7d": upd_7d, + # SUM(work_col) по done-прогонам цикла в окне fresh_days (None для + # data-table режима / без done-прогонов в окне). Additive — база для + # output-floor-проверки, полезно и для UI-диагностики. + "recent_output": recent_output, "age_days": age_days, "thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days}, "freshness": freshness, "status": status, - # Доп. контекст при zero-output-даунгрейде (None в обычном случае). + # Доп. контекст при output-floor-даунгрейде (None в обычном случае). # Additive — не ломает существующий UI/JSON-контракт. "reason": reason, } diff --git a/backend/tests/api/v1/test_admin_scrape_freshness.py b/backend/tests/api/v1/test_admin_scrape_freshness.py index 0e787e85..f5a1c546 100644 --- a/backend/tests/api/v1/test_admin_scrape_freshness.py +++ b/backend/tests/api/v1/test_admin_scrape_freshness.py @@ -4,8 +4,9 @@ nspd / nspd_geo / cadastre — kn и kn_flats делят kn_scrape_runs, разные work_col) и для каждого считает свежесть. Здесь DB замокан: на каждый per-source SELECT возвращаем заранее заданную строку, чтобы проверить классификацию -fresh/stale/critical, маппинг status (ok/stale/failed), zero-output downgrade -(#1945) и агрегат overall_status по критичным источникам. +fresh/stale/critical, маппинг status (ok/stale/failed), output-floor downgrade +(#1945 / redesigned #1947 — recent_output = SUM(work_col) по done-прогонам цикла в +окне fresh_days < min_output_rows) и агрегат overall_status по критичным источникам. """ from __future__ import annotations @@ -68,6 +69,7 @@ def test_freshness_mixed_statuses() -> None: "upd_24h": 1500, "upd_7d": 1500, "last_status": "done", + "recent_output": 376000, # здоровый цикл flats_count → kn_flats ok }, "objective_scrape_runs": { "last_success_at": _ago(10.0), # fresh<7, stale<14 → stale @@ -75,6 +77,7 @@ def test_freshness_mixed_statuses() -> None: "upd_24h": 0, "upd_7d": 325000, "last_status": "done", + "recent_output": 325000, }, # nspd теперь data-table режим (nspd_quarter_dumps.fetched_at_utc): # last_success_at = MAX(timestamp_col), last_status = NULL (нет run-ledger). @@ -150,6 +153,7 @@ def _healthy_run_rows_except_nspd(nspd_row: dict[str, Any]) -> dict[str, dict[st "upd_24h": 10, "upd_7d": 100, "last_status": "done", + "recent_output": 100000, # выше любого floor → flagged-источники ok } rows: dict[str, dict[str, Any]] = {} for s in admin_scrape._FRESHNESS_SOURCES: @@ -224,6 +228,7 @@ def test_freshness_all_healthy() -> None: "upd_24h": 10, "upd_7d": 100, "last_status": "done", + "recent_output": 100000, # выше любого floor → flagged-источники ok } rows = { s.table: dict(fresh_row) for s in admin_scrape._FRESHNESS_SOURCES @@ -249,15 +254,18 @@ def test_freshness_critical_source_failed_drives_overall() -> None: "upd_24h": 1, "upd_7d": 1, "last_status": "done", + "recent_output": 100000, } rows = {s.table: dict(base) for s in admin_scrape._FRESHNESS_SOURCES} - # objective критичный — обнулим успех. + # objective критичный — обнулим успех (last_success=None → нет done-прогона, + # recent_output=None → floor не применяется, status=failed по отсутствию успеха). rows["objective_scrape_runs"] = { "last_success_at": None, "last_attempt_at": _ago(5.0), "upd_24h": 0, "upd_7d": 0, "last_status": "running", + "recent_output": None, } _override_db(_make_db(rows)) try: @@ -281,37 +289,48 @@ def test_classify_freshness_boundaries() -> None: assert f(100.0, 2.0, 5.0) == "critical" -def test_zero_output_downgrades_otherwise_fresh_source() -> None: - """issue #1945: kn_flats прогнался успешно И недавно (был бы fresh→ok по возрасту), - но flats_count=0 за 7d done-прогонов → DOWNGRADE до failed (alertable). +# NB по мокам output-floor: _make_db ключуется по ИМЕНИ таблицы → kn и kn_flats +# делят одну kn_scrape_runs-строку, поэтому оба видят ОДИН recent_output (per-work_col +# дивергенция objects_count vs flats_count есть только в живом SQL, не в моке). Здесь +# мок проверяет Python-сравнение (recent_output < min_output_rows + все guard'ы); +# семантику оконной SQL-SUM подтвердили на проде (objective SUM(rows_lots,7d)=946264, +# kn_flats SUM(flats_count,8d)=9 — gendesign-postgres-1, #1947). - Та же run-таблица kn_scrape_runs у kn (alert_on_zero_output=False) и kn_flats - (True). Возвращаем строку с upd_7d=0 → kn остаётся ok (нет флага = нет ложного - срабатывания), kn_flats падает в failed (zero-output) с пояснением в reason. + +def test_output_floor_downgrades_low_output_cycle() -> None: + """issue #1945 / #1947: kn_flats прогнался успешно И недавно (был бы fresh→ok), но + recent_output=SUM(flats_count) по циклу=9 < min_output_rows=50000 → failed (alertable). + + Прод-сценарий: цикл дал суммарно flats_count=9 (latest done = resume, считает хвост) + при здоровых ~376k. kn БЕЗ флага (alert_on_zero_output=False) → branch не входит, + остаётся ok даже читая ту же низкую recent_output (no false positive). """ - fresh_zero = { + # Сломанный цикл: свежий (1d), но recent_output (SUM flats_count за 8d)=9. + broken_kn = { "last_success_at": _ago(1.0), # < fresh_days=8 → был бы fresh "last_attempt_at": _ago(1.0), - "upd_24h": 0, - "upd_7d": 0, # 0 work-строк за окно done-прогонов → zero-output + "upd_24h": 9, + "upd_7d": 9, "last_status": "done", + "recent_output": 9, # SUM(flats_count) по done-прогонам цикла ≪ 50000 } - # Остальные источники свежие+ненулевые, чтобы изолировать поведение kn/kn_flats. - nonzero_fresh = { + # Остальные источники свежие+здоровые, чтобы изолировать поведение kn/kn_flats. + healthy = { "last_success_at": _ago(0.5), "last_attempt_at": _ago(0.5), - "upd_24h": 10, - "upd_7d": 100, + "upd_24h": 100, + "upd_7d": 1000, "last_status": "done", + "recent_output": 100000, # выше любого floor } rows: dict[str, dict[str, Any]] = {} for s in admin_scrape._FRESHNESS_SOURCES: if s.timestamp_col is not None: - rows[s.table] = {**nonzero_fresh, "last_status": None} + rows[s.table] = {**healthy, "last_status": None, "recent_output": None} elif s.table == "kn_scrape_runs": - rows[s.table] = dict(fresh_zero) + rows[s.table] = dict(broken_kn) else: - rows[s.table] = dict(nonzero_fresh) + rows[s.table] = dict(healthy) _override_db(_make_db(rows)) try: resp = TestClient(app).get(ENDPOINT) @@ -322,14 +341,16 @@ def test_zero_output_downgrades_otherwise_fresh_source() -> None: body = resp.json() by_src = {s["source"]: s for s in body["sources"]} - # kn_flats: otherwise-fresh, но flats_count=0 → DOWNGRADE до failed (alertable). + # kn_flats: otherwise-fresh, но recent_output=9 < 50000 → DOWNGRADE до failed. assert by_src["kn_flats"]["freshness"] == "fresh" # по возрасту был бы fresh - assert by_src["kn_flats"]["status"] == "failed" # zero-output downgrade - assert by_src["kn_flats"]["objects_updated_7d"] == 0 + assert by_src["kn_flats"]["status"] == "failed" # output-floor downgrade + assert by_src["kn_flats"]["recent_output"] == 9 assert by_src["kn_flats"]["reason"] is not None assert "flats_count" in by_src["kn_flats"]["reason"] + assert "9" in by_src["kn_flats"]["reason"] - # kn — та же таблица, upd_7d=0, но БЕЗ флага → остаётся ok (no false positive). + # kn — alert_on_zero_output=False гейтит ВЕСЬ branch → ok, даже читая recent_output=9 + # из той же mocked-строки (per-work_col дивергенция есть только в живом SQL). assert by_src["kn"]["status"] == "ok" assert by_src["kn"]["reason"] is None @@ -338,25 +359,26 @@ def test_zero_output_downgrades_otherwise_fresh_source() -> None: assert body["overall_status"] == "failed" -def test_zero_output_no_false_positive_when_nonzero() -> None: - """alert_on_zero_output=True источник с НЕнулевым выходом → остаётся ok (нет даунгрейда). +def test_output_floor_no_false_positive_when_above_floor() -> None: + """alert_on_zero_output источник с recent_output ВЫШЕ floor → остаётся ok. - Регрессия: zero-output даунгрейд НЕ должен бить по здоровому источнику, который - прогнался и реально извлёк строки. + Регрессия: output-floor НЕ должен бить по здоровому источнику, чей цикл извлёк + нормальный объём (kn_flats ~376k > 50000; objective ~946k > 1000). """ - nonzero_fresh = { + healthy = { "last_success_at": _ago(1.0), "last_attempt_at": _ago(1.0), "upd_24h": 12, - "upd_7d": 1234, # ненулевой выход + "upd_7d": 1234, "last_status": "done", + "recent_output": 376000, # здоровый цикл ≫ всех floor'ов } rows: dict[str, dict[str, Any]] = {} for s in admin_scrape._FRESHNESS_SOURCES: rows[s.table] = ( - {**nonzero_fresh, "last_status": None} + {**healthy, "last_status": None, "recent_output": None} if s.timestamp_col is not None - else dict(nonzero_fresh) + else dict(healthy) ) _override_db(_make_db(rows)) try: @@ -366,38 +388,134 @@ def test_zero_output_no_false_positive_when_nonzero() -> None: assert resp.status_code == 200, resp.text by_src = {s["source"]: s for s in resp.json()["sources"]} - # Все alert_on_zero_output-источники с ненулевым выходом → ok, reason=None. + # Все alert_on_zero_output-источники с выходом выше floor → ok, reason=None. for s in admin_scrape._FRESHNESS_SOURCES: if s.alert_on_zero_output: assert by_src[s.source]["status"] == "ok", s.source assert by_src[s.source]["reason"] is None, s.source -def test_zero_output_not_applied_when_already_stale() -> None: - """Если источник уже stale/failed по возрасту — zero-output НЕ передоунгрейдит - (age уже покрывает), и не двойнит. objective: старый успех (10d → stale) + upd_7d=0 - → остаётся stale (НЕ перепрыгивает в failed через zero-output).""" +def test_output_floor_boundary_equal_floor_stays_ok() -> None: + """Граница: recent_output == min_output_rows → остаётся ok (проверка строгое '<'). + + kn_flats recent_output ровно 50000 (= floor) → НЕ даунгрейдится (порог = минимально- + приемлемый объём, на нём ещё ok). Регрессия на off-by-one в сравнении. + """ + kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats") + at_floor = { + "last_success_at": _ago(1.0), + "last_attempt_at": _ago(1.0), + "upd_24h": 0, + "upd_7d": kn_flats.min_output_rows, + "last_status": "done", + "recent_output": kn_flats.min_output_rows, # ровно на пороге → ok + } + healthy = { + "last_success_at": _ago(0.5), + "last_attempt_at": _ago(0.5), + "upd_24h": 100, + "upd_7d": 1000, + "last_status": "done", + "recent_output": 1000000, + } + rows: dict[str, dict[str, Any]] = {} + for s in admin_scrape._FRESHNESS_SOURCES: + if s.timestamp_col is not None: + rows[s.table] = {**healthy, "last_status": None, "recent_output": None} + elif s.table == "kn_scrape_runs": + rows[s.table] = dict(at_floor) + else: + rows[s.table] = dict(healthy) + _override_db(_make_db(rows)) + try: + resp = TestClient(app).get(ENDPOINT) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + by_src = {s["source"]: s for s in resp.json()["sources"]} + assert by_src["kn_flats"]["recent_output"] == kn_flats.min_output_rows + assert by_src["kn_flats"]["status"] == "ok" # ровно на пороге → НЕ failed + assert by_src["kn_flats"]["reason"] is None + + +def test_output_floor_no_false_positive_on_weekly_window_aging() -> None: + """#1947 fix: здоровый недельный прогон возрастом 7–8d остаётся в окне fresh_days=8d + → recent_output здоров (376k) → НЕ ложно-флагуется. + + Регрессия на самый первый #1947-баг: оконная SUM была хардкод 7d, поэтому прогон + 7–8d (>7d назад) давал SUM=0 при status=ok → ложный downgrade ~раз/неделю. Теперь + окно = fresh_days (8d) → прогон 7.5d ВНУТРИ окна, recent_output=376k > floor → ok. + upd_7d=0 (мок старого 7d-счётчика) НЕ влияет — floor смотрит recent_output. + """ + weekly_aged = { + "last_success_at": _ago(7.5), # < fresh_days=8 → всё ещё fresh, и внутри окна + "last_attempt_at": _ago(7.5), + "upd_24h": 0, + "upd_7d": 0, # старый 7d-счётчик: прогон >7d назад → 0 (раньше ложно флагал) + "last_status": "done", + "recent_output": 376000, # окно fresh_days=8d ловит прогон 7.5d → цикл здоров + } + healthy = { + "last_success_at": _ago(0.5), + "last_attempt_at": _ago(0.5), + "upd_24h": 50, + "upd_7d": 500, + "last_status": "done", + "recent_output": 100000, + } + rows: dict[str, dict[str, Any]] = {} + for s in admin_scrape._FRESHNESS_SOURCES: + if s.timestamp_col is not None: + rows[s.table] = {**healthy, "last_status": None, "recent_output": None} + elif s.table == "kn_scrape_runs": + rows[s.table] = dict(weekly_aged) + else: + rows[s.table] = dict(healthy) + _override_db(_make_db(rows)) + try: + resp = TestClient(app).get(ENDPOINT) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + by_src = {s["source"]: s for s in resp.json()["sources"]} + # kn_flats: fresh по возрасту (7.5<8), upd_7d=0 НЕ важен — floor смотрит + # recent_output (376k > 50000, окно fresh_days покрыло прогон) → ok, без downgrade. + assert by_src["kn_flats"]["freshness"] == "fresh" + assert by_src["kn_flats"]["status"] == "ok" + assert by_src["kn_flats"]["objects_updated_7d"] == 0 + assert by_src["kn_flats"]["recent_output"] == 376000 + assert by_src["kn_flats"]["reason"] is None + + +def test_output_floor_not_applied_when_already_stale() -> None: + """Если источник уже stale/failed по возрасту — output-floor НЕ передоунгрейдит + (age уже покрывает), и не двойнит. objective: старый успех (10d → stale) + + recent_output=0 → остаётся stale (НЕ перепрыгивает в failed через floor).""" base_fresh = { "last_success_at": _ago(0.5), "last_attempt_at": _ago(0.5), "upd_24h": 5, "upd_7d": 50, "last_status": "done", + "recent_output": 500000, } rows: dict[str, dict[str, Any]] = {} for s in admin_scrape._FRESHNESS_SOURCES: rows[s.table] = ( - {**base_fresh, "last_status": None} + {**base_fresh, "last_status": None, "recent_output": None} if s.timestamp_col is not None else dict(base_fresh) ) - # objective: stale по возрасту (10d: fresh<7, stale<14) + 0 выход. + # objective: stale по возрасту (10d: fresh<7, stale<14) + нулевой выход. rows["objective_scrape_runs"] = { "last_success_at": _ago(10.0), "last_attempt_at": _ago(10.0), "upd_24h": 0, "upd_7d": 0, "last_status": "done", + "recent_output": 0, } _override_db(_make_db(rows)) try: @@ -407,19 +525,21 @@ def test_zero_output_not_applied_when_already_stale() -> None: assert resp.status_code == 200, resp.text by_src = {s["source"]: s for s in resp.json()["sources"]} - # zero-output применяется ТОЛЬКО к otherwise-ok; stale остаётся stale, reason=None. + # output-floor применяется ТОЛЬКО к otherwise-ok; stale остаётся stale, reason=None. assert by_src["objective"]["status"] == "stale" assert by_src["objective"]["reason"] is None def test_kn_flats_registry_config() -> None: - """kn_flats: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output, - критичный, недельные пороги. kn (objects_count) остаётся без флага (unchanged).""" + """kn_flats: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output + + min_output_rows=50000, критичный, недельные пороги. kn (objects_count) — без флага.""" kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats") assert kn_flats.table == "kn_scrape_runs" assert kn_flats.work_col == "flats_count" assert kn_flats.timestamp_col is None # run-ledger режим assert kn_flats.alert_on_zero_output is True + # healthy цикл ~376k ≫ floor ≫ broken-sum ≤3670 (прод #1947). + assert kn_flats.min_output_rows == 50000 assert kn_flats.critical is True assert kn_flats.fresh_days == 8.0 assert kn_flats.stale_days == 14.0 @@ -427,14 +547,19 @@ def test_kn_flats_registry_config() -> None: kn = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn") assert kn.work_col == "objects_count" assert kn.alert_on_zero_output is False # существующий источник unchanged + assert kn.min_output_rows == 1 # default — не проверяется без флага objective = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "objective") assert objective.alert_on_zero_output is True + # healthy SUM(rows_lots,7d)~946k по 4 группам; floor 1000 с огромным запасом. + assert objective.min_output_rows == 1000 - # Legitimate-0 источники НЕ должны иметь флаг (избегаем ложных срабатываний). + # Legitimate-low источники НЕ должны иметь флаг (избегаем ложных срабатываний); + # min_output_rows остаётся default=1 и не проверяется. for name in ("nspd", "nspd_geo", "cadastre"): src = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == name) assert src.alert_on_zero_output is False, name + assert src.min_output_rows == 1, name def test_kn_thresholds_match_weekly_cadence() -> None: -- 2.45.3