fix(tradein): апсерт не переписывает неизменившуюся строку — гейт IS DISTINCT FROM + МСК-день (#2992)

Часть #2992. Замер «после» — 22.08 ~09:00 UTC.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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,

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

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)

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,