Prevenção de Falhas em Cascata: Implementando o Padrão Circuit Breaker Distribuído com Amazon DynamoDB e AWS Lambda

resource "aws_dynamodb_table" "circuit_breaker_state" {
name = "DistributedCircuitBreaker"
billing_mode = "PAY_PER_REQUEST"
hash_key = "IntegrationName"

attribute {
name = "IntegrationName"
type = "S"
}

ttl {
attribute_name = "ExpiresAt"
enabled = true
}

tags = {
Architecture = "Resilience"
Pattern = "CircuitBreaker"
}
}

import time
import boto3
from botocore.exceptions import ClientError
from aws_lambda_powertools import Logger

logger = Logger(service="PaymentAdapter")
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("DistributedCircuitBreaker")

class CircuitOpenException(Exception):
pass

class CircuitBreakerAdapter:
def __init__(self, integration_name: str):
self.integration_name = integration_name

def _is_circuit_closed(self) -> bool:
try:
# Leituras fortemente consistentes garantem o estado global exato
response = table.get_item(
Key={"IntegrationName": self.integration_name},
ConsistentRead=True
)
item = response.get("Item")

if not item:
return True

state = item.get("State", "CLOSED")
expires_at = item.get("ExpiresAt", 0)

# Se está ABERTO mas o tempo expirou, permite sondagem
if state == "OPEN" and time.time() > expires_at:
return self._transition_to_half_open()

return state != "OPEN"
except ClientError as e:
logger.error(f"Falha ao consultar estado do circuito: {str(e)}")
return True # Fail-safe: fecha o circuito se o DDB falhar

def _transition_to_half_open(self) -> bool:
# Lógica de transição omitida para o próximo passo focado em concorrência
return True

def execute_call(self, external_call: callable) -> dict:
if not self._is_circuit_closed():
logger.warning(f"Circuito {self.integration_name} ABERTO. Chamada abortada.")
raise CircuitOpenException("A dependência externa encontra-se inacessível.")

try:
# Executa a chamada HTTP real
return external_call()
except Exception as e:
# Delega o registro da falha
self._record_failure()
raise e

def _record_failure(self):
try:
# Incremento atômico que retorna o novo valor em uma única transação
response = table.update_item(
Key={"IntegrationName": self.integration_name},
UpdateExpression="ADD FailureCount :inc SET #S = if_not_exists(#S, :closed)",
ExpressionAttributeNames={"#S": "State"},
ExpressionAttributeValues={":inc": 1, ":closed": "CLOSED"},
ReturnValues="UPDATED_NEW"
)

new_count = response.get("Attributes", {}).get("FailureCount", 0)
logger.info(f"Contador de falhas para {self.integration_name} atingiu {new_count}")

if new_count >= 10:
self._trip_circuit()

except ClientError as e:
logger.error(f"Erro atômico ao registrar falha: {str(e)}")

def _trip_circuit(self):
cooldown_seconds = 60
expires_at = int(time.time()) + cooldown_seconds

try:
# Gravação condicional: Só abre se ainda estiver FECHADO ou HALF_OPEN
table.update_item(
Key={"IntegrationName": self.integration_name},
UpdateExpression="SET #S = :open, ExpiresAt = :exp",
ConditionExpression="#S = :closed OR #S = :half_open",
ExpressionAttributeNames={"#S": "State"},
ExpressionAttributeValues={
":open": "OPEN",
":closed": "CLOSED",
":half_open": "HALF_OPEN",
":exp": expires_at
}
)
logger.critical(f"Circuito {self.integration_name} DISPARADO (OPEN).")
except ClientError as e:
if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
pass # Outra thread concorrente já abriu o circuito
else:
raise