Event-Driven Architecture en AWS, Parte III: lo difícil

Volvamos una vez más al patrón outbox. Resuelve un problema muy común: publicar mensajes después de que la transacción de negocio ya fue confirmada, con el riesgo de que esos eventos nunca lleguen a publicarse. Al implementarlo, te aseguras de que todos los eventos se publiquen al menos una vez. Déjame explicar esa última parte.

Figura 1: el patrón outbox

El relay de publicación toma los eventos de la base de datos operacional y, una vez publicados, los confirma, ya sea eliminando los registros o marcándolos como publicados. ¿Y qué pasa si el proceso falla justo entre la publicación y el marcado del evento como publicado? Pues que en el siguiente intento ese evento se publicará otra vez, y los suscriptores recibirán el mismo evento más de una vez. ¿Es un problema grave? Para nada.

Si usas AWS SQS para procesar los eventos a los que está suscrito un servicio (y deberías), ten presente que el servicio ya garantiza una "entrega at-least-once":

"Amazon SQS almacena copias de tus mensajes en varios servidores para lograr redundancia y alta disponibilidad. En raras ocasiones, uno de los servidores que guarda una copia de un mensaje podría no estar disponible cuando recibes o eliminas un mensaje... Si eso ocurre, podrías volver a recibir esa copia del mensaje al consultar la cola."

Y esto vale no solo para AWS SQS, sino para cualquier bus de mensajes distribuido (que sea honesto). En este punto te puede surgir la duda: ¿y qué pasa con las SQS FIFO Queues? Según la documentación, las SQS FIFO queues están pensadas justamente para resolver este problema:

"A diferencia de las colas estándar, las colas FIFO no introducen mensajes duplicados. Las colas FIFO te ayudan a evitar el envío de duplicados a una cola. Si reintentas la acción SendMessage dentro del intervalo de deduplicación de 5 minutos, Amazon SQS no introduce duplicados en la cola."

Pero esto resuelve el problema solo en parte. Es verdad que, si el outbox publica el mismo mensaje más de una vez (dentro de esos 5 minutos), una cola FIFO detectará el duplicado y lo descartará. Pero pongamos el foco en el lado del consumidor. Procesar mensajes de una cola distribuida implica estos tres pasos:

1. Obtener el siguiente mensaje disponible.
2. Procesar el mensaje.
3. Confirmar el mensaje marcándolo como procesado o, en el caso de SQS, eliminándolo de la cola.

Ahora ponte el sombrero de pesimista extremo y plantéate las tres preguntas que ya hice en el post anterior:

1. ¿Qué puede salir mal aquí?
2. ¿Qué implicaciones tendría?
3. ¿Dirías que este flujo de procesamiento de mensajes es confiable?

Haz una pausa y piénsalo antes de seguir leyendo.

Bien, comparemos respuestas.

Si algo falla entre el procesamiento de un mensaje y su confirmación, ese mensaje se volverá a obtener en la siguiente ejecución y se procesará más de una vez. Aunque las colas FIFO prometen "procesamiento exactly-once", esa garantía aplica al procesamiento del lado de SQS, no al de tu aplicación. ¿Cuál de los dos te importa más? Obviamente, el segundo. Por eso, para diseñar un sistema distribuido confiable hay que asumir que cualquier mensaje puede entregarse a los suscriptores más de una vez.

Si volvemos a la página de documentación de SQS que reconoce la posibilidad de entregas duplicadas, también encontramos cómo abordar el problema:

Diseña tus aplicaciones para que sean idempotentes (no deberían verse afectadas negativamente al procesar el mismo mensaje más de una vez).

Excelente consejo. Si pasar el mismo evento o comando a la lógica de tu aplicación más de una vez produce el mismo resultado, no tienes que preocuparte por los casos límite que generan mensajes duplicados.

¿Es fácil implementar una lógica de procesamiento idempotente, confiable y a prueba de balas? Para nada. Pero antes de contarte cómo se hace, quiero decir un par de cosas sobre cómo no hay que hacerlo.

Muchos buscan lograr procesamiento idempotente implementando el patrón idempotent consumer. La idea detrás del patrón es buena, pero la gran mayoría de las implementaciones se pierden la razón exacta por la que el patrón existe. Así suele implementarse:

Primero, a cada comando/evento que procesa tu aplicación se le asigna un identificador único. Segundo, hace falta un "idempotency store": una base de datos clave/valor que mapea los IDs de las solicitudes a sus resultados (por ejemplo, Redis, DynamoDB, mapas en memoria, etc.).

Una vez cubiertos esos dos requisitos, el procesamiento de una solicitud entrante sigue esta lógica simple:

Cuando llega una solicitud, primero se verifica si su ID ya aparece en el idempotency store. Si está, se devuelve el resultado guardado en la tabla y no hace falta procesar nada. Si el ID no existe en el idempotency store:

1. Se ejecuta la solicitud.
2. Se persisten el ID y el resultado en el idempotency store.

Vuelve a ponerte el sombrero de pesimista y dime qué puede salir mal aquí.

Por supuesto, si por cualquier motivo el proceso falla justo después de completar el procesamiento pero antes de guardar el resultado en el idempotency store, la operación se ejecutará más de una vez.

Quizás dudes que una solución tan ingenua se use en la práctica. Pues échale un vistazo a este blog sobre AWS Lambda Powertools.

Existe una versión un poco más avanzada de este algoritmo, que primero marca la solicitud como "en proceso" y le aplica un timeout pasado cierto intervalo, pero adolece del mismo defecto: si el proceso falla después de completar el procesamiento pero antes de actualizar el idempotency store, la solicitud se procesará otra vez.

En general, mientras la operación use una base de datos que no participe en la misma transacción que el idempotency store, existe la posibilidad de que una solicitud se procese más de una vez.

Llegado este punto puede que te preguntes: bueno, el caso límite del que hablas suena tan raro, ¿por qué tendría que preocuparme siquiera? Lo mismo se podría decir del problema original que estamos tratando de resolver. Puede que para tu sistema concreto sea aceptable. Si es así, ¿para qué sumarle más piezas móviles? En cualquier caso, decidir si el procesamiento duplicado es aceptable o no es una decisión de diseño que hay que tomar de forma consciente.

Entonces, ¿cómo se implementa una lógica de procesamiento idempotente, confiable y a prueba de balas?

Igual que con el patrón outbox, una implementación confiable del patrón idempotent consumer requiere que una transacción atómica abarque tanto la lógica de negocio como el seguimiento de los eventos ya procesados. Y, también como ocurre con el outbox, la implementación del patrón depende del stack tecnológico.

Si tu base de datos operacional soporta transacciones multi-registro, crea una tabla para las idempotency keys, los IDs de las solicitudes entrantes (a esta tabla yo prefiero llamarla "inbox"). Una vez ejecutada la solicitud, actualiza los datos operacionales e inserta la idempotency key en una sola transacción atómica. Eso sí, ten en cuenta que la inserción de la clave debe hacerse con una condición: solo debe tener éxito si ese valor todavía no existe en la tabla. Si ya existe, la operación debe fallar; cuando eso ocurre, sabes que la solicitud entrante ya fue procesada.

...

response = dynamodb.transact_write_items(
    TransactItems=[\
        {\
            'Put': {\
                'TableName': 'operational_data',\
                'Item': {\
                    ...\
                }\
            }\
        },\
        {\
            'Put': {\
                'TableName': 'inbox',\
                'Item': {\
                    'request_id': {'S': event_id},\
                },\
                'ConditionExpression': 'attribute_not_exists(request_id)'\
            }\
        }\
    ]
)

...

Como optimización adicional, puedes verificar la existencia de la idempotency key antes de ejecutar la lógica de procesamiento.

Si la base de datos con la que trabajas no soporta transacciones multi-registro, igual puedes implementar un procesamiento idempotente confiable apoyándote en optimistic concurrency control:

1. Cada operación implica modificar un único registro de la tabla (por ejemplo, un documento JSON).
2. Cada registro tiene un campo de versión para gestionar las excepciones de concurrencia.
3. Cada actualización del registro incrementa el valor de su versión.
4. Al actualizar, la base de datos debe asegurarse de que la versión del registro sobrescrito coincida con la que se leyó inicialmente.

Una vez en marcha el optimistic concurrency control, el registro gestionado se puede ampliar con un campo adicional: processed_events (array). Antes de ejecutar una solicitud, comprueba si el ID de la solicitud existe en el array processed_events. Si no existe, procede con el manejo del evento y agrega su ID al array.

1. Event-Driven Architecture on AWS, Part I: The Basics
2. Event-Driven Architecture on AWS, Part II: The Advanced Basics
3. Event-Driven Architecture on AWS, Part III: The Hard Basics (post actual)

Publicado originalmente en https://vladikk.com .