From e790938f94f7994ba51bbc1b9e2145d2d8d90e2b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 16:04:44 -0500 Subject: [PATCH] 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 --- docs/Historian.md | 17 +++-- .../QuestDbHistorianDataSource.cs | 39 +++++++--- .../QuestDbHistorianDataUpdater.cs | 19 +++-- .../SqlServerHistorianDataUpdater.cs | 8 ++- .../SqlServerHistorianValueWriter.cs | 6 ++ .../OtOpcUaNodeManager.cs | 50 +++++++++++++ .../TrailMarkerTests.cs | 71 ++++++++++++++++--- .../WriteTests.cs | 28 ++++++++ ...ManagerHistoryReadEventsProjectionTests.cs | 44 +++++++++++- 9 files changed, 245 insertions(+), 37 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index 9f84c1ea..ed286b6f 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -599,7 +599,10 @@ Every backend stores the three (SqlServer `AlarmTypeName` / `AlarmName` / `Alarm honoured:** each event is evaluated against it with the SDK's content filter — `OfType` through the server's type tree (a supertype keeps its subtypes), any other operator against the projected fields. Filtering happens after a page's continuation point is taken from the unfiltered page, so paging is unchanged: a filtered page may hold fewer events (even -none) and still carry a continuation point. An unsupported select operand returns a null field (spec-conformant). +none) and still carry a continuation point. An unsupported select operand returns a null field (spec-conformant). An +event filter the SDK's validation rejects (an operand-count mismatch, a dangling or self-referencing element operand, an +unknown operator, …) fails the read with `BadEventFilterInvalid` before any read (#778 — an unknown operator used to fail it with the +generic `BadUnexpectedError`). ### No driver-side history (Gitea #655) @@ -1128,10 +1131,12 @@ modification time and the Part 11 `HistoryUpdateType`. A HistoryReadRaw with `Is the change: a crash (or failed sample INSERT) in between leaves an **orphan trail row** describing a change that never landed — never a change without its trail row. A failed sample write after its trail rows landed logs a **Warning** naming the tag, the row count and the time range. **Applied marker:** the trail row is written `pending = true` and, once the change has landed, re-upserted `pending = false` on - the table's `DEDUP UPSERT KEYS(ts, tag, modified_id)`. A modified read serves a still-`pending` row only when it is the **latest** trail row at - its time and the sample there (a tombstone included) was written at or after the modification — a crash between the change and its marker. - Otherwise it is the orphan of a change that never landed and is **not served** (it used to be served as a change that happened). Residual: - a double fault — a change that landed without its marker, followed by a later change of the same sample — hides the earlier one. + the table's `DEDUP UPSERT KEYS(ts, tag, modified_id)`. A modified read serves a still-`pending` row only when the evidence says its change + landed (a crash or a failed marker write between the change and its marker): as the **latest** trail row at its time, the sample there (a + tombstone included) was written at or after the modification; below a later trail row at that time, that **next row recorded a different + prior value** (kind, value, status or the prior's write time — the next change saw what this one wrote). Otherwise it is the orphan of a + change that never landed and is **not served** (it used to be served as a change that happened). #778: the second rule is new — a change + that landed without its marker used to vanish from modified history as soon as a later change at the same time arrived. - **Rolling upgrade (SqlServer).** The v2 schema is applied by the first node that upgrades. Until its peer is upgraded too, that not-yet-upgraded node still runs the old build, whose HistoryUpdates write **no trail rows** — silent gaps in the audit trail. Upgrade both nodes of a pair together (stop, upgrade, start both); modifications made by an old-build node in the window are never recorded. @@ -2085,7 +2090,7 @@ present **and not deleted**. current value (read by the same existence query), then writes the change. No multi-table transaction exists, so this order is the guarantee: a crash between the two leaves an **orphan trail row**, never an untracked change — the documented deviation from `SqlServer`'s atomic trail. The row is written `pending` and marked applied after the change, so a modified read does not serve the orphan of a change that never -landed. See [Modified-value history](#modified-value-history-part-11). +landed — while a change that landed but lost its marker stays served (#778). See [Modified-value history](#modified-value-history-part-11). The per-item codes come from the shared `HistoryUpdatePlanner`, so they equal `SqlServer`'s (including the duplicate-timestamp-in-one-request rules above). **Every read and aggregate filters `NOT deleted`.** A client Insert at a deleted timestamp re-populates it (the OPC UA intent); only the diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index cd21f6dd..172561b0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -175,7 +175,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource // (ts, modified_id); maxValues + 1 rows are fetched to report HasMore exactly. A row whose status is // NULL had no prior value (an insert) and is served null / BadNoData. Because the trail row lands // first, a crash between it and the sample write leaves an orphan row describing a change that never - // applied (documented). + // applied; on a marker-format trail such a row stays `pending` and is reconciled below. /// public async Task ReadModifiedAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter, @@ -211,20 +211,37 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource "modification_ts, update_type, principal"; // The applied marker: a trail row still `pending` is served only when the change it describes evidently - // landed — it is the LATEST row at its time and the sample there (a tombstone included) was written at or - // after the modification: a crash between the change and its marker. Otherwise it is the orphan of a change - // that never landed (typically followed by the client's successful retry, which is then the latest row) and - // is dropped. The ASOF JOIN reads the tag's samples of the window only (indexed tag, inclusive bounds); the - // resume position filters OUTSIDE the window function so it never cuts a time's partition. + // landed. Two pieces of evidence, one per position in its time's trail: + // - it is the LATEST row at its time and the sample there (a tombstone included) was written at or after + // the modification — a crash, or a failed marker write, between the change and its marker; + // - it is NOT the latest and the NEXT row at its time recorded a different prior value — the next change + // saw what this one wrote (lmxopcua#778: V0→V1 lands, its marker fails, a later V1→V2 must not hide it). + // A change that never landed left the sample alone, so the next row records the same prior value as + // this one (kind, value columns, status and the prior's server_ts — a rewrite always moves server_ts), + // and the row is the orphan of that failed change (typically followed by the client's successful + // retry) and is dropped. + // QuestDB window functions do not take VARCHAR, so the next row's prior is fetched by a LEFT JOIN on + // (ts, next modified_id) rather than lead(). The ASOF JOIN reads the tag's samples of the window only + // (indexed tag, inclusive bounds); the resume position filters OUTSIDE the window functions so it never + // cuts a time's partition. var sql = _bootstrap.SupportsTrailMarker - ? $"SELECT {Columns} FROM (SELECT m.ts ts, m.kind kind, m.num_value num_value, m.int_value int_value, " + + ? $"SELECT {Columns} FROM (SELECT x.ts ts, x.kind kind, x.num_value num_value, x.int_value int_value, " + + "x.str_value str_value, x.status status, x.server_ts server_ts, x.modified_id modified_id, " + + "x.modification_ts modification_ts, x.update_type update_type, x.principal principal, " + + "(NOT x.pending OR (x.next_id = NULL AND x.sample_ts = x.ts AND x.sample_server_ts >= x.modification_ts) " + + "OR (x.next_id != NULL AND NOT (n.kind = x.kind AND n.num_value = x.num_value AND n.int_value = x.int_value " + + "AND n.str_value = x.str_value AND n.status = x.status AND n.server_ts = x.server_ts))) served " + + "FROM (SELECT m.ts ts, m.kind kind, m.num_value num_value, m.int_value int_value, " + "m.str_value str_value, m.status status, m.server_ts server_ts, m.modified_id modified_id, " + "m.modification_ts modification_ts, m.update_type update_type, m.principal principal, m.pending pending, " + - "s.ts sample_ts, s.server_ts sample_server_ts, max(m.modified_id) OVER (PARTITION BY m.ts) latest_id " + + "s.ts sample_ts, s.server_ts sample_server_ts, " + + "lead(m.modified_id) OVER (PARTITION BY m.ts ORDER BY m.modified_id) next_id " + $"FROM {_conn.SampleModifiedTable} m ASOF JOIN (SELECT ts, tag, server_ts FROM {_conn.SampleTable} " + - $"WHERE tag = $1 AND ts >= $2 AND ts <= $3) s ON (tag) WHERE m.tag = $1 AND {window.Replace("ts ", "m.ts ")}) " + - "WHERE (NOT pending OR (sample_ts = ts AND sample_server_ts >= modification_ts AND modified_id = latest_id))" + - $"{resume} ORDER BY ts {order}, modified_id {order}" + $"WHERE tag = $1 AND ts >= $2 AND ts <= $3) s ON (tag) WHERE m.tag = $1 AND {window.Replace("ts ", "m.ts ")}) x " + + "LEFT JOIN (SELECT ts, modified_id, kind, num_value, int_value, str_value, status, server_ts " + + $"FROM {_conn.SampleModifiedTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3) n " + + "ON (x.ts = n.ts AND x.next_id = n.modified_id)) " + + $"WHERE served{resume} ORDER BY ts {order}, modified_id {order}" : $"SELECT {Columns} FROM {_conn.SampleModifiedTable} " + $"WHERE tag = $1 AND {window}{resume} ORDER BY ts {order}, modified_id {order}"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs index fd6e1ae8..69c797ba 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -35,9 +35,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// Applied marker. On a trail table with the marker format /// () the trail row is written pending and, once the /// sample change has landed, re-upserted with pending = false (same (ts, tag, modified_id) — the table's -/// dedup key). A modified read drops a pending row unless the sample at its time was written at or after the -/// modification (the change landed but the marker did not — a crash in between), so an orphan is no longer served as -/// a change that happened. The trail INSERT is re-sent once (same ids) after a transient failure: a first attempt that +/// dedup key). A modified read drops a pending row unless it can tell the change landed: as the latest row at its +/// time, the sample there was written at or after the modification (a crash or a failed marker in between); below a +/// later row at that time, that next row recorded a different prior value (the next change saw this one's result — +/// lmxopcua#778). So an orphan is no longer served as a change that happened, and a change whose marker was lost stays +/// served after later changes at the same time. The trail INSERT is re-sent once (same ids) after a transient failure: a first attempt that /// did apply converges through the dedup key instead of duplicating. /// internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater @@ -361,8 +363,10 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater /// /// The applied marker: re-upserts the trail rows of a change that has landed with pending = false (same - /// (ts, tag, modified_id) key). A failure here does not fail the update — the change landed, and a modified read - /// still serves a pending row whose sample shows the change (logged as a Warning). + /// (ts, tag, modified_id) key). A failure here does not fail the update — the change landed — and is logged as + /// a Warning. Its rows stay pending; a modified read serves each one only while the evidence holds: as the + /// latest row at its time, when the sample there was written at or after the modification; once a later change at + /// that time exists, when that change's trail row recorded a different prior value (lmxopcua#778). /// private async Task MarkAppliedAsync(string tag, IReadOnlyList rows, CancellationToken ct) { @@ -375,8 +379,9 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater } catch (Exception ex) when (ex is not OperationCanceledException) { - _log.LogWarning(ex, "QuestDB applied marker for {Count} trail row(s) of tag {Tag} was not written; the change landed and " + - "modified reads still serve it.", rows.Count, tag); + _log.LogWarning(ex, "QuestDB applied marker for {Count} trail row(s) of tag {Tag} was not written; the change landed, but " + + "the rows stay pending and modified reads serve them only while the sample at their time, or the next change's " + + "recorded prior value, shows the change.", rows.Count, tag); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs index 697b463a..8b10e566 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataUpdater.cs @@ -126,7 +126,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater await tx.CommitAsync(ct).ConfigureAwait(false); } rows += batch; - DeleteRawBatches++; + Interlocked.Increment(ref _deleteRawBatches); if (batch < DeleteRawBatchSize) break; } return HistoryUpdatePlanner.DeleteRawResult(tagKnown: true, rows); @@ -138,8 +138,10 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater /// internal int DeleteRawBatchSize { get; set; } = 4000; - /// DeleteRaw batches run so far (lifetime) — test/diagnostic. - internal int DeleteRawBatches { get; private set; } + private int _deleteRawBatches; + + /// DeleteRaw batches run so far (lifetime) — test/diagnostic. Atomic: concurrent DeleteRaw calls share it. + internal int DeleteRawBatches => Volatile.Read(ref _deleteRawBatches); /// public async Task> DeleteAtTimeAsync(string tag, IReadOnlyList timesUtc, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs index 1dc58c39..cb026202 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianValueWriter.cs @@ -95,6 +95,12 @@ internal sealed class SqlServerHistorianValueWriter : IHistorianValueWriter { throw; // the drain pass was cancelled (shutdown) — not a write failure of this tag } + catch (SqlException ex) when (ct.IsCancellationRequested) + { + // SqlClient reports a command cancelled mid-flight as a SqlException ("Operation cancelled by user"), not an + // OperationCanceledException: still the drain pass's cancellation, never a write failure of this tag. + throw new OperationCanceledException("The SQL historian live-value write was cancelled.", ex, ct); + } catch (Exception ex) { _log.LogWarning("SQL historian live-value write failed ({Exception}); will retry.", ex.GetType().Name); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index b15726a1..6383b1f1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -5703,6 +5703,56 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 NormalizeHistoryValues(nodesToProcess, results); } + /// + /// + /// The SDK validates a HistoryReadEvents event filter before runs and throws the + /// validation's status — BadEventFilterInvalid for a structural fault (operand count, a dangling element + /// operand) — but when the validation itself throws (an unknown filter operator) the request failed with the generic + /// BadUnexpectedError. Any failure of that validation is re-thrown as BadEventFilterInvalid (#778); + /// every other refusal (a bad time range…) and a valid filter pass through unchanged. + /// + protected override void HistoryRead( + ServerSystemContext context, + HistoryReadDetails details, + TimestampsToReturn timestampsToReturn, + bool releaseContinuationPoints, + IList nodesToRead, + IList results, + IList errors, + List nodesToProcess, + IDictionary cache) + { + try + { + base.HistoryRead(context, details, timestampsToReturn, releaseContinuationPoints, nodesToRead, results, errors, + nodesToProcess, cache); + } + catch (ServiceResultException ex) when (ex.StatusCode != StatusCodes.BadEventFilterInvalid + && details is ReadEventDetails { Filter: { } filter } + && !IsValidEventFilter(filter, context)) + { + throw ServiceResultException.Create(StatusCodes.BadEventFilterInvalid, ex, + "The HistoryReadEvents event filter is invalid: {0}", ex.Message); + } + } + + /// The SDK's validation of a HistoryReadEvents event filter; a throw counts as invalid. + /// The event filter. + /// The request's context. + /// true when the filter validates Good. + private bool IsValidEventFilter(EventFilter filter, ServerSystemContext context) + { + try + { + var filterContext = new FilterContext(Server.NamespaceUris, Server.TypeTree, context.PreferredLocales, Server.Telemetry); + return ServiceResult.IsGood(filter.Validate(filterContext).Status); + } + catch (ServiceResultException) + { + return false; + } + } + /// protected override void HistoryReadEvents( ServerSystemContext context, 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 a985b742..98be2e87 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 @@ -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 pending, /// marked applied after the change, keyed (ts, tag, modified_id); 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. /// [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( + () => 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)) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs index 60582f6f..5948460a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/WriteTests.cs @@ -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(() => + 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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsProjectionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsProjectionTests.cs index 2ec7d39c..07b6eeba 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsProjectionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsProjectionTests.cs @@ -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(() => 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 { null! }; nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, new List { 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()