fix(historian): QuestDB modified reads keep a landed change whose marker was lost; H2 review NITs (lmxopcua#778)

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
2026-11-09 16:04:44 -05:00
parent c802f0e7ed
commit e790938f94
9 changed files with 245 additions and 37 deletions
@@ -10,7 +10,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;
 ///     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.
+///     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)
@@ -87,6 +87,57 @@ public sealed class TrailMarkerTests(QuestDbFixture fx)
            [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()
    {
@@ -133,18 +184,22 @@ public sealed class TrailMarkerTests(QuestDbFixture fx)
        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)));
+            // 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);
-            foreach (var r in rows)
+            for (var j = 0; j < rows.Count; j++)
            {
-                var served = !r.Pending || (r.Id == latest.Id && landed);
+                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($"('{ts:O}','t',{r.Id},'{r.Mod:O}',1,'p',NULL,1,1.0,NULL,NULL,0,NULL,{(r.Pending ? "true" : "false")})");
+                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)
            {
@@ -153,8 +208,8 @@ public sealed class TrailMarkerTests(QuestDbFixture fx)
            }
            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);
+                // 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))
@@ -67,6 +67,34 @@ public sealed class WriteTests(SqlHistorianFixture fx)
        (await RowsAsync(s, "tag.race")).Count.ShouldBe(200);
    }

+    [Fact]
+    public async Task A_write_cancelled_mid_statement_rethrows_instead_of_failing_the_tag()
+    {
+        // lmxopcua#778: SqlClient reports a command cancelled mid-flight as a SqlException, which the writer booked as a
+        // failed write (false) — a shutdown could tip a healthy tag over its attempt budget. It must be a cancellation.
+        fx.SkipIfUnavailable();
+        var (w, s) = Make();
+        (await w.WriteLiveValuesAsync("tag.cancel", [new(T0, HistorianValue.FromDouble(0), 0)], CancellationToken.None))
+            .ShouldBeTrue("bootstrap + tag id");
+
+        await using var blocker = new SqlConnection(fx.ConnectionString);
+        var testCt = TestContext.Current.CancellationToken;
+        await blocker.OpenAsync(testCt);
+        await using var tx = (SqlTransaction)await blocker.BeginTransactionAsync(testCt);
+        await using (var hold = new SqlCommand($"SELECT COUNT(*) FROM [{s}].Sample WITH (TABLOCKX, HOLDLOCK)", blocker, tx))
+            await hold.ExecuteScalarAsync(testCt);
+
+        using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(500));
+        var ex = await Should.ThrowAsync<OperationCanceledException>(() =>
+            w.WriteLiveValuesAsync("tag.cancel", [new(T0.AddSeconds(1), HistorianValue.FromDouble(1), 0)], cts.Token));
+        ex.CancellationToken.ShouldBe(cts.Token);
+        await tx.RollbackAsync(testCt);
+
+        // Positive control: with the lock gone the same write succeeds.
+        (await w.WriteLiveValuesAsync("tag.cancel", [new(T0.AddSeconds(1), HistorianValue.FromDouble(1), 0)], CancellationToken.None))
+            .ShouldBeTrue();
+    }
+
    [Fact]
    public async Task A_SQL_failure_returns_false_instead_of_throwing()
    {
@@ -87,6 +87,39 @@ public sealed class NodeManagerHistoryReadEventsProjectionTests : IDisposable
        await host.DisposeAsync();
    }

+    [Theory]
+    [InlineData("unknown operator")]
+    [InlineData("self-referencing element")]
+    [InlineData("operand-count mismatch")]
+    public async Task An_invalid_where_clause_is_BadEventFilterInvalid(string fault)
+    {
+        // lmxopcua#778: an unknown operator made the SDK's pre-validation itself throw, failing the read with the generic
+        // BadUnexpectedError. The structural faults were already BadEventFilterInvalid and must stay so.
+        var (host, server, nm, notifier) = await BootAsync();
+        var where = new ContentFilter();
+        switch (fault)
+        {
+            case "unknown operator":
+                where.Push((FilterOperator)99, new LiteralOperand(new Variant(1)), new LiteralOperand(new Variant(1)));
+                break;
+            case "self-referencing element":
+                where.Push(FilterOperator.Not, new ElementOperand(0));
+                break;
+            default:
+                where.Push(FilterOperator.Equals, new LiteralOperand(new Variant(1)));
+                break;
+        }
+
+        var ex = Should.Throw<ServiceResultException>(() => ReadStatus(server, nm, notifier, Select("EventId"), where));
+        ex.StatusCode.ShouldBe(StatusCodes.BadEventFilterInvalid);
+
+        // Positive control: the same read with a valid where-clause is served.
+        var valid = new ContentFilter();
+        valid.Push(FilterOperator.Equals, new LiteralOperand(new Variant(1)), new LiteralOperand(new Variant(1)));
+        StatusCode.IsGood(ReadStatus(server, nm, notifier, Select("EventId"), valid).Status).ShouldBeTrue();
+        await host.DisposeAsync();
+    }
+
    private static string[] Ids(HistoryEvent h) =>
        h.Events.Select(e => System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value)).ToArray();

@@ -107,6 +140,14 @@ public sealed class NodeManagerHistoryReadEventsProjectionTests : IDisposable

    private static HistoryEvent Read(OtOpcUaSdkServer server, OtOpcUaNodeManager nm, NodeId notifier,
        SimpleAttributeOperandCollection select, ContentFilter? where)
+    {
+        var (status, result) = ReadStatus(server, nm, notifier, select, where);
+        StatusCode.IsGood(status).ShouldBeTrue($"HistoryRead failed: {status}");
+        return (HistoryEvent)ExtensionObject.ToEncodeable(result.HistoryData);
+    }
+
+    private static (StatusCode Status, SdkHistoryReadResult Result) ReadStatus(OtOpcUaSdkServer server, OtOpcUaNodeManager nm,
+        NodeId notifier, SimpleAttributeOperandCollection select, ContentFilter? where)
    {
        var details = new ReadEventDetails
        {
@@ -120,8 +161,7 @@ public sealed class NodeManagerHistoryReadEventsProjectionTests : IDisposable
        var errors = new List<ServiceResult> { null! };
        nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false,
            new List<HistoryReadValueId> { new() { NodeId = notifier } }, results, errors);
-        StatusCode.IsGood(errors[0].StatusCode).ShouldBeTrue($"HistoryRead failed: {errors[0].StatusCode}");
-        return (HistoryEvent)ExtensionObject.ToEncodeable(results[0].HistoryData);
+        return (errors[0].StatusCode, results[0]);
    }

    private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server, OtOpcUaNodeManager Nm, NodeId Notifier)> BootAsync()