From 7c80d36eb6071d47f31995431bd21955f78f5c11 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 11 Oct 2026 09:37:17 -0400 Subject: [PATCH] fix(historian): #668 review follow-ups (explicit modified source time, paging tests, ACL test, QuestDB orphan warning, docs) - HistoryModifiedValue gains explicit SourceTimeUtc (SqlServer + QuestDB populate it); the node manager's resume cursor uses it instead of the prior value's nullable source timestamp (MinValue for inserts). - ACL test now uses a trail-returning data source and asserts Good + data on the allow leg. - Node-manager tests: end-only backward open-ended paging, reverse resume cursor, null-prior cursor. - QuestDB updater logs a Warning (tag + time range) when a sample write fails after its trail rows landed. - docs/Historian.md: rolling-upgrade window, IDENTITY-vs-commit ordering, QuestDB cross-node ModifiedId clock ordering. --- docs/Historian.md | 11 ++++ .../Historian/HistoryModifiedContracts.cs | 7 ++- .../QuestDbHistorianDataSource.cs | 2 +- .../QuestDbHistorianDataUpdater.cs | 31 +++++++++- .../SqlServerHistorianDataSource.cs | 2 +- .../OtOpcUaNodeManager.cs | 2 +- .../HistoryUpdateOverTheWireTests.cs | 2 +- .../NodeManagerAclEnforcementTests.cs | 38 +++++++++++- .../NodeManagerHistoryReadModifiedTests.cs | 58 +++++++++++++++++-- 9 files changed, 138 insertions(+), 15 deletions(-) diff --git a/docs/Historian.md b/docs/Historian.md index 0e66252c..f4c1d8e4 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -673,6 +673,17 @@ modification time and the Part 11 `HistoryUpdateType`. A HistoryReadRaw with `Is so a change and its trail row commit or roll back together. QuestDB has no multi-table transaction, so the trail row is written **before** 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. Treat a QuestDB trail entry whose sample still shows the prior value as such an orphan. + A failed sample write after its trail rows landed logs a **Warning** naming the tag, the row count and the time range, so the orphan + can be found. +- **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. +- **Paging vs a concurrent HistoryUpdate (SqlServer).** `ModifiedId` is an `IDENTITY`, which orders by allocation, not by commit. A modified + read paging at the same moment a HistoryUpdate commits a modification of the **same source timestamp** can have already passed that + row's id and so never see it on this pass (a fresh read does). The cursor is exact for a quiescent trail, not a snapshot guarantee. +- **QuestDB `ModifiedId` across nodes.** Ids are UTC ticks bumped past the last one issued by that process, so a pair's ids interleave by + each node's **wall clock**. Same-source-time modifications made on two nodes order by their clocks (clock skew can reorder them); + within one node the order is strict. - **Retention.** The trail ages out on the same **source-time** cutoff as the samples it describes: `SqlServer` purges `hist.SampleModified` per tag alongside `hist.Sample`; QuestDB puts the same TTL on `{p}sample_modified`. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryModifiedContracts.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryModifiedContracts.cs index aa1c03f7..d3ad6b4c 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryModifiedContracts.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryModifiedContracts.cs @@ -51,12 +51,17 @@ public enum HistoryModificationType : byte /// The kind of modification. /// The principal that made it. /// The backend's ordering key within one source timestamp (paging cursor). +/// +/// The modified sample's source time as stored on the trail row — the paging cursor's time component. Explicit +/// because an insert's has no prior value and so cannot be relied on to carry it. +/// public sealed record HistoryModifiedValue( DataValueSnapshot Value, DateTime ModificationTimeUtc, HistoryModificationType UpdateType, string UserName, - long ModifiedId); + long ModifiedId, + DateTime SourceTimeUtc); /// Resume position for a paged modified-value read: strictly after (or before, reversed) this key. /// The last returned modification's source time. 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 95eb33fb..3f6b939b 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 @@ -221,7 +221,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource values.Add(new HistoryModifiedValue(prior, modifiedAt, (HistoryModificationType)QuestDbConnection.ReadKind(reader, 9), reader.IsDBNull(10) ? string.Empty : reader.GetString(10), - reader.GetInt64(7))); + reader.GetInt64(7), sourceTime)); } } 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 91cdf513..91b060ed 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 @@ -155,7 +155,8 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater // tombstone without its trail row. await WriteTrailAsync(tag, HistoryModificationType.Delete, priors.Select(r => (r.TimeUtc, (LiveRow?)r)).ToList(), context, ct).ConfigureAwait(false); - await WriteTombstonesAsync(tag, page, ct).ConfigureAwait(false); + await AfterTrailAsync(tag, page[0], page[^1], page.Count, + () => WriteTombstonesAsync(tag, page, ct)).ConfigureAwait(false); // The wait between pages is required: without it the next page could re-read rows just tombstoned. await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); total += page.Count; @@ -197,7 +198,9 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater // Trail BEFORE the change (see the class remarks). await WriteTrailAsync(tag, HistoryModificationType.Delete, chunk.Select(t => (live[t].TimeUtc, (LiveRow?)live[t])).ToList(), context, ct).ConfigureAwait(false); - await WriteTombstonesAsync(tag, chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(), ct).ConfigureAwait(false); + var chunkTimes = chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(); + await AfterTrailAsync(tag, chunkTimes[0], chunkTimes[^1], chunkTimes.Count, + () => WriteTombstonesAsync(tag, chunkTimes, ct)).ConfigureAwait(false); } if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); deleted.UnionWith(toDelete); @@ -231,7 +234,8 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater // Trail BEFORE the change (see the class remarks). await WriteTrailAsync(tag, ToModificationType(mode), trail, context, ct).ConfigureAwait(false); foreach (var chunk in toWrite.Chunk(BatchSize)) - await WriteRowsAsync(tag, chunk, ct).ConfigureAwait(false); + await AfterTrailAsync(tag, chunk[0].TimeUtc, chunk[^1].TimeUtc, chunk.Length, + () => WriteRowsAsync(tag, chunk, ct)).ConfigureAwait(false); if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); return outcome; } @@ -367,6 +371,27 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false); } + /// + /// Runs the sample write that follows an already-written trail batch. QuestDB has no multi-table transaction, so a + /// failure here leaves an ORPHAN trail row (a recorded modification that never landed); it is logged with the tag + /// and time range so an operator can reconcile it, then rethrown unchanged. + /// + private async Task AfterTrailAsync(string tag, DateTime from, DateTime to, int count, Func write) + { + try + { + await write().ConfigureAwait(false); + } + catch (Exception ex) + { + _log.LogWarning(ex, + "QuestDB sample write failed AFTER its {Table} trail rows were written; {Count} orphan audit row(s) for tag {Tag} " + + "in [{From:O}, {To:O}] describe a modification that did not land.", + _conn.SampleModifiedTable, count, tag, from, to); + throw; + } + } + private Task WriteRowsAsync(string tag, IReadOnlyList rows, CancellationToken ct) => InsertAsync(rows.Count, ct, (cmd, i) => { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index 530192c7..99182de6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -212,7 +212,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource } values.Add(new HistoryModifiedValue( - prior, modifiedAt, (HistoryModificationType)reader.GetByte(9), reader.GetString(10), reader.GetInt64(7))); + prior, modifiedAt, (HistoryModificationType)reader.GetByte(9), reader.GetString(10), reader.GetInt64(7), sourceTime)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 33d623f7..0c2780b8 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -6218,7 +6218,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { var last = page.Values[^1]; outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState( - tagname, last.Value.SourceTimestampUtc ?? DateTime.MinValue, endUtc, 0, numValuesPerNode, + tagname, last.SourceTimeUtc, endUtc, 0, numValuesPerNode, HistoryCursorKind.Modified, last.ModifiedId, startUtc)); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs index 76e6883f..16f9ccc4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs @@ -313,7 +313,7 @@ public sealed class HistoryUpdateOverTheWireTests private void Record(DateTime t, double? prior, HistoryModificationType type, HistoryUpdateContext who) => _trail.Add(new HistoryModifiedValue( new DataValueSnapshot(prior, prior is null ? StatusCodes.BadNoData : StatusCodes.Good, t, t), - DateTime.UtcNow, type, who.Principal, _trail.Count + 1)); + DateTime.UtcNow, type, who.Principal, _trail.Count + 1, t)); public int UpdateCalls => Volatile.Read(ref _updateCalls); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerAclEnforcementTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerAclEnforcementTests.cs index 65ed1720..9dedcab8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerAclEnforcementTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerAclEnforcementTests.cs @@ -101,7 +101,7 @@ public sealed class NodeManagerAclEnforcementTests : IDisposable try { var nm = server.NodeManager!; - nm.HistorianDataSource = NullHistorianDataSource.Instance; + nm.HistorianDataSource = new OneModificationTrail(); var nodeId = Historized(nm, "Plant/Modbus/dev1/Speed"); nm.Authorizer = Authorizer(enabled: true, GrantOn(NodePermissions.Read)); @@ -110,8 +110,14 @@ public sealed class NodeManagerAclEnforcementTests : IDisposable deniedResults[0].StatusCode.Code.ShouldBe(StatusCodes.BadUserAccessDenied); nm.Authorizer = Authorizer(enabled: true, GrantOn(NodePermissions.HistoryRead)); - var (_, allowedErrors) = ReadRaw(nm, Identity(), modified: true, nodeId); - allowedErrors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + var (allowedResults, allowedErrors) = ReadRaw(nm, Identity(), modified: true, nodeId); + // A real trail entry must come back: Null's BadHistoryOperationUnsupported is also what the SDK pre-seeds, + // so asserting it would pass even if the grant never reached the modified branch. + allowedErrors[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + allowedResults[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + var data = ExtensionObject.ToEncodeable(allowedResults[0].HistoryData).ShouldBeOfType(); + data.DataValues.Count.ShouldBe(1); + data.ModificationInfos[0].UserName.ShouldBe("alice"); } finally { await host.DisposeAsync(); } } @@ -327,6 +333,32 @@ public sealed class NodeManagerAclEnforcementTests : IDisposable } } + /// A backend whose modified-value trail holds exactly one entry; every other read is empty. + private sealed class OneModificationTrail : IHistorianDataSource + { + public Task ReadModifiedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter, + CancellationToken cancellationToken) + { + var t = DateTime.UtcNow.AddMinutes(-5); + var prior = new DataValueSnapshot(1.5, StatusCodes.Good, t, t); + return Task.FromResult(new HistoryModifiedReadResult( + [new HistoryModifiedValue(prior, DateTime.UtcNow, HistoryModificationType.Replace, "alice", 1, t)], false)); + } + + public Task ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); + public Task ReadProcessedAsync( + string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList t, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct); + public Task ReadEventsAsync(string? n, DateTime s, DateTime e, int m, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadEventsAsync(n, s, e, m, ct); + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + public void Dispose() { } + } + private static NodeId Historized(OtOpcUaNodeManager nm, string rawPath) { nm.EnsureVariable(rawPath, parentFolderNodeId: null, displayName: rawPath, dataType: "Float", diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs index dd2ee07c..fc12cc7f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadModifiedTests.cs @@ -105,6 +105,56 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable await host.DisposeAsync(); } + [Fact] + public async Task An_end_only_modified_read_pages_backward_from_the_end() + { + var trail = Enumerable.Range(1, 9).Select(i => Mod(i, seconds: i, prior: i, HistoryModificationType.Update, "u", Epoch)).ToList(); + var (host, nm, node, src) = await SetupAsync(trail); + + // StartTime unset + EndTime set + a cap = "the newest N, walking backward". + Drain(nm, node, DateTime.MinValue, Epoch.AddHours(1), max: 4).ShouldBe(src.Ordered(reverse: true).Select(m => m.ModifiedId)); + src.LastStart.ShouldBe(Epoch.AddHours(1)); + src.LastEnd.ShouldBe(new DateTime(0, DateTimeKind.Utc)); + await host.DisposeAsync(); + } + + [Fact] + public async Task A_reverse_resume_hands_the_backend_the_last_returned_source_time_and_id() + { + // Ids 3 and 4 share a source time; page 1 (reverse, cap 3) ends on id 4, then resumes mid-run. + var trail = new List + { + Mod(1, 1, 1, HistoryModificationType.Update, "u", Epoch), Mod(2, 2, 2, HistoryModificationType.Update, "u", Epoch), + Mod(3, 3, 3, HistoryModificationType.Update, "u", Epoch), Mod(4, 3, 4, HistoryModificationType.Update, "u", Epoch), + Mod(5, 5, 5, HistoryModificationType.Update, "u", Epoch), Mod(6, 6, 6, HistoryModificationType.Update, "u", Epoch), + }; + var (host, nm, node, src) = await SetupAsync(trail); + + var p1 = Read(nm, node, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 3, cp: null); + p1.Ids.ShouldBe([6, 5, 4]); + p1.Cp.ShouldNotBeNull(); + var p2 = Read(nm, node, Epoch.AddHours(1), Epoch.AddSeconds(-1), max: 3, cp: p1.Cp); + src.LastCursor.ShouldBe(new HistoryModifiedCursor(Epoch.AddSeconds(3), 4)); + p2.Ids.ShouldBe([3, 2, 1]); + await host.DisposeAsync(); + } + + [Fact] + public async Task The_resume_cursor_uses_the_explicit_source_time_even_when_the_prior_value_carries_none() + { + // Inserts have no prior value; their snapshot has no source timestamp. The cursor must not fall back to MinValue. + var trail = Enumerable.Range(1, 6).Select(i => new HistoryModifiedValue( + new DataValueSnapshot(null, StatusCodes.BadNoData, null, Epoch), Epoch, HistoryModificationType.Insert, "u", i, + Epoch.AddSeconds(i))).ToList(); + var (host, nm, node, src) = await SetupAsync(trail); + + var p1 = Read(nm, node, Epoch, Epoch.AddHours(1), max: 3, cp: null); + p1.Cp.ShouldNotBeNull(); + Read(nm, node, Epoch, Epoch.AddHours(1), max: 3, cp: p1.Cp); + src.LastCursor.ShouldBe(new HistoryModifiedCursor(Epoch.AddSeconds(3), 3)); + await host.DisposeAsync(); + } + [Fact] public async Task An_open_ended_forward_modified_read_pages_to_the_end() { @@ -172,7 +222,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable var value = prior is null ? new DataValueSnapshot(null, StatusCodes.BadNoData, t, modifiedAt) : new DataValueSnapshot(prior.Value, StatusCodes.Good, t, t); - return new HistoryModifiedValue(value, modifiedAt, type, user, id); + return new HistoryModifiedValue(value, modifiedAt, type, user, id, t); } private static List Drain(OtOpcUaNodeManager nm, NodeId node, DateTime start, DateTime end, uint max) @@ -258,8 +308,8 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable public static long IdOf(DataValue v) => v.ServerTimestamp.Ticks - Epoch.Ticks; public IEnumerable Ordered(bool reverse) => reverse - ? trail.OrderByDescending(m => m.Value.SourceTimestampUtc).ThenByDescending(m => m.ModifiedId) - : trail.OrderBy(m => m.Value.SourceTimestampUtc).ThenBy(m => m.ModifiedId); + ? trail.OrderByDescending(m => m.SourceTimeUtc).ThenByDescending(m => m.ModifiedId) + : trail.OrderBy(m => m.SourceTimeUtc).ThenBy(m => m.ModifiedId); public async Task ReadModifiedAsync( string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter, @@ -273,7 +323,7 @@ public sealed class NodeManagerHistoryReadModifiedTests : IDisposable var (lo, hi) = reverse ? (endUtc, startUtc) : (startUtc, endUtc); var rows = Ordered(reverse).Where(m => { - var t = m.Value.SourceTimestampUtc!.Value; + var t = m.SourceTimeUtc; var inWindow = startUtc == endUtc ? t == lo : reverse ? t > lo && t <= hi : t >= lo && t < hi; if (!inWindow) return false; if (resumeAfter is null) return true;