Joseph Doherty
aa5875bae2
The applied-marker reconciliation changed the modified read's query shape (a window function over each source time, with the resume position applied outside it), which is a paging change. A random trail (1-4 modifications per time; applied, orphaned and landed-without-marker rows; times with no sample) is paged at caps 1..50 in both directions through the cursor and must concatenate to the unpaged read, which must equal an independent oracle of the marker rule. Falsified: dropping the latest-row rule fails it. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
243 lines
13 KiB
C#
243 lines
13 KiB
C#
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
|
|
|
|
/// <summary>
|
|
/// The QuestDB modification trail's applied marker and dedup key. QuestDB has no multi-table transaction, so the trail
|
|
/// row is written BEFORE its change: a failed change left an orphan row that modified reads served as a change that
|
|
/// happened, and a trail INSERT re-sent after a client timeout duplicated. Now the row is written <c>pending</c>,
|
|
/// marked applied after the change, keyed <c>(ts, tag, modified_id)</c>; a modified read drops a pending row unless the
|
|
/// sample shows the change landed.
|
|
/// </summary>
|
|
[Collection(QuestDbCollection.Name)]
|
|
public sealed class TrailMarkerTests(QuestDbFixture fx)
|
|
{
|
|
private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
|
|
private static readonly CancellationToken Ct = CancellationToken.None;
|
|
|
|
private static QuestDbHistorianDataSource Source(UpdateTests.Rig rig) => new(rig.Conn, rig.Tags.Bootstrap, rig.Tags);
|
|
|
|
private static async Task<IReadOnlyList<HistoryModifiedValue>> ModifiedAsync(UpdateTests.Rig rig, string tag) =>
|
|
(await Source(rig).ReadModifiedAsync(tag, T0.AddMinutes(-1), T0.AddMinutes(10), 0, null, Ct)).Values;
|
|
|
|
private static async Task<long> TrailRowsAsync(UpdateTests.Rig rig, string tag)
|
|
{
|
|
await using var conn = await rig.Conn.OpenAsync(Ct);
|
|
await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT count() FROM {rig.Conn.SampleModifiedTable} WHERE tag = $1");
|
|
QuestDbConnection.BindText(cmd, tag);
|
|
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_new_trail_table_has_the_marker_format_and_an_applied_change_reads_back()
|
|
{
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
(await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct))
|
|
.ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
|
|
rig.Tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue();
|
|
|
|
var trail = await ModifiedAsync(rig, "t");
|
|
trail.ShouldHaveSingleItem().UpdateType.ShouldBe(HistoryModificationType.Insert);
|
|
(await TrailRowsAsync(rig, "t")).ShouldBe(1, "the applied marker re-upserts the row, it does not add one");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task The_orphan_of_a_change_that_never_landed_is_not_served()
|
|
{
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
|
|
|
|
rig.Updater.BeforeSampleWrite = () => throw new InvalidOperationException("the sample INSERT failed");
|
|
await Should.ThrowAsync<InvalidOperationException>(
|
|
() => rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct));
|
|
(await TrailRowsAsync(rig, "t")).ShouldBe(2, "precondition: the orphan trail row is stored");
|
|
|
|
var trail = await ModifiedAsync(rig, "t");
|
|
trail.ShouldHaveSingleItem("only the insert that happened").UpdateType.ShouldBe(HistoryModificationType.Insert);
|
|
|
|
// Positive control: the same replace, landing, is served.
|
|
rig.Updater.BeforeSampleWrite = null;
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
|
|
(await ModifiedAsync(rig, "t")).Select(v => v.UpdateType)
|
|
.ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public async Task A_change_that_landed_without_its_marker_is_still_served(bool delete)
|
|
{
|
|
// A crash between the sample change and the marker: the row stays pending, but the sample shows the change.
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
|
|
rig.Updater.SkipAppliedMarker = true;
|
|
|
|
if (delete)
|
|
(await rig.Updater.DeleteAtTimeAsync("t", [T0], UpdateTests.Who, Ct)).ShouldBe([HistorianStatusCodes.Good]);
|
|
else
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
|
|
|
|
(await ModifiedAsync(rig, "t")).Select(v => v.UpdateType).ShouldBe(
|
|
[HistoryModificationType.Insert, delete ? HistoryModificationType.Delete : HistoryModificationType.Replace]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task A_trail_insert_re_sent_after_a_transient_failure_does_not_duplicate()
|
|
{
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
await rig.Tags.Bootstrap.EnsureAsync(Ct);
|
|
var failedOnce = false;
|
|
// The first attempt APPLIES and then fails as a client timeout would: the re-send carries the same ids.
|
|
rig.Updater.AfterTrailInsert = attempt =>
|
|
{
|
|
if (attempt == 0 && !failedOnce) { failedOnce = true; throw new TimeoutException("client timeout"); }
|
|
return Task.CompletedTask;
|
|
};
|
|
|
|
(await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert,
|
|
[UpdateTests.V(T0, 1), UpdateTests.V(T0.AddSeconds(1), 2), UpdateTests.V(T0.AddSeconds(2), 3)], UpdateTests.Who, Ct))
|
|
.ShouldAllBe(s => s == HistorianStatusCodes.GoodEntryInserted);
|
|
|
|
failedOnce.ShouldBeTrue();
|
|
rig.Updater.TrailInserts.ShouldBeGreaterThanOrEqualTo(3, "pending, the re-send, then the marker");
|
|
(await TrailRowsAsync(rig, "t")).ShouldBe(3, "the re-sent rows converge on their dedup key");
|
|
(await ModifiedAsync(rig, "t")).Count.ShouldBe(3);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
[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()
|
|
{
|
|
fx.SkipIfUnavailable();
|
|
var p = fx.UniquePrefix();
|
|
var o = fx.Options(p);
|
|
var c = new QuestDbConnection(o);
|
|
await using (var conn = await c.OpenAsync(Ct))
|
|
{
|
|
foreach (var ddl in QuestDbTables.CreateAll(p, null))
|
|
{
|
|
// The trail as it was: no pending column, no dedup key.
|
|
var old = ddl.Replace(", pending BOOLEAN", "", StringComparison.Ordinal)
|
|
.Replace(" DEDUP UPSERT KEYS(ts, tag, modified_id)", "", StringComparison.Ordinal);
|
|
await using var cmd = c.CreateCommand(conn, old);
|
|
await cmd.ExecuteNonQueryAsync(Ct);
|
|
}
|
|
}
|
|
|
|
// An old-format update first (the bootstrap with AutoCreateSchema=false keeps the old format).
|
|
var legacyOptions = fx.Options(p, autoCreate: false);
|
|
var legacyConn = new QuestDbConnection(legacyOptions);
|
|
var legacyWal = new QuestDbWalVisibility(legacyConn, legacyOptions.VisibilityTimeout);
|
|
var legacyTags = new QuestDbTagCatalog(legacyConn, new QuestDbSchemaBootstrapper(legacyConn, legacyOptions), legacyWal);
|
|
var legacy = new UpdateTests.Rig(legacyConn, new QuestDbHistorianDataUpdater(legacyConn, legacyTags, legacyWal), legacyTags);
|
|
(await legacy.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct))
|
|
.ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
|
|
legacyTags.Bootstrap.SupportsTrailMarker.ShouldBeFalse();
|
|
(await ModifiedAsync(legacy, "t")).Count.ShouldBe(1);
|
|
|
|
// A boot with AutoCreateSchema=true migrates; the old row (pending reads false) is still served, a new one too.
|
|
var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
|
|
var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal);
|
|
var rig = new UpdateTests.Rig(c, new QuestDbHistorianDataUpdater(c, tags, wal), tags);
|
|
await tags.Bootstrap.EnsureAsync(Ct);
|
|
tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue();
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
|
|
(await ModifiedAsync(rig, "t")).Select(v => v.UpdateType)
|
|
.ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]);
|
|
}
|
|
}
|