Merge fix/668-followups: #668 review fix-ups (ACL allow leg, explicit cursor source time, paging tests, QuestDB orphan warning, docs)

2026-10-11 09:37:27 -04:00
parent 062086f9a6 7c80d36eb6
commit 57deaeba16
9 changed files with 138 additions and 15 deletions
@@ -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`.

@@ -51,12 +51,17 @@ public enum HistoryModificationType : byte
 /// <param name="UpdateType">The kind of modification.</param>
 /// <param name="UserName">The principal that made it.</param>
 /// <param name="ModifiedId">The backend's ordering key within one source timestamp (paging cursor).</param>
+/// <param name="SourceTimeUtc">
+///     The modified sample's source time as stored on the trail row — the paging cursor's time component. Explicit
+///     because an insert's <see cref="Value"/> has no prior value and so cannot be relied on to carry it.
+/// </param>
 public sealed record HistoryModifiedValue(
    DataValueSnapshot Value,
    DateTime ModificationTimeUtc,
    HistoryModificationType UpdateType,
    string UserName,
-    long ModifiedId);
+    long ModifiedId,
+    DateTime SourceTimeUtc);

 /// <summary>Resume position for a paged modified-value read: strictly after (or before, reversed) this key.</summary>
 /// <param name="SourceTimeUtc">The last returned modification's source time.</param>
@@ -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));
                }
            }

@@ -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);
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    private async Task AfterTrailAsync(string tag, DateTime from, DateTime to, int count, Func<Task> 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<HistorianSampleRow> rows, CancellationToken ct) =>
        InsertAsync(rows.Count, ct, (cmd, i) =>
        {
@@ -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));
                }
            }

@@ -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));
            }

@@ -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);

@@ -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<HistoryModifiedData>();
+            data.DataValues.Count.ShouldBe(1);
+            data.ModificationInfos[0].UserName.ShouldBe("alice");
        }
        finally { await host.DisposeAsync(); }
    }
@@ -327,6 +333,32 @@ public sealed class NodeManagerAclEnforcementTests : IDisposable
        }
    }

+    /// <summary>A backend whose modified-value trail holds exactly one entry; every other read is empty.</summary>
+    private sealed class OneModificationTrail : IHistorianDataSource
+    {
+        public Task<HistoryModifiedReadResult> 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<ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
+            NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
+        public Task<ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult> ReadProcessedAsync(
+            string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
+            NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
+        public Task<ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
+            NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
+        public Task<HistoricalEventsResult> 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",
@@ -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<HistoryModifiedValue>
+        {
+            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<long> 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<HistoryModifiedValue> 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<HistoryModifiedReadResult> 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;