Merge feat/668-modified-values: Part 11 modified-value history (IsReadModified) on SqlServer + QuestDB (#668)

This commit is contained in:
Joseph Doherty
2026-10-11 09:32:13 -04:00
51 changed files with 2173 additions and 195 deletions
@@ -56,6 +56,13 @@ public class HistoryReadCommand : CommandBase
[CommandOption("interval", Description = "Processing interval in milliseconds for aggregates")]
public double IntervalMs { get; init; } = 3600000;
/// <summary>
/// Gets a value indicating whether to read modified-value history (Part 11 IsReadModified) instead of raw samples:
/// the prior value each HistoryUpdate replaced or deleted, with its update type, user and modification time.
/// </summary>
[CommandOption("modified", Description = "Read modified-value history (what HistoryUpdate changed: prior value, update type, user, time) instead of raw samples")]
public bool Modified { get; init; }
/// <inheritdoc />
public override async ValueTask ExecuteAsync(IConsole console)
{
@@ -64,6 +71,8 @@ public class HistoryReadCommand : CommandBase
throw new CommandException($"--max must be greater than 0 (was {MaxValues}).");
if (!string.IsNullOrEmpty(Aggregate) && IntervalMs <= 0)
throw new CommandException($"--interval must be greater than 0 (was {IntervalMs}).");
if (Modified && !string.IsNullOrEmpty(Aggregate))
throw new CommandException("--modified and --aggregate cannot be combined.");
NodeId nodeId;
try
@@ -119,6 +128,30 @@ public class HistoryReadCommand : CommandBase
var ct = console.RegisterCancellationHandler();
(service, _) = await CreateServiceAndConnectAsync(ct);
if (Modified)
{
await console.Output.WriteLineAsync(
$"Modified history for {NodeId} ({start:yyyy-MM-dd HH:mm} -> {end:yyyy-MM-dd HH:mm})");
var entries = await service.HistoryReadModifiedAsync(nodeId, start, end, MaxValues, ct);
await console.Output.WriteLineAsync();
await console.Output.WriteLineAsync(
$"{"Timestamp",-35} {"Prior value",-15} {"Status",-10} {"Update",-8} {"User",-20} {"Modified at"}");
foreach (var e in entries)
{
var dv = e.Value;
var status = StatusCode.IsGood(dv.StatusCode) ? "Good"
: StatusCode.IsBad(dv.StatusCode) ? "Bad"
: "Uncertain";
await console.Output.WriteLineAsync(
$"{dv.SourceTimestamp.ToString("O"),-35} {dv.Value?.ToString() ?? "(none)",-15} {status,-10} " +
$"{e.UpdateType,-8} {e.UserName ?? "",-20} {e.ModificationTime:O}");
}
await console.Output.WriteLineAsync();
await console.Output.WriteLineAsync($"{entries.Count} modifications returned.");
return;
}
IReadOnlyList<DataValue> values;
if (string.IsNullOrEmpty(Aggregate))
@@ -206,6 +206,74 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
return allValues;
}
/// <inheritdoc />
public async Task<IReadOnlyList<Models.HistoryModifiedEntry>> HistoryReadModifiedAsync(
NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues, CancellationToken ct)
{
var details = new ReadRawModifiedDetails
{
StartTime = startTime,
EndTime = endTime,
NumValuesPerNode = (uint)maxValues,
IsReadModified = true,
ReturnBounds = false
};
var nodesToRead = new HistoryReadValueIdCollection
{
new HistoryReadValueId { NodeId = nodeId }
};
var all = new List<Models.HistoryModifiedEntry>();
byte[]? continuationPoint = null;
do
{
if (continuationPoint != null)
nodesToRead[0].ContinuationPoint = continuationPoint;
var response = await _session.HistoryReadAsync(
null,
new ExtensionObject(details),
TimestampsToReturn.Both,
false,
nodesToRead,
ct).ConfigureAwait(false);
var results = response.Results;
if (results == null || results.Count == 0)
break;
var result = results[0];
HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
if (result.HistoryData is ExtensionObject ext && ext.Body is HistoryModifiedData modified)
{
for (var i = 0; i < modified.DataValues.Count; i++)
{
var info = i < modified.ModificationInfos.Count ? modified.ModificationInfos[i] : null;
all.Add(new Models.HistoryModifiedEntry(
modified.DataValues[i],
info?.ModificationTime ?? DateTime.MinValue,
info?.UpdateType ?? default,
info?.UserName));
}
}
continuationPoint = result.ContinuationPoint;
} while (continuationPoint != null && continuationPoint.Length > 0 && all.Count < maxValues);
// A read stopped by the client's own cap leaves a server cursor behind; release it.
if (continuationPoint is { Length: > 0 })
{
nodesToRead[0].ContinuationPoint = continuationPoint;
await _session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, true,
nodesToRead, ct).ConfigureAwait(false);
}
return all;
}
/// <inheritdoc />
public async Task<IReadOnlyList<DataValue>> HistoryReadAggregateAsync(
NodeId nodeId, DateTime startTime, DateTime endTime, NodeId aggregateId, double intervalMs,
@@ -110,6 +110,20 @@ internal interface ISessionAdapter : IDisposable
Task<IReadOnlyList<DataValue>> HistoryReadRawAsync(NodeId nodeId, DateTime startTime, DateTime endTime,
int maxValues, CancellationToken ct = default);
/// <summary>
/// Reads modified-value history (Part 11, <c>IsReadModified = true</c>): the prior values and modification info
/// the server recorded for every HistoryUpdate in the window, following continuation points.
/// </summary>
/// <param name="nodeId">The historized node.</param>
/// <param name="startTime">Window start.</param>
/// <param name="endTime">Window end.</param>
/// <param name="maxValues">The maximum number of entries to return.</param>
/// <param name="ct">The cancellation token.</param>
/// <returns>The modification entries in server order.</returns>
Task<IReadOnlyList<Models.HistoryModifiedEntry>> HistoryReadModifiedAsync(NodeId nodeId, DateTime startTime,
DateTime endTime, int maxValues, CancellationToken ct = default) =>
throw new NotSupportedException("This session adapter does not support modified-value history reads.");
/// <summary>
/// Sends a HistoryUpdate request for the supplied details.
/// </summary>
@@ -193,6 +193,20 @@ public interface IOpcUaClientService : IDisposable
Task<IReadOnlyList<DataValue>> HistoryReadRawAsync(NodeId nodeId, DateTime startTime, DateTime endTime,
int maxValues = 1000, CancellationToken ct = default);
/// <summary>
/// Reads modified-value history (Part 11, <c>IsReadModified = true</c>): what each HistoryUpdate in the window
/// replaced or deleted, with the server's modification time, update type and user name.
/// </summary>
/// <param name="nodeId">The historized node.</param>
/// <param name="startTime">Window start.</param>
/// <param name="endTime">Window end.</param>
/// <param name="maxValues">The maximum number of entries to return.</param>
/// <param name="ct">The cancellation token.</param>
/// <returns>The modification entries in server order.</returns>
Task<IReadOnlyList<HistoryModifiedEntry>> HistoryReadModifiedAsync(NodeId nodeId, DateTime startTime,
DateTime endTime, int maxValues = 1000, CancellationToken ct = default) =>
throw new NotSupportedException("This client service does not support modified-value history reads.");
/// <summary>
/// Sends a HistoryUpdate request (insert/replace/update or delete) for the supplied details.
/// </summary>
@@ -0,0 +1,17 @@
using Opc.Ua;
namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Models;
/// <summary>
/// One entry of an OPC UA Part 11 modified-value history read (HistoryReadRaw with <c>IsReadModified = true</c>): the
/// value as it was BEFORE the modification plus the server's <see cref="ModificationInfo"/>.
/// </summary>
/// <param name="Value">The prior value at its source timestamp (an insert had none: null value, Bad status).</param>
/// <param name="ModificationTime">When the modification was applied.</param>
/// <param name="UpdateType">The Part 11 update type (Insert / Replace / Update / Delete).</param>
/// <param name="UserName">Who made the modification (the server's recorded principal).</param>
public sealed record HistoryModifiedEntry(
DataValue Value,
DateTime ModificationTime,
HistoryUpdateType UpdateType,
string? UserName);
@@ -559,6 +559,15 @@ public sealed class OpcUaClientService : IOpcUaClientService
return await _session!.HistoryReadRawAsync(nodeId, startTime, endTime, maxValues, ct);
}
/// <inheritdoc />
public async Task<IReadOnlyList<HistoryModifiedEntry>> HistoryReadModifiedAsync(
NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues = 1000, CancellationToken ct = default)
{
ThrowIfDisposed();
ThrowIfNotConnected();
return await _session!.HistoryReadModifiedAsync(nodeId, startTime, endTime, maxValues, ct);
}
/// <inheritdoc />
public async Task<HistoryUpdateOutcome> HistoryUpdateAsync(HistoryUpdateDetails details, CancellationToken ct = default)
{
@@ -0,0 +1,69 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
// OPC UA Part 11 modified-value history (Gitea #668): every HistoryUpdate a backend applies leaves an audit row
// carrying the PRIOR value, who made the change and when; HistoryRead with IsReadModified=true serves them.
/// <summary>Who is performing a HistoryUpdate. Threaded into every <see cref="IHistorianDataUpdater"/> call.</summary>
/// <param name="Principal">
/// The session's authenticated principal (an API key's principal, or the reserved <c>anonymous</c>). Recorded as
/// the modification's user name.
/// </param>
/// <param name="ApiKeyId">The API key id the session authenticated with, or <c>null</c> (anonymous / non-key).</param>
public sealed record HistoryUpdateContext(string Principal, string? ApiKeyId)
{
/// <summary>The longest principal / key id a backend stores; longer values are truncated (never refused).</summary>
public const int MaxIdentityLength = 200;
/// <summary>The principal clamped to <see cref="MaxIdentityLength"/> characters.</summary>
public string StoredPrincipal => Clamp(Principal) ?? string.Empty;
/// <summary>The key id clamped to <see cref="MaxIdentityLength"/> characters.</summary>
public string? StoredApiKeyId => Clamp(ApiKeyId);
private static string? Clamp(string? s) => s is { Length: > MaxIdentityLength } ? s[..MaxIdentityLength] : s;
}
/// <summary>
/// The OPC UA Part 11 <c>HistoryUpdateType</c> of a recorded modification. The numeric values are the spec's and
/// are persisted (SQL <c>tinyint</c>, QuestDB <c>BYTE</c>) — never renumber.
/// </summary>
public enum HistoryModificationType : byte
{
/// <summary>A sample was inserted where none existed (no prior value).</summary>
Insert = 1,
/// <summary>An existing sample was replaced.</summary>
Replace = 2,
/// <summary>An upsert (<c>PerformUpdateType.Update</c>): replaced a sample, or inserted one (then no prior value).</summary>
Update = 3,
/// <summary>A sample was deleted (DeleteAtTime / DeleteRaw).</summary>
Delete = 4,
}
/// <summary>One recorded modification: the value as it was BEFORE the change, plus who changed it and when.</summary>
/// <param name="Value">
/// The prior value at its source timestamp. An insert had no prior value: <c>Value = null</c>,
/// <c>StatusCode = BadNoData</c>.
/// </param>
/// <param name="ModificationTimeUtc">When the modification was applied.</param>
/// <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>
public sealed record HistoryModifiedValue(
DataValueSnapshot Value,
DateTime ModificationTimeUtc,
HistoryModificationType UpdateType,
string UserName,
long ModifiedId);
/// <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>
/// <param name="ModifiedId">The last returned modification's id.</param>
public sealed record HistoryModifiedCursor(DateTime SourceTimeUtc, long ModifiedId);
/// <summary>Result of <see cref="IHistorianDataSource.ReadModifiedAsync"/>.</summary>
/// <param name="Values">Modifications ordered by <c>(SourceTimeUtc, ModifiedId)</c> (descending for a reverse read).</param>
/// <param name="HasMore"><c>true</c> when the cap truncated the result and more modifications remain in the window.</param>
public sealed record HistoryModifiedReadResult(IReadOnlyList<HistoryModifiedValue> Values, bool HasMore);
@@ -122,4 +122,32 @@ public interface IHistorianDataSource : IDisposable
/// Same-timestamp ties are unaffected (they are paged within the timestamp, never stepped over).
/// </remarks>
long TimestampResolutionTicks => 1;
/// <summary>
/// Read modified-value history — OPC UA HistoryReadRaw with <c>IsReadModified = true</c> (Part 11, Gitea #668):
/// every modification a HistoryUpdate applied to the tag inside the window, each carrying the PRIOR value.
/// </summary>
/// <remarks>
/// Window: forward (<c>start &lt; end</c>) <c>[start, end)</c> ascending by <c>(SourceTimeUtc, ModifiedId)</c>;
/// reverse (<c>end &lt; start</c>) <c>(end, start]</c> descending; <c>start == end</c> the modifications at
/// exactly that time. <paramref name="resumeAfter"/> continues strictly past that key in the read's direction.
/// The default throws <see cref="NotSupportedException"/> — a backend that keeps no modification trail
/// (Gateway, Null) — which the server reports as <c>BadHistoryOperationUnsupported</c>.
/// </remarks>
/// <param name="fullReference">The historian tagname.</param>
/// <param name="startUtc">Window start (UTC).</param>
/// <param name="endUtc">Window end (UTC).</param>
/// <param name="maxValues">Cap on returned modifications; <c>0</c> is unbounded.</param>
/// <param name="resumeAfter">Resume cursor from a previous page, or <c>null</c> for the first page.</param>
/// <param name="cancellationToken">Cancellation.</param>
/// <returns>The modifications, with <see cref="HistoryModifiedReadResult.HasMore"/> set when truncated.</returns>
/// <exception cref="NotSupportedException">The backend keeps no modified-value history.</exception>
Task<HistoryModifiedReadResult> ReadModifiedAsync(
string fullReference,
DateTime startUtc,
DateTime endUtc,
uint maxValues,
HistoryModifiedCursor? resumeAfter,
CancellationToken cancellationToken) =>
throw new NotSupportedException("This historian backend keeps no modified-value history.");
}
@@ -17,12 +17,20 @@ public enum HistoryUpdateMode
/// Optional historian capability behind OPC UA HistoryUpdate. A backend that cannot modify history
/// supplies no updater (the server then reports HistoryUpdate as unsupported).
/// </summary>
/// <remarks>
/// <b>Modified-value history (Part 11, Gitea #668).</b> Every sample an implementation inserts, replaces or deletes
/// MUST leave one modification record carrying the prior value (none for an insert), the
/// <see cref="HistoryUpdateContext.Principal"/>, the modification time and the <see cref="HistoryModificationType"/>,
/// readable through <see cref="IHistorianDataSource.ReadModifiedAsync"/>. A refused item (any non-Good code) leaves
/// no record. Live recorder writes and retention purges never leave one.
/// </remarks>
public interface IHistorianDataUpdater
{
/// <summary>Inserts, replaces or upserts samples for one tag.</summary>
/// <param name="tag">Historian tag name.</param>
/// <param name="mode">Update mode.</param>
/// <param name="values">Samples to apply.</param>
/// <param name="context">Who is making the change (recorded in the modification trail).</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>
/// One status code per value, in order: Insert gives <c>GoodEntryInserted</c> or <c>BadEntryExists</c>;
@@ -30,20 +38,24 @@ public interface IHistorianDataUpdater
/// <c>GoodEntryInserted</c> or <c>GoodEntryReplaced</c>; a null timestamp gives <c>BadInvalidTimestamp</c>.
/// </returns>
Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> values, CancellationToken ct);
IReadOnlyList<HistorizationValue> values, HistoryUpdateContext context, CancellationToken ct);
/// <summary>Deletes raw samples in a time range.</summary>
/// <param name="tag">Historian tag name.</param>
/// <param name="startUtc">Range start (UTC).</param>
/// <param name="endUtc">Range end (UTC).</param>
/// <param name="context">Who is making the change (recorded in the modification trail).</param>
/// <param name="ct">Cancellation token.</param>
/// <returns><c>Good</c>, or <c>BadNoData</c> when the range held nothing.</returns>
Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct);
Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, HistoryUpdateContext context,
CancellationToken ct);
/// <summary>Deletes the samples at exact timestamps.</summary>
/// <param name="tag">Historian tag name.</param>
/// <param name="timesUtc">Timestamps to delete (UTC).</param>
/// <param name="context">Who is making the change (recorded in the modification trail).</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>One status code per time, in order: <c>Good</c> or <c>BadNoEntryExists</c>.</returns>
Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct);
Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, HistoryUpdateContext context,
CancellationToken ct);
}
@@ -83,6 +83,8 @@ internal sealed class QuestDbConnection : IAsyncDisposable, IDisposable
public string EventTable => T(QuestDbTables.EventBase);
/// <summary>The prefixed tag table.</summary>
public string TagTable => T(QuestDbTables.TagBase);
/// <summary>The prefixed Part 11 modification-trail table (#668).</summary>
public string SampleModifiedTable => T(QuestDbTables.SampleModifiedBase);
/// <summary>Opens a pooled connection (no bootstrap; callers run the bootstrapper first).</summary>
/// <param name="ct">Cancellation.</param>
@@ -140,6 +140,101 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
}
}
// ------------------------------------------------------------------------------------------ Modified (Part 11, #668)
/// <inheritdoc />
/// <remarks>
/// Reads <c>{prefix}sample_modified</c>, the append-only trail <see cref="QuestDbHistorianDataUpdater"/> writes
/// BEFORE each change. Windows follow <see cref="ReadRawAsync"/> (bounds rounded to µs); rows are ordered by
/// <c>(ts, modified_id)</c>; <c>maxValues + 1</c> rows are fetched to report
/// <see cref="HistoryModifiedReadResult.HasMore"/> exactly. A row whose <c>status</c> is NULL had no prior value
/// (an insert) and is served <c>null</c> / <c>BadNoData</c>. 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).
/// </remarks>
public async Task<HistoryModifiedReadResult> ReadModifiedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter,
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
var bounded = maxValues > 0 && maxValues < int.MaxValue;
try
{
var values = new List<HistoryModifiedValue>();
var hasMore = false;
if (await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false))
{
var start = QuestDbConnection.Round(startUtc);
var end = QuestDbConnection.Round(endUtc);
var reverse = start > end;
string where;
if (start == end) where = "ts = $2 AND ts = $3";
else if (!reverse) where = "ts >= $2 AND ts < $3";
else where = "ts > $2 AND ts <= $3";
var next = 4;
if (resumeAfter is not null)
{
where += reverse
? " AND (ts < $4 OR (ts = $4 AND modified_id < $5))"
: " AND (ts > $4 OR (ts = $4 AND modified_id > $5))";
next = 6;
}
var order = reverse ? "DESC" : "ASC";
var (lo, hi) = start <= end ? (start, end) : (end, start);
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn,
$"SELECT {SampleColumns}, modified_id, modification_ts, update_type, principal FROM {_conn.SampleModifiedTable} " +
$"WHERE tag = $1 AND {where} ORDER BY ts {order}, modified_id {order}" + (bounded ? $" LIMIT ${next}" : ""));
QuestDbConnection.BindText(cmd, fullReference);
QuestDbConnection.BindTimestamp(cmd, lo);
QuestDbConnection.BindTimestamp(cmd, hi);
if (resumeAfter is not null)
{
QuestDbConnection.BindTimestamp(cmd, resumeAfter.SourceTimeUtc);
QuestDbConnection.BindLong(cmd, resumeAfter.ModifiedId);
}
if (bounded)
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = NpgsqlDbType.Integer, Value = (int)maxValues + 1 });
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
if (bounded && values.Count == maxValues)
{
hasMore = true;
break;
}
var sourceTime = QuestDbConnection.ReadUtc(reader, 0);
var modifiedAt = QuestDbConnection.ReadUtc(reader, 8);
DataValueSnapshot prior;
if (reader.IsDBNull(5))
{
prior = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, sourceTime, modifiedAt);
}
else
{
var s = ReadStored(reader, 0);
prior = new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc, ReadServerTime(reader, 6, modifiedAt));
}
values.Add(new HistoryModifiedValue(prior, modifiedAt,
(HistoryModificationType)QuestDbConnection.ReadKind(reader, 9),
reader.IsDBNull(10) ? string.Empty : reader.GetString(10),
reader.GetInt64(7)));
}
}
RecordSuccess();
return new HistoryModifiedReadResult(values, hasMore);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
// ------------------------------------------------------------------------------------------ AtTime
/// <inheritdoc />
@@ -25,6 +25,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// 1970 (read bounds are not clamped).</para>
/// <para><b>Read-your-writes:</b> after writing, <see cref="QuestDbWalVisibility"/> waits for the rows to become
/// visible, bounded by <c>VisibilityTimeout</c>; a timeout only logs a Warning, never changes a code.</para>
/// <para><b>Modified-value history (Part 11, #668).</b> Every sample this updater inserts, replaces or tombstones first
/// gets a row in <c>{prefix}sample_modified</c> carrying its PRIOR live value (the existence read already fetches it;
/// none for an insert — a tombstone is "absent", so an insert over one records no prior value), the caller's
/// <see cref="HistoryUpdateContext"/>, the <see cref="HistoryModificationType"/> and the modification time. QuestDB has
/// no multi-table transaction, so the trail row is written <b>BEFORE</b> the change: a crash (or a failed sample INSERT)
/// between the two leaves an orphan trail row describing a change that never landed — never a change without its
/// trail row. This is the documented deviation from SQL Server, whose trail commits atomically with the change.</para>
/// </summary>
internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
{
@@ -35,6 +42,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
private const int LockStripes = 64;
private readonly SemaphoreSlim[] _locks = Enumerable.Range(0, LockStripes).Select(_ => new SemaphoreSlim(1, 1)).ToArray();
private long _liveRowsRead;
private static long _lastModifiedId;
private readonly QuestDbConnection _conn;
private readonly QuestDbTagCatalog _tags;
private readonly QuestDbWalVisibility _wal;
@@ -71,10 +79,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> values, CancellationToken ct)
IReadOnlyList<HistorizationValue> values, HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
ArgumentNullException.ThrowIfNull(context);
// RC1: a valid-timestamp item before 1970 can never be stored; refuse it and plan the rest.
var final = new uint[values.Count];
@@ -100,7 +109,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
? await _tags.ExistsAsync(tag, ct).ConfigureAwait(false)
: await EnsureAsync(tag, plan.Winners[0].Row.Value.Kind, ct).ConfigureAwait(false);
var outcome = known
? await ApplyAsync(tag, mode, plan.Winners.Select(w => w.Row).ToList(), ct).ConfigureAwait(false)
? await ApplyAsync(tag, mode, plan.Winners.Select(w => w.Row).ToList(), context, ct).ConfigureAwait(false)
: new Dictionary<long, UpsertOutcome>();
planned = HistoryUpdatePlanner.Complete(plan, mode, outcome);
}
@@ -109,9 +118,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
}
/// <inheritdoc/>
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, HistoryUpdateContext context,
CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
if (!await _tags.ExistsAsync(tag, ct).ConfigureAwait(false))
return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
@@ -123,17 +134,27 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
while (cursor < end)
{
var page = new List<DateTime>(BatchSize);
var priors = new List<LiveRow>(BatchSize);
await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
await using (var cmd = _conn.CreateCommand(conn,
$"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts LIMIT {BatchSize}"))
$"SELECT {LiveColumns} FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts LIMIT {BatchSize}"))
{
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindTimestamp(cmd, cursor);
QuestDbConnection.BindTimestamp(cmd, end);
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false)) page.Add(QuestDbConnection.ReadUtc(rd, 0));
while (await rd.ReadAsync(ct).ConfigureAwait(false))
{
var row = ReadLiveRow(rd);
page.Add(row.TimeUtc);
priors.Add(row);
}
}
if (page.Count == 0) break;
// Trail BEFORE the change (no multi-table transaction): a crash here leaves an orphan trail row, never a
// 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);
// 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);
@@ -145,10 +166,12 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct)
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc,
HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag))
{
var refused = new uint[timesUtc.Count];
@@ -168,9 +191,14 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
using var _ = await LockAsync(tag, ct).ConfigureAwait(false);
var live = await ReadLiveAsync(tag, candidates.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(), ct).ConfigureAwait(false);
var toDelete = candidates.Where(live.Contains).OrderBy(x => x).ToList();
var toDelete = candidates.Where(live.ContainsKey).OrderBy(x => x).ToList();
foreach (var chunk in toDelete.Chunk(BatchSize))
{
// 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);
}
if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
deleted.UnionWith(toDelete);
return HistoryUpdatePlanner.DeleteAtTimeResults(rounded, deleted, tagKnown: true);
@@ -183,48 +211,162 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
}
private async Task<Dictionary<long, UpsertOutcome>> ApplyAsync(string tag, HistoryUpdateMode mode,
List<HistorianSampleRow> rows, CancellationToken ct)
List<HistorianSampleRow> rows, HistoryUpdateContext context, CancellationToken ct)
{
// Key = the planner's tick (what Complete looks up); the stored time is Round(of the same row).
var live = await ReadLiveAsync(tag, rows.Select(r => QuestDbConnection.Round(r.TimeUtc)).ToList(), ct).ConfigureAwait(false);
var outcome = new Dictionary<long, UpsertOutcome>();
var toWrite = new List<HistorianSampleRow>();
var trail = new List<(DateTime, LiveRow?)>();
foreach (var r in rows)
{
var present = live.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks);
var stored = QuestDbConnection.Round(r.TimeUtc);
var present = live.TryGetValue(stored.Ticks, out var prior);
if (mode == HistoryUpdateMode.Insert && present) continue;
if (mode == HistoryUpdateMode.Replace && !present) continue;
outcome[r.TimeUtc.Ticks] = present ? UpsertOutcome.Replaced : UpsertOutcome.Inserted;
toWrite.Add(r);
trail.Add((stored, present ? prior : null));
}
// 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);
if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
return outcome;
}
/// <summary>Ticks of the live (present, not deleted) rows near the given times: one range query per cluster (I-2).</summary>
private async Task<HashSet<long>> ReadLiveAsync(string tag, IReadOnlyList<DateTime> times, CancellationToken ct)
/// <summary>The live (present, not deleted) rows near the given times, keyed by stored ticks: one range query per cluster (I-2).</summary>
private async Task<Dictionary<long, LiveRow>> ReadLiveAsync(string tag, IReadOnlyList<DateTime> times, CancellationToken ct)
{
var live = new HashSet<long>();
var live = new Dictionary<long, LiveRow>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
foreach (var (lo, hi) in QuestDbTimeClusters.Cluster(times, QuestDbTimeClusters.ExistenceGap))
{
await using var cmd = _conn.CreateCommand(conn,
$"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted");
$"SELECT {LiveColumns} FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted");
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindTimestamp(cmd, lo);
QuestDbConnection.BindTimestamp(cmd, hi);
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false))
{
live.Add(QuestDbConnection.ReadUtc(rd, 0).Ticks);
var row = ReadLiveRow(rd);
live[row.TimeUtc.Ticks] = row;
Interlocked.Increment(ref _liveRowsRead);
}
}
return live;
}
/// <summary>The columns of a live sample the trail needs (prior value + status + server time).</summary>
private const string LiveColumns = "ts, kind, num_value, int_value, str_value, status, server_ts";
/// <summary>A live sample as it was before a modification.</summary>
private readonly record struct LiveRow(DateTime TimeUtc, HistorianValue Value, uint Status, DateTime? ServerTimeUtc);
private static LiveRow ReadLiveRow(NpgsqlDataReader rd)
{
var value = QuestDbHistorianDataSource.ToValue(
QuestDbConnection.ReadKind(rd, 1),
rd.IsDBNull(2) ? null : rd.GetDouble(2),
rd.IsDBNull(3) ? null : rd.GetInt64(3),
rd.IsDBNull(4) ? null : rd.GetString(4));
var status = rd.IsDBNull(5) ? HistorianStatusCodes.BadNoData : SampleColumnMapper.FromLongStatus(rd.GetInt64(5));
return new LiveRow(QuestDbConnection.ReadUtc(rd, 0), value, status,
rd.IsDBNull(6) ? null : QuestDbConnection.ReadUtc(rd, 6));
}
/// <summary>The Part 11 modification type recorded for an update mode.</summary>
private static HistoryModificationType ToModificationType(HistoryUpdateMode mode) => mode switch
{
HistoryUpdateMode.Insert => HistoryModificationType.Insert,
HistoryUpdateMode.Replace => HistoryModificationType.Replace,
_ => HistoryModificationType.Update,
};
/// <summary>
/// A process-wide, strictly increasing modification id (≈ UTC ticks, bumped past the last issued value). It orders a
/// tag's trail within one source time; ids from two nodes interleave by wall clock, which is the modification order.
/// </summary>
internal static long NextModifiedId()
{
while (true)
{
var last = Interlocked.Read(ref _lastModifiedId);
var next = Math.Max(DateTime.UtcNow.Ticks, last + 1);
if (Interlocked.CompareExchange(ref _lastModifiedId, next, last) == last) return next;
}
}
/// <summary>
/// Appends one modification-trail row per entry (<c>prior = null</c> ⇒ an insert: kind Empty, NULL value and
/// status), then waits for the trail's WAL visibility so a modified read right after the update sees it.
/// </summary>
private async Task WriteTrailAsync(string tag, HistoryModificationType type, IReadOnlyList<(DateTime TimeUtc, LiveRow? Prior)> entries,
HistoryUpdateContext context, CancellationToken ct)
{
if (entries.Count == 0) return;
var now = DateTime.UtcNow;
foreach (var chunk in entries.Chunk(BatchSize))
{
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, "");
var sb = new System.Text.StringBuilder(
$"INSERT INTO {_conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, update_type, principal, " +
"api_key_id, kind, num_value, int_value, str_value, status, server_ts) VALUES ");
for (var i = 0; i < chunk.Length; i++)
{
var (time, prior) = chunk[i];
QuestDbConnection.BindTimestamp(cmd, time, clampToEpoch: true);
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindLong(cmd, NextModifiedId());
QuestDbConnection.BindTimestamp(cmd, now);
QuestDbConnection.BindByte(cmd, (byte)type);
QuestDbConnection.BindText(cmd, context.StoredPrincipal);
QuestDbConnection.BindText(cmd, context.StoredApiKeyId);
if (prior is { } p)
{
var c = SampleColumnMapper.ToColumns(p.Value);
QuestDbConnection.BindByte(cmd, c.ValueKind);
Typed(cmd, NpgsqlDbType.Double, c.NumValue);
Typed(cmd, NpgsqlDbType.Bigint, c.IntValue);
Typed(cmd, NpgsqlDbType.Varchar, c.StrValue);
QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(p.Status));
Typed(cmd, NpgsqlDbType.TimestampTz, p.ServerTimeUtc);
}
else
{
// No prior value: QuestDB BYTE has no NULL, so kind is Empty (0) and the NULL status marks "none".
QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty);
Typed(cmd, NpgsqlDbType.Double, null);
Typed(cmd, NpgsqlDbType.Bigint, null);
Typed(cmd, NpgsqlDbType.Varchar, null);
Typed(cmd, NpgsqlDbType.Bigint, null);
Typed(cmd, NpgsqlDbType.TimestampTz, null);
}
var b = i * 13;
if (i > 0) sb.Append(',');
sb.Append('(');
for (var q = 1; q <= 13; q++) sb.Append(q > 1 ? "," : "").Append('$').Append(b + q);
sb.Append(')');
}
cmd.CommandText = sb.ToString();
try
{
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
_log.LogWarning(ex, "QuestDB modification-trail INSERT of {Count} rows into {Table} failed; the update was not applied.",
chunk.Length, _conn.SampleModifiedTable);
throw;
}
}
await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false);
}
private Task WriteRowsAsync(string tag, IReadOnlyList<HistorianSampleRow> rows, CancellationToken ct) =>
InsertAsync(rows.Count, ct, (cmd, i) =>
{
@@ -6,7 +6,7 @@ using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// Lazily creates the three tables once per instance (<c>CREATE TABLE IF NOT EXISTS</c>), then reconciles the TTL
/// Lazily creates the four tables once per instance (<c>CREATE TABLE IF NOT EXISTS</c>), then reconciles the TTL
/// of the sample and event tables with <see cref="QuestDbHistorianOptions.RetentionDays"/>.
/// <list type="bullet">
/// <item><b>Concurrency:</b> two nodes starting together are safe. If a CREATE still fails with a server error
@@ -15,6 +15,8 @@ internal static class QuestDbTables
public const string EventBase = "event";
/// <summary>Base name of the tag catalog table.</summary>
public const string TagBase = "tag";
/// <summary>Base name of the Part 11 modification-trail table (#668).</summary>
public const string SampleModifiedBase = "sample_modified";
/// <summary>The fixed designated timestamp of every <c>tag</c> row, which makes the dedup key an upsert by name.</summary>
public static readonly DateTime TagEpoch = DateTime.UnixEpoch;
@@ -25,12 +27,17 @@ internal static class QuestDbTables
public static string Event(string prefix) => Name(prefix, EventBase);
/// <summary>The prefixed tag table name.</summary>
public static string Tag(string prefix) => Name(prefix, TagBase);
/// <summary>The prefixed modification-trail table name.</summary>
public static string SampleModified(string prefix) => Name(prefix, SampleModifiedBase);
/// <summary>All three table names in creation order.</summary>
public static string[] All(string prefix) => [Sample(prefix), Event(prefix), Tag(prefix)];
/// <summary>All four table names in creation order.</summary>
public static string[] All(string prefix) => [Sample(prefix), Event(prefix), Tag(prefix), SampleModified(prefix)];
/// <summary>The tables that carry a TTL (sample and event; the tag catalog never expires).</summary>
public static string[] Expiring(string prefix) => [Sample(prefix), Event(prefix)];
/// <summary>
/// The tables that carry a TTL (sample, event and the modification trail, which ages out with the samples it
/// describes; the tag catalog never expires).
/// </summary>
public static string[] Expiring(string prefix) => [Sample(prefix), Event(prefix), SampleModified(prefix)];
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the sample table.</summary>
public static string CreateSample(string prefix, int? retentionDays) =>
@@ -50,9 +57,22 @@ internal static class QuestDbTables
$"CREATE TABLE IF NOT EXISTS {Tag(prefix)} (ts TIMESTAMP, name SYMBOL, data_type BYTE, eng_unit VARCHAR, " +
"description VARCHAR, updated_utc TIMESTAMP) TIMESTAMP(ts) PARTITION BY YEAR WAL DEDUP UPSERT KEYS(ts, name)";
/// <summary>The three CREATE statements in creation order.</summary>
/// <summary>
/// <c>CREATE TABLE IF NOT EXISTS</c> for the Part 11 modification trail (#668): one row per sample a HistoryUpdate
/// inserted, replaced or deleted, carrying the PRIOR value (NULL kind/value/status for an insert). <c>ts</c> is the
/// sample's source time (so the TTL ages it out with the samples); <c>modified_id</c> orders the trail within one
/// source time. Append-only: no DEDUP, never updated or deleted.
/// </summary>
public static string CreateSampleModified(string prefix, int? retentionDays) =>
$"CREATE TABLE IF NOT EXISTS {SampleModified(prefix)} (ts TIMESTAMP, tag SYMBOL, modified_id LONG, " +
"modification_ts TIMESTAMP, update_type BYTE, principal VARCHAR, api_key_id VARCHAR, kind BYTE, num_value DOUBLE, " +
"int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP) " +
$"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL";
/// <summary>The four CREATE statements in creation order (parallel to <see cref="All"/>).</summary>
public static string[] CreateAll(string prefix, int? retentionDays) =>
[CreateSample(prefix, retentionDays), CreateEvent(prefix, retentionDays), CreateTag(prefix)];
[CreateSample(prefix, retentionDays), CreateEvent(prefix, retentionDays), CreateTag(prefix),
CreateSampleModified(prefix, retentionDays)];
/// <summary><c>ALTER TABLE t SET TTL n DAYS</c>; <c>null</c> removes the TTL (<c>SET TTL 0 DAYS</c>).</summary>
public static string AlterTtl(string table, int? retentionDays) =>
@@ -0,0 +1,29 @@
-- OtOpcUa SQL Server historian backend - schema version 2: OPC UA Part 11 modified-value history (Gitea #668).
-- $(Schema) is substituted by SqlServerSchemaBootstrapper (identifier-validated) BEFORE the script is split on GO lines.
-- Every statement is guarded so a re-run is a no-op.
--
-- One row per sample a HistoryUpdate inserted, replaced or deleted, carrying the PRIOR value (all NULL for an
-- insert). Rows are written by the SAME statement that changes hist.Sample (OUTPUT ... INTO), so a failed update
-- leaves no audit row and a committed one always has its row.
-- WARNING: OUTPUT ... INTO forbids, on its target table: enabled triggers, either side of a FOREIGN KEY, and CHECK
-- constraints. Do not add any of those to SampleModified, or every HistoryUpdate fails.
IF OBJECT_ID(N'[$(Schema)].[SampleModified]', N'U') IS NULL
CREATE TABLE [$(Schema)].[SampleModified] (
TagId int NOT NULL,
SourceTimeUtc datetime2(7) NOT NULL,
ModifiedId bigint IDENTITY(1,1) NOT NULL,
ModificationTimeUtc datetime2(7) NOT NULL,
UpdateType tinyint NOT NULL,
Principal nvarchar(200) NOT NULL,
ApiKeyId nvarchar(200) NULL,
ValueKind tinyint NULL,
NumValue float NULL,
IntValue bigint NULL,
StrValue nvarchar(max) NULL,
StatusCode int NULL,
ServerTimeUtc datetime2(7) NULL,
CONSTRAINT PK_SampleModified PRIMARY KEY CLUSTERED (TagId, SourceTimeUtc, ModifiedId))
GO
IF NOT EXISTS (SELECT 1 FROM [$(Schema)].[SchemaVersion] WHERE Version = 2)
INSERT [$(Schema)].[SchemaVersion] (Version, AppliedUtc) VALUES (2, SYSUTCDATETIME())
GO
@@ -135,6 +135,97 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
}
}
// ------------------------------------------------------------------------------------------ Modified (Part 11, #668)
/// <inheritdoc />
/// <remarks>
/// Reads <c>SampleModified</c> (schema v2), the trail <see cref="SqlServerHistorianDataUpdater"/> writes in the
/// same statement as each change. Windows follow <see cref="ReadRawAsync"/>; rows are ordered by
/// <c>(SourceTimeUtc, ModifiedId)</c> (the clustered key after <c>TagId</c>, so a seek). <c>maxValues + 1</c> rows
/// are fetched to report <see cref="HistoryModifiedReadResult.HasMore"/> exactly. An insert's audit row carries no
/// prior value and is served <c>null</c> / <c>BadNoData</c>; its server timestamp is the modification time. An
/// unknown tag is an empty result.
/// </remarks>
public async Task<HistoryModifiedReadResult> ReadModifiedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValues, HistoryModifiedCursor? resumeAfter,
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReference);
var bounded = maxValues > 0 && maxValues < int.MaxValue;
try
{
var values = new List<HistoryModifiedValue>();
var hasMore = false;
var tagId = await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false);
if (tagId is { } id)
{
var reverse = startUtc > endUtc;
string where;
if (startUtc == endUtc) where = "SourceTimeUtc = @lo";
else if (!reverse) where = "SourceTimeUtc >= @lo AND SourceTimeUtc < @hi";
else where = "SourceTimeUtc > @lo AND SourceTimeUtc <= @hi";
if (resumeAfter is not null)
where += reverse
? " AND (SourceTimeUtc < @ct OR (SourceTimeUtc = @ct AND ModifiedId < @cid))"
: " AND (SourceTimeUtc > @ct OR (SourceTimeUtc = @ct AND ModifiedId > @cid))";
var order = reverse ? "DESC" : "ASC";
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn,
$"SELECT {(bounded ? "TOP (@take) " : "")}SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, " +
"ServerTimeUtc, ModifiedId, ModificationTimeUtc, UpdateType, Principal " +
$"FROM {_conn.Q("SampleModified")} WHERE TagId = @id AND {where} " +
$"ORDER BY SourceTimeUtc {order}, ModifiedId {order}");
if (bounded) cmd.Parameters.Add(new SqlParameter("@take", SqlDbType.BigInt) { Value = (long)maxValues + 1 });
cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
SqlHistorianConnection.AddDateTime2(cmd, "@lo", lo);
SqlHistorianConnection.AddDateTime2(cmd, "@hi", hi);
if (resumeAfter is not null)
{
SqlHistorianConnection.AddDateTime2(cmd, "@ct", resumeAfter.SourceTimeUtc);
cmd.Parameters.Add(new SqlParameter("@cid", SqlDbType.BigInt) { Value = resumeAfter.ModifiedId });
}
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
if (bounded && values.Count == maxValues)
{
hasMore = true;
break;
}
var sourceTime = SqlHistorianConnection.GetUtc(reader, 0);
var modifiedAt = SqlHistorianConnection.GetUtc(reader, 8);
DataValueSnapshot prior;
if (reader.IsDBNull(1))
{
// An insert: there was no prior value.
prior = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, sourceTime, modifiedAt);
}
else
{
var s = ReadStored(reader, 0);
prior = new DataValueSnapshot(s.Value.ToObject(), s.Status, s.TimeUtc,
reader.IsDBNull(6) ? modifiedAt : SqlHistorianConnection.GetUtc(reader, 6));
}
values.Add(new HistoryModifiedValue(
prior, modifiedAt, (HistoryModificationType)reader.GetByte(9), reader.GetString(10), reader.GetInt64(7)));
}
}
RecordSuccess();
return new HistoryModifiedReadResult(values, hasMore);
}
catch (Exception ex)
{
RecordFailure(ex);
throw;
}
}
// ------------------------------------------------------------------------------------------ AtTime
/// <inheritdoc />
@@ -22,6 +22,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// is written.</para>
/// <para><b>Unknown tag.</b> Insert/Update create it via <see cref="SqlServerTagCatalog"/>; Replace and the
/// deletes report <c>BadNoEntryExists</c> (DeleteRaw: <c>BadNoData</c>) without creating anything.</para>
/// <para><b>Modified-value history (Part 11, #668).</b> Every row the MERGE inserts or updates and every row a
/// delete removes is copied into <c>SampleModified</c> by an <c>OUTPUT … INTO</c> on the SAME statement — the prior
/// value from <c>deleted.*</c> (NULL for an insert), the caller's <see cref="HistoryUpdateContext"/>, the
/// <see cref="HistoryModificationType"/> and <c>SYSUTCDATETIME()</c>. One statement means the sample change and its
/// audit row commit or roll back together: a failed update leaves no audit row.</para>
/// </summary>
internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
{
@@ -39,10 +44,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> values, CancellationToken ct)
IReadOnlyList<HistorizationValue> values, HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
ArgumentNullException.ThrowIfNull(context);
var plan = HistoryUpdatePlanner.Plan(tag, mode, values);
if (plan.Winners.Count == 0) return HistoryUpdatePlanner.Complete(plan, mode, new Dictionary<long, UpsertOutcome>());
@@ -52,14 +58,16 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
var outcome = tagId is null
? new Dictionary<long, UpsertOutcome>()
: await MergeAsync(tagId.Value, mode, plan.Winners.Select(w => w.Row).ToList(), ct).ConfigureAwait(false);
: await MergeAsync(tagId.Value, mode, plan.Winners.Select(w => w.Row).ToList(), context, ct).ConfigureAwait(false);
return HistoryUpdatePlanner.Complete(plan, mode, outcome);
}
/// <inheritdoc/>
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, HistoryUpdateContext context,
CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
@@ -67,9 +75,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
int rows;
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE FROM {_conn.Q("Sample")} WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
$"DELETE FROM {_conn.Q("Sample")} {AuditInto(HistoryModificationType.Delete)} " +
"WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
rows = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
@@ -79,10 +89,12 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct)
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc,
HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag))
{
var refused = new uint[timesUtc.Count];
@@ -100,9 +112,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE s OUTPUT deleted.SourceTimeUtc FROM {_conn.Q("Sample")} AS s JOIN @times AS t ON s.SourceTimeUtc = t.T WHERE s.TagId = @tagId", tx))
$"DELETE s {AuditInto(HistoryModificationType.Delete)} OUTPUT deleted.SourceTimeUtc " +
$"FROM {_conn.Q("Sample")} AS s JOIN @times AS t ON s.SourceTimeUtc = t.T WHERE s.TagId = @tagId", tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
_conn.AddStructured(cmd, "@times", "TimeRow",
SampleRowMapper.ToTimeRecords(distinct.Select(t => new DateTime(t, DateTimeKind.Utc))));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
@@ -114,7 +128,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
}
private async Task<Dictionary<long, UpsertOutcome>> MergeAsync(int tagId, HistoryUpdateMode mode,
List<HistorianSampleRow> rows, CancellationToken ct)
List<HistorianSampleRow> rows, HistoryUpdateContext context, CancellationToken ct)
{
var sql = $"MERGE {_conn.Q("Sample")} WITH (HOLDLOCK) AS t USING @rows AS r ON t.TagId = @tagId AND t.SourceTimeUtc = r.SourceTimeUtc ";
if (mode != HistoryUpdateMode.Insert)
@@ -123,7 +137,9 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
if (mode != HistoryUpdateMode.Replace)
sql += "WHEN NOT MATCHED THEN INSERT (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
"VALUES (@tagId, r.SourceTimeUtc, r.ValueKind, r.NumValue, r.IntValue, r.StrValue, r.StatusCode, SYSUTCDATETIME()) ";
sql += "OUTPUT $action, r.SourceTimeUtc;";
// The MERGE's deleted.* is NULL for an inserted row, so one OUTPUT INTO records both an insert (no prior value)
// and a replace (the prior value). TagId comes from inserted.* — present on both branches.
sql += $"{AuditInto(ToModificationType(mode), "inserted")} OUTPUT $action, r.SourceTimeUtc;";
var outcome = new Dictionary<long, UpsertOutcome>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
@@ -131,6 +147,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using (var cmd = _conn.CreateCommand(conn, sql, tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId });
AddAuditParameters(cmd, context);
_conn.AddStructured(cmd, "@rows", "SampleRow", SampleRowMapper.ToSampleRecords(rows));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false))
@@ -141,4 +158,33 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await tx.CommitAsync(ct).ConfigureAwait(false);
return outcome;
}
/// <summary>The Part 11 modification type recorded for an update mode.</summary>
/// <param name="mode">The update mode.</param>
/// <returns>Insert ⇒ Insert, Replace ⇒ Replace, Update ⇒ Update.</returns>
internal static HistoryModificationType ToModificationType(HistoryUpdateMode mode) => mode switch
{
HistoryUpdateMode.Insert => HistoryModificationType.Insert,
HistoryUpdateMode.Replace => HistoryModificationType.Replace,
_ => HistoryModificationType.Update,
};
/// <summary>
/// The <c>OUTPUT … INTO SampleModified</c> clause copying each affected row's prior value (<c>deleted.*</c>) into
/// the modification trail. <paramref name="keySource"/> names the pseudo-table that carries <c>TagId</c> /
/// <c>SourceTimeUtc</c> for every affected row (<c>deleted</c> for a DELETE, <c>inserted</c> for the MERGE).
/// </summary>
private string AuditInto(HistoryModificationType type, string keySource = "deleted") =>
$"OUTPUT {keySource}.TagId, {keySource}.SourceTimeUtc, SYSUTCDATETIME(), {(byte)type}, @principal, @apiKeyId, " +
"deleted.ValueKind, deleted.NumValue, deleted.IntValue, deleted.StrValue, deleted.StatusCode, deleted.ServerTimeUtc " +
$"INTO {_conn.Q("SampleModified")} (TagId, SourceTimeUtc, ModificationTimeUtc, UpdateType, Principal, ApiKeyId, " +
"ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc)";
private static void AddAuditParameters(SqlCommand cmd, HistoryUpdateContext context)
{
cmd.Parameters.Add(new SqlParameter("@principal", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
{ Value = context.StoredPrincipal });
cmd.Parameters.Add(new SqlParameter("@apiKeyId", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
{ Value = (object?)context.StoredApiKeyId ?? DBNull.Value });
}
}
@@ -18,7 +18,8 @@ public sealed partial class SqlServerHistorianOptions
/// <summary>
/// <c>true</c> creates the schema and applies missing scripts on first use; <c>false</c> only verifies
/// that schema version 1 exists and fails every historian call with a clear error otherwise.
/// that the required schema version (<c>SqlServerSchemaBootstrapper.RequiredVersion</c>, currently 2) exists and
/// fails every historian call with a clear error otherwise.
/// </summary>
public bool AutoCreateSchema { get; set; } = true;
@@ -7,7 +7,7 @@ using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>
/// Deletes <c>hist.Sample</c> / <c>hist.Event</c> rows older than the retention window, in small
/// Deletes <c>hist.Sample</c> / <c>hist.SampleModified</c> / <c>hist.Event</c> rows older than the retention window, in small
/// <c>DELETE TOP(n)</c> batches (under the lock-escalation threshold). Samples are purged per tag so each
/// delete is a seek on the clustered key <c>(TagId, SourceTimeUtc)</c>; events seek on <c>CX_Event_Time</c>.
/// A session <c>sp_getapplock</c> (timeout 0) serialises purgers that share one database; a loser skips.
@@ -69,7 +69,7 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
return [];
}
long samples = 0;
long samples = 0, modified = 0;
var tagIds = new List<int>();
await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}"))
await using (var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false))
@@ -80,13 +80,23 @@ internal sealed class SqlServerHistorianRetention : IHistorianRetention
samples += await DeleteLoopAsync(conn,
$"DELETE TOP (@n) FROM {_conn.Q("Sample")} WHERE TagId = @id AND SourceTimeUtc < @cutoff",
cutoff, id, ct).ConfigureAwait(false);
// The Part 11 modification trail (#668) ages out on the SAME source-time cutoff as the samples it
// describes; a purge never writes to it (no trigger, no OUTPUT INTO).
modified += await DeleteLoopAsync(conn,
$"DELETE TOP (@n) FROM {_conn.Q("SampleModified")} WHERE TagId = @id AND SourceTimeUtc < @cutoff",
cutoff, id, ct).ConfigureAwait(false);
}
var events = await DeleteLoopAsync(conn,
$"DELETE TOP (@n) FROM {_conn.Q("Event")} WHERE EventTimeUtc < @cutoff",
cutoff, null, ct).ConfigureAwait(false);
return [new RetentionPurgeResult("Sample", samples), new RetentionPurgeResult("Event", events)];
return
[
new RetentionPurgeResult("Sample", samples),
new RetentionPurgeResult("SampleModified", modified),
new RetentionPurgeResult("Event", events),
];
}
finally
{
@@ -14,12 +14,16 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <see cref="BootstrapFailureCache.DefaultDuration"/> retries (<see cref="BootstrapFailureCache"/>). Within that window the last failure is re-thrown without touching
/// SQL, so a missing schema (or a SQL outage) does not serialise every write, read and alarm event behind one
/// semaphore and one round-trip each.
/// With <c>AutoCreateSchema=false</c> it only verifies <c>MAX(Version) &gt;= 1</c>.
/// With <c>AutoCreateSchema=false</c> it only verifies <c>MAX(Version) &gt;= </c><see cref="RequiredVersion"/>.
/// </summary>
internal sealed partial class SqlServerSchemaBootstrapper
{
/// <summary>The schema version this build requires.</summary>
public const int RequiredVersion = 1;
/// <summary>
/// The schema version this build requires: 2 = <c>002_sample_modified.sql</c> (Part 11 modified-value
/// history, Gitea #668). An existing v1 schema is upgraded in place on the next bootstrap when
/// <c>AutoCreateSchema</c> is on; otherwise the script must be applied by hand.
/// </summary>
public const int RequiredVersion = 2;
private const string LockResource = "OtOpcUaHistorianSchema";
private const int LockTimeoutMs = 30_000;
@@ -48,7 +52,7 @@ internal sealed partial class SqlServerSchemaBootstrapper
/// <summary>Ensures the schema is ready; no-op after the first success.</summary>
/// <param name="ct">Cancellation.</param>
/// <returns>A task completing when the schema is usable.</returns>
/// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and version 1 is absent.</exception>
/// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and <see cref="RequiredVersion"/> is absent.</exception>
public async Task EnsureAsync(CancellationToken ct)
{
if (_ready) return;
@@ -94,7 +98,7 @@ internal sealed partial class SqlServerSchemaBootstrapper
if (v < RequiredVersion)
throw new HistorianSchemaMissingException(
$"Historian schema [{_conn.Schema}] is missing version {RequiredVersion} (found {v}) and AutoCreateSchema is false. " +
"Apply docs/historian/sqlserver/001_initial.sql or enable AutoCreateSchema.");
"Apply the docs/historian/sqlserver/NNN_*.sql scripts above version " + v + " in order, or enable AutoCreateSchema.");
}
private async Task ApplyAsync(SqlConnection conn, CancellationToken ct)
@@ -31,11 +31,20 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// prior pages and must be dropped from the head of the next page (tie de-dup).
/// </param>
/// <param name="NumValuesPerNode">The client's per-page cap; re-applied to every resumed page.</param>
/// <param name="IsEvents">
/// <c>true</c> for an EVENT HistoryRead cursor (Gitea #669): <paramref name="Tagname"/> then carries the
/// registered historian event-source name and the boundary fields are over <c>EventTimeUtc</c>. A cursor
/// is only ever resumed by the read kind that issued it, so a raw point presented to an events read
/// (or vice versa) is a <c>BadContinuationPointInvalid</c> miss.
/// <param name="Kind">
/// Which read issued the cursor (Gitea #669 / #668). A cursor is only ever resumed by the read kind that issued it, so a
/// raw point presented to an events or modified read (or any other mismatch) is a <c>BadContinuationPointInvalid</c>
/// miss. <see cref="HistoryCursorKind.Events"/>: <paramref name="Tagname"/> carries the registered historian
/// event-source name and the boundary fields are over <c>EventTimeUtc</c>.
/// </param>
/// <param name="LastModifiedId">
/// <see cref="HistoryCursorKind.Modified"/> only: the <c>ModifiedId</c> of the last returned modification. With
/// <paramref name="NextStartUtc"/> (its source time) it is the exclusive resume key; <paramref name="BoundarySkipCount"/>
/// is unused because the key is unique.
/// </param>
/// <param name="WindowStartUtc">
/// <see cref="HistoryCursorKind.Modified"/> only: the normalised window start, so a resumed page keeps the original
/// window and direction (the modified read resumes by key, not by moving the window start).
/// </param>
internal sealed record HistoryContinuationState(
string Tagname,
@@ -43,7 +52,22 @@ internal sealed record HistoryContinuationState(
DateTime EndUtc,
int BoundarySkipCount,
uint NumValuesPerNode,
bool IsEvents = false);
HistoryCursorKind Kind = HistoryCursorKind.Raw,
long LastModifiedId = 0,
DateTime WindowStartUtc = default);
/// <summary>The HistoryRead kind that issued a <see cref="HistoryContinuationState"/>.</summary>
internal enum HistoryCursorKind
{
/// <summary>HistoryReadRaw (values).</summary>
Raw,
/// <summary>HistoryReadEvents (#669).</summary>
Events,
/// <summary>HistoryReadRaw with <c>IsReadModified = true</c> — the Part 11 modification trail (#668).</summary>
Modified,
}
/// <summary>
/// Pure server-side continuation-point paging decisions for the count-capped variable-history arms
@@ -4977,8 +4977,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{
if (details.IsReadModified)
{
// We never serve modified-value history; mark this node unsupported and move on.
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
// Part 11 modified-value history (#668): the prior values every HistoryUpdate recorded. Gated by the
// same HistoryRead authorization as a raw read (above); a backend with no trail ⇒ unsupported.
work.Add(() => ServeModifiedPagedAsync(
handle, session, nodesToRead, results, errors,
details.StartTime, details.EndTime, details.NumValuesPerNode,
ct));
continue;
}
@@ -5156,7 +5160,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// for THIS node only.
var state = _historyContinuationStore.TryTake(session, inboundCp!);
// A point is bound to the node that issued it: another node's source name is a miss too.
if (state is null || !state.IsEvents || state.Tagname != (sourceName ?? string.Empty))
if (state is null || state.Kind != HistoryCursorKind.Events || state.Tagname != (sourceName ?? string.Empty))
{
errors[handle.Index] = StatusCodes.BadContinuationPointInvalid;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid };
@@ -5237,7 +5241,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (nextStartUtc is { } resumeAt)
{
pendingCursor = new HistoryContinuationState(
sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, IsEvents: true);
sourceName ?? string.Empty, resumeAt, endUtc, nextSkip, numValuesPerNode, HistoryCursorKind.Events);
}
}
else if (truncated && events.Count > 0)
@@ -5247,7 +5251,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
var skip = 0;
for (var i = events.Count - 1; i >= 0 && events[i].EventTimeUtc == boundary; i--) skip++;
pendingCursor = new HistoryContinuationState(
sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, IsEvents: true);
sourceName ?? string.Empty, boundary, endUtc, skip, numValuesPerNode, HistoryCursorKind.Events);
}
var historyEvent = ProjectEvents(events, selectClauses);
@@ -5374,7 +5378,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
var items = (IList<DataValue>?)details.UpdateValues ?? Array.Empty<DataValue>();
return (updater, tagname, ct) => UpdateDataAsync(updater, tagname, mode, items, ct);
return (updater, tagname, who, ct) => UpdateDataAsync(updater, tagname, mode, items, who, ct);
});
}
@@ -5390,17 +5394,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
ServeHistoryUpdate(context, nodesToProcess, results, errors, handle =>
{
var details = nodesToUpdate[handle.Index];
// We never keep modified-value history, so there is none to delete.
// Deleting the modified-value trail itself (Part 11 IsDeleteModified) is not supported: the trail is the
// audit record of every change and is aged out only by retention (#668).
if (details.IsDeleteModified) return null;
var start = details.StartTime;
var end = details.EndTime;
// Part 11 requires both bounds of a raw delete; DateTime.MinValue is the wire's "unspecified". Refuse
// it here rather than let a backend read MinValue as "from the beginning of time".
if (start == DateTime.MinValue || end == DateTime.MinValue)
return (_, _, _) => Task.FromResult(new HistoryUpdateResult { StatusCode = StatusCodes.BadInvalidArgument });
return async (updater, tagname, ct) =>
return (_, _, _, _) => Task.FromResult(new HistoryUpdateResult { StatusCode = StatusCodes.BadInvalidArgument });
return async (updater, tagname, who, ct) =>
{
var code = await updater.DeleteRawAsync(tagname, start, end, ct).ConfigureAwait(false);
var code = await updater.DeleteRawAsync(tagname, start, end, who, ct).ConfigureAwait(false);
return new HistoryUpdateResult { StatusCode = code };
};
});
@@ -5418,11 +5423,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
ServeHistoryUpdate(context, nodesToProcess, results, errors, handle =>
{
var times = nodesToUpdate[handle.Index].ReqTimes?.ToList() ?? new List<DateTime>();
return async (updater, tagname, ct) =>
return async (updater, tagname, who, ct) =>
{
var codes = times.Count == 0
? Array.Empty<uint>()
: await updater.DeleteAtTimeAsync(tagname, times, ct).ConfigureAwait(false);
: await updater.DeleteAtTimeAsync(tagname, times, who, ct).ConfigureAwait(false);
if (codes.Count != times.Count)
throw new InvalidOperationException(
$"historian updater returned {codes.Count} result(s) for {times.Count} time(s)");
@@ -5445,11 +5450,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <param name="tagname">The node's historian tagname.</param>
/// <param name="mode">The update mode.</param>
/// <param name="items">The request's values, in order.</param>
/// <param name="who">The caller, recorded in the modification trail.</param>
/// <param name="ct">The request deadline.</param>
/// <returns>The node's result, carrying one operation result per item.</returns>
private static async Task<HistoryUpdateResult> UpdateDataAsync(
IHistorianDataUpdater updater, string tagname, HistoryUpdateMode mode, IList<DataValue> items,
CancellationToken ct)
HistoryUpdateContext who, CancellationToken ct)
{
var codes = new uint[items.Count];
var toSend = new List<HistorizationValue>(items.Count);
@@ -5478,7 +5484,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (toSend.Count > 0)
{
var backend = await updater.UpdateDataAsync(tagname, mode, toSend, ct).ConfigureAwait(false);
var backend = await updater.UpdateDataAsync(tagname, mode, toSend, who, ct).ConfigureAwait(false);
if (backend.Count != toSend.Count)
throw new InvalidOperationException(
$"historian updater returned {backend.Count} result(s) for {toSend.Count} value(s)");
@@ -5510,11 +5516,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
List<NodeHandle> nodesToProcess,
IList<HistoryUpdateResult> results,
IList<ServiceResult> errors,
Func<NodeHandle, Func<IHistorianDataUpdater, string, CancellationToken, Task<HistoryUpdateResult>>?> plan)
Func<NodeHandle, Func<IHistorianDataUpdater, string, HistoryUpdateContext, CancellationToken, Task<HistoryUpdateResult>>?> plan)
{
nodesToProcess = AuthorizeHistoryUpdate(context, nodesToProcess, results, errors);
if (nodesToProcess.Count == 0) return;
// Who is rewriting history — recorded on every modification-trail row (Part 11 ModificationInfo.UserName, #668).
var who = HistoryUpdateCallerOf(context);
WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () =>
{
using var deadline = CreateDeadline();
@@ -5533,13 +5542,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
var h = handle;
var tag = tagname;
work.Add(() => ServeUpdateNodeAsync(h, results, errors, token => operation(updater!, tag, token), ct));
work.Add(() => ServeUpdateNodeAsync(h, results, errors, token => operation(updater!, tag, who, token), ct));
}
RunBounded(work, HistoryReadBatchParallelism);
});
}
/// <summary>
/// The <see cref="HistoryUpdateContext"/> of the session behind <paramref name="context"/>: its API-key principal
/// (or the reserved <c>anonymous</c> principal) and key id. Only an identity that already passed
/// <see cref="AuthorizeHistoryUpdate"/> reaches a backend, so the anonymous fallback for a missing identity is
/// defence in depth, not a path.
/// </summary>
/// <param name="context">The SDK context carrying the session identity.</param>
/// <returns>The caller recorded in the modification trail.</returns>
internal static HistoryUpdateContext HistoryUpdateCallerOf(ServerSystemContext context)
{
var identity = (context as ISessionOperationContext)?.UserIdentity as RoleCarryingUserIdentity
?? context.OperationContext?.UserIdentity as RoleCarryingUserIdentity;
return identity is null
? new HistoryUpdateContext(OpcUaPrincipals.Anonymous, null)
: new HistoryUpdateContext(identity.Principal, identity.ApiKeyId);
}
/// <summary>
/// Runs one node's HistoryUpdate operation and fills its slots. Never throws (the
/// <see cref="RunBounded"/> contract): a deadline cancellation is <c>BadTimeout</c>, any backend failure
@@ -5950,7 +5976,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// Resume read: take the stored cursor. A miss (unknown / evicted / malformed point) ⇒
// BadContinuationPointInvalid for THIS node.
var state = _historyContinuationStore.TryTake(session, inboundCp);
if (state is null || state.IsEvents
if (state is null || state.Kind != HistoryCursorKind.Raw
|| !TryGetHistorizedTagname(handle.NodeId, out var boundTag) || boundTag != state.Tagname)
{
errors[handle.Index] = StatusCodes.BadContinuationPointInvalid;
@@ -6116,6 +6142,138 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
}
/// <summary>
/// Serve one historized variable for a HistoryReadRaw with <c>IsReadModified = true</c> (Part 11 modified-value
/// history, Gitea #668), with continuation-point paging. The backend orders the trail by
/// <c>(SourceTimeUtc, ModifiedId)</c> and reports truncation itself (<see cref="HistoryModifiedReadResult.HasMore"/>),
/// so the cursor is an exact, exclusive key — there are no ties to trim. Window normalisation matches
/// <see cref="ServeRawPagedAsync"/>: with a cap, an unset EndTime pages forward to an open bound and an unset
/// StartTime pages backward. A cursor is bound to the node's tagname and to the modified read kind; any other
/// point is <c>BadContinuationPointInvalid</c>. A backend that keeps no trail (Gateway, Null) throws
/// <see cref="NotSupportedException"/> ⇒ <c>BadHistoryOperationUnsupported</c> for this node.
/// </summary>
/// <param name="handle">The node handle; <c>handle.Index</c> indexes results/errors.</param>
/// <param name="session">The session (null on the session-less in-process path ⇒ single-shot).</param>
/// <param name="nodesToRead">The per-node read list (inbound continuation points).</param>
/// <param name="results">The service-level results list.</param>
/// <param name="errors">The service-level errors list.</param>
/// <param name="startTimeUtc">Request StartTime.</param>
/// <param name="endUtc">Request EndTime.</param>
/// <param name="numValuesPerNode">Per-page cap; <c>0</c> = all.</param>
/// <param name="ct">Request deadline.</param>
private async Task ServeModifiedPagedAsync(
NodeHandle handle,
ISession? session,
IList<HistoryReadValueId> nodesToRead,
IList<SdkHistoryReadResult> results,
IList<ServiceResult> errors,
DateTime startTimeUtc,
DateTime endUtc,
uint numValuesPerNode,
CancellationToken ct)
{
var inboundCp = nodesToRead[handle.Index].ContinuationPoint;
try
{
if (!TryGetHistorizedTagname(handle.NodeId, out var tagname) || tagname is null)
{
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
return;
}
var startUtc = startTimeUtc;
HistoryModifiedCursor? resumeAfter = null;
if (inboundCp is { Length: > 0 })
{
var state = _historyContinuationStore.TryTake(session, inboundCp);
if (state is null || state.Kind != HistoryCursorKind.Modified || state.Tagname != tagname)
{
errors[handle.Index] = StatusCodes.BadContinuationPointInvalid;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadContinuationPointInvalid };
return;
}
// The cursor carries the NORMALISED window, so a resumed page never re-derives its direction.
startUtc = state.WindowStartUtc;
endUtc = state.EndUtc;
numValuesPerNode = state.NumValuesPerNode;
resumeAfter = new HistoryModifiedCursor(state.NextStartUtc, state.LastModifiedId);
}
else if (numValuesPerNode > 0)
{
if (endUtc == DateTime.MinValue && startUtc != DateTime.MinValue)
endUtc = new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc);
else if (startUtc == DateTime.MinValue && endUtc != DateTime.MinValue)
(startUtc, endUtc) = (endUtc, new DateTime(0, DateTimeKind.Utc));
}
var page = await HistorianDataSource
.ReadModifiedAsync(tagname, startUtc, endUtc, numValuesPerNode, resumeAfter, ct)
.ConfigureAwait(false);
var data = ToHistoryModifiedData(page.Values);
byte[]? outboundCp = null;
if (page.HasMore && numValuesPerNode > 0 && page.Values.Count > 0)
{
var last = page.Values[^1];
outboundCp = _historyContinuationStore.Save(session, new HistoryContinuationState(
tagname, last.Value.SourceTimestampUtc ?? DateTime.MinValue, endUtc, 0, numValuesPerNode,
HistoryCursorKind.Modified, last.ModifiedId, startUtc));
}
results[handle.Index] = new SdkHistoryReadResult
{
StatusCode = page.Values.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good,
HistoryData = new ExtensionObject(data),
ContinuationPoint = outboundCp,
};
errors[handle.Index] = ServiceResult.Good;
}
catch (OperationCanceledException)
{
errors[handle.Index] = StatusCodes.BadTimeout;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
}
catch (NotSupportedException ex)
{
// The backend keeps no modification trail (Gateway / Null) — the spec status, not a server fault.
_logger.LogDebug(ex, "OtOpcUaNodeManager: backend serves no modified-value history for node {NodeId}", handle.NodeId);
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
}
catch (Exception ex)
{
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryRead (modified) failed for node {NodeId}", handle.NodeId);
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
}
}
/// <summary>
/// Project a modification trail into an SDK <see cref="HistoryModifiedData"/>: one prior <see cref="DataValue"/>
/// per modification plus a parallel <see cref="ModificationInfo"/> (time, Part 11 update type, user name).
/// </summary>
/// <param name="values">The modifications, in the order to emit.</param>
/// <returns>The populated <see cref="HistoryModifiedData"/>.</returns>
internal static HistoryModifiedData ToHistoryModifiedData(IReadOnlyList<HistoryModifiedValue> values)
{
var dataValues = new DataValueCollection(values.Count);
var infos = new ModificationInfoCollection(values.Count);
foreach (var m in values)
{
dataValues.Add(ToSdkDataValue(m.Value));
infos.Add(new ModificationInfo
{
ModificationTime = m.ModificationTimeUtc,
UpdateType = (HistoryUpdateType)(int)m.UpdateType,
UserName = m.UserName,
});
}
return new HistoryModifiedData { DataValues = dataValues, ModificationInfos = infos };
}
/// <inheritdoc />
protected override void HistoryReleaseContinuationPoints(
ServerSystemContext context,