diff --git a/backend/app/api/v1/parcels.py b/backend/app/api/v1/parcels.py index 9a296460..d698597c 100644 --- a/backend/app/api/v1/parcels.py +++ b/backend/app/api/v1/parcels.py @@ -952,9 +952,17 @@ def _compute_confidence( # #93 — on-demand cadastre fetch tuning constants. -# _INLINE_FETCH_WAIT_S — суммарно ждём fast-path при analyze fallback. Выше -# 30s — frontend увидит timeout/disconnect; ниже 15s — НСПД не успевает. -_INLINE_FETCH_WAIT_S = 25 +# _INLINE_FETCH_WAIT_S — суммарно ждём fast-path при analyze fallback. +# +# Tradeoff: sync `def analyze_parcel` блокирует один FastAPI threadpool slot +# на это время. Default threadpool в Starlette/FastAPI — 40 slots (anyio +# default). При concurrent burst >40 «миссинг cad» запросов будем saturate +# threadpool — последующие запросы (включая healthcheck) ждут free slot. +# +# 15s выбран как баланс: НСПД ~5-15s avg для quarter, ~10-20s для parcel — +# fast path сработает в ~70% случаев. Остальные 30% получают 202 + +# polling (без блокировки threadpool). +_INLINE_FETCH_WAIT_S = 15 _INLINE_FETCH_POLL_INTERVAL_S = 2 diff --git a/backend/app/services/site_finder/cadastre_fetch.py b/backend/app/services/site_finder/cadastre_fetch.py index c0577cb4..c995fe70 100644 --- a/backend/app/services/site_finder/cadastre_fetch.py +++ b/backend/app/services/site_finder/cadastre_fetch.py @@ -12,8 +12,11 @@ cad_buildings / cad_parcels_geom) — раньше получал 404 «Загр - Smart status: проверяет и БД (ready), и nspd_geo_jobs (fetching/failed), и nspd_geo_targets (per-cad status, distinguishes not_in_nspd vs failed). -Дедупликация: SELECT FOR UPDATE на nspd_geo_jobs WHERE source_kind='auto_on_demand' -AND ... — гарантирует один in-flight job на cad даже при параллельных request'ах. +Дедупликация: `pg_try_advisory_xact_lock(hashtext(cad_num))` оборачивает +шаги "check active job → enqueue new" в `find_or_enqueue_fetch`. Lock держится +до COMMIT текущей транзакции (xact_lock — transaction-scoped). Параллельные +запросы на тот же cad: второй увидит lock=false и фолбэк-чит активный job уже +от первого транзакта (после первого COMMIT). """ from __future__ import annotations @@ -195,8 +198,16 @@ def find_or_enqueue_fetch(db: Session, cad_num: str) -> tuple[FetchStatus, int | - "fetching" → 202 + job_id для polling - "not_in_nspd" / "invalid_format" → 404 с понятным сообщением - "failed" → 503 с retry-after + + Дедупликация (через pg advisory lock): + Шаги 3-6 ниже (check active → check recent → enqueue) оборачиваются + `pg_try_advisory_xact_lock(hashtext(cad_num))`. Lock automatically + released at xact end. Параллельные запросы на тот же cad: + - Первый запрос получает lock → выполняет enqueue → COMMIT → release. + - Второй запрос lock=false → re-проверяет active job (тот что enqueue'нул + первый и уже видим в DB после COMMIT) → возвращает ("fetching", job_id1). """ - # 1) Format check (cheap) + # 1) Format check (cheap, lock-free) if not validate_cad_format(cad_num): return ( "invalid_format", @@ -204,15 +215,30 @@ def find_or_enqueue_fetch(db: Session, cad_num: str) -> tuple[FetchStatus, int | ("Неверный формат кадастрового номера. Ожидается NN:NN:NNNNNN или NN:NN:NNNNNN:NN"), ) - # 2) Already in DB? + # 2) Already in DB? (lock-free read) if cad_exists_in_db(db, cad_num): return ("ready", None, None) - # 3) Existing in-flight job? + # 3-6) Critical section: advisory lock на hashtext(cad_num) гарантирует + # один enqueue per cad даже при concurrent requests. + got_lock = db.execute( + text("SELECT pg_try_advisory_xact_lock(hashtext(:c))"), + {"c": cad_num}, + ).scalar() + + # Lock acquired or not — в обоих случаях сначала проверяем active job + # (если другой xact уже enqueue'нул — мы видим его после его COMMIT). active_job_id = find_active_on_demand_job(db, cad_num) if active_job_id is not None: return ("fetching", active_job_id, None) + if not got_lock: + # Lock потерян concurrent xact но active job ещё не закоммичен. + # Возвращаем "fetching" с None job_id — frontend начнёт polling и + # подцепит job через /fetch-status (там тоже проверка active job). + logger.info("advisory lock contention for cad=%s — caller will poll status", cad_num) + return ("fetching", None, None) + # 4) Recent completed job? Distinguish not_in_nspd vs failed. recent = find_recent_completed_job(db, cad_num) if recent: @@ -237,7 +263,7 @@ def find_or_enqueue_fetch(db: Session, cad_num: str) -> tuple[FetchStatus, int | f"Cannot determine cadastre type from {cad_num}", ) - # 6) Enqueue new on-demand fetch + # 6) Enqueue new on-demand fetch (lock освободится на COMMIT после return) try: job_id = enqueue_on_demand_fetch(db, cad_num, thematic_id) return ("fetching", job_id, None) diff --git a/data/sql/87_on_demand_indexes.sql b/data/sql/87_on_demand_indexes.sql new file mode 100644 index 00000000..b396125c --- /dev/null +++ b/data/sql/87_on_demand_indexes.sql @@ -0,0 +1,36 @@ +-- #93 (PR #95): indexes for on-demand cadastre fetch lookups. +-- +-- Context: +-- `find_active_on_demand_job` и `find_recent_completed_job` в +-- `app/services/site_finder/cadastre_fetch.py` фильтруют +-- `nspd_geo_targets` по cad_num + `nspd_geo_jobs` по source_kind+status. +-- Existing UNIQUE(job_id, cad_num, thematic_id) на targets не помогает +-- при WHERE cad_num=:c — нужен secondary index. +-- +-- Не блокер при текущем размере таблицы (~10K targets), но критично при +-- ускорении on-demand traffic (per #93 trigger любым посетителем +-- /site-finder). +-- +-- Apply: idempotent, CREATE INDEX IF NOT EXISTS — safe to run multiple +-- times. CONCURRENTLY не используется чтобы коммит был атомарным; на +-- prod размере index build занимает <1s. + +BEGIN; + +-- Targets — поиск активных on-demand job'ов для конкретного cad +CREATE INDEX IF NOT EXISTS nspd_geo_targets_cad_num_idx + ON nspd_geo_targets(cad_num); + +-- Jobs — фильтр source_kind='auto_on_demand' + status IN (queued,running) +CREATE INDEX IF NOT EXISTS nspd_geo_jobs_source_status_idx + ON nspd_geo_jobs(source_kind, status); + +COMMIT; + +COMMENT ON INDEX nspd_geo_targets_cad_num_idx IS + 'on-demand cadastre fetch lookups (#93). Used by find_active_on_demand_job ' + 'и find_recent_completed_job в cadastre_fetch service.'; + +COMMENT ON INDEX nspd_geo_jobs_source_status_idx IS + 'Filter source_kind=auto_on_demand + status (#93). Composite чтобы ' + 'покрыть оба predicate сразу.'; diff --git a/frontend/src/hooks/useSiteAnalysis.ts b/frontend/src/hooks/useSiteAnalysis.ts index f1d946c8..7323f45c 100644 --- a/frontend/src/hooks/useSiteAnalysis.ts +++ b/frontend/src/hooks/useSiteAnalysis.ts @@ -93,12 +93,16 @@ export function useSiteAnalysis() { `/api/v1/parcels/${encodeURIComponent(cad)}/fetch-status`, ); if (status.status === "ready") { - // Re-trigger analyze — should be 200 now - setFetchingState(null); + // Re-trigger analyze — should be 200 now. + // setFetchingState(null) ПОСЛЕ await чтобы не было flicker: + // если очистить до await, mutation isPending=true но + // fetchingState=null → пустой экран ~1 RTT. После await + // mutation сразу резолвится с data — render skipped. const second = await apiFetch( `/api/v1/parcels/${encodeURIComponent(cad)}/analyze`, { method: "POST" }, ); + setFetchingState(null); return second; } if (status.status === "not_in_nspd") {