CQRS en AWS: Sincronizando los Servicios de Command y Query con el Estándar Transactional Outbox y la Técnica Polling Publisher

Esta es la segunda parte de la serie sobre las diferentes formas de implementar el estándar CQRS en AWS. En la primera parte, analicé qué es el CQRS y sus beneficios. Para empezar, hablé de un caso de uso en el que tenemos una edición compatible con Amazon Aurora PostgreSQL-Compatible Edition como base de datos del servicio de comandos y uno Amazon Elasticache for Redis como base de datos del servicio de consultas. Cuando cambia el estado del servicio de comandos, se coloca un mensaje en una cola de Amazon SQS, en la que se busca un componente computacional (una función de AWS Lambda, en el ejemplo presentado) que actualice Redis, con la información precalculada lista para ser recuperada.

En esta publicación, usaré un estándar llamado Transactional Outbox (en español, algo así como “bandeja de salida transaccional”). Con él, conservamos un registro que representa un evento junto con el aggregate principal, todo en la misma transacción. Para publicar eventos de forma fiable, hay dos enfoques: Polling Publisher (en español, algo así como “el publicador que consulta”) y Transaction Log Tailing (en español, algo así como “leer la parte inferior del registro de transacciones”). En este post, abordaré la primera técnica.

Introducción

La mayoría de las aplicaciones que creamos necesitan una base de datos para almacenar los datos. Almacenamos el estado de nuestros objetos de dominio y los utilizamos para diversos fines, como procesar y generar informes. En ocasiones, es posible que nuestra base de datos no sea ideal para la recuperación de datos, ya sea por su naturaleza o debido a un modelo de dominio complejo, por ejemplo. Para estos casos, podemos usar el estándar arquitectural CQRS, que sugiere que, para un bounded context (en español, algo así como “contexto delimitado”) de nuestro dominio, podemos tener dos servicios, uno para recibir comandos (es decir, operaciones que cambian de estado) y otro para consultas (que solo recuperan datos). De este modo, cada servicio puede tener la base de datos que mejor se adapte. El desafío consiste en cómo mantener sincronizadas las dos bases de datos, lo que se puede hacer con los eventos publicados desde el servicio de comandos para que los consuma el servicio de consultas.

La publicación confiable de eventos relacionados con cosas que ya han sucedido en una aplicación puede ser un desafío. Tanto si utilizamos el estándar CQRS como si no lo hacemos, si no tenemos cuidado, podemos terminar publicando información que aún no existe o no publicarla en cualquier momento, como ya comentamos en la primera parte de esta serie, y las fuentes de datos pueden perder la sincronización de las fuentes de datos. De hecho, dado que publicamos eventos del servicio de comandos para que los consuma el servicio de consultas, ya existe una consistencial eventual, pero si no tenemos cuidado, es posible que las fuentes de datos no estén sincronizadas durante más tiempo del esperado. En el contexto de una base de datos relacional, el estándar Transactional Outbox nos permite publicar eventos de forma fiable. Cuando se aplica, un evento se confirma en la bandeja de salida en la misma transacción en que se confirman los datos del aggregate principal y, en algún momento, se publica más adelante.

Figura 1. El estándar Transactional Outbox. En este ejemplo, un aggregate de pedidos se compone de la clase Pedido, que a su vez contiene una lista de ArticuloPedido. Al realizar un pedido, el aggregate se confirma junto con un registro que representa el evento de creación del pedido.

Hay dos técnicas para tratar las bandejas de salida. Una de ellas es Polling Publisher, que analizaré en esta parte de esta serie. Una vez que el evento persista en la bandeja de salida, y a intervalos regulares, un componente que pueda realizar cálculos (como una función Lambda) extraerá los eventos que aún no se hayan publicado y los publicará. Después de eso, esos mismos eventos deberán eliminarse antes de la próxima vez que se lean los datos, o será necesario controlar de alguna manera qué eventos ya se han publicado, por ejemplo, con una columna que indique si el evento ya se ha publicado o no.

Es importante tener en cuenta que trabajar con una bandeja de salida significa capturar los cambios en los datos, o CDC, que es la idea de recuperar los cambios que se produjeron en los datos de alguna manera. Hay varias maneras de hacerlo. Una de ellas es consultar la bandeja de salida de tiempos en tiempos para obtener cambios, y es en esta idea en la que se basa la técnica Polling Publisher.

Caso de uso: Amazon Aurora PostgreSQL-Compatible Edition como base de datos del servicio de comandos y Amazon Elasticache for Redis como base de datos del servicio de consultas, mediante el estándar Transactional Outbox y la técnica Polling Publisher

Al igual que en el primer caso que analicé en esta serie de publicaciones, tenemos una Aurora como base de datos del servicio de comando y una Redis como base de datos del servicio de consultas. Aurora contendrá información relacionada con los clientes, los productos y los pedidos, y Redis contendrá información precalculada relacionada con los clientes. Para probar la arquitectura, utilizaremos dos endpoints, uno para guardar los pedidos y otro para recuperar la información relacionada con el cliente.

Resumen de la Solución

En AWS, los servicios de comandos y consultas se pueden implementar con cualquier servicio que pueda realizar cálculos. En este ejemplo, utilizamos lo Amazon API Gateway para recibir solicitudes, que se reenvían a funciones de Lambda, pero que podrían ser, por ejemplo, containers que se ejecutan en Amazon Elastic Kubernetes Service o Amazon Elastic Container Service. El servicio que recibe los comandos conservará la información relacionada con los pedidos en las tablas que hacen referencia al aggregate Pedidos, pero esta vez no publicaremos un mensaje en una cola de Amazon SQS para representar el evento relacionado con la creación del pedido, sino que conservaremos la información relacionada con el pedido y el evento que acaba de ocurrir, todos en la misma transacción. Y eso es todo. En ese momento, el evento aún no se publicará.

La cantidad de información que se conserva dependerá de si el componente que gestionará el evento tiene acceso a las tablas de la base de datos del servicio de comandos o no. Si el componente tiene acceso y puede realizar consultas sobre él, es posible que el evento contenga menos datos, por ejemplo, solo identificadores de registros. Con estos identificadores, las consultas se pueden realizar más adelante y se puede recuperar más información. De lo contrario, el evento debe contener toda la información relevante para poder procesarlo. El evento también puede contener más datos si queremos reducir la base de datos con menos consultas.

En el caso de las bases de datos Aurora, el evento puede persistir como un documento JSON, que es uno de los tipos de datos compatibles. Esto puede resultar interesante, ya que podemos usar la bandeja de salida para conservar eventos de diferentes aggregates y usar metadatos en el documento que indiquen el tipo de evento, que se utilizarán para tratarlo y publicarlo correctamente. Si se usa otra base de datos que no admite este formato, la bandeja de salida puede tener columnas que tengan tipos de datos más “tradicionales”, como varchar, number, timestamp, etc., y será necesario tener n tablas, una para cada tipo de evento que pueda ocurrir en la aplicación. El siguiente documento JSON es un ejemplo de un evento que debe conservarse en una bandeja de salida que solo tiene dos metadatos y el identificador del pedido, cuyos detalles se recuperarán en el momento de la publicación:

{
    "tipoAggregate": "Pedido",
    "tipoEvento": "PedidoCreado",
    "idPedido": 1
}

Un evento como el ejemplificado anteriormente genera menos tráfico de red en el momento de la publicación, pero requiere que los detalles se recuperen para su procesamiento. El evento que se ejemplifica a continuación, por otro lado, genera más tráfico de red en la publicación porque es más grande, pero todos los detalles ya están incluidos en el evento, lo que facilita el procesamiento.

{
    "tipoAggregate": "Pedido",
    "tipoEvento": "PedidoCreado",
    "idPedido": 1,
    "timestamp": 1700836837,
    "nombre": "Bob",
    "email": "bob@anemailprovider.com",
    "productos": [{
        "nombre": "Computador",
        "precio": 1500,
        "cantidad": 1
    }, {
        "nombre": "Phone",
        "precio": 500,
        "cantidad": 3
    }]
}

Los eventos se recuperarán a intervalos predefinidos y cada evento se preparará como un documento JSON para que pueda publicarse en un agente de mensajes. En la solución propuesta en esta entrada de blog, el Amazon EventBridge Scheduler invoca la función Lambda OrderEventTablePollerLambda, lo que le permite ejecutar las tareas programadas de tiempos en tiempos. En EventBridge, también puede definir reglas programadas. En ambas opciones, las tareas se ejecutarán en momentos predeterminados, pero el planificador ofrece una escalabilidad mejorada en comparación con las reglas programadas de EventBridge, con un conjunto más amplio de operaciones de API de destino y servicios de AWS.

Tras recuperar los eventos, la función OrderEventTablePollerLambda de Lambda los publica en un tópico de Amazon SNS, que a su vez los publica en una cola de SQS, que es consultada por la función de Lambda RedisUpdaterLambda, que actualiza o inserta claves en Redis con los datos de cada cliente. Tras las inserciones o actualizaciones, esta función publica mensajes en la cola OrderEventsCleanerQueue de SQS, de modo que todos los registros que ya hayan sido procesados por la función redisUpdaterLambda se puedan eliminar de la bandeja de salida. A continuación, la función OrderEventsCleanerLambda recupera los mensajes de la cola de SQS en lotes de 10 y, dentro de una transacción, elimina todos los registros de los eventos que se publicaron (con una sentencia delete * from public.order_event where id in (<lista de ids de eventos publicados>)), para que no se vuelvan a publicar la próxima vez que se consulte la bandeja de salida.

Hay al menos tres maneras de gestionar una bandeja de salida. La primera consiste simplemente en no actualizar ni eliminar los registros de la tabla y publicar todos los eventos con cada nueva publicación, así como en controlar la idempotencia de la parte del servicio que los consume. Por un lado, tendremos menos operaciones una vez que los eventos se publiquen en el banco. Por otro lado, tendremos costes de procesamiento, almacenamiento y costes innecesarios, tanto para publicar los eventos como para controlarlos en el servicio que los consume. Por lo tanto, esta es la forma menos recomendable de gestionar una bandeja de salida.

La segunda es eliminar los registros publicados con cada publicación. Una vez que el servicio de consultas procese el evento, un componente debe eliminar los registros de la bandeja de salida. La tercera consiste en añadir una columna que indique si un evento ya se ha procesado o no. Con cada publicación, solo se recuperarán y publicarán los registros que aún no se hayan procesado y, después, será necesario actualizar el valor de la columna de control a verdadero. Esta opción puede resultar interesante si quieres volver a ejecutar los eventos para reproducir un error o para cargar un modelo de dominio, por ejemplo.

Si no va a eliminar los registros de la bandeja de salida después de las publicaciones y solo los actualiza para indicar que ya se han procesado, tendrá que pensar en alguna forma de purga de datos para que la base de datos no crezca inadvertidamente, como este enfoque que utiliza la extensión pg_partman de Postgres, junto con Amazon S3. En el caso de MySQL, puede utilizar el archivado por particiones. En el siguiente ejemplo, los registros se eliminan después de cada publicación, por lo que la base de datos siempre tendrá pocos o ningún registro en la bandeja de salida.

Figura 2. Arquitectura propuesta, con Amazon Aurora PostgreSQL-Compatible Edition como base de datos del servicio de comandos y Amazon Elasticache for Redis como base de datos del servicio de consultas.

La ventaja de esta arquitectura es que, aunque es un poco más compleja que la presentada en la primera parte de esta serie, utilizamos el estándar Transactional Outbox, que es una forma fiable de publicar eventos. Como resultado, los eventos persisten y se publican en algún momento.

Hay puntos a tener en cuenta en esta solución. La primera es que, independientemente del número de instancias con las que se ejecute nuestra aplicación o del mecanismo que se utilice para activar la acción de recuperar los registros de la bandeja de salida, lo mejor es tener en cuenta al menos una vez la semántica a la hora de entregar los eventos y, por lo tanto, tener en cuenta que puede haber una publicación duplicada de eventos y garantizar que cada evento se procese solo una vez con algún mecanismo de idempotencia. En segundo lugar, todavía tenemos la publicación de eventos separados de una transacción, que es la transacción que elimina los eventos de la bandeja de salida.

La eliminación de los eventos que ya se han publicado reduce las posibilidades de que un evento se publique por duplicado, además de mantener la tabla lo más limpia posible, lo que ofrece menos costes de almacenamiento y más rendimiento a la hora de recuperar los eventos de la bandeja de salida. Sin embargo, la publicación de los eventos sigue siendo independiente de la transacción, lo que excluye los eventos. Esto significa que los puntos de atención que se plantearon en la primera parte de esta serie aún pueden darse.

Esto supone un problema menor que si no utilizáramos el estándar Transactional Outbox, ya que podemos garantizar que los eventos se publicarán en algún momento. Independientemente del número de instancias en las que se ejecute nuestra aplicación, existe el riesgo remoto de que un evento se publique más de una vez, en caso de que la publicación se haya realizado y los registros de eventos no se eliminen o actualicen debido a un error entre la publicación y la transacción que actualiza o elimina los eventos. En este caso, pueden producirse problemas cuando el componente que recibe el evento publicado realiza operaciones que no son idempotentes, por lo que es importante controlar qué eventos ya han sido procesados por el servicio de consultas de alguna manera.

Una forma de mitigar esto es introducir un mecanismo de idempotencia para gestionar los mensajes duplicados, similar al enfoque adoptado en la primera parte de esta serie. Ahí tenemos claves en Redis que representan los mensajes que ya se han procesado, con un tiempo de caducidad igual a 5 minutos. Podríamos presentar la misma idea aquí, tomando como claves los identificadores de los registros de la bandeja de salida y como valores los identificadores de ejecución de las funciones de Lambda (context.aws_request_id). Antes de publicar un evento, comprobamos que la caché tiene una entrada con el mismo identificador de evento. De lo contrario, podemos publicarla; de lo contrario, podemos eliminar o actualizar el evento de la bandeja de salida porque es un evento que ya se ha procesado.

Al utilizar la técnica Polling Publisher, introduciremos más operaciones en nuestra base de datos consultándola a intervalos regulares. Una opción es leer las réplicas, que es una función disponible en todas las bases de datos de Amazon RDS, de forma que se puedan ver los eventos. Aun así, el componente que recupera la información de una réplica leída competiría por los recursos de la base de datos con otros componentes.

Otro punto es que, con la técnica Polling Publisher, nos limitamos a recuperar los datos que están en la tabla. Este no es el caso que analizamos en este ejemplo, pero no podríamos capturar los metadatos de la transacción en sí. Además, para capturar las exclusiones, tendríamos que mantener de alguna manera los estados entre las búsquedas y compararlos para ver qué se excluyó. Si quisiéramos reutilizar los eventos de la bandeja de salida (lo que no sería práctico), también necesitaríamos tener una columna de marca de tiempo, que se actualizaría mediante alguna lógica de dominio e indicaría cuándo se modificó por última vez un registro en particular y, según ese valor, procederíamos con el tratamiento y la publicación del evento.

Para publicar los eventos, podríamos recuperar todos los registros de la bandeja de salida (con select *) y publicarlos de una vez en un tópico de SNS en un único documento JSON. Luego, podríamos publicarlos todos juntos como un único mensaje en la cola de SQS, desde donde la Lambda que actualiza Redis podría recuperar ese mensaje, repetir los registros e introducir o actualizar una por una las claves correspondientes a cada cliente de Redis. En este enfoque, sería difícil controlar de forma individual las claves que se introdujeron o actualizaron. Además, en el momento de escribir esta entrada de blog, el tamaño máximo de payload de los mensajes en SNS es de 256 KiB, y en SQS es de 256 KiB. Entonces, tendríamos un problema si intentáramos publicar un documento JSON que represente todos los eventos que se van a publicar si el tamaño de ese documento JSON supera los 256 KB. Por lo tanto, lo mejor es recuperar cada evento de la bandeja de salida y publicarlo en un tópico de SNS o en una cola de SQS de forma individual.

Una última consideración se refiere al intervalo de búsqueda en la base de datos. Si utilizamos una base de datos Aurora, una posibilidad sería tener réplicas de lectura y crear un endpoint personalizado, en el que pudiéramos concentrar las réplicas de lectura únicamente para atender las consultas a la bandeja de salida. Si recuperamos los registros que se van a publicar, por ejemplo, cada segundo, aumentaremos el uso de memoria y CPU de nuestras réplicas de lectura, lo que implica una solución más cara. Si hacemos esto, por ejemplo, cada minuto, es posible que le hayan ocurrido muchas cosas a la base de datos del servicio de comandos y que la base de datos del servicio de consultas esté muy desactualizada. En cualquier caso, podemos correr el riesgo de buscar en la bandeja de salida y no recibir nada. En ese caso, estaríamos literalmente sin pagar nada. El rango de consultas debe definirse cuidadosamente caso por caso.

Ejecutando el Ejemplo

Para ejecutar el ejemplo, los lectores deben tener una cuenta de AWS y un usuario con permisos de administrador. A continuación, basta con ejecutar el paso a paso que se proporciona en el repositorio de código para esta serie de entradas de blog sobre CQRS, en AWS Samples, alojadas en Github. Al realizar el proceso paso a paso, los lectores dispondrán de la infraestructura que se presenta aquí en sus propias cuentas.

El ejemplo contiene dos endpoints, uno para recibir información relacionada con los pedidos (que representa nuestro servicio de comando) y el otro para recuperar información relacionada con los clientes (que representa nuestro servicio de consultas). Para comprobar que todo ha funcionado correctamente, ve a la API Gateway y, en la lista de API, introduce la API “OrdersAPI” y, a continuación, Stages. Solo habrá una stage llamada “prod”. Recupere el valor del campo Invoke URL y añada “/orders”. Este es el endpoint que recibe la información relacionada con os pedidos.

Hagamos una solicitud POST a ese endpoint. Podemos usar cualquier herramienta para realizar solicitudes, como cURL o Postman. Como este endpoint está protegido, también necesitamos añadir basic authentication. Si utilizas Postman, tendrás que recuperar el nombre de usuario y la contraseña generados al crear la infraestructura. En la puerta de enlace de API, vaya a “API Keys» y copie el valor de la columna “API Key” de “admin_key”. Este valor contiene el nombre de usuario y la contraseña separados por el carácter “:”, pero está codificado en Base64. Decodifique el valor con una herramienta en línea o con el comando “base64” de Linux. El nombre de usuario está a la izquierda del carácter “:” y la contraseña, a la derecha. Añada una “Authorization” del tipo “Basic Auth” y rellene los campos “Username” y “Password” con los valores recuperados. Añade también un header “Content-Type”, con el valor “application/json”.

El payload para realizar solicitudes a este endpoint es el siguiente:

{
    "id_client": 1,
    "products": [{
        "id_product": 1,
        "quantity": 1
    }, {
        "id_product": 2,
        "quantity": 3
    }]
}

Esto representa un pedido realizado por el cliente con el identificador 1 y que contiene productos con los identificadores 1 y 2. El total de ese pedido es de $3000. Toda esta información se almacenará en Aurora. Al realizar esta solicitud POST, si todo funcionó según lo esperado, debería ver el siguiente resultado:

{
    "statusCode": 200,
    "body": "Order created successfully!"
}

Ahora, verifiquemos que la información relacionada con el cliente se haya enviado a Redis. Al endpoint de API Gateway, que se recuperó anteriormente, añada “/clients/1”. Este es el endpoint que recupera la información relacionada con el cliente. Hagamos una solicitud GET para ese endpoint. Al igual que hicimos con el endpoint “/orders”, necesitamos añadir la autenticación básica. Siga los pasos explicados anteriormente y realice la solicitud GET. En el caso de la infraestructura que puse a disposición en Git, los eventos se recuperan y publican cada minuto, por lo que habrá que esperar al minuto siguiente para ver los cambios en Redis. Si todo ha funcionado según lo esperado, verás un resultado similar al siguiente:

{
    "name": "Bob",
    "email": "bob@anemailprovider.com",
    "total": 3000.0,
    "last_purchase": 1700836837
}

Esto significa que pudimos alimentar correctamente a Redis con información lista para ser leída, mientras la misma información está en Aurora, en otro formato.

Limpiando los Recursos

Para borrar la infraestructura creada, en una terminal, en el mismo directorio donde se creó la infraestructura, basta con ejecutar el comando “cdk destroy” y confirmar. La eliminación tarda aproximadamente 10 minutos.

Conclusión

En esta entrada de blog, presenté la técnica Polling Publisher, que es una forma de implementar el estándar Transactional Outbox. En este enfoque, cada vez que cambiamos el estado de un agregado en el servicio de comandos, guardamos en una bandeja de salida un registro que representa el evento que acaba de ocurrir, junto con el agregado en sí. A intervalos regulares, recuperamos los eventos de la bandeja de salida, los publicamos y los eliminamos o los actualizamos para indicar que ya se han procesado.

Con el estándar Transactional Outbox, en algún momento se publicarán todos los eventos, por lo que no hay riesgo de no publicarlos. Es un enfoque relativamente sencillo, pero cuando se utiliza la técnica Polling Publisher, hay que tener cuidado para garantizar que los eventos se publiquen y gestionen correctamente. Siempre hay que tener en cuenta que los eventos pueden publicarse por duplicado y, por lo tanto, utilizar algún mecanismo de idempotencia que permita procesar cada evento solo una vez.

En la próxima publicación de esta serie, exploraré otra técnica para publicar eventos mediante el estándar Transactional Outbox, que se llama Transaction Log Tailing. Como cualquier otra solución, tiene sus ventajas y desventajas, ¡que analizaremos a continuación!

Este contenido és una traduccíon del blog original en Portugués (enlace acá).

Acerca del Autor
	Roberto Perillo es un arquitecto de soluciones empresariales en AWS Brasil, especializado en sistemas serverless, que presta servicios a clientes del sector financiero y ha estado en la industria del software desde 2001. Trabajó durante casi 20 años como arquitecto de software y desarrollador Java antes de unirse a AWS en 2022. Es licenciado en Ciencia de la Computación, tiene una especialización en Ingeniería de Software y un máster también en Ciencia de la Computación. Un aprendiz eterno. En su tiempo libre, le gusta estudiar, tocar la guitarra y también ¡jugar a los bolos y al fútbol de mesa con su hijo, Lorenzo!
Acerca de los Colaboradores
	Luiz Santos trabaja actualmente como Technical Account Manager (TAM) en AWS y es un entusiasta de la tecnología, siempre busca nuevos conocimientos y tiene una mayor experiencia en el desarrollo de software, el análisis de datos, la seguridad, la tecnología serverless y DevOps. Anteriormente, tenía experiencia como arquitecto de soluciones de AWS y SDE.
	Maria Mendes es arquitecta de soluciones desde agosto de 2022. Anteriormente, tenía experiencia en el área de ingeniería de datos, trabajando con Python, SQL y arquitectura orientada a eventos. En AWS, Maria forma parte de la comunidad técnica centrada en los servicios de DevOps de AWS.
Acerca de las Traductoras
	Gabriela Guimarães es actualmente una de las 10 jóvenes que participan en el primer Tech Apprentice Program – Black Women Edition en AWS. Licenciado en Análisis y Desarrollo de Sistemas, estudia el mundo de la nube con enfoque en backend.
	Maria Lucio es Aprendiz de Tecnología, formando parte del primer Tech Apprentice Program – Black Women Edition en AWS. Licenciada en Técnico en Computación y Licenciada en Sistemas de Información, le apasiona la programación y orienta su trabajo como aprendiz hacia proyectos y estudios que le permitan evolucionar en el área del desarrollo de software.
Acerca de los Revisores
	Ricardo Tasso es arquitecto de soluciones para asociados en AWS, ayuda en el proceso de asociación con AWS y ofrece la mejor solución a los clientes. Ha trabajado con administración de sistemas y soluciones de TI por más de 15 años, con experiencia en redes, almacenamiento, sistemas operativos, colas y mensajería, con enfoque en los últimos 7 años en DevOps, serverless, contenedores, infraestructura como código, CI/CD. , IA para DevOps, así como arquitectura y modernización de aplicaciones.
	Gerson Itiro Hidaka trabaja actualmente como arquitecto de soluciones empresariales en AWS y presta servicios a clientes financieros en Brasil. Entusiasta de tecnologías como el Internet de las cosas (IoT), los drones, DevOps y especialista en tecnologías como la virtualización, la tecnología serverless, los contenedores y Kubernetes. Lleva más de 26 años trabajando con soluciones de TI y tiene experiencia en numerosos proyectos de optimización de infraestructuras, redes, migración, recuperación ante desastres y DevOps incluidos en su cartera.
	Gonzalo Vásquez es Senior Solutions Architect de AWS Chile para clientes de los segmentos Independent software vendor (ISV) y Digital Native Business (DNB) de Argentina, Paraguay y Uruguay. Antes de sumarse a AWS, se desempeñó como desarrollador de software, arquitecto de sistemas, gerente de investigación y desarrollo y CTO en compañías basadas en Chile.