From a9e4aecb543e012ef3fe247d85a5124693e48ed0 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 30 May 2026 17:40:54 +0000 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20data-layer=20tail=20A2+B+C=20?= =?UTF-8?q?=E2=80=94=20trgm=20index,=20FDW=20options,=20matview=20refresh?= =?UTF-8?q?=20(Refs=20#769)=20(#804)=20Co-authored-by:=20bot-backend=20=20Co-committed-by:=20bot-backend=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/services/scheduler.py | 47 +++++++++++++ .../data/sql/086_deals_address_trgm_index.sql | 34 +++++++++ .../data/sql/087_fdw_server_options.sql | 69 +++++++++++++++++++ ..._schedules_seed_search_matview_refresh.sql | 63 +++++++++++++++++ 4 files changed, 213 insertions(+) create mode 100644 tradein-mvp/backend/data/sql/086_deals_address_trgm_index.sql create mode 100644 tradein-mvp/backend/data/sql/087_fdw_server_options.sql create mode 100644 tradein-mvp/backend/data/sql/088_scrape_schedules_seed_search_matview_refresh.sql diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 83cb5c62..fbff867c 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -468,6 +468,51 @@ async def trigger_listing_source_snapshot_run( return run_id +async def trigger_refresh_search_matview_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + запустить REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv. + + listings_search_mv активно читается search_query.py (SELECT FROM listings_search_mv). + Без периодического refresh матвью устаревает по мере появления новых listings. + Задача синхронная (DB-only, БЕЗ внешних HTTP-вызовов) — гоняем в run_in_executor + по образцу trigger_asking_to_sold_ratio_run (#648). + + refresh_search_matview() использует собственное psycopg-соединение с autocommit=True + (REFRESH CONCURRENTLY не может работать внутри транзакции). scrape_runs lifecycle + (mark_done/mark_failed) управляется через отдельную SessionLocal. + + SAFE to enable — schedule seed enabled=true (088), pure internal DB op. + Окно 03:00-04:00 UTC — ночное, до rosreestr_dkp_import (04:00-06:00 UTC). + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.refresh_search_matview import refresh_search_matview + + loop = asyncio.get_event_loop() + # refresh_search_matview() manages its own connection with autocommit=True + # (required for REFRESH MATERIALIZED VIEW CONCURRENTLY outside a transaction). + await loop.run_in_executor(None, refresh_search_matview) + runs_mod.mark_done(run_db, run_id, {}) + except Exception: + logger.exception("scheduler: refresh_search_matview crashed run_id=%d", run_id) + runs_mod.mark_failed(run_db, run_id, "refresh_search_matview failed", {}) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered refresh_search_matview run_id=%d", run_id) + return run_id + + async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task). @@ -782,6 +827,8 @@ async def scheduler_loop() -> None: await trigger_listing_source_snapshot_run(db, sch) elif source == "asking_to_sold_ratio_refresh": await trigger_asking_to_sold_ratio_run(db, sch) + elif source == "refresh_search_matview": + await trigger_refresh_search_matview_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/data/sql/086_deals_address_trgm_index.sql b/tradein-mvp/backend/data/sql/086_deals_address_trgm_index.sql new file mode 100644 index 00000000..9a83bc1e --- /dev/null +++ b/tradein-mvp/backend/data/sql/086_deals_address_trgm_index.sql @@ -0,0 +1,34 @@ +-- 086_deals_address_trgm_index.sql +-- Issue #769 Part A2 (finding #10) — GIN trigram index on deals.address. +-- +-- Context: +-- The deals table (002_core_tables.sql line 84) has an `address text` column used +-- in street-match queries (import_rosreestr_dkp in scheduler.py builds ILIKE / text +-- predicates). Without a trigram index the planner falls back to seq-scan on the +-- full deals table. pg_trgm is already enabled in 050_search_optimization.sql but +-- only for the listings table — not for deals. +-- +-- Idempotency: +-- CREATE EXTENSION IF NOT EXISTS — safe on re-run (ext already active = no-op). +-- CREATE INDEX IF NOT EXISTS — safe on re-run. +-- Both inside a single BEGIN/COMMIT — plain CREATE INDEX (not CONCURRENTLY) is +-- required to stay transaction-safe (CONCURRENTLY cannot run inside a txn block). +-- Acceptable here because deals is populated incrementally via FDW import batches +-- and the table starts small. +-- +-- Dependencies: +-- 002_core_tables.sql (deals table with address column). +-- +-- Deploy order: +-- Apply after 002_core_tables.sql. No application code changes required. +-- Post-deploy QA: EXPLAIN (ANALYZE, BUFFERS) SELECT ... FROM deals WHERE address +-- ILIKE '%улица%'; should show Bitmap Index Scan on deals_address_trgm_idx. + +BEGIN; + +CREATE EXTENSION IF NOT EXISTS pg_trgm; + +CREATE INDEX IF NOT EXISTS deals_address_trgm_idx + ON deals USING gin (address gin_trgm_ops); + +COMMIT; diff --git a/tradein-mvp/backend/data/sql/087_fdw_server_options.sql b/tradein-mvp/backend/data/sql/087_fdw_server_options.sql new file mode 100644 index 00000000..62ced969 --- /dev/null +++ b/tradein-mvp/backend/data/sql/087_fdw_server_options.sql @@ -0,0 +1,69 @@ +-- 087_fdw_server_options.sql +-- Issue #769 Part B (finding #13) — FDW server performance options. +-- +-- Context: +-- SERVER gendesign_remote was created by 060_postgres_fdw_extension.sql. +-- The current options are: host, dbname, port, extensions, connect_timeout. +-- Two planning options are missing: +-- fetch_size — rows fetched per round-trip (default 100, raise to 10000 +-- for bulk cursor-based pagination in import_rosreestr_dkp). +-- use_remote_estimate — let planner query ANALYZE stats from the remote server +-- to produce accurate cost estimates for the 6.8M-row +-- rosreestr_deals foreign table and quarter_price_index MV. +-- +-- Idempotency: +-- ALTER SERVER ... OPTIONS (ADD ...) raises ERROR if the option already exists. +-- ALTER SERVER ... OPTIONS (SET ...) raises ERROR if the option does not exist yet. +-- Therefore a DO $$ block checks pg_foreign_server.srvoptions for each option: +-- - If present → SET (update the value in case it was previously set differently). +-- - If absent → ADD (insert for the first time). +-- This is safe to run repeatedly regardless of prior state. +-- +-- Dependencies: +-- 060_postgres_fdw_extension.sql (gendesign_remote server must exist). +-- +-- Deploy order: +-- Apply after 060_postgres_fdw_extension.sql. No FDW USER MAPPING changes required. +-- Effect is immediate — next FDW query uses the new planner options. + +BEGIN; + +DO $$ +DECLARE + v_opts text[]; +BEGIN + -- Read current srvoptions array for gendesign_remote. + SELECT COALESCE(srvoptions, ARRAY[]::text[]) + INTO v_opts + FROM pg_foreign_server + WHERE srvname = 'gendesign_remote'; + + IF NOT FOUND THEN + RAISE EXCEPTION 'FDW server gendesign_remote not found — apply 060_postgres_fdw_extension.sql first'; + END IF; + + -- fetch_size: rows per FDW cursor round-trip (default 100 → 10000) + IF v_opts @> ARRAY['fetch_size=10000'] THEN + NULL; -- already correct value, no-op + ELSIF EXISTS ( + SELECT 1 FROM unnest(v_opts) AS opt WHERE opt LIKE 'fetch_size=%' + ) THEN + ALTER SERVER gendesign_remote OPTIONS (SET fetch_size '10000'); + ELSE + ALTER SERVER gendesign_remote OPTIONS (ADD fetch_size '10000'); + END IF; + + -- use_remote_estimate: push ANALYZE stats collection to remote for better plans + IF v_opts @> ARRAY['use_remote_estimate=true'] THEN + NULL; -- already correct value, no-op + ELSIF EXISTS ( + SELECT 1 FROM unnest(v_opts) AS opt WHERE opt LIKE 'use_remote_estimate=%' + ) THEN + ALTER SERVER gendesign_remote OPTIONS (SET use_remote_estimate 'true'); + ELSE + ALTER SERVER gendesign_remote OPTIONS (ADD use_remote_estimate 'true'); + END IF; +END +$$; + +COMMIT; diff --git a/tradein-mvp/backend/data/sql/088_scrape_schedules_seed_search_matview_refresh.sql b/tradein-mvp/backend/data/sql/088_scrape_schedules_seed_search_matview_refresh.sql new file mode 100644 index 00000000..71fab41a --- /dev/null +++ b/tradein-mvp/backend/data/sql/088_scrape_schedules_seed_search_matview_refresh.sql @@ -0,0 +1,63 @@ +-- 088_scrape_schedules_seed_search_matview_refresh.sql +-- Issue #769 Part C (finding #14) — seed scrape_schedules row for daily +-- listings_search_mv materialized view refresh. +-- +-- Context: +-- listings_search_mv (created in 050_search_optimization.sql) is actively read by +-- search_query.py (SELECT FROM listings_search_mv) for the /api/v1/search endpoint. +-- The refresh callable (app/tasks/refresh_search_matview.py) exists but was NEVER +-- wired into the in-app scheduler: no scrape_schedules row, no dispatch branch in +-- scheduler_loop(). Without a scheduled refresh the matview becomes stale after the +-- initial CREATE MATERIALIZED VIEW population. +-- +-- Fix: +-- 1. scheduler.py — added trigger_refresh_search_matview_run() + dispatch branch +-- `elif source == "refresh_search_matview"` in scheduler_loop(). +-- 2. This migration — seeds the scrape_schedules row (enabled=true, 03:00-04:00 UTC). +-- +-- Schedule window 03:00-04:00 UTC: +-- Nightly, before rosreestr_dkp_import (04:00-06:00 UTC). Refresh takes <10s on a +-- non-empty matview (CONCURRENTLY — readers are not blocked). +-- +-- next_run_at bootstrapped to tomorrow 03:00 UTC so the scheduler does not fire +-- immediately on deploy (same pattern as 078/079/082). +-- +-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply. +-- +-- Dependencies: +-- 052_scrape_schedules.sql (table + UNIQUE(source)). +-- 050_search_optimization.sql (listings_search_mv must exist in DB). +-- scheduler.py change (trigger_refresh_search_matview_run must be deployed). +-- +-- Deploy order: +-- Apply this migration after deploying the scheduler.py change so the scheduler +-- can dispatch the new source name correctly on first fire. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'refresh_search_matview', + true, -- SAFE: pure internal DB, REFRESH CONCURRENTLY + 3, + 4, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 3)) AT TIME ZONE 'UTC', + '{}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769).'; + +COMMIT;