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.
This commit is contained in:
@@ -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>
|
||||
|
||||
+1
-1
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+28
-3
@@ -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) =>
|
||||
{
|
||||
|
||||
+1
-1
@@ -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));
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user