Invalidation — evict на write, доменные события и distributed-sync (Python)

Как инвалидировать Redis-кеш в FastAPI: явный evict на каждом write, evict по доменному событию и почему distributed sync встроен в Redis без дополнительной обвязки.

Опубликовано 19 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: R-CACHE-INV-1 … R-CACHE-INV-4 и R-CACHE-INV-X1 … R-CACHE-INV-X3 из Caching Rules → раздел 5. Invalidation.

Важно знать

На каждом write-методе того же ресурса — await cache.delete(key) с явным ключом (R-CACHE-INV-1).

Несколько кешей — evict всех затронутых в одном use-case handler-е (R-CACHE-INV-2).

Доменные события — invalidation как side-effect через отдельный обработчик события (R-CACHE-INV-3).

Distributed invalidation встроена в Redis: DEL от одного инстанса виден всем остальным (R-CACHE-INV-4).

FLUSHDB / namespace-wide delete без причины — только при админских операциях, иначе DB-спайк (R-CACHE-INV-X1).

Только TTL для consistency — приемлемо для feature-flags, неприемлемо для money/orders (R-CACHE-INV-X2).

Eventual consistency без декларации в OpenAPI-контракте — нарушение контракта (R-CACHE-INV-X3).

Invalidation — единственная защита от stale-data. TTL даёт верхнюю границу задержки; invalidation делает задержку нулевой. Без явного evict кеш означает «обещаем устаревшую правду на N минут».

В Python нет декларативного @CacheEvict Spring. Evict — явный вызов cache-порта. Это делает invalidation видимой в коде handler-а, что само по себе ценно.

Evict на write-методе

R-CACHE-INV-1: каждый write-handler, меняющий ресурс, инвалидирует соответствующий кеш.

Cache-порт в core/ (Protocol):

from typing import Protocol


class OrderCachePort(Protocol):
    async def get_summary(self, order_id: str) -> dict | None: ...
    async def set_summary(self, order_id: str, data: dict) -> None: ...
    async def evict_summary(self, order_id: str) -> None: ...

Handler ConfirmOrderHandler инвалидирует order-summaries после подтверждения:

from dataclasses import dataclass

from app.core.ports.cache import OrderCachePort
from app.core.ports.repository import OrderRepository


@dataclass
class ConfirmOrderCommand:
    order_id: str
    confirmed_by: str


class ConfirmOrderHandler:
    def __init__(
        self,
        orders: OrderRepository,
        cache: OrderCachePort,
    ) -> None:
        self._orders = orders
        self._cache = cache

    async def handle(self, cmd: ConfirmOrderCommand) -> None:
        order = await self._orders.get(cmd.order_id)
        order.confirm(cmd.confirmed_by)
        await self._orders.save(order)
        await self._cache.evict_summary(cmd.order_id)

Evict происходит после успешного сохранения. Если save бросает исключение — evict не вызывается и кеш остаётся с прежним (правильным) значением.

Ключ evict обязан совпадать с ключом, по которому закешировано значение. Иначе invalidation промахивается и stale-данные остаются до TTL.

Несколько кешей — evict всех затронутых

R-CACHE-INV-2: одно изменение может затрагивать несколько проекций. Evict всех — в одном handler-е.

Пример: обновление профиля клиента инвалидирует customer-profiles и customer-limits (лимиты зависят от статуса профиля):

class UpdateCustomerProfileHandler:
    def __init__(
        self,
        customers: CustomerRepository,
        profile_cache: CustomerProfileCachePort,
        limits_cache: CustomerLimitsCachePort,
    ) -> None:
        self._customers = customers
        self._profile_cache = profile_cache
        self._limits_cache = limits_cache

    async def handle(self, cmd: UpdateCustomerProfileCommand) -> None:
        customer = await self._customers.get(cmd.customer_id)
        customer.apply_profile_update(cmd.display_name, cmd.segment)
        await self._customers.save(customer)

        await self._profile_cache.evict(cmd.customer_id)
        await self._limits_cache.evict(cmd.customer_id)

Два отдельных evict лучше, чем namespace-wide flush: затронуты ровно те ключи, которые изменились.

Если операций evict много — можно сгруппировать в отдельный метод cache-порта:

async def evict_customer_caches(
    profile_cache: CustomerProfileCachePort,
    limits_cache: CustomerLimitsCachePort,
    customer_id: str,
) -> None:
    await profile_cache.evict(customer_id)
    await limits_cache.evict(customer_id)

Invalidation через доменное событие

R-CACHE-INV-3: паттерн «invalidation as side-effect of domain event». Handler события — единственное место, отвечающее за инвалидацию кеша при любом use case, меняющем сущность.

Событие ProductUpdatedEvent может быть поднято из UpdateProductHandler, ApproveProductHandler, BulkImportHandler — все они просто публикуют событие; кто и что инвалидирует — решает обработчик события.

from dataclasses import dataclass


@dataclass(frozen=True)
class ProductUpdatedEvent:
    product_id: str
    category_id: str

Обработчик события (FastAPI + fastapi-events или самописный event-bus):

class ProductCacheInvalidationHandler:
    def __init__(
        self,
        product_cache: ProductCachePort,
        category_cache: CategoryCachePort,
    ) -> None:
        self._product_cache = product_cache
        self._category_cache = category_cache

    async def on_product_updated(self, event: ProductUpdatedEvent) -> None:
        await self._product_cache.evict(event.product_id)
        await self._category_cache.evict_product_list(event.category_id)

Почему так:

Развязка. Handler бизнес-логики не знает, какие кеши надо инвалидировать.
Единое место. Все источники изменения Product (импорт, правка, модерация) публикуют одно событие — обработчик реагирует всегда.
Внешние события. Если ProductUpdatedEvent приходит из Kafka от другого сервиса — тот же обработчик инвалидирует локальный кеш.

Distributed invalidation — встроена в Redis

R-CACHE-INV-4: Redis — общий source of truth. Когда один инстанс удаляет ключ, все остальные инстансы видят miss при следующем чтении. Никакой дополнительной pub/sub-обвязки для invalidation не нужно.

pod-1: await cache.evict("order-summaries:ord-88") → Redis DEL order-summaries:ord-88
pod-2: await cache.get("order-summaries:ord-88")   → miss → load from DB → cache.set(...)

Это работает через адаптер redis.asyncio:

import redis.asyncio as aioredis


class RedisOrderCacheAdapter:
    def __init__(self, redis: aioredis.Redis) -> None:
        self._redis = redis

    async def get_summary(self, order_id: str) -> dict | None:
        raw = await self._redis.get(f"order-summaries:{order_id}")
        if raw is None:
            return None
        return json.loads(raw)

    async def set_summary(self, order_id: str, data: dict, ttl: int) -> None:
        await self._redis.set(
            f"order-summaries:{order_id}",
            json.dumps(data),
            ex=ttl,
        )

    async def evict_summary(self, order_id: str) -> None:
        await self._redis.delete(f"order-summaries:{order_id}")

Сценарий с многоуровневым кешем (L1 = in-process dict, L2 = Redis) — редкий случай, требует явной инвалидации L1 через Redis pub/sub. В стандартной конфигурации UCP-сервисов — только Redis, один уровень.

Что запрещено

`FLUSHDB` / wildcard delete без причины

R-CACHE-INV-X1: операция, сбрасывающая весь namespace или всю БД Redis, — атомная бомба.

# ОПАСНО — сброс всего namespace
keys = await redis.keys("order-summaries:*")
if keys:
    await redis.delete(*keys)

Сценарий: 50 000 заказов в кеше, 500 RPS. Один write сбрасывает всё → следующие 50 000 читателей все промахиваются → 50 000 запросов в БД за секунды → перегрузка.

Допустимо только при явных админских операциях:

Перестройка кеша после миграции структуры.
Ручная очистка в DEV-окружении.

В prod-handler-ах — только точечный delete(key).

Только TTL для consistency

R-CACHE-INV-X2: «обновится само через минуту».

Данные	Только TTL — допустимо?
Feature flags	Да — задержка 60 сек приемлема
Справочник валют	Да — меняется редко
Профиль клиента	Сомнительно — пользователь видит старые данные после сохранения
Сводка по заказу	Нет — клиент видит старый статус, путаница при ожидании
Баланс / лимиты	Нет — money, любой stale = инцидент

Для money/orders — явный evict обязателен, TTL служит только страховкой.

Eventual consistency без декларации в OpenAPI

R-CACHE-INV-X3: если endpoint может вернуть stale-данные — это часть контракта, её документируют.

/products/{product_id}:
  get:
    summary: Получить карточку продукта
    description: |
      Возвращает карточку из кеша.

      **Eventual consistency**: после обновления продукта возможна
      задержка до 10 минут (TTL кеша) до отражения изменений.
      Для немедленной актуальности используйте
      `GET /products/{product_id}?fresh=true`, обходящий кеш.
    responses:
      200:
        description: Карточка продукта (может отставать на TTL)

Без декларации интеграционный тест PATCH + немедленный GET зафиксирует расхождение как баг — который на самом деле является архитектурным выбором.

Что запрещено — таблица

Антипаттерн	Правило	Что взамен
`FLUSHDB` / удаление по маске `*` в prod-handler	`R-CACHE-INV-X1`	`delete(key)` точечно
Только TTL для money/orders	`R-CACHE-INV-X2`	явный evict на каждом write
Eventual consistency без декларации в OpenAPI	`R-CACHE-INV-X3`	`description: 'Возможна задержка...'`
Ключ evict не совпадает с ключом чтения	`R-CACHE-INV-1`	одна функция формирования ключа
Evict после исключения в save	`R-CACHE-INV-1`	evict только после успешного write
Evict одного кеша, когда write затрагивает несколько	`R-CACHE-INV-2`	evict всех затронутых кешей
Каждый handler дублирует логику invalidation	`R-CACHE-INV-3`	отдельный обработчик события
Redis pub/sub для инвалидации между инстансами	`R-CACHE-INV-4`	встроено в Redis, `delete` достаточно

Куда дальше

Cache stampede — distributed lock при одновременных miss на один ключ.
Конфигурация — pydantic-settings для per-cache TTL и backend.
Ключи — формирование ключей: namespace, explicit, kebab-case.
Observability — hit rate метрика, alert при < 70%, evict на DEBUG.
Паттерны — cache-aside, write-through, refresh-ahead в Python.
TTL — TTL как страховка, типовые значения, per-cache конфиг.
Где кешируем — что кешировать: проекции, не агрегаты; не write-path.