vigilcare-clinical/docs/guides/10-transactional-outbox-pattern.md

Guide 10: Transactional Outbox Pattern

What is the Outbox Pattern?

The outbox pattern solves a fundamental problem in distributed systems: how do you reliably update a database AND send a message to a message broker (like Kafka) without losing data?

The Problem

Imagine this naive approach:

// Step 1: Save the observation to the database
await db.Observations.Add(observation);
await db.SaveChangesAsync();

// Step 2: Publish an event to Kafka
await producer.ProduceAsync("observation.recorded", new Message { Value = payload });

What can go wrong?

• Step 1 succeeds, step 2 fails (Kafka is down): The observation is saved, but no downstream consumers (scoring engines, search indexer, data lake) ever learn about it. The patient's NEWS2 score is never updated.
• Step 2 succeeds, step 1 fails (database constraint violation): Consumers process an event for an observation that doesn't exist in the database.
• Application crashes between steps: Same problem — one succeeded, the other didn't.

You can't wrap both in a single transaction because PostgreSQL and Kafka are different systems — there's no "distributed transaction" that spans both atomically (and even if there were, it would be slow and fragile).

The Solution

Instead of publishing to Kafka directly, write a message record (an "outbox event") to a table in the same database, in the same transaction as the domain data:

┌─────────────────────────────────────────────────────┐
│  Single PostgreSQL Transaction                       │
│                                                     │
│  1. INSERT INTO observations (...) VALUES (...)     │
│  2. INSERT INTO outbox_events (...) VALUES (...)    │
│  3. (if critical) INSERT INTO clinical_alerts (...) │
│  4. INSERT INTO outbox_events (...) VALUES (...)    │
│                                                     │
│  COMMIT  ← all or nothing                           │
└─────────────────────────────────────────────────────┘
         │
         │  A background service (the "relay") polls the outbox table
         │  and publishes each event to Kafka separately
         ▼
┌─────────────────────────────────────────────────────┐
│  OutboxRelayService (background)                     │
│                                                     │
│  1. SELECT ... FROM outbox_events                    │
│     WHERE processed_at IS NULL FOR UPDATE SKIP LOCKED│
│  2. ProduceAsync to Kafka                            │
│  3. UPDATE outbox_events SET processed_at = NOW()    │
│  4. COMMIT                                           │
└─────────────────────────────────────────────────────┘

Because both the observation and the outbox event are in the same PostgreSQL transaction, they either both commit or both roll back. The relay can safely retry — if it crashes or Kafka is temporarily down, the unprocessed outbox rows are still in the table, waiting to be picked up.

---

The Outbox Event Entity

The outbox_events table stores messages waiting to be published:

public class OutboxEventConfiguration : IEntityTypeConfiguration<OutboxEvent>
{
    public void Configure(EntityTypeBuilder<OutboxEvent> builder)
    {
        builder.ToTable("outbox_events");
        builder.HasKey(o => o.Id);
        builder.Property(o => o.Id).HasColumnName("id")
            .HasDefaultValueSql("gen_random_uuid()");
        builder.Property(o => o.Topic).HasColumnName("topic")
            .HasMaxLength(200).IsRequired();
        builder.Property(o => o.Payload).HasColumnName("payload")
            .HasColumnType("jsonb").IsRequired();
        builder.Property(o => o.PartitionKey).HasColumnName("partition_key")
            .HasMaxLength(36);
        builder.Property(o => o.CreatedAt).HasColumnName("created_at")
            .HasDefaultValueSql("NOW()");
        builder.Property(o => o.ProcessedAt).HasColumnName("processed_at");
        builder.Property(o => o.RetryCount).HasColumnName("retry_count")
            .HasDefaultValue(0);
        builder.Property(o => o.LastError).HasColumnName("last_error");
        builder.Property(o => o.FailedAt).HasColumnName("failed_at");

        // Partial index: only unprocessed, non-failed events need fast lookup
        builder.HasIndex(o => o.CreatedAt)
            .HasFilter("processed_at IS NULL AND failed_at IS NULL");
    }
}

Each outbox event has:

Column	Purpose
id	Unique identifier
topic	Which Kafka topic to publish to (e.g., observation.recorded)
payload	The JSON message body (stored as JSONB for compactness)
partition_key	Kafka partition key — typically the encounterId, ensuring all events for one encounter are ordered
created_at	When the event was written (used for ordering)
processed_at	Set to NOW() after successful Kafka publish — NULL means "not yet published"
retry_count	How many times the relay has tried and failed to publish this event
last_error	The error message from the most recent failed publish attempt
failed_at	Set when retry_count exceeds the maximum — marks the event as permanently failed

The partial index on created_at only covers rows where processed_at IS NULL AND failed_at IS NULL. This keeps the index small and fast — once an event is processed, it's no longer in the index. The relay's query only touches unprocessed events.

---

Writing Outbox Events (The Write Side)

Every time the application writes domain data that downstream consumers need to know about, it adds an outbox event in the same transaction. Here's the observation ingest path:

// All of this happens inside a single PostgreSQL transaction
await using var tx = await _db.Database.BeginTransactionAsync();

// 1. Insert the observation
_db.Observations.Add(observation);

// 2. Insert an outbox event for the observation
_db.OutboxEvents.Add(BuildOutboxEvent("observation.recorded", new
{
    observationId = observation.Id,
    encounterId,
    patientId = encounter.PatientId,
    observationCode = req.ObservationCode,
    value = req.Value,
    unit = req.Unit,
    recordedAt = req.RecordedAt,
    partitionKey = encounterId.ToString()
}, encounterId.ToString()));

// 3. If critical threshold breach — also insert the alert + its outbox event
if (IsCriticalBreach(req.Value, threshold))
{
    _db.ClinicalAlerts.Add(alert);
    _db.OutboxEvents.Add(BuildOutboxEvent("alert.generated", new
    {
        alertId = alert.Id,
        encounterId,
        alertType = alert.AlertType.ToDbString(),
        severity = alert.Severity.ToDbString(),
        partitionKey = encounterId.ToString()
    }, encounterId.ToString()));
}

// 4. COMMIT — observation, alert, and outbox events are all saved atomically
await _db.SaveChangesAsync();
await tx.CommitAsync();

If any step fails, the entire transaction rolls back — no orphaned events, no missing data.

---

The Outbox Relay Service (The Read Side)

The OutboxRelayService is a background service that runs continuously, polling the outbox_events table for unprocessed events and publishing them to Kafka.

The Idempotent Producer

public override Task StartAsync(CancellationToken cancellationToken)
{
    _producer = new ProducerBuilder<string, string>(new ProducerConfig
    {
        BootstrapServers = _options.BootstrapServers,
        Acks = Acks.All,
        EnableIdempotence = true,
        MessageSendMaxRetries = 3,
        RetryBackoffMs = 100
    }).Build();

    return base.StartAsync(cancellationToken);
}

Why idempotent? If the relay sends a message to Kafka but the acknowledgment is lost (network timeout), the relay retries. Without idempotency, Kafka would store the message twice. With EnableIdempotence = true, Kafka assigns the producer an internal ID and sequence number, recognizing and dropping duplicate deliveries.

The Poll Loop

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    while (!stoppingToken.IsCancellationRequested)
    {
        try
        {
            await ProcessBatchAsync(stoppingToken);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            _logger.LogError(ex, "Outbox relay error — will retry on next poll cycle");
        }

        await Task.Delay(_options.OutboxPollIntervalMs, stoppingToken);  // 1000ms
    }
}

The relay polls every second. If an error occurs (Kafka unreachable, database timeout), it logs the error and tries again on the next cycle. The when (ex is not OperationCanceledException) filter avoids logging expected shutdown cancellations as errors.

Fetching and Locking Events

private async Task ProcessBatchAsync(CancellationToken ct)
{
    using var scope = _services.CreateScope();
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

    await using var tx = await db.Database.BeginTransactionAsync(ct);

    var events = await db.OutboxEvents
        .FromSqlRaw("""
            SELECT id, topic, payload, partition_key, created_at, processed_at,
                   retry_count, last_error, failed_at
            FROM outbox_events
            WHERE processed_at IS NULL AND failed_at IS NULL
            ORDER BY created_at ASC
            LIMIT {0}
            FOR UPDATE SKIP LOCKED
            """, _options.OutboxBatchSize)  // batch size: 100
        .ToListAsync(ct);

    if (events.Count == 0)
    {
        await tx.RollbackAsync(ct);
        return;
    }

What is FOR UPDATE SKIP LOCKED? This is a PostgreSQL feature for safe concurrent access:

• FOR UPDATE locks the selected rows so no other transaction can modify them until this transaction commits or rolls back
• SKIP LOCKED means "if another relay instance already locked some of these rows, skip them instead of waiting"

This lets you run multiple relay instances for higher throughput — each instance grabs a different batch of rows without blocking or duplicating work. Think of it like multiple cashiers at a grocery store, each serving the next available customer.

Publishing Each Event

    foreach (var ev in events)
    {
        try
        {
            var result = await _producer!.ProduceAsync(
                ev.Topic,
                new Message<string, string>
                {
                    Key = ev.PartitionKey ?? string.Empty,
                    Value = ev.Payload
                }, ct);

            ev.ProcessedAt = DateTimeOffset.UtcNow;
        }
        catch (ProduceException<string, string> ex)
        {
            ev.RetryCount++;
            ev.LastError = ex.Error.Reason;

            if (ev.RetryCount >= _options.OutboxMaxRetries)  // 10 retries
            {
                ev.FailedAt = DateTimeOffset.UtcNow;
                _logger.LogError(ex,
                    "Outbox event {Id} permanently failed after {Retries} retries",
                    ev.Id, ev.RetryCount);
            }

            break;  // Stop batch on first failure
        }
    }

    await db.SaveChangesAsync(ct);
    await tx.CommitAsync(ct);
}

Key behaviors:

• On success: ProcessedAt is set. The event won't be picked up again (the WHERE clause filters it out).
• On failure: RetryCount is incremented and LastError is recorded. The event stays unprocessed and will be retried on the next poll cycle.
• After max retries: FailedAt is set, permanently marking the event as failed. This prevents an undeliverable message from blocking the entire queue forever. Failed events need manual investigation.
• Break on first failure: If one event fails to publish, the batch stops. This preserves ordering — events for the same encounter must be delivered in creation order.

Some Events Route to RabbitMQ Instead

Not all outbox events go to Kafka. The relay checks the topic and routes accordingly:

if (ev.Topic == ClinicalSyncOptions.BatchReceivedOutboxTopic)
{
    _rabbitChannel!.BasicPublish(
        exchange: _syncOpts.SyncExchange,
        routingKey: _syncOpts.SyncBatchReceivedRoutingKey,
        basicProperties: _rabbitProps,
        body: Encoding.UTF8.GetBytes(ev.Payload));
}
else
{
    await _producer!.ProduceAsync(ev.Topic, new Message<string, string>
    {
        Key = ev.PartitionKey ?? string.Empty,
        Value = ev.Payload
    }, ct);
}

This keeps the outbox pattern universal — any downstream message, whether Kafka or RabbitMQ, goes through the same transactional guarantee.

---

Monitoring the Outbox

Two monitoring mechanisms track outbox health:

OutboxPendingCollector (Prometheus gauge)

Every 30 seconds, counts how many events haven't been published yet:

var count = await db.OutboxEvents
    .CountAsync(e => e.ProcessedAt == null, ct);
_metrics.OutboxPendingEvents.Set(count);

A rising outbox_pending_events gauge on the Grafana dashboard means the relay is falling behind or Kafka is unreachable.

Outbox Relay Logs

The relay logs each batch:

_logger.LogInformation(
    "Outbox relay published {Count} events. Failed={Failed}",
    published.Count, hadFailure);

---

The Complete Flow

1. HTTP Request arrives
   POST /api/encounters/{id}/observations

2. ObservationService.IngestAsync()
   ┌─── PostgreSQL Transaction ───────────────────┐
   │ INSERT INTO observations (...)                │
   │ INSERT INTO outbox_events (topic='obs.rec.')  │
   │ IF critical:                                  │
   │   INSERT INTO clinical_alerts (...)           │
   │   INSERT INTO outbox_events (topic='alert.')  │
   │ COMMIT                                        │
   └──────────────────────────────────────────────┘

3. HTTP Response returned to caller (201 Created)
   The caller doesn't wait for Kafka — it's decoupled.

4. OutboxRelayService (1 second later)
   ┌─── Poll cycle ──────────────────────────────┐
   │ SELECT FROM outbox_events FOR UPDATE SKIP..  │
   │ ProduceAsync to Kafka (observation.recorded) │
   │ ProduceAsync to Kafka (alert.generated)      │
   │ SET processed_at = NOW()                     │
   │ COMMIT                                        │
   └──────────────────────────────────────────────┘

5. Kafka delivers to 9 consumer groups
   es-indexer, sepsis-engine, news2-scoring, etc.

---

Key Design Decisions

Why Poll Instead of Change Data Capture?

Some implementations use PostgreSQL's logical replication or a CDC (Change Data Capture) tool like Debezium to stream outbox rows to Kafka. This project uses simple polling because:

• Polling is straightforward to implement and debug
• The OutboxPollIntervalMs (1 second) is fast enough for clinical use cases
• No additional infrastructure (Debezium connector, separate process) is needed
• The FOR UPDATE SKIP LOCKED pattern handles concurrency cleanly

Why Break on First Failure?

If event A and event B are for the same encounter, they must arrive at Kafka in order (A before B). If event A fails to publish and we skip it to publish B, consumers would see B first — which could cause incorrect scoring or duplicate alerts. Breaking on first failure maintains ordering at the cost of potentially delaying later events.

Why a Permanent Failure State?

After 10 retries, an event is marked FailedAt and excluded from future relay cycles. Without this, a single undeliverable event (e.g., payload too large for Kafka) would block the entire outbox forever. Failed events are visible in the database and Prometheus metrics for investigation.

---

Key Takeaways

• The outbox pattern guarantees atomicity across database writes and message publishing by keeping both in the same PostgreSQL transaction.
• The relay is eventually consistent — there's a short delay (up to OutboxPollIntervalMs) between the database commit and Kafka delivery. For this project, 1 second is clinically acceptable.
• Idempotent producers handle retry duplicates — network timeouts that cause re-delivery are silently deduplicated by Kafka.
• FOR UPDATE SKIP LOCKED enables horizontal scaling — multiple relay instances can run without coordination.
• Retry tracking prevents infinite loops — events that persistently fail are marked and excluded after a configurable number of attempts.
• The pattern applies to any message broker — the same outbox row can be published to Kafka, RabbitMQ, or any other system. The transactional guarantee is with the database, not with a specific broker.