Fix: Outbox relay has no dead-letter or max retry limit + DataLake writer partial commit inconsistency + ThresholdCacheLoader crashes startup on Redis failure

VigilCareClinicalAPI/BackgroundServices/OutboxRelayService.cs

@@ -68,9 +68,10 @@ public class OutboxRelayService : BackgroundService
 
         var events = await db.OutboxEvents
             .FromSqlRaw("""
-                SELECT id, topic, payload, partition_key, created_at, processed_at
+                SELECT id, topic, payload, partition_key, created_at, processed_at,
+                       retry_count, last_error, failed_at
                 FROM outbox_events
-                WHERE processed_at IS NULL
+                WHERE processed_at IS NULL AND failed_at IS NULL
                 ORDER BY created_at ASC
                 LIMIT {0}
                 FOR UPDATE SKIP LOCKED
@@ -84,7 +85,7 @@ public class OutboxRelayService : BackgroundService
         }
 
         var published = new List<Guid>();
-        var failed = false;
+        var hadFailure = false;
 
         foreach (var ev in events)
         {
@@ -94,48 +95,49 @@ public class OutboxRelayService : BackgroundService
                     ev.Topic,
                     new Message<string, string>
                     {
-                        // The partition key ensures all events for one encounter land on
-                        // the same partition. The sepsis engine depends on this — if two
-                        // observations for the same patient arrive on different partitions,
-                        // a single-partition consumer misses one and SIRS evaluation is wrong.
                         Key = ev.PartitionKey ?? string.Empty,
                         Value = ev.Payload
                     },
                     ct);
 
                 published.Add(ev.Id);
+                ev.ProcessedAt = DateTimeOffset.UtcNow;
 
                 _logger.LogDebug(
                     "Published {Topic} offset={Offset} partition={Partition} key={Key}",
                     ev.Topic, result.Offset.Value, result.Partition.Value, ev.PartitionKey);
             }
             catch (ProduceException<string, string> ex)
+            {
+                ev.RetryCount++;
+                ev.LastError = ex.Error.Reason;
+
+                if (ev.RetryCount >= _options.OutboxMaxRetries)
+                {
+                    ev.FailedAt = DateTimeOffset.UtcNow;
+                    _logger.LogError(ex,
+                        "Outbox event {Id} permanently failed after {Retries} retries — topic={Topic}",
+                        ev.Id, ev.RetryCount, ev.Topic);
+                }
+                else
                 {
                     _logger.LogWarning(ex,
-                    "Kafka produce failed for outbox event {Id} — leaving unprocessed for next cycle",
-                    ev.Id);
-                failed = true;
-                break;  // stop processing this batch; retry from this row next cycle
+                        "Kafka produce failed for outbox event {Id} — retry {Retry}/{Max}",
+                        ev.Id, ev.RetryCount, _options.OutboxMaxRetries);
+                }
+
+                hadFailure = true;
+                break;
             }
         }
 
-        if (published.Count > 0)
-        {
-            var now = DateTimeOffset.UtcNow;
-            foreach (var id in published)
-            {
-                var ev = events.First(e => e.Id == id);
-                ev.ProcessedAt = now;
-            }
         await db.SaveChangesAsync(ct);
-        }
-
         await tx.CommitAsync(ct);
 
         if (published.Count > 0)
             _logger.LogInformation(
                 "Outbox relay published {Count} events. Failed={Failed}",
-                published.Count, failed);
+                published.Count, hadFailure);
     }
 
     public override void Dispose()

VigilCareClinicalAPI/BackgroundServices/ThresholdCacheLoader.cs

@@ -22,9 +22,17 @@ public class ThresholdCacheLoader : IHostedService
     {
         using var scope = _services.CreateScope();
         var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
-        var cache = _redis.GetDatabase();
 
         var thresholds = await db.AlertThresholds.ToListAsync(cancellationToken);
+
+        const int maxAttempts = 3;
+        int[] backoffMs = [2000, 4000, 8000];
+
+        for (var attempt = 0; attempt < maxAttempts; attempt++)
+        {
+            try
+            {
+                var cache = _redis.GetDatabase();
                 var batch = cache.CreateBatch();
 
                 foreach (var t in thresholds)
@@ -42,6 +50,23 @@ public class ThresholdCacheLoader : IHostedService
 
                 batch.Execute();
                 _logger.LogInformation("Loaded {Count} alert thresholds into Redis cache", thresholds.Count);
+                return;
+            }
+            catch (RedisException ex)
+            {
+                _logger.LogWarning(ex,
+                    "Redis unavailable during threshold cache load — attempt {Attempt}/{Max}",
+                    attempt + 1, maxAttempts);
+
+                if (attempt < maxAttempts - 1)
+                    await Task.Delay(backoffMs[attempt], cancellationToken);
+            }
+        }
+
+        _logger.LogError(
+            "Failed to load thresholds into Redis after {Max} attempts — " +
+            "application will start without cache; observation ingest falls back to PostgreSQL",
+            maxAttempts);
     }
 
     public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;

VigilCareClinicalAPI/Configuration/KafkaOptions.cs

@@ -7,5 +7,6 @@ public class KafkaOptions
     public short ReplicationFactor { get; set; } = 3;
     public int OutboxBatchSize { get; set; } = 100;
     public int OutboxPollIntervalMs { get; set; } = 500;
+    public int OutboxMaxRetries { get; set; } = 10;
     public int MaxPoisonRetries { get; set; } = 5;
 }

VigilCareClinicalAPI/Data/Configurations/OutboxEventConfiguration.cs

@@ -13,8 +13,11 @@ public class OutboxEventConfiguration : IEntityTypeConfiguration<OutboxEvent>
         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");
 
         builder.HasIndex(o => o.CreatedAt)
-            .HasFilter("processed_at IS NULL");
+            .HasFilter("processed_at IS NULL AND failed_at IS NULL");
     }
 }

VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs

@@ -122,7 +122,8 @@ public sealed class DataLakeWriterService : BackgroundService
 
     private async Task FlushAsync(IConsumer<string, string> consumer, CancellationToken ct)
     {
-        var filesWritten = 0;
+        var flushedKeys = new HashSet<BufferKey>();
+        var failedPartitions = new HashSet<int>();
 
         foreach (var (key, events) in _buffer)
         {
@@ -135,7 +136,7 @@ public sealed class DataLakeWriterService : BackgroundService
                 var bytes       = await BuildParquetAsync(key.Topic, events, key.Partition);
 
                 await UploadToMinioAsync(objectKey, bytes, ct);
-                filesWritten++;
+                flushedKeys.Add(key);
 
                 _logger.LogInformation(
                     "[DATA-LAKE] Wrote {Count} events → {ObjectKey} ({Bytes} bytes)",
@@ -143,8 +144,8 @@ public sealed class DataLakeWriterService : BackgroundService
             }
             catch (Exception ex)
             {
-                // Log and continue — a failed file for one key must not prevent other
-                // keys from flushing. The uncommitted offsets will cause reprocessing.
+                failedPartitions.Add(key.Partition);
+
                 if (ex is OperationCanceledException && ct.IsCancellationRequested)
                 {
                     _logger.LogInformation(
@@ -157,24 +158,37 @@ public sealed class DataLakeWriterService : BackgroundService
             }
         }
 
-        // Commit only after at least one file uploaded successfully.
-        // Events for any key that failed above will be re-read on next startup.
-        if (filesWritten > 0 && _highWatermarks.Any())
+        // Only commit offsets for topic-partitions that had no failures.
+        var safeOffsets = _highWatermarks
+            .Where(kv => !failedPartitions.Contains(kv.Key.Partition))
+            .Select(kv => kv.Value)
+            .ToList();
+
+        if (safeOffsets.Count > 0)
         {
-            consumer.Commit(_highWatermarks.Values);
+            consumer.Commit(safeOffsets);
             _logger.LogInformation(
                 "[DATA-LAKE] Committed offsets for {PartitionCount} partitions after flushing {FileCount} files",
-                _highWatermarks.Count, filesWritten);
-        }
-        else if (filesWritten == 0 && _highWatermarks.Any())
-        {
-            _logger.LogWarning(
-                "[DATA-LAKE] Skipping offset commit — no files were written ({BufferedPartitions} partitions buffered)",
-                _highWatermarks.Count);
+                safeOffsets.Count, flushedKeys.Count);
         }
 
-        _buffer.Clear();
-        _highWatermarks.Clear();
+        if (failedPartitions.Count > 0)
+        {
+            _logger.LogWarning(
+                "[DATA-LAKE] Retained buffers for {FailedCount} failed partitions — will retry next flush",
+                failedPartitions.Count);
+        }
+
+        // Clear only successfully flushed keys; retain failed ones for retry.
+        foreach (var key in flushedKeys)
+            _buffer.Remove(key);
+
+        // Clear watermarks only for partitions with no failures.
+        foreach (var tp in _highWatermarks.Keys.ToList())
+        {
+            if (!failedPartitions.Contains(tp.Partition))
+                _highWatermarks.Remove(tp);
+        }
     }
 
     private async Task<byte[]> BuildParquetAsync(

VigilCareClinicalAPI/Domains/Entities/OutboxEvent.cs

@@ -6,4 +6,7 @@ public class OutboxEvent
     public string? PartitionKey { get; set; }   // encounter_id for all clinical events
     public DateTimeOffset CreatedAt { get; set; }
     public DateTimeOffset? ProcessedAt { get; set; }
+    public int RetryCount { get; set; }
+    public string? LastError { get; set; }
+    public DateTimeOffset? FailedAt { get; set; }
 }

VigilCareClinicalAPI/Migrations/20260621093835_AddOutboxRetryColumns.cs

@@ -0,0 +1,70 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace VigilCareClinicalAPI.Migrations
+{
+    /// <inheritdoc />
+    public partial class AddOutboxRetryColumns : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropIndex(
+                name: "IX_outbox_events_created_at",
+                table: "outbox_events");
+
+            migrationBuilder.AddColumn<DateTimeOffset>(
+                name: "failed_at",
+                table: "outbox_events",
+                type: "timestamp with time zone",
+                nullable: true);
+
+            migrationBuilder.AddColumn<string>(
+                name: "last_error",
+                table: "outbox_events",
+                type: "text",
+                nullable: true);
+
+            migrationBuilder.AddColumn<int>(
+                name: "retry_count",
+                table: "outbox_events",
+                type: "integer",
+                nullable: false,
+                defaultValue: 0);
+
+            migrationBuilder.CreateIndex(
+                name: "IX_outbox_events_created_at",
+                table: "outbox_events",
+                column: "created_at",
+                filter: "processed_at IS NULL AND failed_at IS NULL");
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropIndex(
+                name: "IX_outbox_events_created_at",
+                table: "outbox_events");
+
+            migrationBuilder.DropColumn(
+                name: "failed_at",
+                table: "outbox_events");
+
+            migrationBuilder.DropColumn(
+                name: "last_error",
+                table: "outbox_events");
+
+            migrationBuilder.DropColumn(
+                name: "retry_count",
+                table: "outbox_events");
+
+            migrationBuilder.CreateIndex(
+                name: "IX_outbox_events_created_at",
+                table: "outbox_events",
+                column: "created_at",
+                filter: "processed_at IS NULL");
+        }
+    }
+}

VigilCareClinicalAPI/Migrations/AppDbContextModelSnapshot.cs

@@ -763,6 +763,14 @@ namespace VigilCareClinicalAPI.Migrations
                         .HasColumnName("created_at")
                         .HasDefaultValueSql("NOW()");
 
+                    b.Property<DateTimeOffset?>("FailedAt")
+                        .HasColumnType("timestamp with time zone")
+                        .HasColumnName("failed_at");
+
+                    b.Property<string>("LastError")
+                        .HasColumnType("text")
+                        .HasColumnName("last_error");
+
                     b.Property<string>("PartitionKey")
                         .HasMaxLength(36)
                         .HasColumnType("character varying(36)")
@@ -777,6 +785,12 @@ namespace VigilCareClinicalAPI.Migrations
                         .HasColumnType("timestamp with time zone")
                         .HasColumnName("processed_at");
 
+                    b.Property<int>("RetryCount")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("integer")
+                        .HasDefaultValue(0)
+                        .HasColumnName("retry_count");
+
                     b.Property<string>("Topic")
                         .IsRequired()
                         .HasMaxLength(200)
@@ -786,7 +800,7 @@ namespace VigilCareClinicalAPI.Migrations
                     b.HasKey("Id");
 
                     b.HasIndex("CreatedAt")
-                        .HasFilter("processed_at IS NULL");
+                        .HasFilter("processed_at IS NULL AND failed_at IS NULL");
 
                     b.ToTable("outbox_events", (string)null);
                 });