Estratégias de cache invalidation em sistemas com múltiplos serviços

1. Fundamentos da Invalidação de Cache em Arquiteturas Distribuídas

1.1. O problema da consistência de dados entre serviços

Em sistemas com múltiplos serviços, cada serviço pode manter seu próprio cache local ou compartilhado. Quando um serviço altera dados no banco de dados principal, os caches de outros serviços que armazenam cópias desses dados tornam-se obsoletos. Esse problema é conhecido como consistência de cache e exige estratégias robustas de invalidação.

Considere um sistema de e-commerce com serviços separados para catálogo, carrinho e estoque. Quando o serviço de estoque reduz a quantidade de um item, o cache do catálogo que exibe a disponibilidade do produto precisa ser atualizado. Sem uma estratégia de invalidação, o usuário pode ver um produto como disponível quando na verdade ele está esgotado.

1.2. Latência vs. coerência: trade-offs em sistemas multi-serviço

A escolha da estratégia de invalidação envolve um trade-off fundamental: latência versus coerência. Estratégias que garantem dados sempre atualizados (alta coerência) geralmente aumentam a latência das operações de leitura e escrita. Por outro lado, estratégias que priorizam baixa latência podem expor dados obsoletos por períodos curtos.

Estratégia Coerência Latência Caso de uso
TTL curto Baixa Baixa Dados não críticos
Invalidação síncrona Alta Alta Transações financeiras
Write-through Alta Média Catálogos de produtos

1.3. Padrões de cache: write-through, write-behind e cache-aside

Os três padrões fundamentais de cache distribuem a responsabilidade entre serviço e cache:

  • Cache-aside: O serviço verifica o cache primeiro. Se não encontrar, busca no banco e armazena.
  • Write-through: Toda escrita no banco é replicada imediatamente no cache.
  • Write-behind: As escritas são acumuladas e enviadas em lote ao cache e banco.

Exemplo de implementação de cache-aside em Python:

def get_user(user_id):
    user = cache.get(f"user:{user_id}")
    if user is None:
        user = database.query("SELECT * FROM users WHERE id = ?", user_id)
        cache.set(f"user:{user_id}", user, ttl=300)
    return user

2. Estratégias Baseadas em TTL (Time-To-Live)

2.1. TTL fixo vs. adaptativo: definindo janelas de expiração

O TTL fixo define um tempo de expiração uniforme para todas as entradas de cache. O TTL adaptativo ajusta dinamicamente o tempo de expiração com base nos padrões de acesso e frequência de alteração dos dados.

# TTL adaptativo baseado em frequência de alteração
def calculate_ttl(entity_type, update_frequency):
    if update_frequency == "high":
        return 30  # segundos
    elif update_frequency == "medium":
        return 300 # 5 minutos
    else:
        return 3600 # 1 hora

2.2. Estratégias de renovação de TTL com base em padrões de acesso

A renovação de TTL (TTL refresh) estende o tempo de vida do cache quando um item é acessado frequentemente. Isso reduz recargas desnecessárias para dados populares.

def get_with_ttl_refresh(key, fetch_func, base_ttl=300):
    data, remaining_ttl = cache.get_with_ttl(key)
    if data is None:
        data = fetch_func()
        cache.set(key, data, ttl=base_ttl)
    elif remaining_ttl < base_ttl * 0.2:
        # Renova o TTL se estiver próximo de expirar
        cache.expire(key, base_ttl)
    return data

2.3. Limitações do TTL em cenários de dados críticos e sincronia

O TTL não garante consistência imediata. Para dados críticos como saldo de conta ou status de pagamento, o TTL pode expor dados obsoletos por segundos ou minutos. Nesses casos, a invalidação por notificação é mais adequada.

3. Invalidação por Notificação entre Serviços

3.1. Uso de message brokers (Redis Pub/Sub, RabbitMQ) para broadcasts de invalidação

Message brokers permitem que serviços publiquem eventos de alteração que outros serviços consomem para invalidar seus caches.

# Serviço de estoque publica evento de alteração
def update_stock(product_id, new_quantity):
    database.update("products", {"stock": new_quantity}, {"id": product_id})
    redis.publish("cache:invalidation", json.dumps({
        "entity": "product",
        "id": product_id,
        "action": "update"
    }))

# Serviço de catálogo consome eventos
def listen_invalidation():
    pubsub = redis.pubsub()
    pubsub.subscribe("cache:invalidation")
    for message in pubsub.listen():
        event = json.loads(message["data"])
        if event["entity"] == "product":
            cache.delete(f"product:{event['id']}")

3.2. Implementação de filas de eventos de alteração com garantia de entrega

Filas como RabbitMQ garantem que eventos de invalidação não sejam perdidos, mesmo que o serviço consumidor esteja temporariamente indisponível.

# Publicação com confirmação
channel.confirm_delivery()
channel.basic_publish(
    exchange='cache_invalidation',
    routing_key='product.update',
    body=json.dumps(event),
    properties=pika.BasicProperties(delivery_mode=2)  # persistente
)

3.3. Padrão de eventos de domínio disparando invalidações em cascata

Eventos de domínio representam mudanças significativas no sistema. Uma alteração em um pedido pode invalidar caches de cliente, estoque e relatórios simultaneamente.

# Evento de domínio: OrderPlaced
event = {
    "type": "OrderPlaced",
    "order_id": 12345,
    "customer_id": 678,
    "products": [{"id": 1, "quantity": 2}, {"id": 3, "quantity": 1}]
}

# Serviços afetados invalidam seus caches
# - Serviço de cliente: invalida cache de pedidos do cliente
# - Serviço de estoque: invalida cache de disponibilidade dos produtos
# - Serviço de relatórios: invalida cache de métricas do dia

4. Invalidação por Versionamento de Dados

4.1. Chaves de cache com versão incremental (ex: user:123:v2)

O versionamento de chave permite invalidar caches sem remover entradas existentes. Basta incrementar a versão e usar a nova chave.

# Ao atualizar um usuário
def update_user(user_id, new_data):
    database.update("users", new_data, {"id": user_id})
    current_version = cache.get(f"user:version:{user_id}") or 1
    new_version = current_version + 1
    cache.set(f"user:version:{user_id}", new_version)
    # A próxima leitura usará a nova versão
    cache.set(f"user:{user_id}:v{new_version}", new_data, ttl=3600)

4.2. Versionamento global vs. por entidade: impacto no armazenamento

O versionamento global usa um número de versão para todo o cache, enquanto o versionamento por entidade mantém versões individuais. O versionamento por entidade é mais eficiente em armazenamento, pois apenas entidades alteradas geram novas versões.

# Versionamento por entidade
def get_cached_user(user_id):
    version = cache.get(f"user:version:{user_id}") or 1
    return cache.get(f"user:{user_id}:v{version}")

# Versionamento global
GLOBAL_VERSION = 1
def get_cached_product(product_id):
    return cache.get(f"product:{product_id}:v{GLOBAL_VERSION}")

4.3. Estratégias de migração de versão sem downtime

Para migrar entre versões de cache sem interromper o serviço, mantenha ambas as versões ativas por um período de transição.

# Migração gradual de v1 para v2
def get_user_migration(user_id):
    # Tenta v2 primeiro
    user = cache.get(f"user:{user_id}:v2")
    if user:
        return user
    # Fallback para v1
    user = cache.get(f"user:{user_id}:v1")
    if user:
        # Atualiza para v2 em background
        cache.set(f"user:{user_id}:v2", user, ttl=3600)
    return user

5. Cache Invalidation com Write-Through e Write-Behind

5.1. Write-through: atualização síncrona do cache e banco

No write-through, a escrita no banco e no cache ocorre na mesma transação, garantindo consistência imediata.

def update_inventory(product_id, quantity):
    with database.transaction():
        database.execute("UPDATE inventory SET quantity=? WHERE product_id=?", 
                        (quantity, product_id))
        cache.set(f"inventory:{product_id}", quantity)

5.2. Write-behind: consolidação de escritas e invalidação assíncrona

No write-behind, as escritas são acumuladas e processadas em lote, reduzindo a latência para o cliente.

# Buffer de escritas
write_buffer = []

def update_inventory_async(product_id, quantity):
    write_buffer.append({"product_id": product_id, "quantity": quantity})
    cache.delete(f"inventory:{product_id}")  # Invalida imediatamente

# Processamento em lote a cada 5 segundos
def flush_writes():
    batch = write_buffer[:100]
    write_buffer = write_buffer[100:]
    with database.transaction():
        for item in batch:
            database.execute("UPDATE inventory SET quantity=? WHERE product_id=?", 
                            (item["quantity"], item["product_id"]))

5.3. Riscos de dados sujos e estratégias de reconciliação

Dados sujos ocorrem quando o cache é atualizado mas o banco falha, ou vice-versa. Estratégias de reconciliação incluem:

  • Retry com backoff: Tentar novamente a operação falha
  • Comparação periódica: Verificar consistência entre cache e banco
  • Versão de dados: Armazenar timestamp da última atualização
def reconcile_cache(entity_type, entity_id):
    cached_data = cache.get(f"{entity_type}:{entity_id}")
    db_data = database.query(f"SELECT * FROM {entity_type} WHERE id=?", entity_id)
    if cached_data and db_data and cached_data["version"] < db_data["version"]:
        cache.set(f"{entity_type}:{entity_id}", db_data)

6. Estratégias de Invalidação por Dependência de Dados

6.1. Grafos de dependência entre entidades e caches relacionados

Entidades frequentemente têm dependências. Um usuário tem posts, que têm comentários. Alterar o usuário pode invalidar caches de posts e comentários.

dependencies = {
    "user": ["user:posts", "user:comments"],
    "post": ["post:comments", "post:author"],
    "comment": ["comment:post"]
}

def invalidate_with_dependencies(entity_type, entity_id):
    cache.delete(f"{entity_type}:{entity_id}")
    for dependent in dependencies.get(entity_type, []):
        cache.delete(f"{dependent}:{entity_id}")

6.2. Invalidação em lote de chaves relacionadas

Para evitar múltiplas chamadas de rede, agrupe invalidações relacionadas em uma única operação.

# Invalidação em lote com Redis pipeline
def invalidate_batch(keys):
    pipeline = cache.pipeline()
    for key in keys:
        pipeline.delete(key)
    pipeline.execute()

# Exemplo: invalidar usuário e todos os seus posts
user_id = 123
post_ids = cache.smembers(f"user:{user_id}:post_ids")
keys_to_invalidate = [f"user:{user_id}"] + [f"post:{pid}" for pid in post_ids]
invalidate_batch(keys_to_invalidate)

6.3. Uso de bloom filters para rastrear dependências sem sobrecarga

Bloom filters permitem verificar rapidamente se uma chave de cache depende de outra, sem armazenar explicitamente todas as dependências.

from pybloom import BloomFilter

# Bloom filter de dependências
dependency_filter = BloomFilter(capacity=100000, error_rate=0.001)

def track_dependency(entity_key, dependent_key):
    dependency_filter.add(f"{entity_key}:{dependent_key}")

def might_have_dependency(entity_key, dependent_key):
    return dependency_filter.contains(f"{entity_key}:{dependent_key}")

7. Monitoramento e Debug de Invalidação em Produção

7.1. Métricas de hit ratio, stale reads e latência de invalidação

Métricas essenciais para monitorar a eficácia da invalidação:

# Coleta de métricas
metrics = {
    "cache_hit_ratio": 0.85,
    "stale_reads_per_minute": 12,
    "invalidation_latency_ms": 45,
    "invalidation_queue_size": 0
}

# Exemplo de monitoramento
def record_cache_access(hit):
    if hit:
        statsd.increment("cache.hit")
    else:
        statsd.increment("cache.miss")
    statsd.gauge("cache.hit_ratio", 
                 statsd.get("cache.hit") / (statsd.get("cache.hit") + statsd.get("cache.miss")))

7.2. Logging estruturado de eventos de cache para rastreamento

Logs estruturados facilitam a depuração de problemas de invalidação.

import structlog
logger = structlog.get_logger()

def invalidate_cache(entity_type, entity_id, reason):
    logger.info("cache.invalidation",
                entity_type=entity_type,
                entity_id=entity_id,
                reason=reason,
                timestamp=time.time())

7.3. Ferramentas de observabilidade: tracing distribuído e dashboards

Ferramentas como Jaeger, Zipkin e Grafana permitem rastrear requisições através de múltiplos serviços e visualizar métricas de cache.

# Exemplo de span para tracing distribuído
from opentelemetry import trace
tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("cache.invalidation") as span:
    span.set_attribute("entity_type", "user")
    span.set_attribute("entity_id", user_id)
    span.set_attribute("invalidation_strategy", "notification")
    # Executa invalidação

Referências