Feature flags dinâmicos em FastAPI: toggles, rollout e cache com Redis

Motivacao e trade-offs

Por que usar feature flags dinâmicas

Feature flags permitem ativar ou desativar comportamentos em runtime sem deploy. Em serviços web isso é crítico para rollout progressivo, testes A/B e mitigacao de risco. Em FastAPI, que ja parte do modelo async, o desafio é aplicar flags de modo consistente, performatico e sem criar gargalos de IO.

Trade-offs principais

Existem escolhas tipicas que impactam latencia, consistencia e custo operacional. Consultar Redis a cada request garante consistencia imediata mas aumenta latencia e custo de rede. Cache local reduz latencia, porem leva a eventual consistency e exige estrategia de expiração ou refresho.

Decisões concretas que você vai confrontar:

• Polling periodico vs TTL: quanto tempo voce aceita inconsistencia?
• Rollout por porcentagem: deterministico por user id ou aleatorio por request?
• Uso de cliente sincrono em rota async: bloqueio e perda de throughput

Recomendacao rapida: evite bloquear o loop principal. Use cliente redis async e um cache local com TTL e refresh em background para alta taxa de requests.

Fundamentos tecnicos

Arquitetura desejada

A arquitetura que proponho combina tres pecas: armazenamento central de flags em Redis, servico de acesso a flags com cliente async e um cache local (in-memory) com TTL e refresh em background. Isso equilibra consistencia, latencia e simplicidade operacional.

Padroes e boas praticas

Padroes importantes:

1. Separar logica de resolucao de flag do handler HTTP
2. Evitar efeito colateral em request handlers (sem blocos IO sincronos)
3. Rollout deterministico usando hashing consistente para mesmo usuario
4. Fallbacks bem definidos para caso de Redis indisponivel

Outra boa pratica: expor um endpoint administrativo seguro para setar e inspecionar flags, mas nunca depender exclusivamente desse endpoint na aplicacao critica sem RBAC.

Exemplo 1: caso simples

Jeito errado (code review)

Aqui um exemplo comum que vejo em code reviews: usar um dicionario global para flags e um cliente Redis sincrono dentro de rotas async. Funciona no local, mas em producao bloqueia workers e produz comportamento nondeterministico sob carga.

Contexto do exemplo: rota que decide se ativa o novo UI consultando um dicionario global e fallback para Redis sincrono.

from fastapi import FastAPI
import redis

app = FastAPI()
r = redis.Redis(host='localhost', port=6379)

# dicionario global mutavel
flags = {}

@app.get('/new-ui')
async def new_ui(user_id: int):
    # problema 1: leitura de estado global sem lock em ambiente concorrente
    # problema 2: r.get eh sincrono e bloqueia o event loop
    val = flags.get('new_ui')
    if val is None:
        rv = r.get('feature:new_ui')
        val = rv == b'true'
        flags['new_ui'] = val
    return {'new_ui': bool(val)}

O que acontece: em carga alta, chamadas a r.get bloqueiam o loop async. Se o processo tem um pool de workers (uvicorn com varios workers), estados locais divergem porque cada worker tem seu proprio dicionario flags. Atualizacoes de flags ficam inconsistentes entre processos.

Correcoes e por que

Corrigir envolve trocar o cliente para a versao async, remover estados globais inseguros ou proteger com locks async, e preferir um padrao de cache local com refresh controlado. Abaixo temos uma implementacao simples e robusta.

Contexto: servico de feature flags que carrega de Redis periodicamente e serve consultas em memoria com garantia de non-blocking.

import asyncio
import time
import hashlib
from typing import Dict, Any
from fastapi import FastAPI
import redis.asyncio as aioredis

app = FastAPI()

class FeatureFlagService:
    def __init__(self, redis_url: str, ttl: int = 5):
        self.redis = aioredis.from_url(redis_url, decode_responses=True)
        self.local: Dict[str, Any] = {}
        self.ttl = ttl
        self._lock = asyncio.Lock()
        self._task = None

    async def start(self):
        if self._task is None:
            self._task = asyncio.create_task(self._refresh_loop())

    async def _refresh_loop(self):
        while True:
            try:
                await self.refresh()
            except Exception:
                # nao explodir o loop em producao, log e retry
                pass
            await asyncio.sleep(self.ttl)

    async def refresh(self):
        async with self._lock:
            raw = await self.redis.hgetall('features')
            now = time.time()
            self.local = {'data': raw, 'expires_at': now + self.ttl}

    async def is_enabled(self, key: str, user_id: int = None) -> bool:
        async with self._lock:
            data = self.local.get('data', {})
            raw = data.get(key)
        if raw is None:
            # fallback direto ao redis em caso de miss
            val = await self.redis.hget('features', key)
            raw = val
        if raw is None:
            return False
        # formato esperado: simple true/false ou json com rollout
        if raw in ('true', 'false'):
            return raw == 'true'
        # formato rollout: 'p:30' significa 30% rollout
        if raw.startswith('p:') and user_id is not None:
            pct = int(raw.split(':', 1)[1])
            bucket = int(hashlib.sha256(f'{user_id}:{key}'.encode()) .hexdigest(), 16) % 100
            return bucket < pct
        return False

flag_service = FeatureFlagService('redis://localhost:6379', ttl=5)

@app.on_event('startup')
async def startup():
    await flag_service.start()

@app.get('/new-ui')
async def new_ui(user_id: int):
    enabled = await flag_service.is_enabled('new_ui', user_id)
    return {'new_ui': enabled}

O que acontece: agora usamos redis.asyncio, temos um loop de refresh que mantém um cache local com TTL, e aplicamos rollout deterministico via hashing. Nao bloqueamos o event loop com chamadas sincronas e temos consistencia eventual controlada pelo TTL.

Exemplo 2: caso avancado do mundo real

Requisitos reais

Imagine um endpoint de recomendacao que precisa aplicar multiplas flags por usuario: experimento A/B, toggle de algoritmo e desconto promocional. Requisitos: latencia 95 perc < 100ms, consistencia eventual aceitavel por 10s, e capacidade de desligar uma feature imediatamente em emergencia.

Arquitetura e implementacao sugerida

Neste caso o padrao evolui: cache local + refresh rapido + sinal de invalidacao imediata via publish/subscribe Redis para bypass do TTL quando necessario. Ao atualizar uma flag no Redis, um produtor publica uma mensagem channel que notifica instancias para invalidar cache instantaneamente.

Contexto: abaixo implementamos o subscriber simple que invalida cache ao receber mensagem. Isso permite both eventual refresh e invalidacao quase imediata.

import asyncio
import json
import hashlib
from fastapi import FastAPI
import redis.asyncio as aioredis

app = FastAPI()

class FeatureFlagWithPubSub(FeatureFlagService):
    def __init__(self, redis_url: str, ttl: int = 5):
        super().__init__(redis_url, ttl)
        self.pubsub = self.redis.pubsub()

    async def start(self):
        await super().start()
        asyncio.create_task(self._subscriber())

    async def _subscriber(self):
        await self.pubsub.subscribe('features:invalidate')
        async for message in self.pubsub.listen():
            if message is None:
                continue
            if message.get('type') != 'message':
                continue
            try:
                payload = message.get('data')
                if isinstance(payload, bytes):
                    payload = payload.decode()
                data = json.loads(payload)
                key = data.get('key')
                # invalidar cache parcial
                async with self._lock:
                    if 'data' in self.local and key in self.local['data']:
                        del self.local['data'][key]
            except Exception:
                pass

flag_service_pub = FeatureFlagWithPubSub('redis://localhost:6379', ttl=10)

@app.on_event('startup')
async def startup_pub():
    await flag_service_pub.start()

@app.get('/recommend')
async def recommend(user_id: int):
    # combinacao de flags
    use_new_algo = await flag_service_pub.is_enabled('new_algo', user_id)
    show_experiment = await flag_service_pub.is_enabled('exp_banner', user_id)
    return {'new_algo': use_new_algo, 'exp_banner': show_experiment}

O que acontece: atualizacoes de flags podem usar HSET em Redis para mudar os valores e publicar uma mensagem no canal features:invalidate com um payload como {'key': 'new_algo'}. Instancias recebem e invalidam o key local, forçando uma leitura direta no Redis no proximo request.

Como atualizar uma flag de forma segura

Workflow recomendado para operadores: 1) Escrever novo valor em Redis HSET features key value 2) PUBLISH features:invalidate '{"key":"key"}' 3) Monitorar logs e metrics. Isso garante invalidacao imediata e rollback rapido.

Debugging e troubleshooting

Erros comuns e armadilhas

Lista de problemas frequentes:

• Usar cliente redis sincrono dentro de função async causando latencia e timeouts
• Manter estado mutavel global sem sincronizacao entre processos ou threads
• Rollout por porcentagem realizado por random sem seed, causando usuarios mudando entre buckets a cada request
• Dependencia exclusiva em cache local sem mecanismo de invalidacao para emergencia

Exemplo de bug sutil: usar hash(user_id) padrao Python para bucket. O hash padrao pode variar entre execucoes do processo por segurança e nao e deterministico entre processos. Use hashing criptografico deterministico como sha256 acima.

Diagnosticos praticos

Passos para investigar problemas:

1. Verifique se existem chamadas sincronas no loop async com strace ou stacktraces
2. Inspecione divergencias entre instancias consultando Redis e comparando conteudo do HSET
3. Simule rollout com varios user_id para validar que bucketizacao e deterministica
4. Teste scenarios de Redis indisponivel e confirme fallback esperado

Ferramentas util: redis-cli para inspecao direta, tcpdump em casos extremos e logs estruturados das invalidacoes publish/subscribe.

Exemplos de rollout e testabilidade

Algoritmo de bucket deterministico

Contexto: garantir que um mesmo usuario fica sempre no mesmo bucket durante o rollout. Aqui o algoritmo usa sha256 para determinismo e uniformidade.

import hashlib

def in_rollout(user_id: int, flag_key: str, percent: int) -> bool:
    # pega um bucket 0-99 deterministico
    h = hashlib.sha256(f'{user_id}:{flag_key}'.encode()).hexdigest()
    bucket = int(h, 16) % 100
    return bucket < percent

# uso
# in_rollout(1234, 'new_ui', 30) -> True se usuario 1234 estiver nos 30% selecionados

O que acontece: o mesmo user_id e flag_key sempre produzem a mesma decisao. Altera-se percent para expandir ou reduzir rollout sem mexer em usuarios individualmente.

Como testar unitariamente

Teste unitario deve cobrir:

• Comportamento para valores true/false simples
• Rollout boundaries (0,1,99,100)
• Falha de Redis e fallback

Exemplo de teste simples para a funcao in_rollout: controlar listagem de user_id esperados para um percent conhecido e validar estabilidade.

Conclusao

Feature flags sao poderosas mas podem introduzir complexidade operacional se implementadas de forma direta e insegura. Erros com clientes sincronos, caches locais inconsistentes e hashing nao deterministico sao comuns e simples de evitar.

Resumo pratico:

• Use cliente Redis async para nao bloquear FastAPI
• Mantenha cache local com TTL e refresh em background para latencia baixa
• Implemente invalidacao via Pub/Sub para emergencias
• Use hashing deterministico para rollouts por porcentagem

Adotar esse padrao reduz latencia, aumenta previsibilidade dos rollouts e fornece mecanismos de emergencia para rollback rapido. A integracao entre Redis e cache in-memory e o ponto de equilibrio entre performance e controle operacional.

Checklist rapido antes do deploy:

1. Remover qualquer cliente redis sincrono do codigo async
2. Garantir que testes cubram rollout e falhas de redis
3. Configurar metrics e alertas para latencia e taxa de misses do cache

Implemente, execute testes de carga e verifique que sua soluçao nao aumenta p95 da aplicacao. Se aumentar, reduza TTL ou melhore o refresh em background.

Boa pratica adicional: documente os formatos aceitos no HSET de Redis (ex: true/false, p:30) e mantenha um endpoint admin que valida o formato antes de aplicar changes em producao.

Fim.