fix(tradein): data-layer tail A2+B+C — trgm index, FDW options, matview refresh (Refs #769) (#804)
All checks were successful
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Successful in 26s
Deploy Trade-In / build-backend (push) Successful in 40s
Deploy Trade-In / deploy (push) Successful in 36s

Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>
This commit is contained in:
bot-backend 2026-05-30 17:40:54 +00:00 committed by bot-reviewer
parent 046febcae8
commit a9e4aecb54
4 changed files with 213 additions and 0 deletions

View file

@ -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:

View file

@ -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;

View file

@ -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;

View file

@ -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;