fix(site-finder): address PR #116 auto-review M1-M5

M1 (mutation risk): replace EMPTY_DUMP_RESULT direct refs with
_make_empty_result() factory. dict(...) shallow copy left nested
nspd_dump shared by reference across concurrent requests — single
mutation pollutes module sentinel for all subsequent calls. Now
каждый caller gets independent dict.

M2 (O(N) spatial scan): SELECT extended denormalized counts
(territorial_zones_count, zouit_count, engineering_count). Each
spatial helper accepts layer_counts and early-returns when count=0
— skips heavy jsonb_array_elements + ST_Transform/ST_Intersects
scan entirely. Critical для quarters с 2000+ features.

M4 (documentation): _trigger_harvest docstring describes known
burst/no-dedup limitation + TODO Redis SETNX (отдельный PR).

M5 (test fragility): _make_db_mock_with_spatial docstring describes
positional-call contract — db.execute order (0=dump, 1=zoning, 2=zouit,
3=engineering) и зависимость от count-values.

+4 new tests (17 total, all pass):
- test_make_empty_result_returns_independent_copies (mutation safety)
- test_make_empty_result_overrides
- test_early_exit_all_counts_zero_no_spatial_queries
- test_early_exit_partial_counts

Per auto-review on 3068a9c.

backend/app/services/site_finder/quarter_dump_lookup.py

@@ -29,7 +29,8 @@ _DUMP_MAX_AGE_DAYS = 180
 # Радиус поиска инженерных сооружений (метры) — договорённость #44 I3.
 _ENGINEERING_RADIUS_M = 200
 
-# Пустой результат — возвращается когда дамп недоступен, устарел или при ошибке.
+# Sentinel для isinstance-проверок и read-only fallback в parcels.py try/except.
+# НИКОГДА не мутировать — использовать _make_empty_result() для новых dict.
 EMPTY_DUMP_RESULT: dict[str, Any] = {
     "nspd_zoning": None,
     "nspd_zouit_overlaps": [],
@@ -44,6 +45,31 @@ EMPTY_DUMP_RESULT: dict[str, Any] = {
 }
 
 
+def _make_empty_result(
+    *,
+    fetched_at_utc: str | None = None,
+    stale: bool = False,
+    harvest_triggered: bool = False,
+    total_features: int | None = None,
+) -> dict[str, Any]:
+    """Создаёт свежую копию empty-dump result с возможностью переопределить поля.
+
+    Вызывать вместо EMPTY_DUMP_RESULT напрямую — чтобы не мутировать singleton.
+    """
+    return {
+        "nspd_zoning": None,
+        "nspd_zouit_overlaps": [],
+        "nspd_engineering_nearby": [],
+        "nspd_dump": {
+            "available": False,
+            "fetched_at_utc": fetched_at_utc,
+            "stale": stale,
+            "harvest_triggered": harvest_triggered,
+            "total_features": total_features,
+        },
+    }
+
+
 def derive_quarter_cad(cad_num: str) -> str | None:
     """3-сегментный кадастровый квартал из любого кадастрового номера.
 
@@ -80,21 +106,23 @@ def get_quarter_dump_data(
     Если parcel_wkt=None — возвращает только freshness metadata (нет geom для
     spatial queries).
     """
-    empty_result = EMPTY_DUMP_RESULT
-
     quarter = derive_quarter_cad(cad_num)
     if quarter is None:
         logger.warning("quarter_dump_lookup: cannot derive quarter from cad=%s", cad_num)
-        return empty_result
+        return _make_empty_result()
 
-    # Читаем строку дампа из БД
+    # Читаем строку дампа из БД. Денормализованные счётчики слоёв используются
+    # для early-exit в spatial helpers (M2 mitigation).
     row = db.execute(
         text(
             """
             SELECT quarter_cad,
                    fetched_at_utc,
                    total_features,
-                   harvest_error
+                   harvest_error,
+                   territorial_zones_count,
+                   zouit_count,
+                   engineering_count
               FROM nspd_quarter_dumps
              WHERE quarter_cad = :q
             """
@@ -108,15 +136,7 @@ def get_quarter_dump_data(
     if row is None:
         # Дампа нет — ставим harvest в очередь
         harvest_triggered = _trigger_harvest(quarter)
-        result = dict(empty_result)
+        return _make_empty_result(harvest_triggered=harvest_triggered)
-        result["nspd_dump"] = {
-            "available": False,
-            "fetched_at_utc": None,
-            "stale": False,
-            "harvest_triggered": harvest_triggered,
-            "total_features": None,
-        }
-        return result
 
     fetched_at: datetime = row[1]
     # Убедимся что timezone-aware для корректного сравнения
@@ -125,6 +145,9 @@ def get_quarter_dump_data(
 
     total_features: int | None = row[2]
     harvest_error: str | None = row[3]
+    territorial_zones_count: int = row[4] or 0
+    zouit_count: int = row[5] or 0
+    engineering_count: int = row[6] or 0
 
     is_stale = (now - fetched_at) > max_age
     has_error = harvest_error is not None
@@ -132,15 +155,12 @@ def get_quarter_dump_data(
     # Устаревший или с ошибкой — триггерим повторный harvest
     if is_stale or has_error:
         harvest_triggered = _trigger_harvest(quarter)
-        result = dict(empty_result)
+        return _make_empty_result(
-        result["nspd_dump"] = {
+            fetched_at_utc=fetched_at.isoformat(),
-            "available": False,
+            stale=is_stale,
-            "fetched_at_utc": fetched_at.isoformat(),
+            harvest_triggered=harvest_triggered,
-            "stale": is_stale,
+            total_features=total_features,
-            "harvest_triggered": harvest_triggered,
+        )
-            "total_features": total_features,
-        }
-        return result
 
     # Свежий дамп без ошибок — извлекаем spatial данные
     dump_meta: dict[str, Any] = {
@@ -160,9 +180,14 @@ def get_quarter_dump_data(
             "nspd_dump": dump_meta,
         }
 
-    nspd_zoning = _get_zoning(db, quarter, parcel_wkt)
+    layer_counts = {
-    nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt)
+        "territorial_zones_count": territorial_zones_count,
-    nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt)
+        "zouit_count": zouit_count,
+        "engineering_count": engineering_count,
+    }
+    nspd_zoning = _get_zoning(db, quarter, parcel_wkt, layer_counts)
+    nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt, layer_counts)
+    nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt, layer_counts)
 
     return {
         "nspd_zoning": nspd_zoning,
@@ -175,11 +200,20 @@ def get_quarter_dump_data(
 # ── Spatial helpers ───────────────────────────────────────────────────────────
 
 
-def _get_zoning(db: Session, quarter: str, parcel_wkt: str) -> dict[str, Any] | None:
+def _get_zoning(
+    db: Session,
+    quarter: str,
+    parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
+) -> dict[str, Any] | None:
     """G1: ПЗЗ территориальная зона по centroid участка из dump.
 
     Geometry в features_json в EPSG:3857 — ST_Transform на чтении → 4326.
+    layer_counts — денормализованные счётчики из строки дампа. Если
+    territorial_zones_count == 0 — пропускаем heavy jsonb_array_elements scan.
     """
+    if layer_counts is not None and layer_counts.get("territorial_zones_count", 1) == 0:
+        return None
     try:
         row = db.execute(
             text(
@@ -228,12 +262,17 @@ def _get_zouit_overlaps(
     db: Session,
     quarter: str,
     parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
 ) -> list[dict[str, Any]]:
     """G3: ЗОУИТ которые пересекают участок.
 
     Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
     zouit_protected, zouit_other.
+    layer_counts — денормализованные счётчики. Если zouit_count == 0 —
+    пропускаем heavy jsonb_array_elements scan.
     """
+    if layer_counts is not None and layer_counts.get("zouit_count", 1) == 0:
+        return []
     try:
         rows = db.execute(
             text(
@@ -285,8 +324,15 @@ def _get_engineering_nearby(
     db: Session,
     quarter: str,
     parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
 ) -> list[dict[str, Any]]:
-    """I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка."""
+    """I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка.
+
+    layer_counts — денормализованные счётчики. Если engineering_count == 0 —
+    пропускаем heavy jsonb_array_elements + ST_DWithin scan.
+    """
+    if layer_counts is not None and layer_counts.get("engineering_count", 1) == 0:
+        return []
     try:
         rows = db.execute(
             text(
@@ -353,6 +399,12 @@ def _trigger_harvest(quarter: str) -> bool:
     """Fire-and-forget harvest_quarter.apply_async(). Возвращает True если enqueue OK.
 
     Ленивый импорт чтобы избежать circular import (tasks → services → tasks).
+
+    Known limitation: burst из N concurrent analyze_parcel запросов на один
+    и тот же ещё не закешированный квартал может поставить N одинаковых задач
+    в очередь (нет дедупликации). UPSERT в harvest_quarter идемпотентен, поэтому
+    данные не портятся, но WAF traffic тратится впустую.
+    TODO: добавить Redis SETNX lock с TTL перед apply_async — отдельная issue/PR.
     """
     try:
         from app.workers.tasks.nspd_sync import harvest_quarter

backend/tests/test_quarter_dump_lookup.py

@@ -7,6 +7,8 @@ Mock-based — без реальной БД и Celery. Проверяет:
 - get_quarter_dump_data: свежий дамп с territorial_zones → nspd_zoning populated
 - get_quarter_dump_data: свежий дамп с 2 zouit → nspd_zouit_overlaps length=2
 - get_quarter_dump_data: parcel_wkt=None → metadata only, no spatial queries
+- _make_empty_result: mutation-safety — каждый вызов возвращает независимый dict
+- early-exit: layer_counts == 0 → spatial query не вызывается
 """
 
 from __future__ import annotations
@@ -16,6 +18,8 @@ from typing import Any
 from unittest.mock import MagicMock, patch
 
 from app.services.site_finder.quarter_dump_lookup import (
+    EMPTY_DUMP_RESULT,
+    _make_empty_result,
     derive_quarter_cad,
     get_quarter_dump_data,
 )
@@ -87,8 +91,9 @@ def test_get_quarter_dump_no_row() -> None:
 def test_get_quarter_dump_stale() -> None:
     """Дамп старше 180 дней → stale=True, harvest triggered, available=False."""
     old_date = datetime.now(UTC) - timedelta(days=200)
-    # row: (quarter_cad, fetched_at_utc, total_features, harvest_error)
+    # row: (quarter_cad, fetched_at_utc, total_features, harvest_error,
-    db = _make_db_mock(row=("66:41:0204016", old_date, 50, None))
+    #        territorial_zones_count, zouit_count, engineering_count)
+    db = _make_db_mock(row=("66:41:0204016", old_date, 50, None, 5, 2, 1))
 
     with patch(
         "app.services.site_finder.quarter_dump_lookup._trigger_harvest",
@@ -106,7 +111,7 @@ def test_get_quarter_dump_stale() -> None:
 def test_get_quarter_dump_with_harvest_error() -> None:
     """Дамп с harvest_error → harvest_triggered=True, available=False."""
     fresh_date = datetime.now(UTC) - timedelta(days=10)
-    db = _make_db_mock(row=("66:41:0204016", fresh_date, 0, "WAF 429"))
+    db = _make_db_mock(row=("66:41:0204016", fresh_date, 0, "WAF 429", 0, 0, 0))
 
     with patch(
         "app.services.site_finder.quarter_dump_lookup._trigger_harvest",
@@ -125,7 +130,7 @@ def test_get_quarter_dump_with_harvest_error() -> None:
 def test_get_quarter_dump_no_parcel_geometry() -> None:
     """parcel_wkt=None → metadata only, нет spatial queries."""
     fresh_date = datetime.now(UTC) - timedelta(days=5)
-    db = _make_db_mock(row=("66:41:0204016", fresh_date, 120, None))
+    db = _make_db_mock(row=("66:41:0204016", fresh_date, 120, None, 10, 3, 2))
 
     result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=None)
 
@@ -149,11 +154,17 @@ def _make_db_mock_with_spatial(
 ) -> MagicMock:
     """Mock DB с разными результатами для разных запросов.
 
-    Порядок вызовов:
+    Порядок вызовов db.execute жёстко привязан к реализации get_quarter_dump_data:
-    1. read dump row
+    0. dump lookup  → .first()
-    2. zoning (first())
+    1. zoning       → .first()   (пропускается если territorial_zones_count == 0)
-    3. zouit (fetchall())
+    2. zouit        → .fetchall() (пропускается если zouit_count == 0)
-    4. engineering (fetchall())
+    3. engineering  → .fetchall() (пропускается если engineering_count == 0)
+
+    Хрупко при реорганизации порядка запросов в production-коде. При изменении
+    порядка в get_quarter_dump_data — обновить индексы call_count здесь.
+    dump_row должен быть 7-элементным tuple:
+    (quarter_cad, fetched_at_utc, total_features, harvest_error,
+     territorial_zones_count, zouit_count, engineering_count).
     """
     db = MagicMock()
     call_count = 0
@@ -185,7 +196,7 @@ def _make_db_mock_with_spatial(
 def test_get_quarter_dump_fresh_zoning() -> None:
     """Свежий дамп с territorial_zone → nspd_zoning populated."""
     fresh_date = datetime.now(UTC) - timedelta(days=30)
-    dump_row = ("66:41:0204016", fresh_date, 100, None)
+    dump_row = ("66:41:0204016", fresh_date, 100, None, 5, 0, 0)
     # zone_props JSONB в виде dict (SQLAlchemy вернёт dict)
     zone_props: dict[str, Any] = {
         "reg_numb_border": "Ж-3",
@@ -214,7 +225,7 @@ def test_get_quarter_dump_fresh_zoning() -> None:
 def test_get_quarter_dump_fresh_zoning_no_match() -> None:
     """Свежий дамп, centroid вне всех territorial_zones → nspd_zoning=None."""
     fresh_date = datetime.now(UTC) - timedelta(days=30)
-    dump_row = ("66:41:0204016", fresh_date, 100, None)
+    dump_row = ("66:41:0204016", fresh_date, 100, None, 3, 0, 0)
 
     db = _make_db_mock_with_spatial(
         dump_row=dump_row,
@@ -234,7 +245,9 @@ def test_get_quarter_dump_fresh_zoning_no_match() -> None:
 def test_get_quarter_dump_fresh_zouit_overlaps() -> None:
     """Свежий дамп с 2 zouit → nspd_zouit_overlaps length=2."""
     fresh_date = datetime.now(UTC) - timedelta(days=10)
-    dump_row = ("66:41:0204016", fresh_date, 200, None)
+    # territorial_zones_count > 0 чтобы все 3 spatial запроса вызывались
+    # (иначе call_count в mock смещается и zouit попадает не в тот слот)
+    dump_row = ("66:41:0204016", fresh_date, 200, None, 1, 5, 0)
 
     props1: dict[str, Any] = {"name": "ОКН Дом купца", "subcategory": "okn"}
     props2: dict[str, Any] = {"name": "ОЗ газопровода", "subcategory": "engineering"}
@@ -267,7 +280,7 @@ def test_get_quarter_dump_fresh_zouit_overlaps() -> None:
 def test_get_quarter_dump_includes_nspd_fields() -> None:
     """Базовая проверка что все 4 nspd-ключа присутствуют в возвращаемом dict."""
     fresh_date = datetime.now(UTC) - timedelta(days=1)
-    dump_row = ("66:41:0204016", fresh_date, 55, None)
+    dump_row = ("66:41:0204016", fresh_date, 55, None, 2, 1, 0)
 
     db = _make_db_mock_with_spatial(
         dump_row=dump_row,
@@ -288,3 +301,89 @@ def test_get_quarter_dump_includes_nspd_fields() -> None:
     assert "stale" in dump
     assert "harvest_triggered" in dump
     assert "total_features" in dump
+
+
+# ── _make_empty_result — mutation safety ─────────────────────────────────────
+
+
+def test_make_empty_result_returns_independent_copies() -> None:
+    """Каждый вызов _make_empty_result() возвращает независимый dict.
+
+    Мутация nspd_dump в одном результате не должна влиять на другой
+    и на модульный sentinel EMPTY_DUMP_RESULT.
+    """
+    a = _make_empty_result()
+    b = _make_empty_result()
+
+    # Вложенный dict — независимый объект
+    assert a["nspd_dump"] is not b["nspd_dump"]
+    assert a["nspd_dump"] is not EMPTY_DUMP_RESULT["nspd_dump"]
+
+    # Мутируем один — другой не трогается
+    a["nspd_dump"]["harvest_triggered"] = True
+    a["nspd_zouit_overlaps"].append({"group_key": "test"})
+
+    assert b["nspd_dump"]["harvest_triggered"] is False
+    assert b["nspd_zouit_overlaps"] == []
+    assert EMPTY_DUMP_RESULT["nspd_dump"]["harvest_triggered"] is False
+    assert EMPTY_DUMP_RESULT["nspd_zouit_overlaps"] == []
+
+
+def test_make_empty_result_overrides() -> None:
+    """_make_empty_result принимает override-аргументы."""
+    r = _make_empty_result(harvest_triggered=True, stale=True, total_features=42)
+    assert r["nspd_dump"]["harvest_triggered"] is True
+    assert r["nspd_dump"]["stale"] is True
+    assert r["nspd_dump"]["total_features"] == 42
+    assert r["nspd_dump"]["available"] is False
+
+
+# ── early-exit when layer_counts == 0 ────────────────────────────────────────
+
+
+def test_early_exit_all_counts_zero_no_spatial_queries() -> None:
+    """Если все denormalized counts == 0 — spatial queries не вызываются."""
+    fresh_date = datetime.now(UTC) - timedelta(days=5)
+    # territorial_zones_count=0, zouit_count=0, engineering_count=0
+    dump_row = ("66:41:0204016", fresh_date, 50, None, 0, 0, 0)
+
+    db = _make_db_mock_with_spatial(dump_row=dump_row)
+
+    result = get_quarter_dump_data(
+        db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))"
+    )
+
+    # Только 1 execute-вызов — чтение dump row; spatial helpers пропущены
+    assert db.execute.call_count == 1
+    assert result["nspd_zoning"] is None
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []
+    assert result["nspd_dump"]["available"] is True
+
+
+def test_early_exit_partial_counts() -> None:
+    """territorial_zones_count > 0, zouit_count == 0 → zoning query вызывается,
+    zouit и engineering — нет."""
+    fresh_date = datetime.now(UTC) - timedelta(days=5)
+    dump_row = ("66:41:0204016", fresh_date, 30, None, 3, 0, 0)
+    zone_props: dict[str, Any] = {"reg_numb_border": "Ж-1", "type_zone": "Зона жилая"}
+    zoning_row_mock = MagicMock()
+    zoning_row_mock.__getitem__ = lambda self, i: zone_props if i == 0 else None
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=zoning_row_mock,
+        zouit_rows=[],
+        engineering_rows=[],
+    )
+
+    result = get_quarter_dump_data(
+        db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))"
+    )
+
+    # dump lookup (0) + zoning (1) = 2 вызова; zouit и engineering — пропущены
+    assert db.execute.call_count == 2
+    assert result["nspd_zoning"] is not None
+    assert result["nspd_zoning"]["zone_code"] == "Ж-1"
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []