Cosmos.EventDriven.CritterStack.AzureServiceBus 3.4.0

Cosmos.EventDriven.CritterStack.AzureServiceBus

Integración de Azure Service Bus con Wolverine para implementar Event Driven Architecture (EDA) en .NET 10.

Descripción

Este paquete provee extensiones sobre WolverineOptions para configurar Azure Service Bus como broker de mensajería en aplicaciones Cosmos EDA, con soporte especial para escenarios serverless mediante publicación inline de IPrivateEvent.

Características

• HabilitarAzureServiceBusParaServerLess: Configura Azure Service Bus para proyectos serverless sin colas de sistema
• AgregarAzureServiceBusNombrado / AgregarAzureServiceBusNombradoServerless: Configura un broker nombrado (para apps que usan más de una conexión)
• PublicarEventoServerless: Publica un IEvent a un topic de Azure Service Bus de forma inline (con sobrecarga para broker nombrado)
• PublicarEventosServerless: Publica múltiples IEvent de un assembly a un topic de Azure Service Bus
• SuscribirseAEvento: Escucha una subscription de un topic con inbox duradero (con sobrecarga para broker nombrado)
• HabilitarClaimCheck: Desvía a Azure Blob Storage el cuerpo de los mensajes de más de 192 KiB, de todos los mensajes o solo de los tipos de evento indicados (opcional; ver Claim check)

Instalación

dotnet add package Cosmos.EventDriven.CritterStack.AzureServiceBus

Uso

Configurar Azure Service Bus en Wolverine (Serverless)

builder.Host.UseWolverine(opts =>
{
    opts.HabilitarAzureServiceBusParaServerLess(
        builder.Configuration.GetConnectionString("AzureServiceBus")!);

    // Publicar un evento específico a un topic
    opts.PublicarEventoServerless<PedidoCreado>("pedidos");

    // Publicar todos los IEvent del assembly a un topic
    opts.PublicarEventosServerless("mi-servicio", typeof(Program).Assembly);
});

Suscribirse a eventos de un servicio productor

builder.Host.UseWolverine(opts =>
{
    opts.HabilitarAzureServiceBusParaServerLess(
        builder.Configuration.GetConnectionString("AzureServiceBus")!);

    // Escuchar una subscription de un topic con inbox duradero
    opts.SuscribirseAEvento("mi-subscription", "pedidos");

    // Sobre un broker nombrado
    opts.AgregarAzureServiceBusNombradoServerless("ventas",
        builder.Configuration.GetConnectionString("VentasServiceBus")!);
    opts.SuscribirseAEvento("ventas", "mi-subscription", "pedidos");
});

Claim check para mensajes grandes

Azure Service Bus Standard acepta mensajes de hasta 256 KB. Sin claim check, un mensaje mayor nunca se envía: en la ruta durable queda en el outbox fallando con MessageTooLargeException en bucle y bloquea (latch) el agente de envío del topic. HabilitarClaimCheck desvía a Blob Storage el cuerpo de los mensajes de más de 192 KiB (256 KB menos los 64 KB máximos de headers; umbral fijo, UmbralClaimCheckEnBytes). Por Service Bus viaja el cuerpo vacío y el header claim-check.$body = <blobId>|<content-type>|<bytes>. Los mensajes por debajo del umbral viajan igual que antes.

builder.Host.UseWolverine(opts =>
{
    opts.HabilitarAzureServiceBus(builder.Configuration.GetConnectionString("AzureServiceBus")!);
    opts.PublicarEvento<PedidoCreado>("pedidos");

    // Emisor y receptores: misma cuenta de Storage y mismo contenedor
    opts.HabilitarClaimCheck(
        builder.Configuration.GetConnectionString("ClaimCheckStorage")!,
        "claim-checks");
});

Reglas:

• Llamarlo después de configurar el serializer (UseSystemTextJsonForSerialization, UseNewtonsoftForSerialization o DefaultSerializer): una configuración de serializer posterior reemplaza el serializer global y desactiva el claim check sin avisar.
• Aplica a todos los transportes de la aplicación, porque envuelve el serializer global de Wolverine, no solo a Service Bus.
• Una sola llamada por servicio: una segunda llamada a cualquier sobrecarga de HabilitarClaimCheck lanza InvalidOperationException al arrancar (Wolverine reemplaza la configuración del claim check en cada llamada, así que la segunda borraría la primera en silencio).
• El contenedor debe existir; se declara con Terraform (el paquete no lo crea).
• Permisos: HabilitarClaimCheck recibe una cadena de conexión, y es ella la que otorga el acceso. Con la account key, emisor y receptor tienen acceso total a la cuenta (no hay separación por rol). Para mínimo privilegio, usa una SAS limitada al contenedor: emisor con crear/escribir (más borrar), receptor con leer. Receta de la SAS sin verificar (no se probó en el spike). Como la API recibe una cadena de conexión, los roles RBAC de Entra ID (p. ej. Storage Blob Data Reader/Contributor) no aplican a esta configuración: el acceso lo gobierna la account key o la SAS de la cadena de conexión.
• Borrado: Wolverine 6.54.0 borra el blob, en la medida de lo posible (best-effort), solo cuando el envío que lo subió falla (doc XML de Wolverine 6.54.0, ClaimCheckMessageSerializer). Nunca lo borra tras consumirlo con éxito; para eso está la regla de ciclo de vida (ver más abajo).
• El paquete depende de WolverineFx.AzureBlobStorage (trae Azure.Storage.Blobs) aunque no habilites el claim check.

Claim check para eventos específicos

Si solo algunos eventos pueden pasar de 192 KiB, pasa sus tipos en la misma llamada: solo el cuerpo de esos eventos se desvía a Blob Storage (cuando supera el umbral); el resto de los mensajes viaja siempre completo, aunque sea grande.

// Un tipo
opts.HabilitarClaimCheck(
    builder.Configuration.GetConnectionString("ClaimCheckStorage")!,
    "claim-checks",
    [typeof(InformeGenerado)]);

// Varios tipos: todos en la MISMA llamada
opts.HabilitarClaimCheck(
    builder.Configuration.GetConnectionString("ClaimCheckStorage")!,
    "claim-checks",
    [typeof(InformeGenerado), typeof(CatalogoPublicado)]);

Reglas adicionales:

• Todos los tipos en una sola llamada. No hay sobrecarga genérica por tipo a propósito: cada llamada reemplaza la configuración anterior, así que llamadas separadas descartarían los tipos previos. Una segunda llamada falla al arrancar.
• El receptor usa la sobrecarga por tipo, en una sola llamada, con AL MENOS los tipos que desvía cada emisor que consume (la unión de sus listas; una lista superset lee bien, comprobado en ClaimCheckTests.HabilitarClaimCheckConLista_ReceptorConSupersetDeTipos_RecuperaElEvento). El emisor por tipo estampa en el mensaje la clave del store por tipo (header claim-check.$store = msg:<NombreCompletoDelTipo>) y el receptor busca el store por esa clave. Un receptor con el claim check global no la conoce y no puede leer mensajes por tipo: falla con InvalidOperationException (comprobado en ClaimCheckTests.HabilitarClaimCheck_ReceptorConClaimCheckGlobal_NoLeeEventoDesviadoPorTipo, con Azurite). Al revés sí funciona: un emisor global solo estampa claim-check.$body, y un receptor por tipo lo lee aunque su lista no incluya ese tipo, porque la sobrecarga por tipo también deja el mismo Blob como store por defecto (ClaimCheckTests.HabilitarClaimCheckConLista_ReceptorPorTipo_LeeEventoDeEmisorGlobal). La clave depende del tipo, no del orden de la lista.
• Tipos concretos exactos; no se aplica a subtipos: cada tipo se compara por igualdad exacta con el del mensaje, así que se rechazan interfaces, tipos abstractos y genéricos abiertos (p. ej. typeof(IPublicEvent) no desviaría nada). Los tipos deben implementar IEvent y no ser genéricos abiertos (typeof(G<>)); la lista no puede estar vacía ni tener nulos (ArgumentException).

¿Global o por tipo? Usa el global si cualquier mensaje del servicio puede crecer o no quieres mantener la lista; es lo más simple y el receptor no necesita conocer los tipos. Usa el de tipos cuando sabes exactamente qué eventos son grandes y quieres que el resto nunca dependa de Blob Storage. Un receptor por tipo lee tanto a emisores por tipo como globales; un receptor global solo lee a emisores globales. Como solo se permite una llamada por servicio, todos los productores que un receptor consume con claim check deben compartir la misma cuenta de Storage y el mismo contenedor. Los tipos se comparan por su nombre completo CLR (normalmente sin problema: los receptores referencian el ensamblado de contratos del productor).

Si configuras un receptor: pregunta al emisor

Antes de suscribirte a un topic, pregunta al equipo del emisor si tiene activo el claim check. Si lo tiene, el receptor debe llamar también a HabilitarClaimCheck con la misma cuenta de Storage y el mismo contenedor: con la sobrecarga global si el emisor es global, o con la de tipos (al menos los tipos que desvía cada emisor que consume, en una sola llamada) si es por tipo. Si el emisor es global, también sirve la de tipos. Se recomienda que el emisor lo deje dicho en el README o contrato de sus eventos.

Por eso, al activarlo en un emisor existente, despliega primero los receptores con el claim check habilitado y después el emisor.

Ejemplo de un receptor (solo suscripción):

builder.Host.UseWolverine(opts =>
{
    opts.HabilitarAzureServiceBus(builder.Configuration.GetConnectionString("AzureServiceBus")!);
    opts.SuscribirseAEvento("mi-subscription", "pedidos");

    // Si configuras un serializer propio, hazlo antes de esta línea; misma cuenta y contenedor que el emisor
    opts.HabilitarClaimCheck(
        builder.Configuration.GetConnectionString("ClaimCheckStorage")!,
        "claim-checks");
});

Qué pasa si el receptor no lo habilita (comprobado en un spike con Wolverine 6.54.0, emulador de Service Bus y Azurite, con UseDurableInbox(), que es lo que usa SuscribirseAEvento):

• Los mensajes grandes fallan al deserializar con System.Text.Json.JsonException ("The input does not contain any JSON tokens"); el handler no se ejecuta y no hay reintentos.
• El mensaje va al dead-letter de Wolverine (tabla wolverine_dead_letters de la base del inbox durable, replayable = false), no al DLQ de Service Bus.
• Los mensajes por debajo del umbral se procesan normal, así que el fallo solo aparece con los grandes.
• Es recuperable: habilita el claim check en el receptor, despliega y marca las filas como replayable = true; Wolverine las re-ejecuta y descarga el cuerpo del blob, siempre que el blob siga existiendo.

Ciclo de vida de los blobs (obligatorio)

Wolverine no borra los blobs tras consumirlos (en el spike seguían ahí después de ser consumidos). Configura una lifecycle policy en el contenedor (Terraform) con un TTL mayor que el peor escenario de reintento o dead-letter en el que quieras poder re-ejecutar mensajes; un blob expirado hace irrecuperable el mensaje.

Durabilidad y fallos

• Alcance de la verificación: el spike solo verificó la ruta durable (PublicarEvento con outbox durable + SuscribirseAEvento con inbox durable). La ruta serverless inline (PublicarEventoServerless, SendInline) está sin verificar. Además, la ruta de escritura síncrona de Wolverine sube el blob bloqueando al llamador (doc XML de ClaimCheckMessageSerializer).
• El outbox durable guarda el mensaje ya desviado (solo el token, unos cientos de bytes): el blob se sube al publicar, aunque Service Bus esté caído, y al recuperar el outbox se envía el mensaje pequeño.
• Si Blob Storage no responde, PublishAsync lanza WolverineSerializationException tras agotar reintentos (medido ~103 s con la conexión a Blob rechazada; otros fallos, como throttling o errores 5xx, pueden variar; bloquea al llamador) y el mensaje no queda en el outbox: falla ruidosamente, sin pérdida silenciosa. Sin verificar: el comportamiento dentro de un handler con outbox transaccional de Marten.

Definir un Evento Privado

using Cosmos.EventDriven.Abstractions;

public record PedidoCreado(
    Guid PedidoId,
    string ClienteId,
    DateTime CreadoEn
) : IPrivateEvent;

Requisitos

• .NET 10.0 o superior
• Azure Service Bus accesible desde la aplicación
• Cadena de conexión configurada en appsettings.json

Licencia

Copyright © Cosmos. Todos los derechos reservados.