fix: add sort guard and stale-event detection in ApiPollerWorker

Author: udmada

docs/local-setup.md

@@ -26,7 +26,7 @@ docker exec -it $(docker ps -q -f ancestor=mcr.microsoft.com/azure-sql-edge) \
   -Q "CREATE DATABASE ScanEvents"
 ```
 
-The worker auto-creates the `ProcessingState` and `ParcelSummary` tables on startup via `DatabaseInitializer`.
+The worker auto-creates the `ProcessingState` and `ParcelSummary` tables on startup via `DatabaseInitialiser`.
 
 ### 3. Set dummy AWS credentials (LocalStack doesn't validate them)
 

src/ScanEventWorker/Infrastructure/ApiClient/ApiJsonContext.cs

@@ -3,7 +3,7 @@
 
 namespace ScanEventWorker.Infrastructure.ApiClient;
 
-// CamelCase policy applies to ScanEvent (SQS serialization/deserialization) - symmetrical both ways.
+// CamelCase policy applies to ScanEvent (SQS serialisation/deserialisation) - symmetrical both ways.
 // ScanEventApiResponse/ScanEventDto use explicit [JsonPropertyName] overrides and are unaffected.
 // Do NOT add [JsonPropertyName] to ScanEvent properties - they inherit camelCase from this policy.
 [JsonSerializable(typeof(ScanEventApiResponse))]

…cture/Persistence/DatabaseInitializer.cs …cture/Persistence/DatabaseInitialiser.cssrc/ScanEventWorker/Infrastructure/Persistence/DatabaseInitializer.cs renamed to src/ScanEventWorker/Infrastructure/Persistence/DatabaseInitialiser.cs src/ScanEventWorker/Infrastructure/Persistence/DatabaseInitializer.cs renamed to src/ScanEventWorker/Infrastructure/Persistence/DatabaseInitialiser.cs

@@ -4,11 +4,11 @@
 namespace ScanEventWorker.Infrastructure.Persistence;
 
 [DapperAot]
-public sealed class DatabaseInitializer(string connectionString, ILogger<DatabaseInitializer> logger)
+public sealed class DatabaseInitialiser(string connectionString, ILogger<DatabaseInitialiser> logger)
 {
-    public async Task InitializeAsync(CancellationToken ct)
+    public async Task InitialiseAsync(CancellationToken ct)
     {
-        logger.LogInformation("Initializing database schema");
+        logger.LogInformation("Initialising database schema");
 
         await using var connection = new SqlConnection(connectionString);
         await connection.OpenAsync(ct);
@@ -42,6 +42,6 @@ DeliveredAtUtc DATETIMEOFFSET NULL
                                                                 END
                                                                 """, cancellationToken: ct));
 
-        logger.LogInformation("Database schema initialized");
+        logger.LogInformation("Database schema initialised");
     }
 }

src/ScanEventWorker/Infrastructure/Persistence/ScanEventRepository.cs

@@ -22,7 +22,7 @@ public async Task<long> GetLastEventIdAsync(CancellationToken ct)
         {
             logger.LogWarning(
                 "ProcessingState row not found - defaulting LastEventId to 1. " +
-                "Re-run DatabaseInitializer or INSERT INTO ProcessingState (Id, LastEventId) VALUES (1, 1).");
+                "Re-run DatabaseInitialiser or INSERT INTO ProcessingState (Id, LastEventId) VALUES (1, 1).");
             return 1L;
         }
 

src/ScanEventWorker/Program.cs

@@ -7,6 +7,16 @@
 using ScanEventWorker.Services;
 using ScanEventWorker.Workers;
 
+// Prevent multiple instances from running simultaneously (see Assumptions: single-instance constraint).
+// The Mutex is held for the lifetime of the process; released automatically on process exit.
+using var instanceMutex = new Mutex(initiallyOwned: true, name: "Global\\ScanEventWorker_Instance", out bool mutexAcquired);
+if (!mutexAcquired)
+{
+    // Another instance is already running — log to stderr and exit cleanly.
+    Console.Error.WriteLine("FATAL: Another ScanEventWorker instance is already running. Exiting.");
+    return 1;
+}
+
 HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);
 
 // Configuration
@@ -30,7 +40,7 @@
 
 // Infrastructure — Persistence
 builder.Services.AddSingleton(sp =>
-    new DatabaseInitializer(connectionString, sp.GetRequiredService<ILogger<DatabaseInitializer>>()));
+    new DatabaseInitialiser(connectionString, sp.GetRequiredService<ILogger<DatabaseInitialiser>>()));
 builder.Services.AddSingleton<IScanEventRepository>(sp =>
     new ScanEventRepository(
         connectionString,
@@ -69,8 +79,10 @@
 
 IHost host = builder.Build();
 
-// Initialize database schema on startup
-DatabaseInitializer dbInitializer = host.Services.GetRequiredService<DatabaseInitializer>();
-await dbInitializer.InitializeAsync(CancellationToken.None);
+// Initialise database schema on startup
+DatabaseInitialiser dbInitialiser = host.Services.GetRequiredService<DatabaseInitialiser>();
+await dbInitialiser.InitialiseAsync(CancellationToken.None);
 
 host.Run();
+
+return 0;

src/ScanEventWorker/Workers/ApiPollerWorker.cs

@@ -49,12 +49,31 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                 continue;
             }
 
+            // Guard: sort defensively if API returns events out of EventId order (Assumption 1)
+            if (events.Count > 1 && events.Zip(events.Skip(1)).Any(pair => pair.First.EventId.Value > pair.Second.EventId.Value))
+            {
+                logger.LogWarning(
+                    "API returned {Count} events out of EventId order — sorting defensively",
+                    events.Count);
+                events = [.. events.OrderBy(e => e.EventId.Value)];
+            }
+
+            // Guard: warn if API returns events older than lastEventId (Assumption 2)
+            int staleCount = events.Count(e => e.EventId.Value < lastEventId);
+            if (staleCount > 0)
+            {
+                logger.LogWarning(
+                    "API returned {StaleCount} stale events with EventId < {FromId} — possible FromEventId contract violation",
+                    staleCount,
+                    lastEventId);
+            }
+
             foreach (ScanEvent scanEvent in events)
             {
                 await messageQueue.SendAsync(scanEvent, stoppingToken);
             }
 
-            long maxEventId = events[^1].EventId.Value;
+            long maxEventId = Math.Max(lastEventId, events[^1].EventId.Value);
             await repository.UpdateLastEventIdAsync(maxEventId, stoppingToken);
             lastEventId = maxEventId;
 

tests/ScanEventWorker.Tests/Workers/ApiPollerWorkerTests.cs

@@ -210,4 +210,58 @@ public async Task WhenApiReturnsEvents_CallsUpdateExactlyOnce()
 
         await _repository.Received(1).UpdateLastEventIdAsync(Arg.Any<long>(), Arg.Any<CancellationToken>());
     }
+
+    [Fact(Timeout = 5000)]
+    public async Task WhenApiReturnsOutOfOrderEvents_SortsDefensivelyAndAdvancesCorrectly()
+    {
+        var cts = new CancellationTokenSource();
+        // Deliberately out of order: 10 first, then 5
+        var events = new List<ScanEvent> { MakeScanEvent(10), MakeScanEvent(5) };
+
+        _ = _repository.GetLastEventIdAsync(Arg.Any<CancellationToken>()).Returns(0L);
+        _ = _apiClient.GetScanEventsAsync(Arg.Any<long>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Result<IReadOnlyList<ScanEvent>>.Success(events));
+        _ = _repository.UpdateLastEventIdAsync(Arg.Any<long>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo =>
+            {
+                cts.Cancel();
+                return Task.CompletedTask;
+            });
+
+        ApiPollerWorker worker = CreateWorker();
+        await worker.StartAsync(cts.Token);
+        await worker.ExecuteTask!;
+        await worker.StopAsync(CancellationToken.None);
+
+        // After sort [5, 10], events[^1] is EventId=10 — the correct max
+        await _repository.Received(1).UpdateLastEventIdAsync(10L, Arg.Any<CancellationToken>());
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task WhenApiReturnsStaleEvents_ContinuesProcessingNormally()
+    {
+        var cts = new CancellationTokenSource();
+        // Both events are older than lastEventId=20
+        var events = new List<ScanEvent> { MakeScanEvent(5), MakeScanEvent(10) };
+
+        _ = _repository.GetLastEventIdAsync(Arg.Any<CancellationToken>()).Returns(20L);
+        _ = _apiClient.GetScanEventsAsync(Arg.Any<long>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Result<IReadOnlyList<ScanEvent>>.Success(events));
+        _ = _repository.UpdateLastEventIdAsync(Arg.Any<long>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo =>
+            {
+                cts.Cancel();
+                return Task.CompletedTask;
+            });
+
+        ApiPollerWorker worker = CreateWorker();
+        await worker.StartAsync(cts.Token);
+        await worker.ExecuteTask!;
+        await worker.StopAsync(CancellationToken.None);
+
+        // Stale events are processed normally — idempotent MERGE handles dedup.
+        // lastEventId must not regress: advance marker stays at 20, not events[^1]=10.
+        await _repository.Received(1).UpdateLastEventIdAsync(20L, Arg.Any<CancellationToken>());
+        await _queue.Received(2).SendAsync(Arg.Any<ScanEvent>(), Arg.Any<CancellationToken>());
+    }
 }