Joseph Doherty
e790938f94
QuestDB modified-history reads dropped every still-pending trail row that was not the latest at its source time, so V0->V1 (applied marker never written) followed by V1->V2 hid the V0->V1 change for good. A pending row below the latest is now served when the NEXT trail row at that time recorded a different prior value (kind, value, status or the prior's server_ts) - the next change saw what this one wrote; an orphan of a change that never landed leaves the same prior for the next row and stays hidden. QuestDB window functions reject VARCHAR, so the next row's prior comes from a LEFT JOIN on (ts, lead(id)). The updater's comment and marker-failure log no longer claim modified reads "still serve it" unconditionally. Regression tests (QuestDB 10.0.1 throwaway container): marker lost then a replace / delete at the same time; an orphan followed by a marker-lost retry then a later change (orphan still dropped); the random marker-mix fuzz now varies the recorded priors. All four fail on the old filter. NITs: - SqlServerHistorianValueWriter: a SqlException raised while the token is cancelled (SqlClient's "Operation cancelled by user") is rethrown as OperationCanceledException instead of booked as a failed write. Integration test holds a TABLOCKX on Sample and cancels mid-INSERT; it returned false before. - SqlServerHistorianDataUpdater.DeleteRawBatches is an Interlocked counter. - HistoryReadEvents: an event filter whose SDK validation throws (unknown operator) failed with the generic BadUnexpectedError; the node manager now re-throws any failed event-filter validation as BadEventFilterInvalid. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
298 lines
17 KiB
C#
298 lines
17 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 (latest row) or the next row's recorded prior value (any earlier row, lmxopcua#778) 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]);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public async Task A_change_whose_marker_was_lost_stays_served_after_a_later_change_at_its_time(bool delete)
|
|
{
|
|
// lmxopcua#778: V0→V1 lands but its applied marker is never written; a later V1→V2 (or a delete) at the same time
|
|
// used to hide it — the pending row was no longer the latest at its time, so it was dropped as an orphan.
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 0)], UpdateTests.Who, Ct);
|
|
rig.Updater.SkipAppliedMarker = true;
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
|
|
rig.Updater.SkipAppliedMarker = false;
|
|
(await ModifiedAsync(rig, "t")).Count.ShouldBe(2, "precondition: served while it is the latest row");
|
|
|
|
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);
|
|
|
|
var trail = await ModifiedAsync(rig, "t");
|
|
trail.Select(v => v.UpdateType).ShouldBe(
|
|
[HistoryModificationType.Insert, HistoryModificationType.Replace, delete ? HistoryModificationType.Delete : HistoryModificationType.Replace]);
|
|
trail.Select(v => v.Value.Value).ShouldBe([null, 0.0, 1.0], "each row carries the value its change replaced");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task An_orphan_followed_by_a_landed_change_whose_marker_was_lost_is_still_dropped()
|
|
{
|
|
// The other side of lmxopcua#778's rule: V0→V1 never lands (orphan), then the retry V0→V1 lands without its marker,
|
|
// then V1→V2. The orphan's next row recorded the SAME prior value (V0) ⇒ dropped; the retry's next row recorded V1
|
|
// ⇒ served.
|
|
fx.SkipIfUnavailable();
|
|
var rig = UpdateTests.MakeRig(fx);
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 0)], 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, 1)], UpdateTests.Who, Ct));
|
|
rig.Updater.BeforeSampleWrite = null;
|
|
rig.Updater.SkipAppliedMarker = true;
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
|
|
rig.Updater.SkipAppliedMarker = false;
|
|
await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
|
|
(await TrailRowsAsync(rig, "t")).ShouldBe(4, "precondition: the orphan trail row is stored");
|
|
|
|
var trail = await ModifiedAsync(rig, "t");
|
|
trail.Select(v => v.UpdateType)
|
|
.ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace, HistoryModificationType.Replace]);
|
|
trail.Select(v => v.Value.Value).ShouldBe([null, 0.0, 1.0]);
|
|
}
|
|
|
|
[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);
|
|
// Each row's recorded prior value is 1.0 or 2.0: a pending row below the latest is served exactly when the
|
|
// next row recorded a different prior (lmxopcua#778).
|
|
var rows = new List<(long Id, bool Pending, DateTime Mod, double Prior)>();
|
|
for (var k = rng.Next(1, 5); k > 0; k--) rows.Add((id++, rng.Next(3) == 0, mod0.AddSeconds(id), rng.Next(2) + 1.0));
|
|
// 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);
|
|
for (var j = 0; j < rows.Count; j++)
|
|
{
|
|
var r = rows[j];
|
|
var served = !r.Pending || (r.Id == latest.Id ? landed : rows[j + 1].Prior != r.Prior);
|
|
if (served) expected.Add((ts, r.Id));
|
|
if (trailSql.Length > 0) trailSql.Append(',');
|
|
trailSql.Append(FormattableString.Invariant(
|
|
$"('{ts:O}','t',{r.Id},'{r.Mod:O}',1,'p',NULL,1,{r.Prior:0.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: the latest row, if pending, is not served (earlier rows are unaffected).
|
|
expected.RemoveAll(e => e.Ts == ts && e.Id == latest.Id && latest.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]);
|
|
}
|
|
}
|