Gestión de Conexiones WebSocket Persistentes con Leases Distribuidas en DynamoDB

La gestión de conexiones de red persistentes y con estado, como los WebSockets, en un entorno distribuido presenta un problema fundamental de coordinación: cómo asegurar que exactamente un worker sea propietario de cada conexión en todo momento, incluso frente a fallos, despliegues y escalado. A diferencia de las solicitudes HTTP sin estado, las conexiones WebSocket requieren que el servidor mantenga un estado continuo, lo que introduce desafíos operacionales significativos como la pérdida de datos durante fallos de workers, interrupciones en despliegues rolling, y el riesgo de doble reclamación de conexiones.

Este patrón resuelve el problema utilizando un mecanismo de lease distribuido basado en escrituras condicionales atómicas en una base de datos NoSQL. Al asignar una 'lease' (un reclamo de propiedad con un tiempo limitado) a cada conexión y requerir que los workers la renueven continuamente, el sistema puede detectar y reasignar automáticamente las conexiones huérfanas cuando un worker falla o se retira. Este enfoque evita la necesidad de un servicio de coordinación distribuida dedicado, simplificando la arquitectura y reduciendo la latencia de recuperación.

Arquitectura del Sistema

La arquitectura propuesta se basa en un fleet de workers ejecutándose en Amazon ECS con AWS Fargate, coordinados principalmente por Amazon DynamoDB. Los eventos de inicio/parada de sesiones de streaming se ingieren a través de Amazon API Gateway y son procesados por una función AWS Lambda, que actualiza el estado de la conexión en DynamoDB y encola una notificación en Amazon SQS. DynamoDB actúa como la fuente de verdad para el estado de la conexión y la propiedad de la lease, utilizando escrituras condicionales para garantizar la atomicidad en la adquisición y renovación de leases.

Cada conexión se representa como un ítem en DynamoDB con atributos como connection_id (Partition Key), desired_state, ws_url, lease_owner, lease_expires_at_ms y last_seq. Un Global Secondary Index (GSI) sobre desired_state y lease_expires_at_ms permite consultas eficientes para identificar conexiones huérfanas. Los workers realizan un 'heartbeat' continuo para renovar sus leases y un 'reconciliation loop' periódico para detectar y adquirir leases expiradas o no reclamadas. SQS se utiliza como un canal de notificación rápido para nuevas conexiones, permitiendo a los workers reclamarlas sin esperar el ciclo de reconciliación. CloudWatch se emplea para métricas personalizadas que impulsan el autoescalado de la flota de workers.

El ciclo de vida de la lease incluye cuatro estados: Adquirir (mediante una escritura condicional que verifica la inexistencia o expiración de la lease), Renovar (actualizando el timestamp de expiración si el worker sigue siendo el propietario), Liberar (estableciendo lease_expires_at_ms = 0 durante el apagado elegante) y Expirar (cuando un worker falla y no renueva la lease, lo que la hace elegible para ser reclamada por otro worker a través del bucle de reconciliación).

response = self.dynamodb_client.update_item(
    TableName=self.table_name,
    Key={'Pk': {'S': connection_id}},
    UpdateExpression='SET #owner = :owner, #expires = :expires, #state = :state, #url = :url',
    ConditionExpression='attribute_not_exists(Pk) OR #expires < :now',
    ExpressionAttributeNames={
        '#owner': 'lease_owner',
        '#expires': 'lease_expires_at_ms',
        '#state': 'desired_state',
        '#url': 'ws_url'
    },
    ExpressionAttributeValues={
        ':owner': {'S': self.worker_id},
        ':expires': {'N': str(lease_expires_at_ms)},
        ':state': {'S': 'STARTED'},
        ':url': {'S': ws_url},
        ':now': {'N': str(current_time_ms)}
    }
)

response = self.dynamodb_client.update_item(
    TableName=self.table_name,
    Key={'Pk': {'S': connection_id}},
    UpdateExpression='SET #expires = :expires',
    ConditionExpression='#owner = :owner AND #expires > :now',
    ExpressionAttributeNames={
        '#owner': 'lease_owner',
        '#expires': 'lease_expires_at_ms'
    },
    ExpressionAttributeValues={
        ':owner': {'S': self.worker_id},
        ':expires': {'N': str(new_expiration_time_ms)},
        ':now': {'N': str(current_time_ms)}
    }
)

async def graceful_shutdown(self):
    print("Received SIGTERM, initiating graceful shutdown...")
    self.shutdown_event.set()
    # Wait for all connection tasks to complete their cleanup
    await asyncio.gather(*[conn.close() for conn in self.active_connections.values()])
    print("All connections closed and leases released.")

Fundamentos Teóricos

El concepto de 'lease' o 'lock' distribuido es un pilar en la computación distribuida para garantizar la exclusividad de acceso a un recurso compartido. Este patrón se alinea con los principios establecidos en trabajos fundamentales sobre sistemas distribuidos y tolerancia a fallos. La utilización de un timestamp de expiración para una lease es una técnica común para manejar fallos de procesos, permitiendo que otros procesos asuman la propiedad después de un período de gracia. Esto se relaciona con los conceptos de 'fencing tokens' y 'leases' como se discute en el contexto de sistemas de archivos distribuidos y bases de datos, donde la coordinación de acceso a recursos compartidos es crítica para la consistencia y la disponibilidad.

Aunque no se cita un paper específico, la implementación de escrituras condicionales para lograr un bloqueo distribuido sin un servicio de coordinación explícito evoca los principios de algoritmos de consenso como Paxos o Raft en su objetivo de lograr un acuerdo sobre el estado. Sin embargo, en lugar de un consenso completo, se enfoca en la atomicidad de una operación de escritura para establecer la propiedad. La dependencia de relojes sincronizados para la expiración de leases es un recordatorio de los desafíos de la sincronización de relojes en sistemas distribuidos, un tema explorado por Leslie Lamport en su trabajo sobre 'Logical Clocks' y 'Distributed Snapshots', aunque aquí se confía en la sincronización de NTP para la precisión práctica.