feature: Explainable Alerts
This commit is contained in:
voltsrage
@@ -0,0 +1,403 @@
|
||||
# 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:
|
||||
|
||||
```csharp
|
||||
// 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:
|
||||
|
||||
```csharp
|
||||
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:
|
||||
|
||||
```csharp
|
||||
// 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
|
||||
|
||||
```csharp
|
||||
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
|
||||
|
||||
```csharp
|
||||
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
|
||||
|
||||
```csharp
|
||||
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
|
||||
|
||||
```csharp
|
||||
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:
|
||||
|
||||
```csharp
|
||||
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:
|
||||
|
||||
```csharp
|
||||
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:
|
||||
|
||||
```csharp
|
||||
_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.
|
||||
Reference in New Issue
Block a user