diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs index 745e159b..a985b742 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs @@ -111,6 +111,94 @@ public sealed class TrailMarkerTests(QuestDbFixture fx) (await ModifiedAsync(rig, "t")).Count.ShouldBe(3); } + /// + /// The paged-vs-unpaged fuzz for the reconciled modified read: a random trail (several modifications per source + /// time, a random mix of applied, orphaned and landed-without-marker rows) paged at random caps, forward and + /// reverse, through the cursor, must concatenate to the unpaged read — the window function's partition must never be + /// cut by the resume position — and the unpaged read must equal an independent oracle of the marker rule. + /// + [Fact] + public async Task Paged_modified_reads_equal_the_unpaged_read_over_a_random_marker_mix() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Tags.Bootstrap.EnsureAsync(Ct); + await rig.Tags.EnsureAsync("t", HistorianDataTypes.FromKind(HistorianValueKind.Double), Ct); + var rng = new Random(746); + var expected = new List<(DateTime Ts, long Id)>(); + var trailSql = new System.Text.StringBuilder(); + var sampleSql = new System.Text.StringBuilder(); + long id = 1_000; + var mod0 = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc); + for (var t = 0; t < 40; t++) + { + var ts = T0.AddSeconds(t); + var rows = new List<(long Id, bool Pending, DateTime Mod)>(); + for (var k = rng.Next(1, 5); k > 0; k--) rows.Add((id++, rng.Next(3) == 0, mod0.AddSeconds(id))); + // The sample's write time: after every modification (the latest landed), or before the latest (it did not). + var latest = rows[^1]; + var landed = rng.Next(2) == 0; + var sampleServer = landed ? latest.Mod.AddSeconds(1) : latest.Mod.AddSeconds(-1); + foreach (var r in rows) + { + var served = !r.Pending || (r.Id == latest.Id && landed); + if (served) expected.Add((ts, r.Id)); + if (trailSql.Length > 0) trailSql.Append(','); + trailSql.Append($"('{ts:O}','t',{r.Id},'{r.Mod:O}',1,'p',NULL,1,1.0,NULL,NULL,0,NULL,{(r.Pending ? "true" : "false")})"); + } + if (rng.Next(6) > 0) + { + if (sampleSql.Length > 0) sampleSql.Append(','); + sampleSql.Append($"('{ts:O}','t',1,1.0,NULL,NULL,0,false,'{sampleServer:O}')"); + } + else + { + // No sample at this time at all: only non-pending rows are served. + expected.RemoveAll(e => e.Ts == ts && rows.Single(r => r.Id == e.Id).Pending); + } + } + await using (var conn = await rig.Conn.OpenAsync(Ct)) + { + await using (var cmd = rig.Conn.CreateCommand(conn, $"INSERT INTO {rig.Conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, " + + $"update_type, principal, api_key_id, kind, num_value, int_value, str_value, status, server_ts, pending) VALUES {trailSql}")) + await cmd.ExecuteNonQueryAsync(Ct); + if (sampleSql.Length > 0) + await using (var cmd = rig.Conn.CreateCommand(conn, $"INSERT INTO {rig.Conn.SampleTable} (ts, tag, kind, num_value, " + + $"int_value, str_value, status, deleted, server_ts) VALUES {sampleSql}")) + await cmd.ExecuteNonQueryAsync(Ct); + } + var wal = new QuestDbWalVisibility(rig.Conn, TimeSpan.FromSeconds(10)); + await wal.WaitAsync(rig.Conn.SampleModifiedTable, Ct); + await wal.WaitAsync(rig.Conn.SampleTable, Ct); + + var src = Source(rig); + var (lo, hi) = (T0.AddSeconds(-1), T0.AddSeconds(41)); + foreach (var reverse in new[] { false, true }) + { + var (start, end) = reverse ? (hi, lo) : (lo, hi); + var truth = (await src.ReadModifiedAsync("t", start, end, 0, null, Ct)).Values.Select(v => (v.SourceTimeUtc, v.ModifiedId)).ToList(); + var oracle = reverse ? expected.AsEnumerable().Reverse().ToList() : expected; + truth.ShouldBe(oracle, $"unpaged vs the marker oracle (reverse {reverse})"); + foreach (var cap in new uint[] { 1, 2, 3, 5, 7, 13, 50 }) + { + var paged = new List<(DateTime, long)>(); + HistoryModifiedCursor? cursor = null; + for (var page = 0; ; page++) + { + page.ShouldBeLessThan(500, "paging must terminate"); + var r = await src.ReadModifiedAsync("t", start, end, cap, cursor, Ct); + r.Values.Count.ShouldBeLessThanOrEqualTo((int)cap); + paged.AddRange(r.Values.Select(v => (v.SourceTimeUtc, v.ModifiedId))); + if (!r.HasMore) break; + cursor = new HistoryModifiedCursor(r.Values[^1].SourceTimeUtc, r.Values[^1].ModifiedId); + } + paged.ShouldBe(truth, $"paged ≠ unpaged at cap {cap} (reverse {reverse})"); + } + } + expected.Count.ShouldBeGreaterThan(0); + ((long)expected.Count).ShouldBeLessThan(id - 1_000, "the mix must hide some rows, or the reconciliation is untested"); + } + [Fact] public async Task A_pre_marker_trail_table_is_migrated_on_boot_and_its_old_rows_stay_visible() {