Light1YT
8206a0b067
Cold §22 forecast measured ~215-233s on prod: §9.x layers re-execute the same horizon/segment-invariant DB loads with identical args hundreds of times per report (profiled: get_competitors x69, market_metrics x124, get_monthly_macro x290). Add a per-report ContextVar cache (forecast_cache(), opened once in the orchestrator) + @cached(key_builder) on the expensive §9.x loaders so each unique load runs ONCE and reuses the same frozen, read-only instance. Output is byte-identical (memoized producers are frozen dataclasses / read-only Pydantic, callers never mutate; cache is per-report, discarded on exit; no-op outside the report build). No concurrency, no signature changes. - forecast_request_cache.py: ContextVar cache + cached() decorator (no-op outside context, reentrant, _MISS sentinel for cached None) - @cached on competitors/future_supply/market_metrics/macro_series/ sales_series/macro_coefficient/demand_normalization/regression loaders - orchestrator: wrap build_site_finder_report in forecast_cache() - 58 tests: key discrimination (call-counting regression guard), no-op-outside, per-report isolation, reentrancy, frozen-producer canary, amplification proof (real get_monthly_macro xN->1) code-reviewer APPROVE (keys correct, mutation-safe, output identical). 1265 forecast/cache tests green. No new deps. Refs #1129.
135 lines
8.8 KiB
Python
135 lines
8.8 KiB
Python
"""Request-scoped memoization для §22-форсайта (#1129) — БЕЗ смены сигнатур сервисов.
|
||
|
||
ПРОБЛЕМА (профиль #1129): один холодный §22-отчёт (`build_site_finder_report`,
|
||
orchestrator.py) собирает 8 §9.x-слоёв, и эти слои МАССОВО пере-вызывают ОДНИ И ТЕ ЖЕ
|
||
горизонт-инвариантные БД-загрузки с ИДЕНТИЧНЫМИ аргументами. Пример амплификации на
|
||
один отчёт (сетка 6/12/18/24, default what_to_build-сетка 3×5):
|
||
• `compute_scenarios` гоняет `compute_demand_supply_forecast` ТРИЖДЫ (conservative/
|
||
base/aggressive) × 4 горизонта;
|
||
• `special_indices`/`recommendation` дважды зовут `rank_segments` → 15 ячеек ×
|
||
`compute_demand_supply_forecast`;
|
||
• КАЖДАЯ `compute_demand_supply_forecast`-ячейка внутри зовёт §9.4
|
||
`compute_demand_normalization` (→ §9.6 `compute_rate_regime_sensitivity` →
|
||
`get_monthly_macro` + 2× `build_sales_series`) + §9.3
|
||
`compute_future_supply_pressure` (→ `compute_market_metrics`) + §9.7
|
||
`get_competitors`.
|
||
Итог — сотни повторных roundtrip'ов к БД, почти все с СОВПАДАЮЩИМИ аргументами
|
||
(один и тот же `district`, один и тот же `spec`, один и тот же `cad_num`,
|
||
горизонт-инвариантный макро-ряд). Это case (c) аудита #1129: ОБЩАЯ дорогая загрузка
|
||
пере-выполняется N раз — мемоизировать её ОДИН раз на отчёт убирает дублирование БЕЗ
|
||
конкурентности и БЕЗ изменения результата.
|
||
|
||
РЕШЕНИЕ: per-request (per-отчёт) кэш на `ContextVar`. Контекст-менеджер
|
||
`forecast_cache()` активирует свежий dict на время сборки ОДНОГО отчёта; декоратор
|
||
`cached(...)` оборачивает дорогие §9.x-сервисы и:
|
||
• ВНУТРИ активного контекста → результат считается один раз на уникальный ключ
|
||
аргументов и переиспользуется (тот же объект — все наши результаты frozen-dataclass
|
||
либо read-only-список frozen-dataclass'ов, см. ниже: callers их НЕ мутируют, поэтому
|
||
шаринг инстанса детерминирован и безопасен);
|
||
• ВНЕ контекста (обычные standalone-вызовы: API `/competitors`, юнит-тесты, прочие
|
||
эндпоинты) → функция выполняется КАК ОБЫЧНО, без кэша. Нулевое изменение поведения
|
||
за пределами сборки §22-отчёта.
|
||
|
||
ПОЧЕМУ ContextVar, а не аргумент-кэш: не надо протаскивать cache-параметр через ~10
|
||
гетерогенных сигнатур §9.x. ContextVar автоматически скоупится сборкой отчёта и
|
||
изолирован per-thread/per-task (важно, если позже сборка уйдёт в ThreadPool — у
|
||
каждого потока свой контекст; здесь же §22 идёт в одном Celery-`def`-воркере).
|
||
|
||
КОРРЕКТНОСТЬ (жёсткий гейт #1129): кэш возвращает ТОТ ЖЕ инстанс для совпадающих
|
||
ключей. Все мемоизируемые продьюсеры возвращают НЕИЗМЕНЯЕМЫЕ значения —
|
||
`@dataclass(frozen=True)` (MarketMetrics / FutureSupplyPressure / DemandSupplyForecast /
|
||
DemandNormalization / RateSensitivity / MacroCoefficient / SalesSeries / MonthlyMacro) и
|
||
Pydantic-модель CompetitorsResponse — а вызывающий код их только ЧИТАЕТ (атрибуты,
|
||
list-comprehension, `for`), нигде не мутирует список/поля на месте. Поэтому шаринг
|
||
кэш-инстанса даёт ПОБАЙТОВО тот же собранный отчёт, что и без кэша.
|
||
|
||
Ключи строятся ЯВНЫМ key-builder'ом на каждую функцию (не «магический» по сигнатуре):
|
||
нормализуем только те аргументы, что РЕАЛЬНО влияют на результат, в hashable-tuple.
|
||
`db`-сессия в ключ НЕ входит (на время отчёта она одна; данные за сборку не меняются).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections.abc import Callable, Iterator
|
||
from contextlib import contextmanager
|
||
from contextvars import ContextVar
|
||
from functools import wraps
|
||
from typing import Any, TypeVar, cast
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Активный per-отчёт кэш (или None вне сборки отчёта). Хранит {ключ: результат}.
|
||
# default=None ⇔ «контекст не активен» → cached(...) выполняет функцию без кэша.
|
||
_CACHE: ContextVar[dict[Any, Any] | None] = ContextVar("forecast_request_cache", default=None)
|
||
|
||
F = TypeVar("F", bound=Callable[..., Any])
|
||
|
||
# Сентинел для «значение отсутствует» — отличаем закэшированный None от промаха.
|
||
_MISS = object()
|
||
|
||
|
||
@contextmanager
|
||
def forecast_cache() -> Iterator[dict[Any, Any]]:
|
||
"""Активировать свежий per-отчёт кэш на время блока (один §22-отчёт).
|
||
|
||
Внутри блока все `cached(...)`-сервисы мемоизируются в общий dict; на выходе он
|
||
отбрасывается (кэш живёт ровно одну сборку отчёта — никакой кросс-отчётной утечки
|
||
устаревших данных). Реентерабельность: вложенный `forecast_cache()` НЕ заводит
|
||
новый кэш — переиспользует уже активный (idempotent), чтобы мемоизация была общей
|
||
на всю сборку, даже если вход обёрнут дважды.
|
||
"""
|
||
existing = _CACHE.get()
|
||
if existing is not None:
|
||
# Уже внутри активного контекста — переиспользуем (не сбрасываем накопленное).
|
||
yield existing
|
||
return
|
||
cache: dict[Any, Any] = {}
|
||
token = _CACHE.set(cache)
|
||
try:
|
||
yield cache
|
||
finally:
|
||
_CACHE.reset(token)
|
||
logger.debug("forecast_cache: closed (entries=%d)", len(cache))
|
||
|
||
|
||
def cached(key: Callable[..., Any], *, label: str) -> Callable[[F], F]:
|
||
"""Декоратор: мемоизировать результат функции per-отчёт по явному ключу аргументов.
|
||
|
||
Args:
|
||
key: builder ключа — принимает те же *args/**kwargs, что обёрнутая функция
|
||
(включая `db` первым позиционным — его игнорируем), и возвращает hashable
|
||
идентификатор того, что РЕАЛЬНО влияет на результат. Должен быть
|
||
детерминирован и полностью покрывать варьирующиеся входы.
|
||
label: короткое имя для namespacing ключа и для debug-лога (имя функции).
|
||
|
||
Returns:
|
||
Декоратор, оборачивающий sync-функцию `(db, ...) -> Result`.
|
||
|
||
Вне активного `forecast_cache()` обёртка просто зовёт оригинал (без кэша) — обычные
|
||
standalone-вызовы (API/тесты) не затрагиваются.
|
||
"""
|
||
|
||
def decorate(fn: F) -> F:
|
||
@wraps(fn)
|
||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||
cache = _CACHE.get()
|
||
if cache is None:
|
||
# Нет активного контекста → поведение по умолчанию, без мемоизации.
|
||
return fn(*args, **kwargs)
|
||
cache_key = (label, key(*args, **kwargs))
|
||
hit = cache.get(cache_key, _MISS)
|
||
if hit is not _MISS:
|
||
return hit
|
||
result = fn(*args, **kwargs)
|
||
cache[cache_key] = result
|
||
return result
|
||
|
||
return cast(F, wrapper)
|
||
|
||
return decorate
|
||
|
||
|
||
def is_active() -> bool:
|
||
"""Активен ли сейчас per-отчёт кэш (для диагностики/тестов). PURE-ish."""
|
||
return _CACHE.get() is not None
|