IronAlpine.Framework.EventBus 3.1.1

IronAlpine.Framework.EventBus

Kafka event bus with a transactional outbox, an idempotent inbox, retries with backoff, dead letter tables, replay, retention cleanup and readiness health - all stored in the service's own EF Core database.

• Package id: IronAlpine.Framework.EventBus (3.1.1, net10.0)
• Depends on: Kernel, Mediator, Data, Confluent.Kafka 2.12

How it works

publisher service                                   consumer service
-----------------                                   ----------------
handler / domain event handler                      KafkaInboxWriterHostedService
  IOutboxStore.AddAsync(event)  --> OutboxMessages     consumes subscribed topics
                                      |                writes InboxMessages (idempotency key = event id)
KafkaOutboxPublisherHostedService     |                commits the Kafka offset
  claims pending rows (lease) --------+--> Kafka --->
  produces (acks=all, idempotent)                   InboxProcessorHostedService
  marks processed / retry / dead letter               claims pending rows (lease)
                                                      deserializes to the registered CLR type
                                                      IMediator.Publish(event) -> INotificationHandler<TEvent>
                                                      marks processed / retry / dead letter

Delivery is at least once end to end. Consumers must be idempotent.

Installation

<PackageReference Include="IronAlpine.Framework.EventBus" Version="3.1.1" />

Registration

using IronAlpine.EventBus.Extensions;

builder.Services.AddIronAlpineData<OrdersDbContext>(builder.Configuration, cfg => /* ... */);

builder.Services.AddIronAlpineEventBusKafka(
    builder.Configuration,
    options => options.ConsumerGroupId = "orders-service",            // optional code overrides
    topology =>
    {
        topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle");
        topology.Publish<OrderCancelledIntegrationEvent>("orders.lifecycle");
        topology.Consume<PaymentCapturedIntegrationEvent>("payments.lifecycle");
    });

Overloads: AddIronAlpineEventBusKafka(configuration, configure, configureTopology, registerStorage = true, enableCleanupHostedService = true) and AddIronAlpineEventBusKafka(configuration, configureTopology, ...). (AddIronAlpineEventbusKafka with a lower-case b is obsolete.)

It registers:

The stores use the scoped DbContext alias registered by AddIronAlpineData. Add a migration after registering the event bus; design-time factories must call AddIronAlpineEventBusKafka too, or the tables are missing.

Start-up validation (when Enabled): consume routes require Writer.Enabled and Processor.Enabled; publish routes require Outbox.Enabled. The legacy section IronAlpine:EventBus:Kafka:Kafka throws.

Configuration

{
  "IronAlpine": {
    "EventBus": {
      "Kafka": {
        "Enabled": true,
        "BootstrapServers": "kafka:9092",
        "ConsumerGroupId": "orders-service",
        "ClientId": "orders-service",
        "SecurityProtocol": "SaslSsl",
        "SaslMechanism": "ScramSha512",
        "SaslUsername": "orders",
        "SaslPassword": "<secret>",
        "SslCaLocation": "/certs/ca.crt",
        "SslCertLocation": null,
        "SslKeyLocation": null,
        "SslKeyPassword": null,
        "SslEndpointIdentificationAlgorithm": "Https",
        "EnableAutoCommit": false,
        "EnableAutoOffsetStore": false,
        "AutoOffsetReset": "Earliest",
        "Provisioning": { "Enabled": true, "DefaultPartitions": 3, "DefaultReplicationFactor": 1, "MetadataWaitTimeoutSeconds": 20, "MetadataPollIntervalMilliseconds": 500 },
        "Writer": { "Enabled": true, "StartupDelaySeconds": 5, "PollTimeoutMilliseconds": 1000, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2 },
        "Processor": { "Enabled": true, "BatchSize": 64, "LockSeconds": 30, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2, "MaxRetryCount": 100, "FastRetryCount": 3, "FastRetryDelaySeconds": 5, "BackoffBaseDelaySeconds": 15, "MaxRetryDelaySeconds": 500 },
        "Outbox": { "Enabled": true, "BatchSize": 64, "LockSeconds": 30, "IdleDelayMilliseconds": 100, "WorkerErrorDelaySeconds": 2, "MaxRetryCount": 50, "FastRetryCount": 3, "FastRetryDelaySeconds": 5, "BackoffBaseDelaySeconds": 15, "MaxRetryDelaySeconds": 500, "MaxMessageBytes": 950000 },
        "Cleanup": { "Enabled": true, "OutboxProcessedRetentionDays": 0, "InboxProcessedRetentionDays": 14, "DeadLetterRetentionDays": 60, "ReplayAuditRetentionDays": 180, "IntervalHours": 24, "BatchSize": 1000 },
        "Schema": {
          "InboxTableName": "InboxMessages",
          "OutboxTableName": "OutboxMessages",
          "InboxDeadLetterTableName": "InboxDeadLetters",
          "OutboxDeadLetterTableName": "OutboxDeadLetters",
          "ReplayOperationTableName": "EventReplayOperations",
          "RequireIdempotencyUniqueIndex": true,
          "RequireConcurrencyToken": true
        }
      }
    }
  }
}

Values above are the defaults except for the connection/security examples. Notes:

• ConsumerGroupId defaults to ia-eventbus-inbox (EventBusKafkaOptions.DefaultConsumerGroupId). Set a unique group per service: services sharing a group split each topic's messages between them. A service with consume routes logs a start-up warning while it uses the default (the default itself is unchanged for compatibility).
• SecurityProtocol / SaslMechanism / SslEndpointIdentificationAlgorithm are Confluent enum names (Plaintext, Ssl, SaslPlaintext, SaslSsl; Plain, ScramSha256, ScramSha512, ...; None, Https). SASL protocols require username and password; SSL protocols require SslCaLocation.
• Schema.*MaxLength options set column lengths (MessageTypeMaxLength 512, TopicMaxLength 256, KeyMaxLength 256, IdempotencyKeyMaxLength 256, LastErrorMaxLength 4000, LastErrorDetailMaxLength 8000, CorrelationIdMaxLength 128, CausationIdMaxLength 128, ...).
• The producer uses EnableIdempotence = true, Acks = All, MessageTimeoutMs = 30000. The consumer commits offsets manually after the inbox row is stored.

Integration events

using IronAlpine.EventBus.Contracts;
using IronAlpine.Kernel.Primitives;

public sealed record OrderPlacedIntegrationEvent(Guid OrderId, Guid CustomerId, decimal Total)
    : IntegrationEventBase, INotification;

IntegrationEventBase sets EventId, OccurredOnUtc, EventType (CLR full name) and EventVersion = 1, and exposes setters: SetEventId, SetOccurredOnUtc, SetEventType, SetCorrelationId, SetCausationId, SetPartitionKey, SetEventVersion, SetTargetConsumer.

Event time = save instant

OccurredOnUtc defaults to SaveInstantScope.Current ?? DateTime.UtcNow (since 3.0.6). While SaveChangesAsync dispatches domain events, SaveInstantScope.Current is the save instant of that save - the exact value the Data package writes to CreatedOnUtc/ModifiedOnUtc (see the Data README, "Save instant"). So an event created in a domain event handler carries the same DateTime as the aggregate's audit stamp, and so does its outbox row.

This lets a consumer check whether a projection is in sync with the source: the source aggregate's ModifiedOnUtc ?? CreatedOnUtc equals the OccurredOnUtc of its last full-state event, which the projection stores (IOccurredOnUtc). Rule: create full-state events in a domain event handler (inside SaveChanges). An event created outside a save takes the current time; SetOccurredOnUtc(...) always wins over both.

Targeted delivery (TargetConsumer)

var repair = new OrderSnapshotIntegrationEvent(order.Id, order.Status, order.Total);
repair.SetTargetConsumer("billing-service");   // the target service's Kafka ConsumerGroupId
await outbox.AddAsync(repair, ct);

• The value travels as the target-consumer header (EventEnvelopeFactory.TargetConsumerHeader) and in the payload.
• The inbox writer of every other consumer group commits the offset and does not write the message to its inbox (Debug log). The comparison with ConsumerGroupId is ordinal (Kafka group ids are case-sensitive).
• Without the header (the default) every subscriber processes the event as before.
• The handler sees TargetConsumer on the deserialized event, so it can tell a targeted (for example repair) event from a regular one.
• IIntegrationEvent.TargetConsumer has a default implementation returning null, so custom implementations of the interface keep compiling.

Deserialization on the consumer

The base properties EventId, OccurredOnUtc, CorrelationId, CausationId, PartitionKey, EventVersion and TargetConsumer are marked [JsonInclude], so the event a handler receives carries the publisher's values. Up to 3.0.5 their private setters were ignored by System.Text.Json: a consumer saw a new EventId, its own receive time as OccurredOnUtc and empty correlation fields. EventType keeps the consumer's CLR name.

Implement INotification on every event a service consumes: the inbox processor dispatches with IMediator.Publish, which rejects non-notifications (the message is retried and finally dead-lettered).

Wire format:

• Kafka key: PartitionKey, else the event id (N format). Set a partition key (for example the aggregate id) to keep the order of events of one entity.
• value: the event serialized with System.Text.Json web defaults; eventType, correlationId, causationId, partitionKey are rewritten in the payload;
• headers: correlation-id, causation-id, traceparent, tracestate, baggage, partition-key, event-id, event-type, event-version, and target-consumer when set. Correlation id falls back to the current Activity trace id, then the event id.

Topology

topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle");
topology.Publish<OrderPlacedIntegrationEvent>("orders.lifecycle", new EventTopicSpecification(Partitions: 6, ReplicationFactor: 3));
topology.Consume<PaymentCapturedIntegrationEvent>("payments.lifecycle", eventTypeName: "Payments.PaymentCaptured");
topology.Consume(typeof(PaymentRefundedIntegrationEvent), "payments.lifecycle");   // EventTopologyBuilderTypeExtensions

• The event type name defaults to the CLR full name (assembly-qualified suffixes are stripped). Publisher and consumer must agree on it: share the contract assembly, or pass the same eventTypeName on both sides.
• One CLR type maps to exactly one topic and one type name; one type name maps to one CLR type; topic specifications must be identical wherever a topic is declared. Violations throw when the registry is built.
• Many event types may share a topic; consumers route by (topic, event type).
• With Provisioning.Enabled the writer and publisher create missing topics and wait for metadata. An existing topic whose partition count or replication factor differs from the declared/default values is used as is and a warning is logged (up to 3.0.4 this threw out of the hosted service and stopped the host). Align the declared values with the real topic to silence the warning.

Publishing

public sealed class OrderPlacedDomainEventHandler(IOutboxStore outbox) : INotificationHandler<OrderPlacedDomainEvent>
{
    public async Task Handle(OrderPlacedDomainEvent notification, CancellationToken cancellationToken)
    {
        var integrationEvent = new OrderPlacedIntegrationEvent(notification.OrderId, notification.CustomerId, notification.Total);
        integrationEvent.SetPartitionKey(notification.OrderId.ToString());
        await outbox.AddAsync(integrationEvent, cancellationToken);   // no SaveChanges here
    }
}

• IOutboxStore.AddAsync(IIntegrationEvent) builds the envelope (the type must be registered with Publish) and adds an OutboxMessageEntity (OccurredOnUtc, Key, CorrelationId, CausationId and the full HeadersJson).
• Inside domain event dispatch (DomainEventDispatchScope.IsActive) it does not save: the row is written by the SaveChangesAsync that dispatched the domain event, atomically with the aggregate change.
• Outside that scope it calls SaveChangesAsync immediately, which also saves every other pending change of the DbContext. Call it after the business changes are complete, or publish from a domain event handler.
• IEventBus.PublishAsync(event) does the same through IOutboxStore.

Consuming

public sealed class PaymentCapturedHandler(IRepository<Order> orders, IUnitOfWork unitOfWork)
    : INotificationHandler<PaymentCapturedIntegrationEvent>
{
    public async Task Handle(PaymentCapturedIntegrationEvent message, CancellationToken cancellationToken)
    {
        var order = await orders.GetByIdAsync(message.OrderId, cancellationToken);
        if (order is null || order.IsPaid)
        {
            return;                       // idempotent: duplicates and replays are no-ops
        }

        order.MarkPaid(message.OccurredOnUtc);
        await unitOfWork.SaveChangesResilientAsync(cancellationToken);
    }
}

• Each message is published in a new DI scope; all handlers of the event run sequentially in that scope.
• Mediator pipeline behaviours (logging, validation, transaction) do not run for these handlers; each SaveChanges is its own transaction.
• A handler exception schedules a retry: attempts 1..FastRetryCount wait FastRetryDelaySeconds, later attempts wait BackoffBaseDelaySeconds * 2^n capped at MaxRetryDelaySeconds. After MaxRetryCount the message is dead-lettered with RETRY_EXHAUSTED. The exception type does not matter (any exception is retried).
• Unregistered event types are skipped (since 3.0.6): a message on a subscribed topic whose type has no Consume route for that topic is marked processed without calling a handler and without a dead letter (Warning log with EventType, Topic, MessageId). Many event types may share a topic (for example one topic per aggregate) and each consumer registers only the types it needs. Up to 3.0.5 such messages were dead-lettered as TYPE_NOT_REGISTERED; that reason code is removed. While the topology is not yet ready the message is retried as before. Only messages whose type name is known are skipped: a message without any event type name in its headers or payload is malformed and is dead-lettered as TYPE_UNRESOLVED.
• Other dead letter reasons (ReasonCodes): TYPE_UNRESOLVED (no event type name in headers or payload), TYPE_MISMATCH (header and payload types differ), DESERIALIZATION_ERROR, TRANSPORT_NON_RETRIABLE (outbox: message too large or non-retriable produce error).
• If several handlers exist and one fails, the others may already have saved; on retry all run again.
• Retries and parallel workers do not preserve ordering. Use OccurredOnUtc / versions in handlers to ignore stale events.

Dead letters and replay

var page = await operationReader.GetDeadLettersAsync(new DeadLetterQuery
{
    Direction = DeadLetterDirection.Inbox,
    EventType = typeof(PaymentCapturedIntegrationEvent).FullName,
    PageNumber = 1,
    PageSize = 50                                   // max 200
}, ct);

var result = await replayService.ReplayInboxDeadLettersAsync(new ReplayRequest
{
    EventType = typeof(PaymentCapturedIntegrationEvent).FullName,
    OccurredFromUtc = DateTime.UtcNow.AddDays(-1),  // filters on FailedOnUtc
    MaxItems = 100,                                 // default 1000, max 10000
    DryRun = false,
    RequestedBy = userName,
    Reason = "handler fixed",
    Source = "admin-api"
}, ct);

Replay re-queues matching dead letters and records an EventReplayOperations audit row (also for dry runs). The dead-lettered inbox/outbox row is kept until its table's processed retention (InboxProcessedRetentionDays / OutboxProcessedRetentionDays) and still owns the idempotency key, so replay re-arms that row (pending again, retry count and errors cleared) instead of inserting a duplicate key; only when the row no longer exists is a new pending row inserted (with Key, CorrelationId, CausationId taken from the dead letter). Several dead letters of the same event re-queue it once. The dead letter rows are kept. IMessageOperationReader also lists replay operations (GetReplayOperationsAsync(EventReplayOperationQuery)). The service has its own PagedResult<T> type (IronAlpine.EventBus.Contracts), distinct from the Kernel one.

Read APIs for admin screens

IMessageOperationReader (scoped, read-only, AsNoTracking) also serves operations screens (since 3.0.7). The host application exposes it through its own authorized endpoints; the package adds no endpoints.

Message search (event journey)

var page = await operationReader.SearchMessagesAsync(new MessageSearchQuery
{
    Key = orderId.ToString(),        // Kafka key = partition key (usually the aggregate id)
    // CorrelationId = "...",        // or the correlation id
    // EventId = "...",              // or the integration event id ("D" or "N" GUID format)
    MessageType = typeof(OrderPlacedIntegrationEvent).FullName,   // optional
    FromUtc = DateTime.UtcNow.AddDays(-7),                        // optional, on TimestampUtc
    OldestFirst = true,              // default false (newest first)
    PageNumber = 1,
    PageSize = 50                    // max 200
}, ct);

• Outbox, inbox, inbox dead letters and outbox dead letters are returned in one list; Source tells the table (Outbox, Inbox, InboxDeadLetter, OutboxDeadLetter). TimestampUtc is outbox OccurredOnUtc, inbox ReceivedOnUtc, dead letter FailedOnUtc. Rows with the same timestamp follow the journey order (outbox, outbox dead letter, inbox, inbox dead letter).
• At least one of Key, CorrelationId, EventId is required (query.HasSearchKey); without one the result is an empty page, never a table scan. All given filters are combined with AND.
• EventId: outbox and inbox rows store the event-id header as IdempotencyKey (unique index), so the event is found by index. Dead letters keep it only in HeadersJson; they are matched by OriginalMessageId (the outbox/inbox row that owned the event, or the event id itself for inbox rows, whose id is the GUID event id).
• Key and CorrelationId use the outbox/inbox indexes added in 3.0.6 (dead letter tables index CorrelationId but not Key). Rows written before 3.0.6 have NULL in the new columns (inbox Key, outbox/inbox CorrelationId), so such rows are found by the other filters only.
• Each source is read in the final order up to the end of the requested page and merged in memory; deep pages cost more, so narrow the search with MessageType / FromUtc / ToUtc instead of paging far.
• The payload is not returned. LastError is cut to 500 characters; EventId and TargetConsumer come from the headers.

Status is derived:

Details of one row (payload and headers JSON) for a detail panel:

MessagePayloadDto? detail = await operationReader.GetMessagePayloadAsync(item.Source, item.Id, ct); // null if gone

Dead letter summary

IReadOnlyList<DeadLetterSummaryItemDto> summary = await operationReader.GetDeadLetterSummaryAsync(
    new DeadLetterSummaryQuery { FromUtc = DateTime.UtcNow.AddDays(-1) },   // FromUtc/ToUtc optional, on FailedOnUtc
    ct);

One row per Direction (Inbox/Outbox), MessageType and ReasonCode with Count, FirstFailedOnUtc and LastFailedOnUtc. Grouping runs in the database (GROUP BY); the largest group comes first, then the one that failed last.

Payloads and headers may hold personal data: authorize these endpoints for operators only.

Health and diagnostics

• IEventBusHealthState: TopologyReady, TransportReady, writer/processor heartbeats, LastError, package version and assembly hash. ia-eventbus-ready reports Unhealthy (topology/transport not ready), Degraded (recent error) or Healthy; it is always Healthy while Enabled = false. It is tagged ready, so /health/ready from MapIronAlpineHealthChecks (Web package) includes it (since 3.0.5). It can still be mapped on its own:

  app.MapHealthChecks("/health/eventbus", new HealthCheckOptions { Predicate = r => r.Name == "ia-eventbus-ready" });
  

• Activities are emitted from the ActivitySource named IronAlpine.Eventbus (eventbus.outbox.publish, eventbus.inbox.write, eventbus.inbox.process). Add the source to your tracer provider to export them.

• Metrics are emitted from the meter IronAlpine.EventBus (since 3.1.0): outbox published/failed, inbox received/processed/failed/skipped and dead letter counters tagged with topic (dead letters also direction and reason), plus pending-count and oldest-pending-age gauges for the outbox and inbox. The gauges are sampled every 30 seconds by a hosted service (EventBusBacklogSampler, registered with the storage) and read from memory, so a scrape never queries the database. Every pod reports the same gauge value: use max(). Names, PromQL and alert rules: Observability README, section Metrics. AddIronAlpineTelemetry already listens to this meter.

• Logs use the [IA EventBus] prefix.

Cleanup

EventBusCleanupHostedService runs every Cleanup.IntervalHours and deletes in batches, with one retention per table:

0 means the table is never cleaned; negative values fail start-up validation. Disable the whole service with enableCleanupHostedService: false or Cleanup.Enabled = false.

Upgrading from 3.0.5: ProcessedRetentionDays is removed. Configuration keys that are no longer bound are ignored silently, so rename it to InboxProcessedRetentionDays (and set OutboxProcessedRetentionDays if the outbox must still be cleaned; the new default keeps it forever).

Extension points

• Replace any store (IOutboxStore, IInboxStore, IDeadLetterStore, IReplayService, IMessageOperationReader) by registering it before AddIronAlpineEventBusKafka (registrations use TryAdd), or pass registerStorage: false and register storage yourself. A custom IMessageOperationReader must implement the 3.0.9 members (SearchMessagesAsync, GetMessagePayloadAsync, GetDeadLetterSummaryAsync).
• EventEnvelopeFactory.Create(payload, registry) and EventTypeCanonicalizer are public for custom transports.
• ModelBuilderExtensions.ApplyIronAlpineEventbus(modelBuilder, schema) maps the tables manually.

Schema changes in 3.0.6

The package ships no migrations; the consuming service adds one after upgrading:

• OutboxMessages: new columns CorrelationId, CausationId (length CorrelationIdMaxLength / CausationIdMaxLength, default 128); new indexes on Key, CorrelationId, CausationId.
• InboxMessages: new columns Key (KeyMaxLength, default 256), CorrelationId, CausationId; new indexes on Key, CorrelationId, CausationId.

The columns are filled on write (outbox from the envelope headers, inbox from the Kafka message key and headers, replay from the dead letter). Over-long values are cut to the column length; the full value stays in HeadersJson. Rows written before the upgrade keep NULL; HeadersJson is unchanged and still holds every header.

Pitfalls

• Consumed event types must implement INotification.
• Renaming or moving an event class changes its default type name; consumers then skip it as an unregistered type (since 3.0.6 it is marked processed with only a Warning log; earlier it was dead-lettered). Pin eventTypeName before refactoring contracts and watch for "Unregistered event type skipped" warnings after a deployment.
• IOutboxStore.AddAsync outside domain event dispatch saves the whole DbContext.
• In a publish-only service (no Consume routes) the inbox writer stays idle and does not touch TransportReady (up to 3.0.4 it kept marking the transport unavailable between publishes).
• ia-eventbus-ready is part of /health/ready since 3.0.5: a service whose transport or topology is not ready is reported not ready.
• The outbox claim order is by CreatedOnUtc, the inbox claim order by ReceivedOnUtc; there is no per-key ordering guarantee once retries happen.

Correlation and causation

Every message carries two ids that let you rebuild what happened:

• CorrelationId — the same for the whole chain. It starts with the incoming HTTP request (the trace id) and every event published while a consumed message is handled inherits it.
• CausationId — the EventId of the message that directly caused this event. Following causation ids from child to parent gives the tree of a chain (request → OrderPlaced → InvoiceCreated → ...).

Values set explicitly with SetCorrelationId / SetCausationId win. The consumer activity continues the producer's traceparent, so the chain is also a single trace in your tracing backend.