test(reconciliation): pin inclusive (>=) cursor reads at the site (audit + site-calls); EventId/upsert dedup absorbs boundary duplicates

2026-07-08 21:58:57 -04:00
parent 846279f6a8
commit ae94b22d7d
2 changed files with 61 additions and 0 deletions
@@ -681,4 +681,43 @@ public class OperationTrackingStoreTests
        // And a third async — also a no-op.
        await store.DisposeAsync();
    }
+
+    // ── Task 24: pin the inclusive (>=) reconciliation cursor boundary ──
+
+    [Fact]
+    public async Task ReadChangedSince_BoundaryTimestamp_IsInclusive_UpsertOnNewerAbsorbsDuplicates()
+    {
+        // Two rows share one UpdatedAtUtc; a pull whose sinceUtc equals that instant
+        // MUST return both (inclusive >= read). Central ingest is insert-if-not-exists
+        // + upsert-on-newer, so re-reading the boundary row is a safe no-op — but an
+        // exclusive (>) read would silently and permanently drop the same-timestamp
+        // row. This test pins the >= contract.
+        var (store, dataSource) = CreateStore(
+            nameof(ReadChangedSince_BoundaryTimestamp_IsInclusive_UpsertOnNewerAbsorbsDuplicates));
+        await using var _store = store;
+
+        await store.RecordEnqueueAsync(TrackedOperationId.New(),
+            kind: nameof(AuditKind.ApiCallCached), targetSummary: "ERP.A",
+            sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+        await store.RecordEnqueueAsync(TrackedOperationId.New(),
+            kind: nameof(AuditKind.ApiCallCached), targetSummary: "ERP.B",
+            sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+
+        // Force both rows to share one exact UpdatedAtUtc instant, formatted the same
+        // way the read normalises the cursor ('Z'-suffixed round-trip "o").
+        var boundary = DateTime.UtcNow;
+        var boundaryText = DateTime.SpecifyKind(boundary, DateTimeKind.Utc)
+            .ToString("o", System.Globalization.CultureInfo.InvariantCulture);
+        using (var conn = OpenVerifierConnection(dataSource))
+        using (var cmd = conn.CreateCommand())
+        {
+            cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $ts";
+            cmd.Parameters.AddWithValue("$ts", boundaryText);
+            cmd.ExecuteNonQuery();
+        }
+
+        var rows = await store.ReadChangedSinceAsync(boundary, batchSize: 10);
+
+        Assert.Equal(2, rows.Count);
+    }
 }