From 6cdf56820d5db14a26723cdfca3487671b5f3abc Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 21 Aug 2026 13:57:09 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20=D0=B0=D0=BF=D1=81=D0=B5=D1=80?= =?UTF-8?q?=D1=82=20=D0=BD=D0=B5=20=D0=BF=D0=B5=D1=80=D0=B5=D0=BF=D0=B8?= =?UTF-8?q?=D1=81=D1=8B=D0=B2=D0=B0=D0=B5=D1=82=20=D0=BD=D0=B5=D0=B8=D0=B7?= =?UTF-8?q?=D0=BC=D0=B5=D0=BD=D0=B8=D0=B2=D1=88=D1=83=D1=8E=D1=81=D1=8F=20?= =?UTF-8?q?=D1=81=D1=82=D1=80=D0=BE=D0=BA=D1=83=20=E2=80=94=20=D0=B3=D0=B5?= =?UTF-8?q?=D0=B9=D1=82=20IS=20DISTINCT=20FROM=20+=20=D0=9C=D0=A1=D0=9A-?= =?UTF-8?q?=D0=B4=D0=B5=D0=BD=D1=8C=20(#2992)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (полное описание — в PR) Co-Authored-By: Claude Opus 5 --- .../backend/app/services/matching/listings.py | 20 ++ tradein-mvp/backend/tests/skip_allowlist.txt | 12 + .../tests/test_2992_upsert_unchanged_gate.py | 258 ++++++++++++++++++ .../scraper-kit/src/scraper_kit/base.py | 86 +++++- 4 files changed, 374 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_2992_upsert_unchanged_gate.py diff --git a/tradein-mvp/backend/app/services/matching/listings.py b/tradein-mvp/backend/app/services/matching/listings.py index ceaae20c..398bbefa 100644 --- a/tradein-mvp/backend/app/services/matching/listings.py +++ b/tradein-mvp/backend/app/services/matching/listings.py @@ -287,6 +287,26 @@ def _upsert_listing_source( floor = COALESCE(EXCLUDED.floor, listing_sources.floor), rooms_count = COALESCE(EXCLUDED.rooms_count, listing_sources.rooms_count), raw_payload = COALESCE(EXCLUDED.raw_payload, listing_sources.raw_payload) + -- #2992: симметричный гейт к listings-апсерту (scraper_kit/base.py). Вторая + -- по нагрузке таблица базы: 10.27 млн апдейтов на 102 тыс. строк, HOT 0.10 %. + -- Апдейт идёт, если итоговое значение хоть одной колонки отличается ИЛИ + -- last_seen_at ещё не сегодняшний по МСК. Тот же МСК-критерий, что у listings, + -- — иначе разъехались бы listings.last_seen_at и listing_sources.last_seen_at, + -- равные сегодня у 100 % пар (см. докстринг выше). + WHERE ( + listing_sources.confidence, listing_sources.price_rub, + listing_sources.area_m2, listing_sources.floor, + listing_sources.rooms_count, listing_sources.raw_payload + ) IS DISTINCT FROM ( + GREATEST(EXCLUDED.confidence, listing_sources.confidence), + COALESCE(EXCLUDED.price_rub, listing_sources.price_rub), + COALESCE(EXCLUDED.area_m2, listing_sources.area_m2), + COALESCE(EXCLUDED.floor, listing_sources.floor), + COALESCE(EXCLUDED.rooms_count, listing_sources.rooms_count), + COALESCE(EXCLUDED.raw_payload, listing_sources.raw_payload) + ) + OR (listing_sources.last_seen_at AT TIME ZONE 'Europe/Moscow')::date + IS DISTINCT FROM (statement_timestamp() AT TIME ZONE 'Europe/Moscow')::date """), { "lid": listing_id, diff --git a/tradein-mvp/backend/tests/skip_allowlist.txt b/tradein-mvp/backend/tests/skip_allowlist.txt index ee3c6e89..4883fa72 100644 --- a/tradein-mvp/backend/tests/skip_allowlist.txt +++ b/tradein-mvp/backend/tests/skip_allowlist.txt @@ -93,3 +93,15 @@ tests/test_coverage_probe_endpoint.py::test_major1_cohort_excludes_novostroyki_a # `days_on_market <= :max_age_days` встречается в SQL дважды). На мок-лэйне # (deploy-tradein.yml, DSN-заглушка) реальной БД нет — self-skip. tests/test_coverage_probe_endpoint.py::test_max_age_outlier_excluded_from_median_live + +# #2992 — гейт неизменной строки в апсертах listings / listing_sources. Шесть live-тестов +# идут через реальный save_listings / upsert_listing_source на живом Postgres и меряют, +# переписана ли строка, по ctid (save_listings коммитит — транзакционные счётчики +# бесполезны). Без БД — skip; в CI Trade-In идут на postgres-сервисе (#2745). Свои +# строки t2992-* тесты удаляют явно в finally. +tests/test_2992_upsert_unchanged_gate.py::test_unchanged_rescrape_same_day_does_not_update_the_row +tests/test_2992_upsert_unchanged_gate.py::test_changed_price_still_updates +tests/test_2992_upsert_unchanged_gate.py::test_coalesce_backfill_still_updates +tests/test_2992_upsert_unchanged_gate.py::test_next_day_rescrape_updates_even_if_unchanged +tests/test_2992_upsert_unchanged_gate.py::test_skipped_row_still_yields_listing_id_for_downstream +tests/test_2992_upsert_unchanged_gate.py::test_listing_sources_unchanged_rescrape_same_day_does_not_update diff --git a/tradein-mvp/backend/tests/test_2992_upsert_unchanged_gate.py b/tradein-mvp/backend/tests/test_2992_upsert_unchanged_gate.py new file mode 100644 index 00000000..0b3a0703 --- /dev/null +++ b/tradein-mvp/backend/tests/test_2992_upsert_unchanged_gate.py @@ -0,0 +1,258 @@ +"""Повторный скрейп неизменившегося объявления не переписывает строку (#2992). + +Апсерт listings безусловно присваивал 41 колонку при каждом повторном скрейпе — +заново тостил description и плодил TOAST-чанки. Прод 2026-08-20: n_tup_upd +20.87 млн на 107 тыс. строк (198 апдейтов на строку за 91 день), HOT 0.44 %, +TOAST 15 ГБ при ~230 МБ полезного, WAL 7 ГБ/сутки. + +Гейт в `ON CONFLICT DO UPDATE … WHERE`: апдейт идёт, если итоговое +(post-COALESCE) значение хоть одной контентной колонки отличается ИЛИ +last_seen_at ещё не сегодняшний по МСК. Второе условие обязательно: метка +живости должна сдвигаться хотя бы раз в сутки — на неё завязаны деактиватор, +снапшоты, эстиматор (#2206), монитор свежести. + +Проверяется на ЖИВОМ Postgres через реальный save_listings, в транзакции с +откатом. Счётчик апдейтов — pg_stat_xact_user_tables (текущая транзакция), +то есть ровно «повторный скрейп не увеличивает n_tup_upd» из acceptance. +Без БД — skip (как соседние live-тесты; в CI Trade-In есть postgres-сервис). +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import uuid +from typing import Any +from unittest.mock import MagicMock + +import pytest +from scraper_kit.base import ScrapedLot, save_listings +from sqlalchemy import text + + +def _live_session() -> Any | None: + try: + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "") + if not dsn or "localhost:5432/test" in dsn: + return None + engine = create_engine(dsn, future=True) + conn = engine.connect() + conn.execute(text("SELECT 1")) + conn.close() + return sessionmaker(bind=engine, future=True)() + except Exception: + return None + + +def _matcher() -> MagicMock: + m = MagicMock() + m.match_or_create_house.return_value = (None, 0.0, "no_address") + m.upsert_listing_source.return_value = None + return m + + +def _lot( + src_id: str, + price: int = 5_000_000, + description: str | None = "тестовое описание", + address: str | None = None, +) -> ScrapedLot: + return ScrapedLot( + source="cian", + source_url=f"https://ekb.cian.ru/sale/flat/t2992-{src_id}/", + source_id=f"t2992-{src_id}", + price_rub=price, + description=description, + address=address, + ) + + +def _row(db: Any, src_id: str) -> Any: + """Строка + её физическая версия (ctid). save_listings КОММИТИТ, поэтому + транзакционные счётчики pg_stat_xact_* тут бесполезны (сбрасываются). UPDATE + всегда создаёт новую версию строки → ctid меняется; пропущенный апдейт → ctid + тот же. Это и есть прямой замер «переписана ли строка».""" + return db.execute( + text( + "SELECT id, ctid::text AS ctid, price_rub, city, description, last_seen_at " + "FROM listings WHERE source = 'cian' AND source_id = :sid" + ), + {"sid": f"t2992-{src_id}"}, + ).fetchone() + + +def _cleanup(db: Any) -> None: + """save_listings коммитит — rollback не спасает. Убираем свои строки явно.""" + try: + db.rollback() + db.execute( + text( + "DELETE FROM listings_snapshots WHERE listing_id IN " + "(SELECT id FROM listings WHERE source='cian' AND source_id LIKE 't2992-%')" + ) + ) + db.execute(text("DELETE FROM listing_sources WHERE ext_id LIKE 't2992-%'")) + db.execute(text("DELETE FROM listings WHERE source='cian' AND source_id LIKE 't2992-%'")) + db.commit() + finally: + db.close() + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_unchanged_rescrape_same_day_does_not_update_the_row() -> None: + """Головной: второй апсерт неизменного лота в тот же МСК-день — n_tup_upd не растёт. + + На origin/main (без гейта) второй save_listings переписывает строку — счётчик +1. + """ + db = _live_session() + sid = uuid.uuid4().hex[:8] + try: + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66) + first = _row(db, sid) + assert first is not None, "первая вставка не прошла" + ins, upd = save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66) + second = _row(db, sid) + assert ( + second.ctid == first.ctid + ), f"повторный скрейп неизменного лота переписал строку: ctid {first.ctid}→{second.ctid}" + assert (ins, upd) == (0, 0), f"счётчики: inserted={ins} updated={upd}, ждали 0/0" + assert second.last_seen_at == first.last_seen_at, "last_seen_at сдвинулся без нужды" + finally: + _cleanup(db) + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_changed_price_still_updates() -> None: + """Контроль: изменение контента проходит — гейт не глушит настоящие изменения.""" + db = _live_session() + sid = uuid.uuid4().hex[:8] + try: + save_listings(db, [_lot(sid, price=5_000_000)], matcher=_matcher(), region_code=66) + first = _row(db, sid) + save_listings(db, [_lot(sid, price=5_100_000)], matcher=_matcher(), region_code=66) + second = _row(db, sid) + assert second.ctid != first.ctid, "изменение цены не записалось (строка не переписана)" + assert second.price_rub == 5_100_000 + finally: + _cleanup(db) + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_coalesce_backfill_still_updates() -> None: + """Контроль: COALESCE-дозаполнение (город NULL → значение) проходит. + + Именно за это забраковали гейт по card_hash: он бы блокировал дозаполнение. + Сравнение идёт по ИТОГОВЫМ post-COALESCE значениям — если дописал, строка + отличается. + """ + db = _live_session() + sid = uuid.uuid4().hex[:8] + try: + # city приходит ПАРАМЕТРОМ save_listings (развёртка города), не полем лота. + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66, city=None) + first = _row(db, sid) + assert first.city is None + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66, city="Екатеринбург") + second = _row(db, sid) + assert second.ctid != first.ctid, "дозаполнение города заблокировано гейтом" + assert second.city == "Екатеринбург" + finally: + _cleanup(db) + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_next_day_rescrape_updates_even_if_unchanged() -> None: + """Живость: если last_seen_at вчерашний — апдейт идёт и без изменений контента. + + Иначе деактиватор/снапшоты/эстиматор потеряли бы живой инвентарь. Симулируем + «вчера», отодвинув last_seen_at у только что вставленной строки. + """ + db = _live_session() + sid = uuid.uuid4().hex[:8] + try: + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66) + db.execute( + text( + "UPDATE listings SET last_seen_at = last_seen_at - interval '1 day' " + "WHERE source='cian' AND source_id = :sid" + ), + {"sid": f"t2992-{sid}"}, + ) + db.commit() + first = _row(db, sid) + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66) + second = _row(db, sid) + assert second.ctid != first.ctid, "суточное ре-подтверждение живости заблокировано" + assert second.last_seen_at > first.last_seen_at, "last_seen_at не сдвинулся" + finally: + _cleanup(db) + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_skipped_row_still_yields_listing_id_for_downstream() -> None: + """При пропуске апдейта listing_id всё равно известен — downstream (снапшот, + матчинг, listing_sources) идёт; иначе гейт молча отключил бы сопоставление.""" + db = _live_session() + sid = uuid.uuid4().hex[:8] + m = _matcher() + lot = _lot(sid, address="Екатеринбург, ул. Тестовая, 1") # матчинг идёт только с адресом/гео + try: + save_listings(db, [lot], matcher=m, region_code=66) + calls_before = m.match_or_create_house.call_count + assert ( + calls_before == 1 + ), "контроль сконструирован неверно: матчинг не зовётся и в первый раз" + save_listings(db, [lot], matcher=m, region_code=66) + assert ( + m.match_or_create_house.call_count == calls_before + 1 + ), "при пропущенном апдейте матчинг не вызван — listing_id потерян" + finally: + _cleanup(db) + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB") +def test_listing_sources_unchanged_rescrape_same_day_does_not_update() -> None: + """Симметричный гейт на listing_sources — второй по нагрузке таблице (10.27 млн + апдейтов, HOT 0.10 %). Повторный upsert_listing_source того же лота в тот же + МСК-день не переписывает строку; изменение цены — переписывает.""" + from app.services.matching.listings import upsert_listing_source + + db = _live_session() + sid = uuid.uuid4().hex[:8] + try: + save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66) + lid = int(_row(db, sid).id) + + def ls_row() -> Any: + return db.execute( + text( + "SELECT ctid::text AS ctid, price_rub, last_seen_at FROM listing_sources " + "WHERE ext_source='cian' AND ext_id = :e" + ), + {"e": f"t2992-{sid}"}, + ).fetchone() + + kw = dict(ext_source="cian", ext_id=f"t2992-{sid}", method="test", confidence=0.9) + upsert_listing_source(db, listing_id=lid, price_rub=5_000_000, **kw) + db.commit() + first = ls_row() + assert first is not None + upsert_listing_source(db, listing_id=lid, price_rub=5_000_000, **kw) + db.commit() + second = ls_row() + assert ( + second.ctid == first.ctid + ), f"listing_sources переписана без изменений: ctid {first.ctid}→{second.ctid}" + upsert_listing_source(db, listing_id=lid, price_rub=5_100_000, **kw) + db.commit() + third = ls_row() + assert ( + third.ctid != second.ctid and third.price_rub == 5_100_000 + ), "изменение цены не записалось" + finally: + _cleanup(db) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py index 58ad9b0d..d250af19 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py @@ -382,6 +382,7 @@ def save_listings( updated = 0 skipped = 0 reconciled = 0 # UPDATE by (source,source_id) при dedup_hash-дрейфе + unchanged_today = 0 # #2992: гейт апсерта пропустил неизменную, уже виденную сегодня строку matched = 0 match_failures = 0 geo_guard_dropped = 0 # city NULL'ен из-за geo-guard (лот вне city_radius_km от anchor'а) @@ -423,7 +424,7 @@ def save_listings( # never the old ones, so a lightweight SELECT is the clean way. # None → row did not exist yet (fresh insert) → always process. prior_row = db.execute( - text("SELECT card_hash, last_seen_at FROM listings WHERE dedup_hash = :dedup"), + text("SELECT id, card_hash, last_seen_at FROM listings WHERE dedup_hash = :dedup"), {"dedup": dedup}, ).fetchone() prior_card_hash = prior_row.card_hash if prior_row is not None else None @@ -681,6 +682,81 @@ def save_listings( EXCLUDED.newbuilding_url, listings.newbuilding_url ), card_hash = EXCLUDED.card_hash + -- #2992: гейт неизменной строки. Без него каждый повторный скрейп + -- переписывал 41 колонку, заново тостил description и плодил TOAST-чанки: + -- прод 2026-08-20 — n_tup_upd 20.87 млн на 107 тыс. строк (198 апдейтов на + -- строку за 91 день), HOT 0.44 %, TOAST 15 ГБ при ~230 МБ полезного, WAL + -- 7 ГБ/сутки. Детектор изменений (card_hash) есть рядом, но гейта в апсерте + -- не было. + -- + -- ДВА условия, при любом из которых апдейт ИДЁТ: + -- (1) итоговое (post-COALESCE) значение хотя бы одной контентной колонки + -- отличается от текущего — IS DISTINCT FROM по кортежу, NULL-safe. + -- Сравниваются ИТОГОВЫЕ значения, а не EXCLUDED: так COALESCE- + -- дозаполнение (адрес, город, сегмент…) по-прежнему проходит — если + -- дописал, строка отличается. Гейт по card_hash был забракован именно + -- за то, что ломает дозаполнение. + -- (2) last_seen_at ещё не сегодняшний по МСК — метка живости ОБЯЗАНА + -- сдвигаться хотя бы раз в сутки: на неё завязаны деактиватор (TTL), + -- снапшоты (is_active = last_seen_at > now()-7d), эстиматор + -- (scraped_at > NOW()-14d, #2206), монитор свежести. Это та же + -- семантика, что у skip_seen_today (МСК-день), но в SQL, race-free и + -- для ВСЕХ путей записи, а не трёх full_load. + -- + -- При пропуске RETURNING пуст → listing_id берётся из pre-read (prior_row.id), + -- downstream (снапшот, матчинг, listing_sources) идёт как прежде. + -- listing_sources получает симметричный гейт (matching/listings.py) — иначе + -- разъехались бы listings.last_seen_at и listing_sources.last_seen_at, + -- которые сегодня равны у 100 % пар. + WHERE ( + listings.price_rub, listings.price_per_m2, listings.living_area_m2, + listings.bedrooms_count, listings.balconies_count, listings.loggias_count, + listings.description_minhash, listings.cadastral_number, + listings.building_cadastral_number, listings.phones, listings.is_homeowner, + listings.is_pro_seller, listings.bargain_allowed, listings.sale_type, + listings.metro_stations, listings.listing_date, listings.area_m2, + listings.address, listings.city, listings.kitchen_area_m2, + listings.ceiling_height_m, listings.mortgage_available, + listings.is_apartments, listings.is_rosreestr_checked, + listings.listing_segment, listings.publish_date, listings.days_on_market, + listings.description, listings.agency_name, listings.yandex_offer_id, + listings.predicted_price_rub, listings.predicted_price_min, + listings.predicted_price_max, listings.price_trend, + listings.price_previous_rub, listings.newbuilding_id, + listings.newbuilding_url, listings.card_hash, listings.is_active + ) IS DISTINCT FROM ( + EXCLUDED.price_rub, EXCLUDED.price_per_m2, EXCLUDED.living_area_m2, + EXCLUDED.bedrooms_count, EXCLUDED.balconies_count, EXCLUDED.loggias_count, + EXCLUDED.description_minhash, EXCLUDED.cadastral_number, + EXCLUDED.building_cadastral_number, EXCLUDED.phones, EXCLUDED.is_homeowner, + EXCLUDED.is_pro_seller, EXCLUDED.bargain_allowed, EXCLUDED.sale_type, + EXCLUDED.metro_stations, + COALESCE(EXCLUDED.listing_date, listings.listing_date), + COALESCE(EXCLUDED.area_m2, listings.area_m2), + COALESCE(listings.address, EXCLUDED.address), + COALESCE(EXCLUDED.city, listings.city), + COALESCE(EXCLUDED.kitchen_area_m2, listings.kitchen_area_m2), + COALESCE(EXCLUDED.ceiling_height_m, listings.ceiling_height_m), + COALESCE(EXCLUDED.mortgage_available, listings.mortgage_available), + COALESCE(EXCLUDED.is_apartments, listings.is_apartments), + COALESCE(EXCLUDED.is_rosreestr_checked, listings.is_rosreestr_checked), + COALESCE(EXCLUDED.listing_segment, listings.listing_segment), + COALESCE(EXCLUDED.publish_date, listings.publish_date), + COALESCE(EXCLUDED.days_on_market, listings.days_on_market), + COALESCE(EXCLUDED.description, listings.description), + COALESCE(EXCLUDED.agency_name, listings.agency_name), + COALESCE(EXCLUDED.yandex_offer_id, listings.yandex_offer_id), + COALESCE(EXCLUDED.predicted_price_rub, listings.predicted_price_rub), + COALESCE(EXCLUDED.predicted_price_min, listings.predicted_price_min), + COALESCE(EXCLUDED.predicted_price_max, listings.predicted_price_max), + COALESCE(EXCLUDED.price_trend, listings.price_trend), + COALESCE(EXCLUDED.price_previous_rub, listings.price_previous_rub), + COALESCE(EXCLUDED.newbuilding_id, listings.newbuilding_id), + COALESCE(EXCLUDED.newbuilding_url, listings.newbuilding_url), + EXCLUDED.card_hash, true + ) + OR (listings.last_seen_at AT TIME ZONE 'Europe/Moscow')::date + IS DISTINCT FROM (statement_timestamp() AT TIME ZONE 'Europe/Moscow')::date RETURNING id, (xmax = 0) AS inserted """ ) @@ -809,6 +885,11 @@ def save_listings( inserted += 1 else: updated += 1 + elif prior_row is not None: + # #2992: гейт пропустил апдейт — строка не изменилась и уже видена + # сегодня. listing_id известен из pre-read; downstream идёт как прежде. + listing_id = int(prior_row.id) + unchanged_today += 1 # ── Snapshot: point-in-time observation in listings_snapshots ─── # Fault-tolerant: failure here MUST NOT abort the listings batch. @@ -881,11 +962,12 @@ def save_listings( db.commit() logger.info( - "save_listings: source=%s inserted=%d updated=%d reconciled=%d " + "save_listings: source=%s inserted=%d updated=%d unchanged_today=%d reconciled=%d " "skipped_seen_today=%d matched=%d match_failures=%d geo_guard_dropped=%d (total %d)", lots[0].source if lots else "?", inserted, updated, + unchanged_today, reconciled, skipped, matched,