fix(scrape-objective): stream-parse lots_pf via ijson to avoid OOM #315
No reviewers
Labels
No labels
admin
analytics
auth
automation
bug
business
chore
ci
compliance
data
data-moat
docs
duplicate
dx
enhancement
Fable 5 ревью
feedback/max
generative
GG-форсайт
needs-discussion
needs-human
observability
pause-bots
performance
priority/p0
priority/p1
priority/p2
priority/p3
scope/backend
scope/db
scope/devops
scope/frontend
scope/qa
scrapers
security
site-finder
stage/1
stage/2
status/blocked
status/done
status/needs-analysis
status/needs-fix
status/qa
status/ready
status/review
status/wip
tech-debt
tradein
ux
week ревью 1
wontfix
вторичка
ИРД
No milestone
No project
No assignees
1 participant
Notifications
Due date
No due date set.
Dependencies
No dependencies set.
Reference: lekss361/gendesign#315
Loading…
Add table
Reference in a new issue
No description provided.
Delete branch "fix/obj-lots-pf-streaming"
Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?
Bug
После активации live scraper (PR #311 + #313) worker SIGKILL'нут host OOM-killer'om при первом же live trigger'е sync. Glitchtip:
fce48a8):Process 'ForkPoolWorker-7' pid:17 exited with 'signal 9 (SIGKILL)'WorkerLostError: signal 9 (SIGKILL)(билиард вmark_as_worker_lost)Root cause
ObjectiveClient._request_authed()используетr.json()который декодирует 607 МБ JSON ответа Objectivelots_pfцеликом вlist[dict]из 324 217 записей → 2-3 ГБ Python-объектов → host OOM. VPS не имеет mem_limit для worker, но физической памяти не хватает.run_id=1зафиксирован в логах: corp_sum (7 МБ) распарсил 3 435 строк успешно, fetch lots_pf → SIGKILL до того как_save_rawиparse_lots_pfуспели сработать.Fix — ijson streaming
backend/pyproject.toml+uv.lockijson>=3.2.0backend/app/services/scrapers/objective.py_open_stream()(httpx send stream=True с retry/auth), +stream_report()(@contextmanager — yieldhttpx.Response)data/sql/70_parse_objective_raw.pyparse_lots_pf_stream(stream, raw_id, snap, db, dry_run, batch_size=500)—ijson.items(..., 'result.item')+ batched UPSERT + commit каждые 5 batch (~2 500 rows)backend/app/workers/tasks/scrape_objective.py_save_raw()принимаетpayload=None. lots_pf branch теперь —with client.stream_report(...) as resp: parse_lots_pf_stream(resp.iter_bytes(65536), ...). corp_sum — без изменений (7МБ — нет проблемы)data/sql/79_relax_objective_raw_payload_null.sqlALTER TABLE objective_raw_reports ALTER COLUMN payload DROP NOT NULL— streamed reports не хранят payload целиком (иначе 600МБ × 4 группы × еженедельно = 96ГБ за полгода). meta/size/rows_extracted остаются.RAM budget
Acceptance после merge + deploy
run_id=2появляется соstatus='running'objective_raw_reportsдля lots_pf имеетpayload IS NULL,payload_size ≈ 607M,rows_extracted ≈ 324kobjective_lots.raw_id IS NOT NULLcount растёт батчами по 2500 пока sync идётobjective_lots_historyпишется (delta tracking)fce48a8Risk
DROP NOT NULLна nullable column безопасен)parse_lots_pf(payload, ...)сохранена (используется CLI dry-run из локальных JSON)References
code/patterns/Objective_Data_Source_May17.md— single source of truthevents/Objective_Integration_May07_2026.md— live scrape integration historyfce48a8, 2026-05-17 16:20Z)Deep Code Review — verdict ✅ APPROVE (с 1 follow-up issue для tracking)
Files reviewed: 6 (P0: 2 SQL+migration, P1: 2 backend code, P2: 2 deps)
Lines: +411 / -52
Mergeable: true, no conflicts, head
4875ef1against basefce48a8Risk score: Medium (новый streaming path, но fallback на старый код сохранён)
1. ijson API correctness ✅
ijson.items(stream, 'result.item')— корректный prefix для JSON{"result": [{...}, ...]}. Подтверждено по официальной документации ijson: array elements добавляют литералitemк prefix'у.result[]— это синтаксис jq, ijson его не понимает. Не путать.2. Stream lifecycle ✅ (с одной micro-nit)
@contextmanager stream_report()корректно закрываетresp.close()вfinally— даже при exception в caller'е connection вернётся в pool._open_stream()reproduces auth/rate-limit/retry-логику_request_authed()верно: 401 → re-auth и retry, 429/5xx → exponential backoff с Retry-After header, network error → exponential retry. Re-auth происходит на статусе ответа (до начала чтения тела) — это правильно, не во время streaming.objective.py:308-311):body_preview = r.read()[:200]на не-200 status — для нормальных ошибок body маленький, OK. Если API сломается и вернёт 200 OK с error JSON в начале —parse_lots_pf_streamсам поймает ijson parse error и rollback. Acceptable.3. Batched UPSERT — 🟡 1 MEDIUM (pre-existing, не регрессия)
:objective_lot_id, etc) через SQLAlchemytext(). Нет:x::typecast trap._LOTS_UPSERTи_LOTS_HISTORY_INSERTпринимаютlist[dict]— корректный executemany-pattern.70_parse_objective_raw.py:564-595новый код,:489-491старый). По.claude/rules/backend.mdper-row INSERT в loop требуетwith db.begin_nested():. Сейчас одна плохая row в batch'е (constraint violation, type mismatch) → весь batch fail →db.rollback()откатит всю outer tx, счётчикиn_lots/n_historyуже инкрементированы → inconsistent state. НО: старыйparse_lots_pfимеет идентичный pattern → это pre-existing tech debt, не регрессия от этого PR. Acceptable для emergency OOM fix, но создать follow-up issue на refactor с SAVEPOINT иexecutemany_values-style чанками.--latest-runдогонит) + history INSERT имеетON CONFLICT DO NOTHING. Recoverable.4. Migration ordering ✅
114_22c_flat_plans.sql. PR #314 ушёл с миграциями 110-114. PR #315 добавляет79_relax_objective_raw_payload_null.sql.DROP NOT NULLидемпотентен — повторное применение безопасно (PostgreSQL no-op).5.
payload IS NULLimpact ✅Grep по
backend/+data/sql/показывает только 3 consumer'аobjective_raw_reports.payload:_save_raw()— INSERT (writer, новая логика handle NULL)data/sql/70_parse_objective_raw.py:606-619— CLI--raw-id/--latest-runre-parse. Внимание:detect_kind(NULL)вернётNone→kind неизвестен, skipветка. Это означает что CLI re-parse для streamed lots_pf не сработает — но это by design (нет смысла re-parse'ить 600МБ payload который мы намеренно не сохранили).parse_lots_pfожидаетdict→payload.get("result")упадёт с AttributeError приNone. Защищено черезdetect_kindearly-return.objective_raw_reports.payloadcolum-comment обновлён через migration 79 — самодокументируется.Не нашёл consumers ждущих non-null jsonb. ✅
6.
_request_authedcorp_sum call-site ✅scrape_objective.py:236-237—method = getattr(client, fn_name); payload = method(**params)— для corp_sum (7МБ) логика идентичная пре-PR._save_raw(...payload=payload)— payload передаётся как dict, веткаif payload is not Noneсериализует черезjson.dumps. Backwards-compatible. ✅7. Backwards compat CLI ✅
Старый
parse_lots_pf(payload, ...)(70_parse_objective_raw.py:474-509) сохранён. CLI--file-lots-pf(70_parse_objective_raw.py:563-565) использует именно его (читает локальный JSON черезjson.loads+parse_lots_pf). Smoke-test без БД работает. ✅8. GHA path filter ✅
backend/**triggersdeploy.yml(backend-changeddetection)data/sql/**.sqltriggers SQL auto-apply step (scripts/apply_sql_migrations.shили эквивалент)9. SQL injection ✅
_row_to_lot(row)возвращает params-dict с bind-name keys, не f-string. Все значения текстовых полей Objective (русский язык, кавычки, спец-символы) проходят через SQLAlchemy parameter binding → психопг-v3 escape'ит на уровне protocol. Safe. ✅raw_props = json.dumps(row, ensure_ascii=False)— JSON serialization, не SQL concat. Safe. ✅10. Memory peak ✅
Расчёт совпадает:
Cross-file impact ✅
_save_raw()signature changed:payload: Any→payload: Any | None = None. Backwards-compat (positional аргументы продолжают работать, default не required). Только 2 call-site (corp_sum + lots_pf) — оба обновлены в этом же PR.ObjectiveClient.stream_report()— новый method, никто кромеscrape_objective.py:264не зовёт. ✅parse_lots_pf_stream— новый, импортируется только через late-import в task. ✅Pre-flight ✅
fix/obj-lots-pf-streaming, не main--no-verify/--force/--amendследовpsycopg2import (grep clean)print()в prod code (толькоlogger.*)Positive observations
ON CONFLICT DO UPDATE/ON CONFLICT DO NOTHINGкорректно использованы_open_stream()дублирует retry/auth/rate-limit logic_request_authed()— некрасиво, но безопасно (можно отрефакторить в shared helper в follow-up)parse_lots_pf+ CLI--file-lots-pfживы)DROP NOT NULLна nullable = no-op)Follow-up (не блокирующее merge)
Создать issue "objective parser: add SAVEPOINT per-row (or per-batch) in parse_lots_pf{,_stream}" — pre-existing tech debt, по
.claude/rules/backend.mdобязательно для loop upsert. Refactor scope: ~30-50 строк, использоватьwith db.begin_nested(): db.execute(_LOTS_UPSERT, [row])per-row либо try/except вокруг каждого batch с graceful skip.Verdict: ✅ APPROVE → squash-merge.
Emergency OOM fix valid, RAM budget математически корректен, ijson API верный, backwards-compat сохранён. SAVEPOINT — pre-existing issue (не регрессия), создам follow-up.
Merged через deep-code-reviewer — verdict APPROVE (P0 critical, OOM fix).
Ключевые проверки:
_open_stream()—client.send(build_request(...), stream=True)корректный httpx Client API, retry/auth логика идентична_request_authedstream_report()—@contextmanagerсfinally: resp.close()— connection close гарантированresult.item— соответствует структуре payload (result: [...]подтверждено по существующемуparse_lots_pf)CAST(:raw_props AS jsonb)— без::typeколоночного trap_save_raw(payload=None)backward compat для corp_sum веткиNits (не блокеры):
_heartbeatне вызывается во время stream — long-running parse (130 batches) без mid-stream heartbeat, но это monitoring-only gapAcceptance после deploy:
run_id=2соstatus='running'objective_raw_reportslots_pf:payload IS NULL,payload_size ≈ 607M,rows_extracted ≈ 324kobjective_lots.raw_idcount растёт батчами по 2500fce48a8Merge commit:
389e141e