perf(tradein): cian full-load incremental save + concurrent pagination #929

Merged

bot-reviewer merged 2 commits from feat/cian-fullload-fast-incremental into main

2026-05-31 20:27:28 +00:00

bot-backend commented

2026-05-31 20:20:13 +00:00

Collaborator

Summary

Save-at-end → save per leaf-bucket: on_bucket callback вызывает save_listings (с db.commit) сразу после каждого leaf-бакета — краш/бан больше не теряет накопленное.
Параллельная пагинация бакета: asyncio.gather + Semaphore(concurrency=5) запускает все страницы одновременно вместо sequential sleep-loop — кратное ускорение.
Снижен delay floor в CianFullLoadRequest: 3.0→1.5s (ge=0.5), выделенный прокси + IP-ротация позволяют агрессивнее.
on_bucket поддерживает async и sync callback (inspect.isawaitable).
Cancel per-bucket: RuntimeError('cancelled') из on_bucket → graceful mark_done (не mark_failed).
fetch_all_secondary стартует с hi=_MAX_PRICE (убирает лишний probe None→split).
Новые тесты: on_bucket вызывается per-leaf, cancel прерывает прогон, дедуп работает при параллельной пагинации.

Test plan

pytest tests/scrapers/test_cian_exhaustive.py — 6/6 passed
ruff check — all checks passed
POST /scrape/cian-full-load: request_delay_sec default снижен до 1.5, concurrency поле добавлено

## Summary - Save-at-end → save per leaf-bucket: on_bucket callback вызывает save_listings (с db.commit) сразу после каждого leaf-бакета — краш/бан больше не теряет накопленное. - Параллельная пагинация бакета: asyncio.gather + Semaphore(concurrency=5) запускает все страницы одновременно вместо sequential sleep-loop — кратное ускорение. - Снижен delay floor в CianFullLoadRequest: 3.0→1.5s (ge=0.5), выделенный прокси + IP-ротация позволяют агрессивнее. - on_bucket поддерживает async и sync callback (inspect.isawaitable). - Cancel per-bucket: RuntimeError('cancelled') из on_bucket → graceful mark_done (не mark_failed). - fetch_all_secondary стартует с hi=_MAX_PRICE (убирает лишний probe None→split). - Новые тесты: on_bucket вызывается per-leaf, cancel прерывает прогон, дедуп работает при параллельной пагинации. ## Test plan - [ ] pytest tests/scrapers/test_cian_exhaustive.py — 6/6 passed - [ ] ruff check — all checks passed - [ ] POST /scrape/cian-full-load: request_delay_sec default снижен до 1.5, concurrency поле добавлено

bot-backend added 1 commit 2026-05-31 20:20:14 +00:00

perf(tradein): cian full-load incremental save + concurrent pagination a9f71686a9

Save-at-end → save per leaf-bucket (краш/бан больше не теряет накопленное).
Параллельная пагинация бакета (asyncio.gather + semaphore, concurrency=5) +
снижен delay floor (1.5s, выделенный прокси) → кратное ускорение. on_bucket
callback коммитит в БД сразу; cancel проверяется per-bucket.

lekss361 referenced this pull request from a commit

2026-05-31 20:24:49 +00:00

fix(tradein): drop undefined _cancelled assign in cian-full-load cancel path

lekss361 added 1 commit 2026-05-31 20:24:50 +00:00

fix(tradein): drop undefined _cancelled assign in cian-full-load cancel path 2ebc3eeead

Review #929: _cancelled was removed earlier in the PR but a stray assignment
remained → NameError on every cooperative cancel (common exit path). Variable
unused; mark_done is called unconditionally next line.

bot-reviewer approved these changes 2026-05-31 20:27:17 +00:00

bot-reviewer left a comment

Collaborator

✅ APPROVE — корректный concurrency + incremental-save, additive. Мержу.

Verified

Нет data race: asyncio single-thread; gather собирает page-результаты → seen/bucket_lots мержатся последовательно после. Semaphore(concurrency) лимитит in-flight. Дедуп по source_id целостен (тест concurrent_pages_deduped: page3 дублирует page1 → 20 unique). ✓
Терминация сохранена: hi теперь всегда int (старт _MAX_PRICE, None-ветка убрана чисто); split halves → bounded depth. min_bracket_guard тест обновлён на ≤55 (probe + до 54 параллельных стр) — всё ещё валидирует отсутствие бесконечной рекурсии. ✓
Incremental save: on_bucket→save_listings(commit) per leaf → краш не теряет накопленное. Upsert гасит boundary-дубли между бакетами. ✓
Cancel: on_bucket кидает RuntimeError("cancelled") → ловится except RuntimeError (стоит ПЕРЕД generic except Exception, проверяется первым) → mark_done с частичными уже-сохранёнными результатами. Не-"cancelled" RuntimeError → mark_failed+raise. Корректный порядок. ✓
return_exceptions=True → упавшие страницы логируются и скипаются (не рушат бакет). Cross-file: на базе #926, save_listings sig не менялся.

🟡 Advisory (не блокер)

Per-page IP-rotation-retry убран в concurrent-пагинации (транзиентный page-fail скипается, не ретраится) — но это даже устойчивее старого abort-bucket-on-fail, и митигировано incremental-save + re-runnable.
concurrency=5 @ delay 1.5s агрессивно для одного egress — но operator-tunable (admin endpoint), выделенный прокси #923 + rotation на probe.
counters.unique_fetched double-count'ит cross-bucket дубли (cosmetic, только heartbeat).

Squash-merge + delete branch.

✅ **APPROVE** — корректный concurrency + incremental-save, additive. Мержу. ## Verified - **Нет data race**: asyncio single-thread; `gather` собирает page-результаты → `seen`/`bucket_lots` мержатся последовательно после. `Semaphore(concurrency)` лимитит in-flight. Дедуп по source_id целостен (тест `concurrent_pages_deduped`: page3 дублирует page1 → 20 unique). ✓ - **Терминация сохранена**: `hi` теперь всегда int (старт `_MAX_PRICE`, None-ветка убрана чисто); split halves → bounded depth. `min_bracket_guard` тест обновлён на ≤55 (probe + до 54 параллельных стр) — всё ещё валидирует отсутствие бесконечной рекурсии. ✓ - **Incremental save**: `on_bucket`→`save_listings`(commit) per leaf → краш не теряет накопленное. Upsert гасит boundary-дубли между бакетами. ✓ - **Cancel**: `on_bucket` кидает `RuntimeError("cancelled")` → ловится `except RuntimeError` (стоит ПЕРЕД generic `except Exception`, проверяется первым) → `mark_done` с частичными уже-сохранёнными результатами. Не-"cancelled" RuntimeError → mark_failed+raise. Корректный порядок. ✓ - `return_exceptions=True` → упавшие страницы логируются и скипаются (не рушат бакет). Cross-file: на базе #926, `save_listings` sig не менялся. ## 🟡 Advisory (не блокер) - Per-page IP-rotation-retry убран в concurrent-пагинации (транзиентный page-fail скипается, не ретраится) — но это даже устойчивее старого abort-bucket-on-fail, и митигировано incremental-save + re-runnable. - `concurrency=5 @ delay 1.5s` агрессивно для одного egress — но operator-tunable (admin endpoint), выделенный прокси #923 + rotation на probe. - `counters.unique_fetched` double-count'ит cross-bucket дубли (cosmetic, только heartbeat). Squash-merge + delete branch.